From 911d425092684e79c7e90bae6dadb187079cb4fc Mon Sep 17 00:00:00 2001 From: zml1206 Date: Wed, 8 Jul 2026 11:36:19 +0800 Subject: [PATCH] [VL] Fix SPARK-54439 in GlutenKeyGroupedPartitioningSuite for Spark 4.1 --- .../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 7048695d3ef..9595222c5cf 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 58941a3b497..e5bea385523 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))) + } + } }