From 7f3bb6d20eb20af7bb62014108731459160d72fd Mon Sep 17 00:00:00 2001 From: Naveen Kumar Puppala Date: Mon, 20 Jul 2026 22:16:48 -0700 Subject: [PATCH 1/3] [SPARK-52246][SQL][TESTS] Add bucket transform regression test for one-side shuffle with join key tail of partition keys ### What changes were proposed in this pull request? Add a regression test to KeyGroupedPartitioningSuite for the scenario reported in SPARK-52246 (test adapted from the reproduction in the JIRA description): a storage-partitioned join where one side is auto-shuffled, the partitioned side uses a bucket transform, and the partition key is the tail of the join keys. ### Why are the changes needed? SPARK-52246 reported silent incorrect results (empty join output) for this scenario on 4.0.0/4.0.1. The defect was fixed by SPARK-54439, but the tests added there cover identity and years transforms only, not bucket. This locks in coverage for the bucket case. ### Does this PR introduce any user-facing change? No. ### How was this patch tested? New test passes on master; verified it fails on v4.0.1 (pre-SPARK-54439) reproducing the original wrong results. ### Was this patch authored or co-authored using generative AI tooling? Yes. --- .../KeyGroupedPartitioningSuite.scala | 31 +++++++++++++++++++ 1 file changed, 31 insertions(+) diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala index e765b86301892..9a64e16733c38 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala @@ -2724,6 +2724,37 @@ class KeyGroupedPartitioningSuite extends DistributionAndOrderingSuiteBase with } } + test("SPARK-52246: one-side shuffle with join key tail part of the partition keys") { + val items_partitions = Array(bucket(2, "id")) + createTable(items, itemsColumns, items_partitions) + + sql(s"INSERT INTO testcat.ns.$items VALUES " + + "(1, 'aa', 40.0, cast('2020-01-01' as timestamp)), " + + "(1, 'aa', 30.0, cast('2020-01-02' 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)), " + + "(1, 89.0, cast('2020-01-03' as timestamp)), " + + "(3, 19.5, cast('2020-02-01' as timestamp)), " + + "(5, 26.0, cast('2023-01-01' as timestamp)), " + + "(6, 50.0, cast('2023-02-01' as timestamp))") + + withSQLConf( + SQLConf.REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION.key -> "false", + SQLConf.V2_BUCKETING_SHUFFLE_ENABLED.key -> "true", + SQLConf.V2_BUCKETING_PUSH_PART_VALUES_ENABLED.key -> "true", + SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED.key -> "false", + SQLConf.V2_BUCKETING_ALLOW_KEYS_SUBSET_OF_PARTITION_KEYS.key -> "true") { + val df = createJoinTestDF(Seq("arrive_time" -> "time", "id" -> "item_id")) + val shuffles = collectShuffles(df.queryExecution.executedPlan) + assert(shuffles.size == 1, "SPJ should be triggered") + checkAnswer(df, Seq(Row(1, "aa", 40.0, 42.0))) + } + } + test("SPARK-48949: test partition filters inner join") { val items_partitions = Array(bucket(8, "id"), days("arrive_time")) createTable(items, itemsColumns, items_partitions) From 4a3e8fc0ab2df57f63423f5088e6cb357d8730ab Mon Sep 17 00:00:00 2001 From: Naveen Kumar Puppala Date: Fri, 24 Jul 2026 13:27:15 -0700 Subject: [PATCH 2/3] Keep only V2_BUCKETING_SHUFFLE_ENABLED, matching the SPARK-54439 tests --- .../spark/sql/connector/KeyGroupedPartitioningSuite.scala | 7 +------ 1 file changed, 1 insertion(+), 6 deletions(-) diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala index 9a64e16733c38..ceaac9729ddd1 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala @@ -2742,12 +2742,7 @@ class KeyGroupedPartitioningSuite extends DistributionAndOrderingSuiteBase with "(5, 26.0, cast('2023-01-01' as timestamp)), " + "(6, 50.0, cast('2023-02-01' as timestamp))") - withSQLConf( - SQLConf.REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION.key -> "false", - SQLConf.V2_BUCKETING_SHUFFLE_ENABLED.key -> "true", - SQLConf.V2_BUCKETING_PUSH_PART_VALUES_ENABLED.key -> "true", - SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED.key -> "false", - SQLConf.V2_BUCKETING_ALLOW_KEYS_SUBSET_OF_PARTITION_KEYS.key -> "true") { + withSQLConf(SQLConf.V2_BUCKETING_SHUFFLE_ENABLED.key -> "true") { val df = createJoinTestDF(Seq("arrive_time" -> "time", "id" -> "item_id")) val shuffles = collectShuffles(df.queryExecution.executedPlan) assert(shuffles.size == 1, "SPJ should be triggered") From 91d8ab1bfed6aeed7931a24fe8f5383c5843bdba Mon Sep 17 00:00:00 2001 From: Naveen Kumar Puppala Date: Sat, 25 Jul 2026 09:39:06 -0700 Subject: [PATCH 3/3] Retrigger CI