Test ascending and descending sort orders.

Author: JoshRosen

sql/catalyst/src/main/java/org/apache/spark/sql/execution/UnsafeExternalRowSorter.java

@@ -178,6 +178,7 @@ public Iterator<InternalRow> sort(Iterator<InternalRow> inputIterator) throws IO
    * Return true if UnsafeExternalRowSorter can sort rows with the given schema, false otherwise.
    */
   public static boolean supportsSchema(StructType schema) {
+    // TODO: add spilling note.
     for (StructField field : schema.fields()) {
       if (UnsafeColumnWriter.forType(field.dataType()) instanceof ObjectUnsafeColumnWriter) {
         return false;

sql/core/src/test/scala/org/apache/spark/sql/execution/UnsafeExternalSortSuite.scala

@@ -37,25 +37,28 @@ class UnsafeExternalSortSuite extends SparkPlanTest with BeforeAndAfterAll {
   }
 
   // Test sorting on different data types
-  (DataTypeTestUtils.atomicTypes ++ Set(NullType)).foreach{ dataType =>
-    for (nullable <- Seq(true, false)) {
-      RandomDataGenerator.forType(dataType, nullable).foreach { randomDataGenerator =>
-        test(s"sorting on $dataType with nullable=$nullable") {
-          val inputData = Seq.fill(1024)(randomDataGenerator()).filter {
-            case d: Double => !d.isNaN
-            case f: Float => !java.lang.Float.isNaN(f)
-            case x => true
-          }
-          val inputDf = TestSQLContext.createDataFrame(
-            TestSQLContext.sparkContext.parallelize(Random.shuffle(inputData).map(v => Row(v))),
-            StructType(StructField("a", dataType, nullable = true) :: Nil)
-          )
-          checkAnswer(
-            inputDf,
-            UnsafeExternalSort('a.asc :: Nil, global = false, _: SparkPlan),
-            Sort('a.asc :: Nil, global = false, _: SparkPlan)
-          )
+  // TODO: randomized spilling to ensure that merging is tested at least once for every data type.
+  (DataTypeTestUtils.atomicTypes ++ Set(NullType)).foreach { dataType =>
+    for (
+      nullable <- Seq(true, false);
+      sortOrder <- Seq('a.asc :: Nil, 'a.desc :: Nil);
+      randomDataGenerator <- RandomDataGenerator.forType(dataType, nullable)
+    ) {
+      test(s"sorting on $dataType with nullable=$nullable, sortOrder=$sortOrder") {
+        val inputData = Seq.fill(1024)(randomDataGenerator()).filter {
+          case d: Double => !d.isNaN
+          case f: Float => !java.lang.Float.isNaN(f)
+          case x => true
         }
+        val inputDf = TestSQLContext.createDataFrame(
+          TestSQLContext.sparkContext.parallelize(Random.shuffle(inputData).map(v => Row(v))),
+          StructType(StructField("a", dataType, nullable = true) :: Nil)
+        )
+        checkAnswer(
+          inputDf,
+          UnsafeExternalSort(sortOrder, global = false, _: SparkPlan),
+          Sort(sortOrder, global = false, _: SparkPlan)
+        )
       }
     }
   }