spark-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sro...@apache.org
Subject spark git commit: [SPARK-7181] [CORE] fix inifite loop in Externalsorter's mergeWithAggregation
Date Wed, 29 Apr 2015 22:52:32 GMT
Repository: spark
Updated Branches:
  refs/heads/branch-1.3 bf4ca12f7 -> 5b893bd60


[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

(cherry picked from commit 7f4b583733714bbecb43fb0823134bf2ec720a17)
Signed-off-by: Sean Owen <sowen@cloudera.com>


Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/5b893bd6
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/5b893bd6
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/5b893bd6

Branch: refs/heads/branch-1.3
Commit: 5b893bd60900ac56d71d3d8660fc96e2db6d95ad
Parents: bf4ca12
Author: Qiping Li <liqiping1991@gmail.com>
Authored: Wed Apr 29 23:52:16 2015 +0100
Committer: Sean Owen <sowen@cloudera.com>
Committed: Wed Apr 29 23:52:28 2015 +0100

----------------------------------------------------------------------
 .../scala/org/apache/spark/util/collection/ExternalSorter.scala | 3 ++-
 .../org/apache/spark/util/collection/ExternalSorterSuite.scala  | 5 ++++-
 2 files changed, 6 insertions(+), 2 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/spark/blob/5b893bd6/core/src/main/scala/org/apache/spark/util/collection/ExternalSorter.scala
----------------------------------------------------------------------
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 244d4af..c8da612 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
@@ -510,7 +510,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)
         }

http://git-wip-us.apache.org/repos/asf/spark/blob/5b893bd6/core/src/test/scala/org/apache/spark/util/collection/ExternalSorterSuite.scala
----------------------------------------------------------------------
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 72d9679..7d91951 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)


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org


Mime
View raw message