for testing only, change default to always spill after read

Author: squito

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

@@ -39,7 +39,7 @@ case class Aggregator[K, V, C] (
       context: TaskContext): Iterator[(K, C)] = {
     val combiners = new ExternalAppendOnlyMap[K, V, C](createCombiner, mergeValue, mergeCombiners)
     combiners.insertAll(iter)
-    if (SparkEnv.get.conf.getBoolean("spark.shuffle.spillAfterRead", false)) {
+    if (SparkEnv.get.conf.getBoolean("spark.shuffle.spillAfterRead", true)) {
       combiners.spill()
     }
     updateMetrics(context, combiners)
@@ -51,7 +51,7 @@ case class Aggregator[K, V, C] (
       context: TaskContext): Iterator[(K, C)] = {
     val combiners = new ExternalAppendOnlyMap[K, C, C](identity, mergeCombiners, mergeCombiners)
     combiners.insertAll(iter)
-    if (SparkEnv.get.conf.getBoolean("spark.shuffle.spillAfterRead", false)) {
+    if (SparkEnv.get.conf.getBoolean("spark.shuffle.spillAfterRead", true)) {
       combiners.spill()
     }
     updateMetrics(context, combiners)

core/src/main/scala/org/apache/spark/shuffle/BlockStoreShuffleReader.scala

@@ -103,7 +103,7 @@ private[spark] class BlockStoreShuffleReader[K, C](
         val sorter =
           new ExternalSorter[K, C, C](context, ordering = Some(keyOrd), serializer = dep.serializer)
         sorter.insertAll(aggregatedIter)
-        if (SparkEnv.get.conf.getBoolean("spark.shuffle.spillAfterRead", false)) {
+        if (SparkEnv.get.conf.getBoolean("spark.shuffle.spillAfterRead", true)) {
           sorter.spill()
         }
         context.taskMetrics().incMemoryBytesSpilled(sorter.memoryBytesSpilled)