From 7f4b583733714bbecb43fb0823134bf2ec720a17 Mon Sep 17 00:00:00 2001
From: Qiping Li <liqiping1991@gmail.com>
Date: Wed, 29 Apr 2015 23:52:16 +0100
Subject: [PATCH] [SPARK-7181] [CORE] fix inifite loop in Externalsorter's
mergeWithAggregation
see [SPARK-7181](https://issues.apache.org/jira/browse/SPARK-7181).
Author: Qiping Li <liqiping1991@gmail.com>
Closes #5737 from chouqin/externalsorter and squashes the following commits:
2924b93 [Qiping Li] fix inifite loop in Externalsorter's mergeWithAggregation
---
.../org/apache/spark/util/collection/ExternalSorter.scala | 3 ++-
.../apache/spark/util/collection/ExternalSorterSuite.scala | 5 ++++-
2 files changed, 6 insertions(+), 2 deletions(-)
diff --git a/core/src/main/scala/org/apache/spark/util/collection/ExternalSorter.scala b/core/src/main/scala/org/apache/spark/util/collection/ExternalSorter.scala
index ef3cac6225..4ed8a740f9 100644
--- a/core/src/main/scala/org/apache/spark/util/collection/ExternalSorter.scala
+++ b/core/src/main/scala/org/apache/spark/util/collection/ExternalSorter.scala
@@ -527,7 +527,8 @@ private[spark] class ExternalSorter[K, V, C](
val k = elem._1
var c = elem._2
while (sorted.hasNext && sorted.head._1 == k) {
- c = mergeCombiners(c, sorted.head._2)
+ val pair = sorted.next()
+ c = mergeCombiners(c, pair._2)
}
(k, c)
}
diff --git a/core/src/test/scala/org/apache/spark/util/collection/ExternalSorterSuite.scala b/core/src/test/scala/org/apache/spark/util/collection/ExternalSorterSuite.scala
index 9ff067f86a..de26aa351b 100644
--- a/core/src/test/scala/org/apache/spark/util/collection/ExternalSorterSuite.scala
+++ b/core/src/test/scala/org/apache/spark/util/collection/ExternalSorterSuite.scala
@@ -506,7 +506,10 @@ class ExternalSorterSuite extends FunSuite with LocalSparkContext with PrivateMe
val agg = new Aggregator[Int, Int, Int](i => i, (i, j) => i + j, (i, j) => i + j)
val ord = implicitly[Ordering[Int]]
val sorter = new ExternalSorter(Some(agg), Some(new HashPartitioner(3)), Some(ord), None)
- sorter.insertAll((0 until 100000).iterator.map(i => (i / 2, i)))
+
+ // avoid combine before spill
+ sorter.insertAll((0 until 50000).iterator.map(i => (i , 2 * i)))
+ sorter.insertAll((0 until 50000).iterator.map(i => (i, 2 * i + 1)))
val results = sorter.partitionedIterator.map{case (p, vs) => (p, vs.toSet)}.toSet
val expected = (0 until 3).map(p => {
(p, (0 until 50000).map(i => (i, i * 4 + 1)).filter(_._1 % 3 == p).toSet)
--
GitLab