Change sort flag into Option

Author: jerryshao

core/src/main/scala/org/apache/spark/Dependency.scala

@@ -63,7 +63,7 @@ class ShuffleDependency[K, V, C](
     val keyOrdering: Option[Ordering[K]] = None,
     val aggregator: Option[Aggregator[K, V, C]] = None,
     val mapSideCombine: Boolean = false,
-    val ascending: Boolean = true)
+    val ascending: Option[Boolean] = None)
   extends Dependency(rdd.asInstanceOf[RDD[Product2[K, V]]]) {
 
   val shuffleId: Int = rdd.context.newShuffleId()

core/src/main/scala/org/apache/spark/rdd/ShuffledRDD.scala

@@ -51,7 +51,7 @@ class ShuffledRDD[K, V, C, P <: Product2[K, C] : ClassTag](
 
   private var mapSideCombine: Boolean = false
 
-  private var ascending: Boolean = true
+  private var ascending: Option[Boolean] = None
 
   /** Set a serializer for this RDD's shuffle, or null to use the default (spark.serializer) */
   def setSerializer(serializer: Serializer): ShuffledRDD[K, V, C, P] = {
@@ -79,7 +79,7 @@ class ShuffledRDD[K, V, C, P <: Product2[K, C] : ClassTag](
 
   /** Set sort flag for RDD's sorting. */
   def setSortFlag(ascending: Boolean): ShuffledRDD[K, V, C, P] = {
-    this.ascending = ascending
+    this.ascending = Option(ascending)
     this
   }
 

core/src/main/scala/org/apache/spark/shuffle/hash/HashShuffleReader.scala

@@ -50,16 +50,16 @@ class HashShuffleReader[K, C](
       iter
     }
 
-    dep.keyOrdering.map { ordering =>
+    val sortedIter = for (asc <- dep.ascending; ordering <- dep.keyOrdering) yield {
       val buf = aggregatedIter.toArray
-      if (dep.ascending) {
+      if (asc) {
         buf.sortWith((x, y) => ordering.lt(x._1, y._1)).iterator
       } else {
         buf.sortWith((x, y) => ordering.gt(x._1, y._1)).iterator
       }
-    }.getOrElse {
-      aggregatedIter
     }
+
+    sortedIter.getOrElse(aggregatedIter)
   }
 
   /** Close this reader */