This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 8c4c00d9fe86 [SPARK-59027][SQL] Share grouped partition key ordering
between `createShuffleSpec` and `GroupPartitionsExec`
8c4c00d9fe86 is described below
commit 8c4c00d9fe864c9e15eae11b313e0357a54fd734
Author: Dongjoon Hyun <[email protected]>
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 <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../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: [email protected]
For additional commands, e-mail: [email protected]
