This is an automated email from the ASF dual-hosted git repository.
baibaichen pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new d3325b0000 [GLUTEN-13174][VL] Re-enable the Spark 4.2
storage-partitioned join tests (#13243)
d3325b0000 is described below
commit d3325b000053c48864442e0452af64ef26ac9b61
Author: Akshay Tayal <[email protected]>
AuthorDate: Thu Oct 15 19:32:41 2026 -0700
[GLUTEN-13174][VL] Re-enable the Spark 4.2 storage-partitioned join tests
(#13243)
* [GLUTEN-13174][VL] Re-enable the Spark 4.2 storage-partitioned join tests
Remove the 29 GlutenKeyGroupedPartitioningSuite excludes that #13163 added
for
Spark 4.2, so the spark42 UT CI shows which of them still fail on current
main. Fixes for the remaining failures follow in this PR.
Co-authored-by: Copilot <[email protected]>
Copilot-Session: 3a1d1e30-b32c-45b1-9f0e-d7aa4654eab9
* [GLUTEN-13174][VL] Leave the SPJ sorted-merge coalescing tests to a
separate PR
Keep the 4 SPARK-55715 / SPARK-56549 tests excluded here. They need a
different fix and have their own PR.
Co-authored-by: Copilot <[email protected]>
Copilot-Session: 3a1d1e30-b32c-45b1-9f0e-d7aa4654eab9
* [GLUTEN-13174][VL] Look for GroupPartitionsExec under
SortMergeJoinExecTransformer
Spark 4.2 collectGroupPartitions only looks under a vanilla
SortMergeJoinExec,
which Gluten replaces with SortMergeJoinExecTransformer. Override it the
same
way as collectShuffles. This fixes 8 of the re-enabled tests.
Co-authored-by: Copilot <[email protected]>
Copilot-Session: 3a1d1e30-b32c-45b1-9f0e-d7aa4654eab9
* [GLUTEN-13174][VL] Port the Gluten SPJ test copies to Spark 4.2
Spark 4.2 moved SPJ partition grouping from BatchScanExec into
GroupPartitionsExec and updated its KeyGroupedPartitioningSuite for it.
Bring the Gluten copies in GlutenKeyGroupedPartitioningSuite in line:
- SPARK-42038 (8 tests), SPARK-44647 (3 tests) and SPARK-47094: check
the GroupPartitionsExec output partitions instead of the raw scan
split counts, with the Spark 4.2 expectations (no GroupPartitionsExec
when SPJ is not used). The SPARK-42038 checks also require a
non-empty result.
- SPARK-53322: Spark 4.2 uses checkpointed scans for SPJ. Replace the
"aren't used for SPJ" and "can't shuffle other children on SPJ" copies
with the Spark 4.2 "are used for SPJ" and "can shuffle other children
on SPJ" tests, and remove "can be shuffled by children on SPJ", which
Spark 4.2 dropped. Add the Spark 4.2 GroupPartitionsExec checks to
"avoid shuffles for aggregates".
- collectAllShuffles returns vanilla and columnar shuffles again, as the
inherited tests expect. Native-only checks use the new
collectAllColumnarShuffles.
- Add Gluten variants of the two SPARK-56241 tests, which look for a
vanilla SortMergeJoinExec, and of "SPARK-55848: checkpointed
partially-clustered join with dedup", which looks for a vanilla
RDDScanExec that Gluten offloads to an RDDScanTransformer. The vanilla
tests are excluded.
Co-authored-by: Copilot <[email protected]>
Copilot-Session: 3a1d1e30-b32c-45b1-9f0e-d7aa4654eab9
---------
Co-authored-by: Akshay Tayal <[email protected]>
Co-authored-by: Copilot <[email protected]>
Copilot-Session: 3a1d1e30-b32c-45b1-9f0e-d7aa4654eab9
---
.../gluten/utils/velox/VeloxTestSettings.scala | 80 +----
.../GlutenKeyGroupedPartitioningSuite.scala | 329 ++++++++++++++++-----
2 files changed, 253 insertions(+), 156 deletions(-)
diff --git
a/gluten-ut/spark42/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
b/gluten-ut/spark42/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
index 2023bf3236..08fbd9c469 100644
---
a/gluten-ut/spark42/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
+++
b/gluten-ut/spark42/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
@@ -111,6 +111,7 @@ class VeloxTestSettings extends BackendTestSettings {
.exclude("partitioned join: number of buckets mismatch should trigger
shuffle")
.exclude("partitioned join: only one side reports partitioning")
.exclude("partitioned join: join with two partition keys and different #
of partition keys")
+ .exclude("SPARK-55848: checkpointed partially-clustered join with dedup")
.excludeByPrefix("SPARK-47094")
.excludeByPrefix("SPARK-48655")
.excludeByPrefix("SPARK-48012")
@@ -121,58 +122,13 @@ class VeloxTestSettings extends BackendTestSettings {
// disable due to check for SMJ node
.excludeByPrefix("SPARK-41413: partitioned join:")
.excludeByPrefix("SPARK-42038: partially clustered:")
- .exclude("SPARK-44641: duplicated records when SPJ is not triggered")
- .exclude(
- "Gluten - SPARK-42038: partially clustered: full outer join is not
applicable"
- ) // GLUTEN-12569 Spark 4.2 UT enablement (CI-confirmed)
- .exclude(
- "Gluten - SPARK-42038: partially clustered: left outer join"
- ) // GLUTEN-12569 Spark 4.2 UT enablement (CI-confirmed)
- .exclude(
- "Gluten - SPARK-42038: partially clustered: right outer join"
- ) // GLUTEN-12569 Spark 4.2 UT enablement (CI-confirmed)
- .exclude(
- "Gluten - SPARK-42038: partially clustered: with different partition
keys and both sides partially clustered"
- ) // GLUTEN-12569 Spark 4.2 UT enablement (CI-confirmed)
- .exclude(
- "Gluten - SPARK-42038: partially clustered: with different partition
keys and missing keys on left-hand side"
- ) // GLUTEN-12569 Spark 4.2 UT enablement (CI-confirmed)
- .exclude(
- "Gluten - SPARK-42038: partially clustered: with different partition
keys and missing keys on right-hand side"
- ) // GLUTEN-12569 Spark 4.2 UT enablement (CI-confirmed)
- .exclude(
- "Gluten - SPARK-42038: partially clustered: with same partition keys and
both sides partially clustered"
- ) // GLUTEN-12569 Spark 4.2 UT enablement (CI-confirmed)
- .exclude(
- "Gluten - SPARK-42038: partially clustered: with same partition keys and
one side fully clustered"
- ) // GLUTEN-12569 Spark 4.2 UT enablement (CI-confirmed)
- .exclude(
- "Gluten - SPARK-44647: SPJ: test join key is subset of cluster key with
push values and partially-clustered"
- ) // GLUTEN-12569 Spark 4.2 UT enablement (CI-confirmed)
- .exclude(
- "Gluten - SPARK-44647: test join key is the second partition key and a
transform"
- ) // GLUTEN-12569 Spark 4.2 UT enablement (CI-confirmed)
- .exclude(
- "Gluten - SPARK-47094: Compatible buckets does not support SPJ with
push-down values or partially-clustered"
- ) // GLUTEN-12569 Spark 4.2 UT enablement (CI-confirmed)
- .exclude(
- "Gluten - SPARK-53322: checkpointed scans aren't used for SPJ"
- ) // GLUTEN-12569 Spark 4.2 UT enablement (CI-confirmed)
.exclude(
- "Gluten - SPARK-53322: checkpointed scans can be shuffled by children on
SPJ"
- ) // GLUTEN-12569 Spark 4.2 UT enablement (CI-confirmed)
- .exclude(
- "Gluten - SPARK-53322: checkpointed scans can't shuffle other children
on SPJ"
- ) // GLUTEN-12569 Spark 4.2 UT enablement (CI-confirmed)
- .exclude(
- "SPARK-48065: SPJ: allowJoinKeysSubsetOfPartitionKeys is too strict"
- ) // GLUTEN-12569 Spark 4.2 UT enablement (CI-confirmed)
- .exclude(
- "SPARK-55535: Multi table join granular partition grouping"
- ) // GLUTEN-12569 Spark 4.2 UT enablement (CI-confirmed)
+ "SPARK-56241: GroupPartitionsExec non-coalescing passes through child
ordering, " +
+ "no pre-join SortExec needed before SortMergeJoin")
.exclude(
- "SPARK-55535: Multi table join partial clustering"
- ) // GLUTEN-12569 Spark 4.2 UT enablement (CI-confirmed)
+ "SPARK-56241: GroupPartitionsExec coalescing derives ordering from key
expressions, " +
+ "no pre-join SortExec needed before SortMergeJoin")
+ .exclude("SPARK-44641: duplicated records when SPJ is not triggered")
.exclude(
"SPARK-55715: Custom metrics of sorted-merge coalesced partitions"
) // GLUTEN-12569 Spark 4.2 UT enablement (CI-confirmed)
@@ -182,33 +138,9 @@ class VeloxTestSettings extends BackendTestSettings {
.exclude(
"SPARK-55715: preserve outputOrdering when coalescing
transform-partitioned splits"
) // GLUTEN-12569 Spark 4.2 UT enablement (CI-confirmed)
- .exclude(
- "SPARK-55848: Window dedup after SPJ with partial clustering"
- ) // GLUTEN-12569 Spark 4.2 UT enablement (CI-confirmed)
- .exclude(
- "SPARK-55848: checkpointed partially-clustered join with dedup"
- ) // GLUTEN-12569 Spark 4.2 UT enablement (CI-confirmed)
- .exclude(
- "SPARK-55848: dropDuplicates after SPJ with partial clustering"
- ) // GLUTEN-12569 Spark 4.2 UT enablement (CI-confirmed)
- .exclude(
- "SPARK-56241: GroupPartitionsExec coalescing derives ordering from key
expressions, no pre-join SortExec needed before SortMergeJoin"
- ) // GLUTEN-12569 Spark 4.2 UT enablement (CI-confirmed)
- .exclude(
- "SPARK-56241: GroupPartitionsExec non-coalescing passes through child
ordering, no pre-join SortExec needed before SortMergeJoin"
- ) // GLUTEN-12569 Spark 4.2 UT enablement (CI-confirmed)
.exclude(
"SPARK-56549: k-way merge enabled only when parent requires ordering"
) // GLUTEN-12569 Spark 4.2 UT enablement (CI-confirmed)
- .exclude(
- "partitioned join: exact distribution (same number of buckets) from both
sides"
- ) // GLUTEN-12569 Spark 4.2 UT enablement (CI-confirmed)
- .exclude(
- "partitioned join: join with two partition keys and matching & sorted
partitions"
- ) // GLUTEN-12569 Spark 4.2 UT enablement (CI-confirmed)
- .exclude(
- "partitioned join: join with two partition keys and unsorted partitions"
- ) // GLUTEN-12569 Spark 4.2 UT enablement (CI-confirmed)
enableSuite[GlutenLocalScanSuite]
enableSuite[GlutenMetadataColumnSuite]
enableSuite[GlutenSupportsCatalogOptionsSuite]
diff --git
a/gluten-ut/spark42/src/test/scala/org/apache/spark/sql/connector/GlutenKeyGroupedPartitioningSuite.scala
b/gluten-ut/spark42/src/test/scala/org/apache/spark/sql/connector/GlutenKeyGroupedPartitioningSuite.scala
index 9d03138809..4c665e5dee 100644
---
a/gluten-ut/spark42/src/test/scala/org/apache/spark/sql/connector/GlutenKeyGroupedPartitioningSuite.scala
+++
b/gluten-ut/spark42/src/test/scala/org/apache/spark/sql/connector/GlutenKeyGroupedPartitioningSuite.scala
@@ -17,7 +17,7 @@
package org.apache.spark.sql.connector
import org.apache.gluten.config.GlutenConfig
-import org.apache.gluten.execution.SortMergeJoinExecTransformer
+import org.apache.gluten.execution.{SortExecTransformer,
SortMergeJoinExecTransformer}
import org.apache.spark.SparkConf
import org.apache.spark.sql.{DataFrame, GlutenSQLTestsBaseTrait, Row}
@@ -26,8 +26,8 @@ import org.apache.spark.sql.connector.catalog.{Column,
Identifier, InMemoryTable
import org.apache.spark.sql.connector.distributions.Distributions
import org.apache.spark.sql.connector.expressions.Expressions.{bucket, days,
identity, years}
import org.apache.spark.sql.connector.expressions.Transform
-import org.apache.spark.sql.execution.{ColumnarShuffleExchangeExec, SparkPlan}
-import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
+import org.apache.spark.sql.execution.{ColumnarShuffleExchangeExec,
RDDScanExec, RDDScanTransformer, SortExec, SparkPlan}
+import org.apache.spark.sql.execution.datasources.v2.GroupPartitionsExec
import org.apache.spark.sql.execution.exchange.{ShuffleExchangeExec,
ShuffleExchangeLike}
import org.apache.spark.sql.execution.joins.SortMergeJoinExec
import org.apache.spark.sql.functions.{col, max}
@@ -100,7 +100,22 @@ class GlutenKeyGroupedPartitioningSuite
})
}
- override protected def collectAllShuffles(plan: SparkPlan):
Seq[ColumnarShuffleExchangeExec] = {
+ override protected def collectGroupPartitions(plan: SparkPlan):
Seq[GroupPartitionsExec] = {
+ // here we skip collecting group partitions operators that are not
associated with SMJ
+ collect(plan) {
+ case s: SortMergeJoinExec => s
+ case s: SortMergeJoinExecTransformer => s
+ }.flatMap(smj => collect(smj) { case g: GroupPartitionsExec => g })
+ }.toSet.toSeq
+
+ override protected def collectAllShuffles(plan: SparkPlan):
Seq[ShuffleExchangeLike] = {
+ collect(plan) {
+ case s: ShuffleExchangeExec => s
+ case s: ColumnarShuffleExchangeExec => s
+ }
+ }
+
+ private def collectAllColumnarShuffles(plan: SparkPlan):
Seq[ColumnarShuffleExchangeExec] = {
collect(plan) { case s: ColumnarShuffleExchangeExec => s }
}
@@ -108,10 +123,6 @@ class GlutenKeyGroupedPartitioningSuite
collect(plan) { case s: ShuffleExchangeExec => s }
}
- private def collectScans(plan: SparkPlan): Seq[BatchScanExec] = {
- collect(plan) { case s: BatchScanExec => s }
- }
-
private def selectWithMergeJoinHint(t1: String, t2: String): String = {
s"SELECT /*+ MERGE($t1, $t2) */ "
}
@@ -359,8 +370,11 @@ class GlutenKeyGroupedPartitioningSuite
val shuffles =
collectColumnarShuffleExchangeExec(df.queryExecution.executedPlan)
assert(shuffles.isEmpty, "should not contain any shuffle")
if (pushDownValues) {
- val scans = collectScans(df.queryExecution.executedPlan)
- assert(scans.forall(_.inputRDD.partitions.length == expected))
+ val groupPartitions =
collectGroupPartitions(df.queryExecution.executedPlan)
+ .map(_.outputPartitioning.numPartitions)
+ assert(
+ groupPartitions.nonEmpty && groupPartitions.forall(_ ==
expected),
+ s"Expected $expected but got $groupPartitions")
}
checkAnswer(
df,
@@ -415,8 +429,11 @@ class GlutenKeyGroupedPartitioningSuite
val shuffles =
collectColumnarShuffleExchangeExec(df.queryExecution.executedPlan)
assert(shuffles.isEmpty, "should not contain any shuffle")
if (pushDownValues) {
- val scans = collectScans(df.queryExecution.executedPlan)
- assert(scans.forall(_.inputRDD.partitions.length == expected))
+ val groupPartitions =
collectGroupPartitions(df.queryExecution.executedPlan)
+ .map(_.outputPartitioning.numPartitions)
+ assert(
+ groupPartitions.nonEmpty && groupPartitions.forall(_ ==
expected),
+ s"Expected $expected but got $groupPartitions")
}
checkAnswer(
df,
@@ -481,8 +498,11 @@ class GlutenKeyGroupedPartitioningSuite
val shuffles =
collectColumnarShuffleExchangeExec(df.queryExecution.executedPlan)
if (pushDownValues) {
assert(shuffles.isEmpty, "should not contain any shuffle")
- val scans = collectScans(df.queryExecution.executedPlan)
- assert(scans.forall(_.inputRDD.partitions.length == expected))
+ val groupPartitions =
collectGroupPartitions(df.queryExecution.executedPlan)
+ .map(_.outputPartitioning.numPartitions)
+ assert(
+ groupPartitions.nonEmpty && groupPartitions.forall(_ ==
expected),
+ s"Expected $expected but got $groupPartitions")
} else {
assert(
shuffles.nonEmpty,
@@ -550,8 +570,11 @@ class GlutenKeyGroupedPartitioningSuite
val shuffles =
collectColumnarShuffleExchangeExec(df.queryExecution.executedPlan)
if (pushDownValues) {
assert(shuffles.isEmpty, "should not contain any shuffle")
- val scans = collectScans(df.queryExecution.executedPlan)
- assert(scans.forall(_.inputRDD.partitions.length == expected))
+ val groupPartitions =
collectGroupPartitions(df.queryExecution.executedPlan)
+ .map(_.outputPartitioning.numPartitions)
+ assert(
+ groupPartitions.nonEmpty && groupPartitions.forall(_ ==
expected),
+ s"Expected $expected but got $groupPartitions")
} else {
assert(
shuffles.nonEmpty,
@@ -608,8 +631,11 @@ class GlutenKeyGroupedPartitioningSuite
val shuffles =
collectColumnarShuffleExchangeExec(df.queryExecution.executedPlan)
if (pushDownValues) {
assert(shuffles.isEmpty, "should not contain any shuffle")
- val scans = collectScans(df.queryExecution.executedPlan)
- assert(scans.forall(_.inputRDD.partitions.length == expected))
+ val groupPartitions =
collectGroupPartitions(df.queryExecution.executedPlan)
+ .map(_.outputPartitioning.numPartitions)
+ assert(
+ groupPartitions.nonEmpty && groupPartitions.forall(_ ==
expected),
+ s"Expected $expected but got $groupPartitions")
} else {
assert(
shuffles.nonEmpty,
@@ -663,10 +689,11 @@ class GlutenKeyGroupedPartitioningSuite
val shuffles =
collectColumnarShuffleExchangeExec(df.queryExecution.executedPlan)
if (pushDownValues) {
assert(shuffles.isEmpty, "should not contain any shuffle")
- val scans = collectScans(df.queryExecution.executedPlan)
+ val groupPartitions =
collectGroupPartitions(df.queryExecution.executedPlan)
+ .map(_.outputPartitioning.numPartitions)
assert(
- scans.forall(_.inputRDD.partitions.length == expected),
- s"Expected $expected but got
${scans.head.inputRDD.partitions.length}")
+ groupPartitions.nonEmpty && groupPartitions.forall(_ ==
expected),
+ s"Expected $expected but got $groupPartitions")
} else {
assert(
shuffles.nonEmpty,
@@ -727,11 +754,11 @@ class GlutenKeyGroupedPartitioningSuite
val shuffles =
collectColumnarShuffleExchangeExec(df.queryExecution.executedPlan)
if (pushDownValues) {
assert(shuffles.isEmpty, "should not contain any shuffle")
- val scans = collectScans(df.queryExecution.executedPlan)
- assert(scans.map(_.inputRDD.partitions.length).toSet.size == 1)
+ val groupPartitions =
collectGroupPartitions(df.queryExecution.executedPlan)
+ .map(_.outputPartitioning.numPartitions)
assert(
- scans.forall(_.inputRDD.partitions.length == expected),
- s"Expected $expected but got
${scans.head.inputRDD.partitions.length}")
+ groupPartitions.nonEmpty && groupPartitions.forall(_ ==
expected),
+ s"Expected $expected but got $groupPartitions")
} else {
assert(
shuffles.nonEmpty,
@@ -790,11 +817,11 @@ class GlutenKeyGroupedPartitioningSuite
val shuffles =
collectColumnarShuffleExchangeExec(df.queryExecution.executedPlan)
if (pushDownValues) {
assert(shuffles.isEmpty, "should not contain any shuffle")
- val scans = collectScans(df.queryExecution.executedPlan)
- assert(scans.map(_.inputRDD.partitions.length).toSet.size == 1)
+ val groupPartitions =
collectGroupPartitions(df.queryExecution.executedPlan)
+ .map(_.outputPartitioning.numPartitions)
assert(
- scans.forall(_.inputRDD.partitions.length == expected),
- s"Expected $expected but got
${scans.head.inputRDD.partitions.length}")
+ groupPartitions.nonEmpty && groupPartitions.forall(_ ==
expected),
+ s"Expected $expected but got $groupPartitions")
} else {
assert(
shuffles.nonEmpty,
@@ -1113,7 +1140,7 @@ class GlutenKeyGroupedPartitioningSuite
keyGroupedShuffles.nonEmpty,
"KeyGroupedPartitioning shuffle should fall back to a vanilla
ShuffleExchangeExec")
- val columnarKeyGroupedShuffles = collectAllShuffles(plan)
+ val columnarKeyGroupedShuffles = collectAllColumnarShuffles(plan)
.filter(_.outputPartitioning.isInstanceOf[KeyedPartitioning])
assert(
columnarKeyGroupedShuffles.isEmpty,
@@ -1320,23 +1347,23 @@ class GlutenKeyGroupedPartitioningSuite
assert(shuffles.nonEmpty, "SPJ should not be
triggered")
}
- val scannedPartitions =
collectScans(df.queryExecution.executedPlan)
- .map(_.inputRDD.partitions.length)
+ val groupPartitions =
collectGroupPartitions(df.queryExecution.executedPlan)
+ .map(_.outputPartitioning.numPartitions)
(allowJoinKeysSubsetOfPartitionKeys, partiallyClustered,
filter) match {
// SPJ, partially-clustered, with filter
- case (true, true, true) => assert(scannedPartitions ==
Seq(6, 6))
+ case (true, true, true) => assert(groupPartitions ==
Seq(6, 6))
// SPJ, partially-clustered, no filter
- case (true, true, false) => assert(scannedPartitions
== Seq(8, 8))
+ case (true, true, false) => assert(groupPartitions ==
Seq(8, 8))
// SPJ and not partially-clustered, with filter
- case (true, false, true) => assert(scannedPartitions
== Seq(2, 2))
+ case (true, false, true) => assert(groupPartitions ==
Seq(2, 2))
// SPJ and not partially-clustered, no filter
- case (true, false, false) => assert(scannedPartitions
== Seq(4, 4))
+ case (true, false, false) => assert(groupPartitions ==
Seq(4, 4))
// No SPJ
- case _ => assert(scannedPartitions == Seq(5, 4))
+ case _ => assert(groupPartitions.isEmpty)
}
checkAnswer(
@@ -1419,13 +1446,13 @@ class GlutenKeyGroupedPartitioningSuite
assert(shuffles.nonEmpty, "SPJ should not be triggered")
}
- val scans = collectScans(df.queryExecution.executedPlan)
- .map(_.inputRDD.partitions.length)
+ val partitions =
collectGroupPartitions(df.queryExecution.executedPlan)
+ .map(_.outputPartitioning.numPartitions)
(pushDownValues, allowJoinKeysSubsetOfPartitionKeys,
partiallyClustered) match {
// SPJ and partially-clustered
- case (true, true, true) => assert(scans == Seq(3, 3))
+ case (_, true, _) => assert(partitions == Seq(3, 3))
// non-SPJ or SPJ/partially-clustered
- case _ => assert(scans == Seq(3, 3))
+ case _ => assert(partitions.isEmpty)
}
}
}
@@ -1479,15 +1506,15 @@ class GlutenKeyGroupedPartitioningSuite
assert(shuffles.nonEmpty, "SPJ should not be triggered")
}
- val scans = collectScans(df.queryExecution.executedPlan)
- .map(_.inputRDD.partitions.length)
+ val partitions =
collectGroupPartitions(df.queryExecution.executedPlan)
+ .map(_.outputPartitioning.numPartitions)
(allowJoinKeysSubsetOfPartitionKeys, partiallyClustered)
match {
// SPJ and partially-clustered
- case (true, true) => assert(scans == Seq(5, 5))
+ case (true, true) => assert(partitions == Seq(5, 5))
// SPJ and not partially-clustered
- case (true, false) => assert(scans == Seq(3, 3))
+ case (true, false) => assert(partitions == Seq(3, 3))
// No SPJ
- case _ => assert(scans == Seq(4, 4))
+ case _ => assert(partitions.isEmpty)
}
checkAnswer(
@@ -1599,16 +1626,16 @@ class GlutenKeyGroupedPartitioningSuite
|""".stripMargin)
val shuffles =
collectColumnarShuffleExchangeExec(df.queryExecution.executedPlan)
- val scans =
-
collectScans(df.queryExecution.executedPlan).map(_.inputRDD.partitions.length)
+ val partitions =
collectGroupPartitions(df.queryExecution.executedPlan)
+ .map(_.outputPartitioning.numPartitions)
(allowPushDown, partiallyClustered) match {
case (true, false) =>
assert(shuffles.isEmpty, "SPJ should be triggered")
- assert(scans == Seq(2, 2))
+ assert(partitions == Seq(2, 2))
case (_, _) =>
assert(shuffles.nonEmpty, "SPJ should not be triggered")
- assert(scans == Seq(3, 2))
+ assert(partitions.isEmpty)
}
checkAnswer(
@@ -1900,10 +1927,13 @@ class GlutenKeyGroupedPartitioningSuite
assert(
shuffles.isEmpty,
"should not contain shuffle when not grouping by partition values")
+ val groupPartitions =
collectAllGroupPartitions(df.queryExecution.executedPlan)
+ assert(groupPartitions.size === 1)
+ assert(groupPartitions.head.outputPartitioning.numPartitions == 3)
}
}
- testGluten("SPARK-53322: checkpointed scans aren't used for SPJ") {
+ testGluten("SPARK-53322: checkpointed scans are used for SPJ") {
withTempDir {
dir =>
spark.sparkContext.setCheckpointDir(dir.getPath)
@@ -1943,14 +1973,22 @@ class GlutenKeyGroupedPartitioningSuite
.orderBy("id", "purchase_price", "sale_price")
checkAnswer(df, Seq(Row(1, "aa", 41.0, 40.0), Row(3, "cc", 15.5,
25.5)))
- // One shuffle for sort and two shuffles for join are expected.
- assert(collectAllShuffles(df.queryExecution.executedPlan).length
=== 3)
+ val plan = df.queryExecution.executedPlan
+ if (pushdownValues) {
+ // 1 shuffle for SORT and 2 group partitions for JOIN are
expected.
+ assert(collectAllShuffles(plan).length === 1)
+ assert(collectAllGroupPartitions(plan).length === 2)
+ } else {
+ // 1 shuffle for SORT and 2 shuffles for JOIN are expected.
+ assert(collectAllShuffles(plan).length === 3)
+ assert(collectAllGroupPartitions(plan).isEmpty)
+ }
}
}
}
}
- testGluten("SPARK-53322: checkpointed scans can't shuffle other children on
SPJ") {
+ testGluten("SPARK-53322: checkpointed scans can shuffle other children on
SPJ") {
withTempDir {
dir =>
spark.sparkContext.setCheckpointDir(dir.getPath)
@@ -1985,51 +2023,76 @@ class GlutenKeyGroupedPartitioningSuite
.orderBy("id", "purchase_price", "sale_price")
checkAnswer(df, Seq(Row(1, "aa", 41.0, 40.0), Row(3, "cc", 15.5,
25.5)))
- // One shuffle for sort and two shuffles for join are expected.
- assert(collectAllShuffles(df.queryExecution.executedPlan).length
=== 3)
+ val plan = df.queryExecution.executedPlan
+ // 1 shuffle for SORT and 1 shuffle for JOIN are expected.
+ assert(collectAllShuffles(plan).length === 2)
+ // The JOIN shuffle has KeyedPartitioning and falls back to
vanilla Spark
+ // (GLUTEN-10992). The SORT shuffle is native.
+ assert(collectVanillaShuffles(plan).length === 1)
+ assert(collectAllColumnarShuffles(plan).length === 1)
+ // 0 group partitions are expected because both sides of the
join are clustered
+ assert(collectAllGroupPartitions(plan).isEmpty)
}
}
}
}
- testGluten("SPARK-53322: checkpointed scans can be shuffled by children on
SPJ") {
+ testGluten("SPARK-55848: checkpointed partially-clustered join with dedup") {
withTempDir {
dir =>
spark.sparkContext.setCheckpointDir(dir.getPath)
- val itemsPartitions = Array(identity("id"))
- createTable(items, itemsColumns, itemsPartitions)
+ val items_partitions = Array(identity("id"))
+ createTable(items, itemsColumns, items_partitions)
sql(
s"INSERT INTO testcat.ns.$items VALUES " +
- "(1, 'aa', 41.0, cast('2020-01-01' as timestamp)), " +
- "(2, 'bb', 10.0, cast('2020-01-02' as timestamp)), " +
- "(3, 'cc', 15.5, cast('2020-01-03' as timestamp))")
+ "(1, 'aa', 40.0, cast('2020-01-01' as timestamp)), " +
+ "(1, 'aa', 41.0, cast('2020-01-15' as timestamp)), " +
+ "(2, 'bb', 10.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 'cc', 15.5, cast('2020-02-01' as timestamp))")
- createTable(purchases, purchasesColumns, Array(identity("item_id")))
+ val purchases_partitions = Array(identity("item_id"))
+ createTable(purchases, purchasesColumns, purchases_partitions)
sql(
s"INSERT INTO testcat.ns.$purchases VALUES " +
- "(1, 40.0, cast('2020-01-01' as timestamp)), " +
- "(3, 25.5, cast('2020-01-03' as timestamp)), " +
- "(4, 20.0, cast('2020-01-04' as timestamp))")
+ "(1, 42.0, cast('2020-01-01' as timestamp)), " +
+ "(1, 50.0, cast('2020-01-02' as timestamp)), " +
+ "(2, 11.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 19.5, cast('2020-02-01' as timestamp))")
withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
- SQLConf.V2_BUCKETING_SHUFFLE_ENABLED.key -> "true",
- SQLConf.V2_BUCKETING_PUSH_PART_VALUES_ENABLED.key -> "true"
+ SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED.key ->
true.toString
) {
- val scanDF1 =
spark.read.table(s"testcat.ns.$items").checkpoint().as("i")
- val scanDF2 = spark.read.table(s"testcat.ns.$purchases").as("p")
-
- val df = scanDF1
- .join(scanDF2, col("id") === col("item_id"))
- .selectExpr("id", "name", "i.price AS purchase_price", "p.price AS
sale_price")
- .orderBy("id", "purchase_price", "sale_price")
- checkAnswer(df, Seq(Row(1, "aa", 41.0, 40.0), Row(3, "cc", 15.5,
25.5)))
-
- // One shuffle for sort and one vanilla shuffle for one side of join
are expected.
- val plan = df.queryExecution.executedPlan
- assert(collectAllShuffles(plan).length === 1)
- // KeyGroupedPartitioning shuffle falls back to vanilla
ShuffleExchangeExec
- assert(collectVanillaShuffles(plan).length === 1)
+ // Checkpoint the JOIN result (not the scan) so the checkpoint node
carries the
+ // partially-clustered KeyGroupedPartitioning. The dedup on top must
still insert
+ // the required grouping operator because partially-clustered
partitioning does not
+ // satisfy ClusteredDistribution.
+ val joinedDf = sql(s"""${selectWithMergeJoinHint("i", "p")} i.id,
i.name, i.price
+ |FROM testcat.ns.$items i
+ |JOIN testcat.ns.$purchases p ON i.id =
p.item_id""".stripMargin)
+ val checkpointedDf = joinedDf.checkpoint()
+ val df = checkpointedDf.select("id").distinct()
+ checkAnswer(df, Seq(Row(1), Row(2), Row(3)))
+
+ // Gluten offloads the checkpoint's RDDScanExec to an
RDDScanTransformer, which keeps
+ // the output partitioning of the RDDScanExec.
+ val checkpointScans = collect(df.queryExecution.executedPlan) {
+ case r: RDDScanExec => r
+ case r: RDDScanTransformer => r
+ }
+ assert(
+ checkpointScans.exists(_.outputPartitioning match {
+ case kp: KeyedPartitioning => !kp.isGrouped
+ case _ => false
+ }),
+ "checkpoint scan should have ungrouped KeyedPartitioning"
+ )
+
+ val allGroupPartitions =
collectAllGroupPartitions(df.queryExecution.executedPlan)
+ assert(
+ allGroupPartitions.size === 1,
+ "expected 1 GroupPartitionsExec above the checkpointed join for
dedup")
}
}
}
@@ -2085,4 +2148,106 @@ class GlutenKeyGroupedPartitioningSuite
checkAnswer(df, Seq(Row(1, "aa", 40.0, 42.0)))
}
}
+
+ private def assertNoSortBeforeSortMergeJoin(plan: SparkPlan): Unit = {
+ val smjs = collect(plan) { case j: SortMergeJoinExecTransformer => j }
+ assert(smjs.nonEmpty, "expected SortMergeJoinExecTransformer in plan")
+ smjs.foreach {
+ smj =>
+ val sorts = smj.children.flatMap {
+ child =>
+ collect(child) {
+ case s: SortExec => s
+ case s: SortExecTransformer => s
+ }
+ }
+ assert(sorts.isEmpty, s"should not add a sort before SMJ, but found:
$sorts")
+ }
+ }
+
+ testGluten(
+ "SPARK-56241: GroupPartitionsExec non-coalescing passes through child
ordering, " +
+ "no pre-join SortExec needed before SortMergeJoin") {
+ // Non-identical key sets force GroupPartitionsExec to be inserted on both
sides align them,
+ // but each group has exactly one partition — no coalescing.
+ val items_partitions = Array(identity("id"))
+ createTable(items, itemsColumns, items_partitions)
+ sql(
+ s"INSERT INTO testcat.ns.$items VALUES " +
+ "(1, 'aa', 10.0, cast('2021-01-01' as timestamp)), " +
+ "(2, 'bb', 20.0, cast('2021-01-01' as timestamp)), " +
+ "(3, 'cc', 30.0, cast('2021-01-01' as timestamp))")
+
+ val purchases_partitions = Array(identity("item_id"))
+ createTable(purchases, purchasesColumns, purchases_partitions)
+ sql(
+ s"INSERT INTO testcat.ns.$purchases VALUES " +
+ "(1, 100.0, cast('2021-01-01' as timestamp)), " +
+ "(2, 200.0, cast('2021-01-01' as timestamp))")
+
+ // GroupPartitionsExec passes through the child's key-derived
outputOrdering.
+ // EnsureRequirements checks outputOrdering directly so no sort should be
inserted before
+ // the SMJ.
+ withSQLConf(SQLConf.V2_BUCKETING_PARTITION_KEY_ORDERING_ENABLED.key ->
"true") {
+ val df = sql(s"""
+ |${selectWithMergeJoinHint("i", "p")}
+ |i.id, i.name
+ |FROM testcat.ns.$items i JOIN testcat.ns.$purchases p
ON p.item_id = i.id
+ |""".stripMargin)
+
+ checkAnswer(df, Seq(Row(1, "aa"), Row(2, "bb")))
+
+ val plan = df.queryExecution.executedPlan
+ val groupPartitions = collectGroupPartitions(plan)
+ assert(groupPartitions.nonEmpty, "expected GroupPartitionsExec in plan")
+ assert(
+ groupPartitions.forall(_.groupedPartitions.forall(_._2.size <= 1)),
+ "expected non-coalescing GroupPartitionsExec")
+ assertNoSortBeforeSortMergeJoin(plan)
+ }
+ }
+
+ testGluten(
+ "SPARK-56241: GroupPartitionsExec coalescing derives ordering from key
expressions, " +
+ "no pre-join SortExec needed before SortMergeJoin") {
+ // Duplicate key 1 on both sides causes coalescing.
+ val items_partitions = Array(identity("id"))
+ createTable(items, itemsColumns, items_partitions)
+ sql(
+ s"INSERT INTO testcat.ns.$items VALUES " +
+ "(1, 'aa', 10.0, cast('2021-01-01' as timestamp)), " +
+ "(1, 'ab', 11.0, cast('2021-06-01' as timestamp)), " +
+ "(2, 'bb', 20.0, cast('2021-01-01' as timestamp))")
+
+ val purchases_partitions = Array(identity("item_id"))
+ createTable(purchases, purchasesColumns, purchases_partitions)
+ sql(
+ s"INSERT INTO testcat.ns.$purchases VALUES " +
+ "(1, 100.0, cast('2021-01-01' as timestamp)), " +
+ "(1, 110.0, cast('2021-06-01' as timestamp)), " +
+ "(2, 200.0, cast('2021-01-01' as timestamp))")
+
+ // GroupPartitionsExec derives outputOrdering from the key expressions
after coalescing.
+ // EnsureRequirements checks outputOrdering directly so no sort should be
inserted before
+ // the SMJ.
+ withSQLConf(
+ SQLConf.V2_BUCKETING_PARTITION_KEY_ORDERING_ENABLED.key -> "true",
+ SQLConf.V2_BUCKETING_PRESERVE_KEY_ORDERING_ON_COALESCE_ENABLED.key ->
"true") {
+ val df = sql(s"""
+ |${selectWithMergeJoinHint("i", "p")}
+ |i.id, i.name
+ |FROM testcat.ns.$items i JOIN testcat.ns.$purchases p
ON p.item_id = i.id
+ |""".stripMargin)
+
+ checkAnswer(df, Seq(Row(1, "aa"), Row(1, "aa"), Row(1, "ab"), Row(1,
"ab"), Row(2, "bb")))
+
+ val plan = df.queryExecution.executedPlan
+ val groupPartitions = collectGroupPartitions(plan)
+ assert(groupPartitions.nonEmpty, "expected GroupPartitionsExec in plan")
+ assert(
+ groupPartitions.exists(_.groupedPartitions.exists(_._2.size > 1)),
+ "expected coalescing GroupPartitionsExec")
+ assertNoSortBeforeSortMergeJoin(plan)
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
