add more comments

Author: cloud-fan

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/misc.scala

@@ -23,8 +23,6 @@ import java.util.zip.CRC32
 import org.apache.commons.codec.digest.DigestUtils
 
 import org.apache.spark.sql.catalyst.expressions.codegen._
-import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.catalyst.util.{MapData, ArrayData}
 import org.apache.spark.sql.types._
 import org.apache.spark.unsafe.types.UTF8String
 

sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriter.scala

@@ -119,8 +119,6 @@ final class DataFrameWriter private[sql](df: DataFrame) {
    * Partitions the output by the given columns on the file system. If specified, the output is
    * laid out on the file system similar to Hive's partitioning scheme.
    *
-   * This is only applicable for Parquet at the moment.
-   *
    * @since 1.4.0
    */
   @scala.annotation.varargs
@@ -129,13 +127,24 @@ final class DataFrameWriter private[sql](df: DataFrame) {
     this
   }
 
+  /**
+   * Buckets the output by the given columns on the file system. If specified, the output is
+   * laid out on the file system similar to Hive's bucketing scheme.
+   *
+   * @since 2.0
+   */
   @scala.annotation.varargs
   def bucketBy(numBuckets: Int, colName: String, colNames: String*): DataFrameWriter = {
     this.numBuckets = Option(numBuckets)
     this.bucketingColumns = Option(colName +: colNames)
     this
   }
 
+  /**
+   * Sorts the bucketed output by the given columns.
+   *
+   * @since 2.0
+   */
   @scala.annotation.varargs
   def sortBy(colName: String, colNames: String*): DataFrameWriter = {
     this.sortingColumns = Option(colName +: colNames)

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/WriterContainer.scala

@@ -327,9 +327,10 @@ private[sql] class DynamicPartitionWriterContainer(
     val getKey: InternalRow => UnsafeRow = if (bucketSpec.isEmpty) {
       val projection = UnsafeProjection.create(partitionColumns, inputSchema)
       row => projection(row)
-    } else {
+    } else { // If it's bucketed, we should also consider bucket id as part of the key.
       val bucketColumns = bucketSpec.get.resolvedBucketingColumns(inputSchema)
       val getBucketKey = UnsafeProjection.create(bucketColumns, inputSchema)
+      // Leave an empty int slot at the last of the result row, so that we can set bucket id later.
       val getResultRow = UnsafeProjection.create(partitionColumns :+ Literal(-1), inputSchema)
       row => {
         val bucketId = math.abs(getBucketKey(row).hashCode()) % bucketSpec.get.numBuckets
@@ -341,7 +342,7 @@ private[sql] class DynamicPartitionWriterContainer(
 
     val keySchema = if (bucketSpec.isEmpty) {
       StructType.fromAttributes(partitionColumns)
-    } else {
+    } else { // If it's bucketed, we should also consider bucket id as part of the key.
       StructType.fromAttributes(partitionColumns).add("bucketId", IntegerType, nullable = false)
     }
 

sql/hive/src/test/scala/org/apache/spark/sql/sources/BucketedWriteSuite.scala

@@ -120,8 +120,8 @@ class BucketedWriteSuite extends QueryTest with SQLTestUtils with TestHiveSingle
 
               for (row <- rows) {
                 assert(row.isInstanceOf[UnsafeRow])
-                val actuaBucketId = math.abs(row.hashCode()) % 8
-                assert(actuaBucketId == bucketId)
+                val actualBucketId = math.abs(row.hashCode()) % 8
+                assert(actualBucketId == bucketId)
               }
             }
           }
@@ -151,8 +151,8 @@ class BucketedWriteSuite extends QueryTest with SQLTestUtils with TestHiveSingle
 
             for (row <- rows) {
               assert(row.isInstanceOf[UnsafeRow])
-              val actuaBucketId = math.abs(row.hashCode()) % 8
-              assert(actuaBucketId == bucketId)
+              val actualBucketId = math.abs(row.hashCode()) % 8
+              assert(actualBucketId == bucketId)
             }
           }
         }
@@ -183,8 +183,8 @@ class BucketedWriteSuite extends QueryTest with SQLTestUtils with TestHiveSingle
 
             for (row <- rows) {
               assert(row.isInstanceOf[UnsafeRow])
-              val actuaBucketId = math.abs(row.hashCode()) % 8
-              assert(actuaBucketId == bucketId)
+              val actualBucketId = math.abs(row.hashCode()) % 8
+              assert(actualBucketId == bucketId)
             }
           }
         }