Skip to content

Commit 0f7c9e8

Browse files
committed
[SPARK-18189] [SQL] [Followup] Move test from ReplSuite to prevent java.lang.ClassCircularityError
closes apache#15774
1 parent 0e3312e commit 0f7c9e8

File tree

2 files changed

+12
-17
lines changed

2 files changed

+12
-17
lines changed

repl/scala-2.11/src/test/scala/org/apache/spark/repl/ReplSuite.scala

-17
Original file line numberDiff line numberDiff line change
@@ -473,21 +473,4 @@ class ReplSuite extends SparkFunSuite {
473473
assertDoesNotContain("AssertionError", output)
474474
assertDoesNotContain("Exception", output)
475475
}
476-
477-
test("SPARK-18189: Fix serialization issue in KeyValueGroupedDataset") {
478-
val resultValue = 12345
479-
val output = runInterpreter("local",
480-
s"""
481-
|val keyValueGrouped = Seq((1, 2), (3, 4)).toDS().groupByKey(_._1)
482-
|val mapGroups = keyValueGrouped.mapGroups((k, v) => (k, 1))
483-
|val broadcasted = sc.broadcast($resultValue)
484-
|
485-
|// Using broadcast triggers serialization issue in KeyValueGroupedDataset
486-
|val dataset = mapGroups.map(_ => broadcasted.value)
487-
|dataset.collect()
488-
""".stripMargin)
489-
assertDoesNotContain("error:", output)
490-
assertDoesNotContain("Exception", output)
491-
assertContains(s": Array[Int] = Array($resultValue, $resultValue)", output)
492-
}
493476
}

sql/core/src/test/scala/org/apache/spark/sql/DatasetSuite.scala

+12
Original file line numberDiff line numberDiff line change
@@ -923,6 +923,18 @@ class DatasetSuite extends QueryTest with SharedSQLContext {
923923
.groupByKey(_.a).flatMapGroups { case (x, iter) => List[Int]() })
924924
}
925925

926+
test("SPARK-18189: Fix serialization issue in KeyValueGroupedDataset") {
927+
val resultValue = 12345
928+
val keyValueGrouped = Seq((1, 2), (3, 4)).toDS().groupByKey(_._1)
929+
val mapGroups = keyValueGrouped.mapGroups((k, v) => (k, 1))
930+
val broadcasted = spark.sparkContext.broadcast(resultValue)
931+
932+
// Using broadcast triggers serialization issue in KeyValueGroupedDataset
933+
val dataset = mapGroups.map(_ => broadcasted.value)
934+
935+
assert(dataset.collect() sameElements Array(resultValue, resultValue))
936+
}
937+
926938
Seq(true, false).foreach { eager =>
927939
def testCheckpointing(testName: String)(f: => Unit): Unit = {
928940
test(s"Dataset.checkpoint() - $testName (eager = $eager)") {

0 commit comments

Comments
 (0)