This is an automated email from the ASF dual-hosted git repository.
zml1206 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 75d6d1f0d0 [VL] Fix SPARK-53322 in GlutenKeyGroupedPartitioningSuite
for Spark 4.1 (#12469)
75d6d1f0d0 is described below
commit 75d6d1f0d0265535a6aafde544400649dbe7e9f7
Author: Mingliang Zhu <[email protected]>
AuthorDate: Thu Jul 9 13:32:39 2026 +0800
[VL] Fix SPARK-53322 in GlutenKeyGroupedPartitioningSuite for Spark 4.1
(#12469)
---
.../gluten/utils/velox/VeloxTestSettings.scala | 2 +-
.../GlutenKeyGroupedPartitioningSuite.scala | 160 +++++++++++++++++++++
2 files changed, 161 insertions(+), 1 deletion(-)
diff --git
a/gluten-ut/spark41/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
b/gluten-ut/spark41/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
index 045df1600d..0d373793ad 100644
---
a/gluten-ut/spark41/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
+++
b/gluten-ut/spark41/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
@@ -80,12 +80,12 @@ class VeloxTestSettings extends BackendTestSettings {
.excludeByPrefix("SPARK-48012")
.excludeByPrefix("SPARK-44647")
.excludeByPrefix("SPARK-41471")
+ .excludeByPrefix("SPARK-53322")
// 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")
// TODO: fix on Spark-4.1
- .excludeByPrefix("SPARK-53322") // see
https://github.com/apache/spark/pull/53132
.excludeByPrefix("SPARK-54439") // see
https://github.com/apache/spark/pull/53142
enableSuite[GlutenLocalScanSuite]
enableSuite[GlutenMetadataColumnSuite]
diff --git
a/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/connector/GlutenKeyGroupedPartitioningSuite.scala
b/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/connector/GlutenKeyGroupedPartitioningSuite.scala
index 5f6793e0e8..58941a3b49 100644
---
a/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/connector/GlutenKeyGroupedPartitioningSuite.scala
+++
b/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/connector/GlutenKeyGroupedPartitioningSuite.scala
@@ -30,6 +30,7 @@ import
org.apache.spark.sql.execution.{ColumnarShuffleExchangeExec, SparkPlan}
import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
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}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types._
@@ -103,6 +104,10 @@ class GlutenKeyGroupedPartitioningSuite
collect(plan) { case s: ColumnarShuffleExchangeExec => s }
}
+ private def collectVanillaShuffles(plan: SparkPlan):
Seq[ShuffleExchangeExec] = {
+ collect(plan) { case s: ShuffleExchangeExec => s }
+ }
+
private def collectScans(plan: SparkPlan): Seq[BatchScanExec] = {
collect(plan) { case s: BatchScanExec => s }
}
@@ -1874,4 +1879,159 @@ class GlutenKeyGroupedPartitioningSuite
}
}
+ testGluten("SPARK-53322: checkpointed scans avoid shuffles for aggregates") {
+ withTempDir {
+ dir =>
+ spark.sparkContext.setCheckpointDir(dir.getPath)
+ val itemsPartitions = Array(identity("id"))
+ createTable(items, itemsColumns, itemsPartitions)
+ sql(
+ s"INSERT INTO testcat.ns.$items VALUES " +
+ "(1, 'aa', 40.0, cast('2020-01-01' as timestamp)), " +
+ "(1, 'aa', 41.0, cast('2020-01-02' as timestamp)), " +
+ "(2, 'bb', 10.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 'cc', 15.5, cast('2020-02-01' as timestamp))")
+
+ val scanDF = spark.read.table(s"testcat.ns.$items").checkpoint()
+ val df = scanDF.groupBy("id").agg(max("price").as("res")).select("res")
+ checkAnswer(df.sort("res"), Seq(Row(10.0), Row(15.5), Row(41.0)))
+
+ val shuffles = collectAllShuffles(df.queryExecution.executedPlan)
+ assert(
+ shuffles.isEmpty,
+ "should not contain shuffle when not grouping by partition values")
+ }
+ }
+
+ testGluten("SPARK-53322: checkpointed scans aren't used for SPJ") {
+ withTempDir {
+ dir =>
+ spark.sparkContext.setCheckpointDir(dir.getPath)
+ val itemsPartitions = Array(identity("id"))
+ createTable(items, itemsColumns, itemsPartitions)
+ 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))")
+
+ val purchasePartitions = Array(identity("item_id"))
+ createTable(purchases, purchasesColumns, purchasePartitions)
+ 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))")
+
+ for {
+ pushdownValues <- Seq(true, false)
+ checkpointBothScans <- Seq(true, false)
+ } {
+ withSQLConf(
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.V2_BUCKETING_PUSH_PART_VALUES_ENABLED.key ->
pushdownValues.toString) {
+ val scanDF1 =
spark.read.table(s"testcat.ns.$items").checkpoint().as("i")
+ val scanDF2 = if (checkpointBothScans) {
+ spark.read.table(s"testcat.ns.$purchases").checkpoint().as("p")
+ } else {
+ 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 two shuffles for join are expected.
+ assert(collectAllShuffles(df.queryExecution.executedPlan).length
=== 3)
+ }
+ }
+ }
+ }
+
+ testGluten("SPARK-53322: checkpointed scans can't shuffle other children on
SPJ") {
+ withTempDir {
+ dir =>
+ spark.sparkContext.setCheckpointDir(dir.getPath)
+ val itemsPartitions = Array(identity("id"))
+ createTable(items, itemsColumns, itemsPartitions)
+ 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))")
+
+ createTable(purchases, purchasesColumns, Array.empty)
+ 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))")
+
+ Seq(true, false).foreach {
+ pushdownValues =>
+ withSQLConf(
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.V2_BUCKETING_SHUFFLE_ENABLED.key -> "true",
+ SQLConf.V2_BUCKETING_PUSH_PART_VALUES_ENABLED.key ->
pushdownValues.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 two shuffles for join are expected.
+ assert(collectAllShuffles(df.queryExecution.executedPlan).length
=== 3)
+ }
+ }
+ }
+ }
+
+ testGluten("SPARK-53322: checkpointed scans can be shuffled by children on
SPJ") {
+ withTempDir {
+ dir =>
+ spark.sparkContext.setCheckpointDir(dir.getPath)
+ val itemsPartitions = Array(identity("id"))
+ createTable(items, itemsColumns, itemsPartitions)
+ 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))")
+
+ createTable(purchases, purchasesColumns, Array(identity("item_id")))
+ 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))")
+
+ withSQLConf(
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.V2_BUCKETING_SHUFFLE_ENABLED.key -> "true",
+ SQLConf.V2_BUCKETING_PUSH_PART_VALUES_ENABLED.key -> "true"
+ ) {
+ 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)
+ }
+ }
+ }
+
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
