You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ul...@apache.org on 2026/08/26 08:34:07 UTC
(spark) branch branch-4.x updated: [SPARK-58988][SQL] Fix PartitioningCollection invariant violation exception when allowKeysSubsetOfPartitionKeys
This is an automated email from the ASF dual-hosted git repository.
ulysses-you pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 1fb16ad8a2d0 [SPARK-58988][SQL] Fix PartitioningCollection invariant violation exception when allowKeysSubsetOfPartitionKeys
1fb16ad8a2d0 is described below
commit 1fb16ad8a2d08b07846c5e480243f2cb7d8d7543
Author: Xiduo You <ul...@gmail.com>
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 <ul...@gmail.com>
Signed-off-by: Xiduo You <ul...@apache.org>
(cherry picked from commit 19f59da405c204f8610d42aafa1cfe8437fd9f82)
Signed-off-by: Xiduo You <ul...@apache.org>
---
.../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 e400d890fa50..426ad8e658fb 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
@@ -4207,6 +4207,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: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org