Skip to content

Commit dab1b0a

Browse files
jerryshaomateiz
authored andcommitted
[SPARK-3032][Shuffle] Fix key comparison integer overflow introduced sorting exception
Previous key comparison in `ExternalSorter` will get wrong sorting result or exception when key comparison overflows, details can be seen in [SPARK-3032](https://issues.apache.org/jira/browse/SPARK-3032). Here fix this and add a unit test to prove it. Author: jerryshao <[email protected]> Closes apache#2514 from jerryshao/SPARK-3032 and squashes the following commits: 6f3c302 [jerryshao] Improve the unit test according to comments 01911e6 [jerryshao] Change the test to show the contract violate exception 83acb38 [jerryshao] Minor changes according to comments fa2a08f [jerryshao] Fix key comparison integer overflow introduced sorting exception
1 parent 587a0cd commit dab1b0a

File tree

2 files changed

+56
-1
lines changed

2 files changed

+56
-1
lines changed

core/src/main/scala/org/apache/spark/util/collection/ExternalSorter.scala

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -144,7 +144,7 @@ private[spark] class ExternalSorter[K, V, C](
144144
override def compare(a: K, b: K): Int = {
145145
val h1 = if (a == null) 0 else a.hashCode()
146146
val h2 = if (b == null) 0 else b.hashCode()
147-
h1 - h2
147+
if (h1 < h2) -1 else if (h1 == h2) 0 else 1
148148
}
149149
})
150150

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

Lines changed: 55 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,8 @@ import org.scalatest.{PrivateMethodTester, FunSuite}
2424
import org.apache.spark._
2525
import org.apache.spark.SparkContext._
2626

27+
import scala.util.Random
28+
2729
class ExternalSorterSuite extends FunSuite with LocalSparkContext with PrivateMethodTester {
2830
private def createSparkConf(loadDefaults: Boolean): SparkConf = {
2931
val conf = new SparkConf(loadDefaults)
@@ -707,4 +709,57 @@ class ExternalSorterSuite extends FunSuite with LocalSparkContext with PrivateMe
707709
Some(agg), Some(new HashPartitioner(FEW_PARTITIONS)), None, None)
708710
assertDidNotBypassMergeSort(sorter4)
709711
}
712+
713+
test("sort without breaking sorting contracts") {
714+
val conf = createSparkConf(true)
715+
conf.set("spark.shuffle.memoryFraction", "0.01")
716+
conf.set("spark.shuffle.manager", "sort")
717+
sc = new SparkContext("local-cluster[1,1,512]", "test", conf)
718+
719+
// Using wrongOrdering to show integer overflow introduced exception.
720+
val rand = new Random(100L)
721+
val wrongOrdering = new Ordering[String] {
722+
override def compare(a: String, b: String) = {
723+
val h1 = if (a == null) 0 else a.hashCode()
724+
val h2 = if (b == null) 0 else b.hashCode()
725+
h1 - h2
726+
}
727+
}
728+
729+
val testData = Array.tabulate(100000) { _ => rand.nextInt().toString }
730+
731+
val sorter1 = new ExternalSorter[String, String, String](
732+
None, None, Some(wrongOrdering), None)
733+
val thrown = intercept[IllegalArgumentException] {
734+
sorter1.insertAll(testData.iterator.map(i => (i, i)))
735+
sorter1.iterator
736+
}
737+
738+
assert(thrown.getClass() === classOf[IllegalArgumentException])
739+
assert(thrown.getMessage().contains("Comparison method violates its general contract"))
740+
sorter1.stop()
741+
742+
// Using aggregation and external spill to make sure ExternalSorter using
743+
// partitionKeyComparator.
744+
def createCombiner(i: String) = ArrayBuffer(i)
745+
def mergeValue(c: ArrayBuffer[String], i: String) = c += i
746+
def mergeCombiners(c1: ArrayBuffer[String], c2: ArrayBuffer[String]) = c1 ++= c2
747+
748+
val agg = new Aggregator[String, String, ArrayBuffer[String]](
749+
createCombiner, mergeValue, mergeCombiners)
750+
751+
val sorter2 = new ExternalSorter[String, String, ArrayBuffer[String]](
752+
Some(agg), None, None, None)
753+
sorter2.insertAll(testData.iterator.map(i => (i, i)))
754+
755+
// To validate the hash ordering of key
756+
var minKey = Int.MinValue
757+
sorter2.iterator.foreach { case (k, v) =>
758+
val h = k.hashCode()
759+
assert(h >= minKey)
760+
minKey = h
761+
}
762+
763+
sorter2.stop()
764+
}
710765
}

0 commit comments

Comments
 (0)