ExternalAppendOnlyMapSuite code style fixes (minor)

Author: andrewor14

core/src/test/scala/org/apache/spark/util/collection/ExternalAppendOnlyMapSuite.scala

@@ -25,10 +25,13 @@ import org.apache.spark._
 import org.apache.spark.SparkContext._
 
 class ExternalAppendOnlyMapSuite extends FunSuite with LocalSparkContext {
+  private def createCombiner[T](i: T) = ArrayBuffer[T](i)
+  private def mergeValue[T](buffer: ArrayBuffer[T], i: T): ArrayBuffer[T] = buffer += i
+  private def mergeCombiners[T](buf1: ArrayBuffer[T], buf2: ArrayBuffer[T]): ArrayBuffer[T] =
+    buf1 ++= buf2
 
-  private def createCombiner(i: Int) = ArrayBuffer[Int](i)
-  private def mergeValue(buffer: ArrayBuffer[Int], i: Int) = buffer += i
-  private def mergeCombiners(buf1: ArrayBuffer[Int], buf2: ArrayBuffer[Int]) = buf1 ++= buf2
+  private def createExternalMap[T] = new ExternalAppendOnlyMap[T, T, ArrayBuffer[T]](
+    createCombiner[T], mergeValue[T], mergeCombiners[T])
 
   private def createSparkConf(loadDefaults: Boolean): SparkConf = {
     val conf = new SparkConf(loadDefaults)
@@ -42,37 +45,33 @@ class ExternalAppendOnlyMapSuite extends FunSuite with LocalSparkContext {
   }
 
   test("simple insert") {
-    val conf = createSparkConf(false)
+    val conf = createSparkConf(loadDefaults = false)
     sc = new SparkContext("local", "test", conf)
-
-    val map = new ExternalAppendOnlyMap[Int, Int, ArrayBuffer[Int]](createCombiner,
-      mergeValue, mergeCombiners)
+    val map = createExternalMap[Int]
 
     // Single insert
     map.insert(1, 10)
     var it = map.iterator
     assert(it.hasNext)
     val kv = it.next()
-    assert(kv._1 == 1 && kv._2 == ArrayBuffer[Int](10))
+    assert(kv._1 === 1 && kv._2 === ArrayBuffer[Int](10))
     assert(!it.hasNext)
 
     // Multiple insert
     map.insert(2, 20)
     map.insert(3, 30)
     it = map.iterator
     assert(it.hasNext)
-    assert(it.toSet == Set[(Int, ArrayBuffer[Int])](
+    assert(it.toSet === Set[(Int, ArrayBuffer[Int])](
       (1, ArrayBuffer[Int](10)),
       (2, ArrayBuffer[Int](20)),
       (3, ArrayBuffer[Int](30))))
   }
 
   test("insert with collision") {
-    val conf = createSparkConf(false)
+    val conf = createSparkConf(loadDefaults = false)
     sc = new SparkContext("local", "test", conf)
-
-    val map = new ExternalAppendOnlyMap[Int, Int, ArrayBuffer[Int]](createCombiner,
-      mergeValue, mergeCombiners)
+    val map = createExternalMap[Int]
 
     map.insertAll(Seq(
       (1, 10),
@@ -84,30 +83,27 @@ class ExternalAppendOnlyMapSuite extends FunSuite with LocalSparkContext {
     val it = map.iterator
     assert(it.hasNext)
     val result = it.toSet[(Int, ArrayBuffer[Int])].map(kv => (kv._1, kv._2.toSet))
-    assert(result == Set[(Int, Set[Int])](
+    assert(result === Set[(Int, Set[Int])](
       (1, Set[Int](10, 100, 1000)),
       (2, Set[Int](20, 200)),
       (3, Set[Int](30))))
   }
 
   test("ordering") {
-    val conf = createSparkConf(false)
+    val conf = createSparkConf(loadDefaults = false)
     sc = new SparkContext("local", "test", conf)
 
-    val map1 = new ExternalAppendOnlyMap[Int, Int, ArrayBuffer[Int]](createCombiner,
-      mergeValue, mergeCombiners)
+    val map1 = createExternalMap[Int]
     map1.insert(1, 10)
     map1.insert(2, 20)
     map1.insert(3, 30)
 
-    val map2 = new ExternalAppendOnlyMap[Int, Int, ArrayBuffer[Int]](createCombiner,
-      mergeValue, mergeCombiners)
+    val map2 = createExternalMap[Int]
     map2.insert(2, 20)
     map2.insert(3, 30)
     map2.insert(1, 10)
 
-    val map3 = new ExternalAppendOnlyMap[Int, Int, ArrayBuffer[Int]](createCombiner,
-      mergeValue, mergeCombiners)
+    val map3 = createExternalMap[Int]
     map3.insert(3, 30)
     map3.insert(1, 10)
     map3.insert(2, 20)
@@ -119,33 +115,32 @@ class ExternalAppendOnlyMapSuite extends FunSuite with LocalSparkContext {
     var kv1 = it1.next()
     var kv2 = it2.next()
     var kv3 = it3.next()
-    assert(kv1._1 == kv2._1 && kv2._1 == kv3._1)
-    assert(kv1._2 == kv2._2 && kv2._2 == kv3._2)
+    assert(kv1._1 === kv2._1 && kv2._1 === kv3._1)
+    assert(kv1._2 === kv2._2 && kv2._2 === kv3._2)
 
     kv1 = it1.next()
     kv2 = it2.next()
     kv3 = it3.next()
-    assert(kv1._1 == kv2._1 && kv2._1 == kv3._1)
-    assert(kv1._2 == kv2._2 && kv2._2 == kv3._2)
+    assert(kv1._1 === kv2._1 && kv2._1 === kv3._1)
+    assert(kv1._2 === kv2._2 && kv2._2 === kv3._2)
 
     kv1 = it1.next()
     kv2 = it2.next()
     kv3 = it3.next()
-    assert(kv1._1 == kv2._1 && kv2._1 == kv3._1)
-    assert(kv1._2 == kv2._2 && kv2._2 == kv3._2)
+    assert(kv1._1 === kv2._1 && kv2._1 === kv3._1)
+    assert(kv1._2 === kv2._2 && kv2._2 === kv3._2)
   }
 
   test("null keys and values") {
-    val conf = createSparkConf(false)
+    val conf = createSparkConf(loadDefaults = false)
     sc = new SparkContext("local", "test", conf)
 
-    val map = new ExternalAppendOnlyMap[Int, Int, ArrayBuffer[Int]](createCombiner,
-      mergeValue, mergeCombiners)
+    val map = createExternalMap[Int]
     map.insert(1, 5)
     map.insert(2, 6)
     map.insert(3, 7)
     assert(map.size === 3)
-    assert(map.iterator.toSet == Set[(Int, Seq[Int])](
+    assert(map.iterator.toSet === Set[(Int, Seq[Int])](
       (1, Seq[Int](5)),
       (2, Seq[Int](6)),
       (3, Seq[Int](7))
@@ -155,7 +150,7 @@ class ExternalAppendOnlyMapSuite extends FunSuite with LocalSparkContext {
     val nullInt = null.asInstanceOf[Int]
     map.insert(nullInt, 8)
     assert(map.size === 4)
-    assert(map.iterator.toSet == Set[(Int, Seq[Int])](
+    assert(map.iterator.toSet === Set[(Int, Seq[Int])](
       (1, Seq[Int](5)),
       (2, Seq[Int](6)),
       (3, Seq[Int](7)),
@@ -167,7 +162,7 @@ class ExternalAppendOnlyMapSuite extends FunSuite with LocalSparkContext {
     map.insert(nullInt, nullInt)
     assert(map.size === 5)
     val result = map.iterator.toSet[(Int, ArrayBuffer[Int])].map(kv => (kv._1, kv._2.toSet))
-    assert(result == Set[(Int, Set[Int])](
+    assert(result === Set[(Int, Set[Int])](
       (1, Set[Int](5)),
       (2, Set[Int](6)),
       (3, Set[Int](7)),
@@ -177,100 +172,90 @@ class ExternalAppendOnlyMapSuite extends FunSuite with LocalSparkContext {
   }
 
   test("simple aggregator") {
-    val conf = createSparkConf(false)
+    val conf = createSparkConf(loadDefaults = false)
     sc = new SparkContext("local", "test", conf)
 
     // reduceByKey
     val rdd = sc.parallelize(1 to 10).map(i => (i%2, 1))
     val result1 = rdd.reduceByKey(_+_).collect()
-    assert(result1.toSet == Set[(Int, Int)]((0, 5), (1, 5)))
+    assert(result1.toSet === Set[(Int, Int)]((0, 5), (1, 5)))
 
     // groupByKey
     val result2 = rdd.groupByKey().collect().map(x => (x._1, x._2.toList)).toSet
-    assert(result2.toSet == Set[(Int, Seq[Int])]
+    assert(result2.toSet === Set[(Int, Seq[Int])]
       ((0, List[Int](1, 1, 1, 1, 1)), (1, List[Int](1, 1, 1, 1, 1))))
   }
 
   test("simple cogroup") {
-    val conf = createSparkConf(false)
+    val conf = createSparkConf(loadDefaults = false)
     sc = new SparkContext("local", "test", conf)
     val rdd1 = sc.parallelize(1 to 4).map(i => (i, i))
     val rdd2 = sc.parallelize(1 to 4).map(i => (i%2, i))
     val result = rdd1.cogroup(rdd2).collect()
 
     result.foreach { case (i, (seq1, seq2)) =>
       i match {
-        case 0 => assert(seq1.toSet == Set[Int]() && seq2.toSet == Set[Int](2, 4))
-        case 1 => assert(seq1.toSet == Set[Int](1) && seq2.toSet == Set[Int](1, 3))
-        case 2 => assert(seq1.toSet == Set[Int](2) && seq2.toSet == Set[Int]())
-        case 3 => assert(seq1.toSet == Set[Int](3) && seq2.toSet == Set[Int]())
-        case 4 => assert(seq1.toSet == Set[Int](4) && seq2.toSet == Set[Int]())
+        case 0 => assert(seq1.toSet === Set[Int]() && seq2.toSet === Set[Int](2, 4))
+        case 1 => assert(seq1.toSet === Set[Int](1) && seq2.toSet === Set[Int](1, 3))
+        case 2 => assert(seq1.toSet === Set[Int](2) && seq2.toSet === Set[Int]())
+        case 3 => assert(seq1.toSet === Set[Int](3) && seq2.toSet === Set[Int]())
+        case 4 => assert(seq1.toSet === Set[Int](4) && seq2.toSet === Set[Int]())
       }
     }
   }
 
   test("spilling") {
-    val conf = createSparkConf(true)  // Load defaults, otherwise SPARK_HOME is not found
+    val conf = createSparkConf(loadDefaults = true)  // Load defaults for Spark home
     conf.set("spark.shuffle.memoryFraction", "0.001")
     sc = new SparkContext("local-cluster[1,1,512]", "test", conf)
 
     // reduceByKey - should spill ~8 times
     val rddA = sc.parallelize(0 until 100000).map(i => (i/2, i))
     val resultA = rddA.reduceByKey(math.max).collect()
-    assert(resultA.length == 50000)
-    resultA.foreach { case(k, v) =>
-      if (v != k * 2 + 1) {
-        fail(s"Value for ${k} was wrong: expected ${k * 2 + 1}, got ${v}")
-      }
+    assert(resultA.length === 50000)
+    resultA.foreach { case (k, v) =>
+      assert(v === k * 2 + 1, s"Value for $k was wrong: expected ${k * 2 + 1}, got $v")
     }
 
     // groupByKey - should spill ~17 times
     val rddB = sc.parallelize(0 until 100000).map(i => (i/4, i))
     val resultB = rddB.groupByKey().collect()
-    assert(resultB.length == 25000)
-    resultB.foreach { case(i, seq) =>
+    assert(resultB.length === 25000)
+    resultB.foreach { case (i, seq) =>
       val expected = Set(i * 4, i * 4 + 1, i * 4 + 2, i * 4 + 3)
-      if (seq.toSet != expected) {
-        fail(s"Value for ${i} was wrong: expected ${expected}, got ${seq.toSet}")
-      }
+      assert(seq.toSet === expected,
+        s"Value for $i was wrong: expected $expected, got ${seq.toSet}")
     }
 
     // cogroup - should spill ~7 times
     val rddC1 = sc.parallelize(0 until 10000).map(i => (i, i))
     val rddC2 = sc.parallelize(0 until 10000).map(i => (i%1000, i))
     val resultC = rddC1.cogroup(rddC2).collect()
-    assert(resultC.length == 10000)
-    resultC.foreach { case(i, (seq1, seq2)) =>
+    assert(resultC.length === 10000)
+    resultC.foreach { case (i, (seq1, seq2)) =>
       i match {
         case 0 =>
-          assert(seq1.toSet == Set[Int](0))
-          assert(seq2.toSet == Set[Int](0, 1000, 2000, 3000, 4000, 5000, 6000, 7000, 8000, 9000))
+          assert(seq1.toSet === Set[Int](0))
+          assert(seq2.toSet === Set[Int](0, 1000, 2000, 3000, 4000, 5000, 6000, 7000, 8000, 9000))
         case 1 =>
-          assert(seq1.toSet == Set[Int](1))
-          assert(seq2.toSet == Set[Int](1, 1001, 2001, 3001, 4001, 5001, 6001, 7001, 8001, 9001))
+          assert(seq1.toSet === Set[Int](1))
+          assert(seq2.toSet === Set[Int](1, 1001, 2001, 3001, 4001, 5001, 6001, 7001, 8001, 9001))
         case 5000 =>
-          assert(seq1.toSet == Set[Int](5000))
-          assert(seq2.toSet == Set[Int]())
+          assert(seq1.toSet === Set[Int](5000))
+          assert(seq2.toSet === Set[Int]())
         case 9999 =>
-          assert(seq1.toSet == Set[Int](9999))
-          assert(seq2.toSet == Set[Int]())
+          assert(seq1.toSet === Set[Int](9999))
+          assert(seq2.toSet === Set[Int]())
         case _ =>
       }
     }
   }
 
   test("spilling with hash collisions") {
-    val conf = createSparkConf(true)
+    val conf = createSparkConf(loadDefaults = true)
     conf.set("spark.shuffle.memoryFraction", "0.001")
     sc = new SparkContext("local-cluster[1,1,512]", "test", conf)
-
-    def createCombiner(i: String) = ArrayBuffer[String](i)
-    def mergeValue(buffer: ArrayBuffer[String], i: String) = buffer += i
-    def mergeCombiners(buffer1: ArrayBuffer[String], buffer2: ArrayBuffer[String]) =
-      buffer1 ++= buffer2
-
-    val map = new ExternalAppendOnlyMap[String, String, ArrayBuffer[String]](
-      createCombiner, mergeValue, mergeCombiners)
+    val map = createExternalMap[String]
 
     val collisionPairs = Seq(
       ("Aa", "BB"),                   // 2112
@@ -315,10 +300,9 @@ class ExternalAppendOnlyMapSuite extends FunSuite with LocalSparkContext {
   }
 
   test("spilling with many hash collisions") {
-    val conf = createSparkConf(true)
+    val conf = createSparkConf(loadDefaults = true)
     conf.set("spark.shuffle.memoryFraction", "0.0001")
     sc = new SparkContext("local-cluster[1,1,512]", "test", conf)
-
     val map = new ExternalAppendOnlyMap[FixedHashObject, Int, Int](_ => 1, _ + _, _ + _)
 
     // Insert 10 copies each of lots of objects whose hash codes are either 0 or 1. This causes
@@ -340,12 +324,10 @@ class ExternalAppendOnlyMapSuite extends FunSuite with LocalSparkContext {
   }
 
   test("spilling with hash collisions using the Int.MaxValue key") {
-    val conf = createSparkConf(true)
+    val conf = createSparkConf(loadDefaults = true)
     conf.set("spark.shuffle.memoryFraction", "0.001")
     sc = new SparkContext("local-cluster[1,1,512]", "test", conf)
-
-    val map = new ExternalAppendOnlyMap[Int, Int, ArrayBuffer[Int]](
-      createCombiner, mergeValue, mergeCombiners)
+    val map = createExternalMap[Int]
 
     (1 to 100000).foreach { i => map.insert(i, i) }
     map.insert(Int.MaxValue, Int.MaxValue)
@@ -358,12 +340,10 @@ class ExternalAppendOnlyMapSuite extends FunSuite with LocalSparkContext {
   }
 
   test("spilling with null keys and values") {
-    val conf = createSparkConf(true)
+    val conf = createSparkConf(loadDefaults = true)
     conf.set("spark.shuffle.memoryFraction", "0.001")
     sc = new SparkContext("local-cluster[1,1,512]", "test", conf)
-
-    val map = new ExternalAppendOnlyMap[Int, Int, ArrayBuffer[Int]](
-      createCombiner, mergeValue, mergeCombiners)
+    val map = createExternalMap[Int]
 
     map.insertAll((1 to 100000).iterator.map(i => (i, i)))
     map.insert(null.asInstanceOf[Int], 1)