You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2026/08/27 00:33:13 UTC
(spark) branch branch-4.x updated: [SPARK-59027][SQL] Share grouped partition key ordering between `createShuffleSpec` and `GroupPartitionsExec`
This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun 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 a1d7b2a96478 [SPARK-59027][SQL] Share grouped partition key ordering between `createShuffleSpec` and `GroupPartitionsExec`
a1d7b2a96478 is described below
commit a1d7b2a9647865fe15a09a79b4443b21208f08a1
Author: Dongjoon Hyun <do...@apache.org>
AuthorDate: Wed Aug 26 17:32:57 2026 -0700
[SPARK-59027][SQL] Share grouped partition key ordering between `createShuffleSpec` and `GroupPartitionsExec`
### What changes were proposed in this pull request?
Extract the ordering of grouped partition keys into a shared helper, `KeyedPartitioning.groupedKeyRowOrdering`, and use it from both `KeyedPartitioning.keyRowOrdering` (used by `toGrouped`, and thus by `createShuffleSpec`'s subset-keys branch) and `GroupPartitionsExec.groupAndSortByKeys`. Add tests pinning that the two orders agree: a unit test comparing them directly, and a LEFT OUTER join test guarding the silent failure mode end to end.
### Why are the changes needed?
With `spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys.enabled`, one join side may be shuffled onto the key order declared by `createShuffleSpec`, while the keyed side's physical layout is produced by `GroupPartitionsExec`. The two sorts must be identical. If they diverged, inner joins would fail loudly at planning time (`ShuffledJoin` wraps both sides' partitionings into a `PartitioningCollection`, whose invariant requires equal partition keys), but join types that expos [...]
Today the two sorts agree only because both call `RowOrdering.createNaturalAscendingOrdering` on the same data types, and the requirement is recorded only in a comment. This fragility was noticed during the review of #58311 (SPARK-59022), which fixed a related but distinct hazard: an order that must follow another being derived independently. This PR is behavior-neutral and makes the contract shared, documented, and tested.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Pass the CIs with the newly added test coverage to ensure this contract.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Fable 5
Closes #58318 from dongjoon-hyun/SPARK-59027.
Authored-by: Dongjoon Hyun <do...@apache.org>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
(cherry picked from commit 8c4c00d9fe864c9e15eae11b313e0357a54fd734)
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
.../sql/catalyst/plans/physical/partitioning.scala | 28 +++++++++++++---
.../datasources/v2/GroupPartitionsExec.scala | 6 ++--
.../connector/KeyGroupedPartitioningSuite.scala | 39 ++++++++++++++++++++++
.../datasources/v2/GroupPartitionsExecSuite.scala | 27 ++++++++++++++-
4 files changed, 92 insertions(+), 8 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 31861559f6ec..12511c1c4224 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
@@ -545,7 +545,7 @@ case class KeyedPartitioning(
@transient lazy val expressionDataTypes: Seq[DataType] = expressions.map(_.dataType)
@transient lazy val keyRowOrdering =
- RowOrdering.createNaturalAscendingOrdering(expressionDataTypes)
+ KeyedPartitioning.groupedKeyRowOrdering(expressionDataTypes)
@transient lazy val keyOrdering = keyRowOrdering.on((t: InternalRowComparableWrapper) => t.row)
@@ -620,10 +620,10 @@ case class KeyedPartitioning(
val joinKeyPositions = result.keyPositions.map(_.nonEmpty).zipWithIndex.filter(_._1).map(_._2)
val projectedExpressions = joinKeyPositions.map(expressions)
val projectedKeys = projectKeys(joinKeyPositions)._2
- // 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.
+ // Sort the distinct projected keys the same way `GroupPartitionsExec` does (both sort with
+ // `KeyedPartitioning.groupedKeyRowOrdering`). 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 =
new KeyedPartitioning(projectedExpressions, projectedKeys, isGrouped = false).toGrouped
result.copy(partitioning = projectedPartitioning, joinKeyPositions = Some(joinKeyPositions))
@@ -668,6 +668,24 @@ object KeyedPartitioning {
}
}
+ /**
+ * The ascending ordering in which grouped partition keys are laid out, for keys of the given
+ * data types.
+ *
+ * This is a shared contract, not a convenience: with `allowKeysSubsetOfPartitionKeys`,
+ * `createShuffleSpec` declares the keyed side's projected keys in this order (via `toGrouped`),
+ * and the other side of the join may be shuffled onto exactly those keys, while the
+ * `GroupPartitionsExec` inserted on the keyed side independently re-groups its partitions with
+ * the same key positions and sorts them with this same ordering
+ * (`GroupPartitionsExec.groupAndSortByKeys`). If the two sorts diverged, inner joins would fail
+ * loudly at planning time -- `ShuffledJoin` wraps both sides' partitionings into a
+ * `PartitioningCollection`, whose invariant requires equal partition keys -- but join types that
+ * expose only one side's partitioning (e.g. LEFT OUTER) run nothing that compares the two
+ * orders, and silently return wrong results.
+ */
+ def groupedKeyRowOrdering(dataTypes: Seq[DataType]): BaseOrdering =
+ RowOrdering.createNaturalAscendingOrdering(dataTypes)
+
/**
* Projects a sequence of partition keys by selecting only the specified positions.
*/
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala
index ccff1f5b3311..6f671bfcb80b 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala
@@ -101,12 +101,14 @@ case class GroupPartitionsExec(
}
/**
- * Groups and sorts partitions by their keys in ascending order.
+ * Groups and sorts partitions by their keys in ascending order. The sort must match
+ * `KeyedPartitioning.toGrouped`, which is why both use
+ * `KeyedPartitioning.groupedKeyRowOrdering` -- see its documentation for the contract.
*/
private def groupAndSortByKeys(
keyMap: Map[InternalRowComparableWrapper, Seq[Int]],
dataTypes: Seq[DataType]) = {
- val keyOrdering = RowOrdering.createNaturalAscendingOrdering(dataTypes)
+ val keyOrdering = KeyedPartitioning.groupedKeyRowOrdering(dataTypes)
keyMap.toSeq.sorted(keyOrdering.on((t: (InternalRowComparableWrapper, _)) => t._1.row))
}
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 64c0d0f2bc90..9ae0e636fef5 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
@@ -4388,6 +4388,45 @@ class KeyGroupedPartitioningSuite extends DistributionAndOrderingSuiteBase with
}
}
+ test("SPARK-59027: v2 bucketed table with subset join keys left-outer joining v1 table") {
+ // Same shape as the SPARK-58988 test above, but LEFT OUTER: `ShuffledJoin` then exposes only
+ // the left side's partitioning, so no `PartitioningCollection` invariant compares the two
+ // sides' declared keys at planning time. If the order declared by `createShuffleSpec` and the
+ // physical layouts of the two sides (`GroupPartitionsExec` on the keyed side, the shuffle
+ // partitioner on the other) ever diverged, this join would silently lose matches instead of
+ // failing planning, so the answer check is the guard here. The unmatched row distinguishes a
+ // legitimate outer-join null from a lost match.
+ 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_t3", cols, partitions)
+ sql("INSERT INTO testcat.ns.iceberg_t3 VALUES " +
+ "(2, 'cc', '2020'), (1, 'aa', '2021'), (3, 'ee', '2022')")
+
+ 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_t3 t0 LEFT 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"),
+ Row(3L, "ee", "2022", null, null)))
+ }
+ }
+ }
+
test("SPARK-57881: storage-partitioned join leverages union output KeyedPartitioning to " +
"avoid shuffle") {
val cols = Array(
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExecSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExecSuite.scala
index cd09fca7c702..be772bc3f28e 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExecSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExecSuite.scala
@@ -20,7 +20,7 @@ package org.apache.spark.sql.execution.datasources.v2
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.{Ascending, Attribute, AttributeReference, SortOrder}
-import org.apache.spark.sql.catalyst.plans.physical.{KeyedPartitioning, Partitioning, PartitioningCollection, UnknownPartitioning}
+import org.apache.spark.sql.catalyst.plans.physical.{ClusteredDistribution, KeyedPartitioning, KeyedShuffleSpec, Partitioning, PartitioningCollection, UnknownPartitioning}
import org.apache.spark.sql.execution.{DummySparkPlan, LeafExecNode, SafeForKWayMerge}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
@@ -255,6 +255,31 @@ class GroupPartitionsExecSuite extends SharedSparkSession {
}
}
+ test("SPARK-59027: createShuffleSpec subset-keys spec orders keys the same as this node's " +
+ "grouping") {
+ // With `allowKeysSubsetOfPartitionKeys`, `EnsureRequirements` may shuffle the other join side
+ // onto the spec's projected keys while this side is re-grouped by a `GroupPartitionsExec`
+ // carrying the spec's `joinKeyPositions`. The two key orders must agree (see
+ // `KeyedPartitioning.groupedKeyRowOrdering`), or the sides are mis-aligned -- a planning-time
+ // `PartitioningCollection` invariant failure for inner joins, silent wrong results for join
+ // types that expose only one side's partitioning.
+ // First-appearance order of the projected keys ([3], [1], [2]) differs from their sorted
+ // order ([1], [2], [3]), so the assertion discriminates the sort each side uses.
+ val partitionKeys = Seq(row(3, 30), row(1, 10), row(2, 20), row(1, 99))
+ val partitioning = KeyedPartitioning(Seq(exprA, exprB), partitionKeys)
+
+ withSQLConf(SQLConf.V2_BUCKETING_ALLOW_KEYS_SUBSET_OF_PARTITION_KEYS.key -> "true") {
+ val spec = partitioning.createShuffleSpec(ClusteredDistribution(Seq(exprA)))
+ .asInstanceOf[KeyedShuffleSpec]
+ assert(spec.joinKeyPositions === Some(Seq(0)))
+
+ val gpe = GroupPartitionsExec(
+ DummySparkPlan(outputPartitioning = partitioning),
+ joinKeyPositions = spec.joinKeyPositions)
+ assert(gpe.groupedPartitions.map(_._1) === spec.partitioning.partitionKeys)
+ }
+ }
+
test("SPARK-56549: tryEnableSortedMerge returns None when no coalescing occurs") {
val partitionKeys = Seq(row(1), row(2), row(3))
val childOrdering = Seq(SortOrder(exprA, Ascending))
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org