This is an automated email from the ASF dual-hosted git repository.
zhouyuan 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 fadf00629d [VL] Fix SPARK-54439 in GlutenKeyGroupedPartitioningSuite
for Spark 4.1 (#12482)
fadf00629d is described below
commit fadf00629d66033609736ea6bbd9de695de51691
Author: Mingliang Zhu <[email protected]>
AuthorDate: Fri Jul 10 17:53:18 2026 +0800
[VL] Fix SPARK-54439 in GlutenKeyGroupedPartitioningSuite for Spark 4.1
(#12482)
This PR updates GlutenKeyGroupedPartitioningSuite for Spark 4.1 to handle
the new SPARK-54439 test coverage.
Spark 4.1 added tests for KeyGroupedPartitioning cases where the reported
partitioning key size does not match the join key size. Since Gluten does not
currently support these cases with native columnar shuffle, this PR excludes
the original Spark-side SPARK-54439 tests from the Velox settings and adds
Gluten-aware versions that validate the expected fallback behavior.
---
.../gluten/utils/velox/VeloxTestSettings.scala | 3 +-
.../GlutenKeyGroupedPartitioningSuite.scala | 51 ++++++++++++++++++++++
2 files changed, 52 insertions(+), 2 deletions(-)
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 8eeff48a41..e8a3006f78 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
@@ -82,12 +82,11 @@ class VeloxTestSettings extends BackendTestSettings {
.excludeByPrefix("SPARK-44647")
.excludeByPrefix("SPARK-41471")
.excludeByPrefix("SPARK-53322")
+ .excludeByPrefix("SPARK-54439")
// 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-54439") // see
https://github.com/apache/spark/pull/53142
enableSuite[GlutenLocalScanSuite]
enableSuite[GlutenMetadataColumnSuite]
enableSuite[GlutenSupportsCatalogOptionsSuite]
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 58941a3b49..e5bea38552 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
@@ -2034,4 +2034,55 @@ class GlutenKeyGroupedPartitioningSuite
}
}
+ testGluten("SPARK-54439: KeyGroupedPartitioning and join key size mismatch")
{
+ val items_partitions = Array(identity("id"))
+ createTable(items, itemsColumns, items_partitions)
+
+ sql(s"INSERT INTO testcat.ns.$items VALUES " +
+ "(1, 'aa', 40.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 'bb', 10.0, cast('2020-01-01' as timestamp)), " +
+ "(4, 'cc', 15.5, cast('2020-02-01' as timestamp))")
+
+ createTable(purchases, purchasesColumns, Array.empty)
+ sql(s"INSERT INTO testcat.ns.$purchases VALUES " +
+ "(1, 42.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 19.5, cast('2020-02-01' as timestamp))")
+
+ withSQLConf(SQLConf.V2_BUCKETING_SHUFFLE_ENABLED.key -> "true") {
+ // `time` and `item_id` in the required `ClusteredDistribution` for
`purchases`, but `item` is
+ // storage partitioned only by `id`
+ val df = createJoinTestDF(Seq("arrive_time" -> "time", "id" ->
"item_id"))
+ val shuffles = collectShuffles(df.queryExecution.executedPlan)
+ assert(shuffles.size == 1, "only shuffle one side not report
partitioning")
+
+ checkAnswer(df, Seq(Row(1, "aa", 40.0, 42.0)))
+ }
+ }
+
+ testGluten("SPARK-54439: KeyGroupedPartitioning with transform and join key
size mismatch") {
+ // Do not use `bucket()` in "one side partition" tests as its
implementation in
+ // `InMemoryBaseTable` conflicts with `BucketFunction`
+ val items_partitions = Array(years("arrive_time"))
+ createTable(items, itemsColumns, items_partitions)
+
+ sql(s"INSERT INTO testcat.ns.$items VALUES " +
+ "(1, 'aa', 40.0, cast('2020-01-01' as timestamp)), " +
+ "(1, 'bb', 10.0, cast('2021-01-01' as timestamp)), " +
+ "(4, 'cc', 15.5, cast('2021-02-01' as timestamp))")
+
+ createTable(purchases, purchasesColumns, Array.empty)
+ sql(s"INSERT INTO testcat.ns.$purchases VALUES " +
+ "(1, 42.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 19.5, cast('2021-02-01' as timestamp))")
+
+ withSQLConf(SQLConf.V2_BUCKETING_SHUFFLE_ENABLED.key -> "true") {
+ // `item_id` and `time` in the required `ClusteredDistribution` for
`purchases`, but `item` is
+ // storage partitioned only by `year(arrive_time)`
+ val df = createJoinTestDF(Seq("id" -> "item_id", "arrive_time" ->
"time"))
+ val shuffles = collectShuffles(df.queryExecution.executedPlan)
+ assert(shuffles.size == 1, "only shuffle one side not report
partitioning")
+
+ checkAnswer(df, Seq(Row(1, "aa", 40.0, 42.0)))
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
