Copilot commented on code in PR #12469:
URL: https://github.com/apache/gluten/pull/12469#discussion_r3535654060
##########
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")
Review Comment:
The suite is adding new SPARK-53322 regression tests, but VeloxTestSettings
still excludes all tests with the "SPARK-53322" prefix. This means the newly
added tests won’t run for the Velox backend, which contradicts the PR’s stated
intent to fix/validate SPARK-53322 on Spark 4.1. Please remove or narrow this
exclusion so the new coverage is exercised.
##########
gluten-ut/spark41/src/test/scala/org/apache/spark/sql/connector/GlutenKeyGroupedPartitioningSuite.scala:
##########
@@ -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"))
Review Comment:
spark.sparkContext.setCheckpointDir mutates global SparkContext state, but
the directory comes from withTempDir and will be deleted at the end of the
block. This can leak an invalid checkpointDir into later tests and cause
order-dependent failures. Consider using a helper that saves/restores the
previous checkpointDir and manages temp directory lifetime explicitly (create +
delete in finally), or at least restore the previous checkpointDir after each
test.
##########
gluten-ut/spark41/src/test/scala/org/apache/spark/sql/connector/GlutenKeyGroupedPartitioningSuite.scala:
##########
@@ -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)
Review Comment:
This assertion counts only ColumnarShuffleExchangeExec (Gluten shuffles).
The comment above says the plan should have 3 shuffles total (sort + join), so
this should count all ShuffleExchangeLike nodes; otherwise a vanilla
ShuffleExchangeExec could be present and the test would either fail
unexpectedly or miss the intended condition.
##########
gluten-ut/spark41/src/test/scala/org/apache/spark/sql/connector/GlutenKeyGroupedPartitioningSuite.scala:
##########
@@ -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")
Review Comment:
This assertion message is incorrect (the query *is* grouping by the
partition key `id`), and the shuffle check only looks for
ColumnarShuffleExchangeExec. If the plan introduces a vanilla
ShuffleExchangeExec (e.g., due to a fallback), this test would still pass even
though a shuffle occurred. Consider asserting over all ShuffleExchangeLike
nodes and fix the message.
##########
gluten-ut/spark41/src/test/scala/org/apache/spark/sql/connector/GlutenKeyGroupedPartitioningSuite.scala:
##########
@@ -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)
Review Comment:
Same issue as above: this is asserting on ColumnarShuffleExchangeExec only,
but the intent is to assert the total shuffle count (sort + join). Counting all
ShuffleExchangeLike nodes makes the test robust to vanilla vs columnar exchange
implementations.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
