This is an automated email from the ASF dual-hosted git repository.
ulysses-you pushed a commit to branch branch-4.3
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.3 by this push:
new bfeaa492738e [SPARK-58988][SQL] Fix PartitioningCollection invariant
violation exception when allowKeysSubsetOfPartitionKeys
bfeaa492738e is described below
commit bfeaa492738e254d03bd97903a7a64c08a5379d1
Author: Xiduo You <[email protected]>
AuthorDate: Wed Aug 26 16:33:39 2026 +0800
[SPARK-58988][SQL] Fix PartitioningCollection invariant violation exception
when allowKeysSubsetOfPartitionKeys
### What changes were proposed in this pull request?
`KeyedPartitioning.createShuffleSpec` now sorts the distinct projected
partition keys (via
`toGrouped`) so they follow the same natural ascending ordering as
`GroupPartitionsExec`.
### Why are the changes needed?
SPARK-56877 added a check in `PartitioningCollection.fromPartitionings`
requiring all
`KeyedPartitioning`s to share equal `partitionKeys`. In a
storage-partitioned join whose join keys
are a subset of the partition keys (e.g. a v2 table partitioned by `[dt,
bucket(16, c1)]` joined
on `c1`), with `spark.sql.sources.v2.bucketing.shuffle.enabled` enabled so
only the non-keyed side
is re-shuffled, the keyed side's projected keys are sorted by
`GroupPartitionsExec` while
`createShuffleSpec` kept them in first-occurrence order. The two sides then
carry the same keys in
different orders and the query fails with:
```
java.lang.IllegalArgumentException: requirement failed: All
KeyedPartitionings in a
PartitioningCollection must have equal partitionKeys
```
### Does this PR introduce _any_ user-facing change?
No by default.
### How was this patch tested?
Added a regression test in `KeyGroupedPartitioningSuite`
(`SPARK-58988: v2 bucketed table with subset join keys joining v1 table`)
that reproduces the
failure and passes with the fix. Also ran `KeyGroupedPartitioningSuite`,
`EnsureRequirementsSuite`,
`GroupPartitionsExecSuite`, and `ProjectedOrderingAndPartitioningSuite`.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code
Closes #58273 from ulysses-you/SPARK-58988.
Authored-by: Xiduo You <[email protected]>
Signed-off-by: Xiduo You <[email protected]>
(cherry picked from commit 19f59da405c204f8610d42aafa1cfe8437fd9f82)
Signed-off-by: Xiduo You <[email protected]>
---
.../sql/catalyst/plans/physical/partitioning.scala | 7 ++--
.../connector/KeyGroupedPartitioningSuite.scala | 39 ++++++++++++++++++++++
2 files changed, 44 insertions(+), 2 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala
index 76e667590a33..7376c0ff4466 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala
@@ -613,9 +613,12 @@ case class KeyedPartitioning(
val joinKeyPositions =
result.keyPositions.map(_.nonEmpty).zipWithIndex.filter(_._1).map(_._2)
val projectedExpressions = joinKeyPositions.map(expressions)
val projectedKeys = projectKeys(joinKeyPositions)._2
- val distinctProjectedKeys = projectedKeys.distinct
+ // Sort the distinct projected keys the same way `GroupPartitionsExec`
does. Otherwise, when
+ // only the keyed side is grouped and the other side is re-shuffled
using this spec, the two
+ // `KeyedPartitioning`s carry the same keys in a different order and
+ // `PartitioningCollection.fromPartitionings` rejects them.
val projectedPartitioning =
- KeyedPartitioning(projectedExpressions, distinctProjectedKeys,
isGrouped = true)
+ new KeyedPartitioning(projectedExpressions, projectedKeys, isGrouped =
false).toGrouped
result.copy(partitioning = projectedPartitioning, joinKeyPositions =
Some(joinKeyPositions))
} else {
result
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 ceaac9729ddd..f5e2b70725bd 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
@@ -4217,6 +4217,45 @@ class KeyGroupedPartitioningSuite extends
DistributionAndOrderingSuiteBase with
}
}
+ test("SPARK-58988: v2 bucketed table with subset join keys joining v1
table") {
+ // The v2 table is partitioned by an extra identity key `dt` plus
`bucket(16, c1)`, while the
+ // join is only on `c1`. allowKeysSubsetOfPartitionKeys lets the operation
key `c1` be a subset
+ // of the partition keys `[dt, bucket(16, c1)]`, so EnsureRequirements
projects the keyed side
+ // to `[bucket(16, c1)]`. v2BucketingShuffleEnabled then re-shuffles only
the v1 side using that
+ // projected KeyedPartitioning. ShuffledJoin wraps the two output
partitionings into a
+ // PartitioningCollection, which requires all KeyedPartitionings to share
equal partitionKeys.
+ // The v2 side's keys are sorted by GroupPartitionsExec, while the keys
re-used for the v1 side
+ // keep their first-occurrence order from createShuffleSpec, so the two
sequences disagree and
+ // the collection construction used to fail.
+ val cols = Array(
+ Column.create("c1", LongType),
+ Column.create("c2", StringType),
+ Column.create("dt", StringType))
+ val partitions = Array(identity("dt"), bucket(16, "c1"))
+
+ createTable("iceberg_t2", cols, partitions)
+ sql("INSERT INTO testcat.ns.iceberg_t2 VALUES (2, 'cc', '2020'), (1, 'aa',
'2021')")
+
+ withTable("t1") {
+ sql("CREATE TABLE t1 (c1 BIGINT, c2 STRING) USING parquet")
+ sql("INSERT INTO t1 VALUES (1, 'aa'), (2, 'cc')")
+
+ withSQLConf(
+ SQLConf.V2_BUCKETING_ALLOW_KEYS_SUBSET_OF_PARTITION_KEYS.key ->
"true",
+ SQLConf.V2_BUCKETING_SHUFFLE_ENABLED.key -> "true",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ val df = sql("SELECT * FROM testcat.ns.iceberg_t2 t0 JOIN t1 ON t0.c1
= t1.c1")
+ val plan = df.queryExecution.executedPlan
+ // Only the v1 side is re-shuffled; the v2 side is regrouped onto the
join key instead.
+ assert(collectShuffles(plan).length == 1)
+ assert(collectGroupPartitions(plan).length == 1)
+ checkAnswer(df, Seq(
+ Row(1L, "aa", "2021", 1L, "aa"),
+ Row(2L, "cc", "2020", 2L, "cc")))
+ }
+ }
+ }
+
test("SPARK-57881: storage-partitioned join leverages union output
KeyedPartitioning to " +
"avoid shuffle") {
val cols = Array(
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
