This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun 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 266ee891d92b [SPARK-59905][4.3][SQL] Fix storage-partitioned join
sorting partitions the wrong way for an ORDER BY over a join-key expression
266ee891d92b is described below
commit 266ee891d92b57413c4be486026eb22f3ecbd5f3
Author: Peter Toth <[email protected]>
AuthorDate: Fri Oct 2 08:13:23 2026 -0700
[SPARK-59905][4.3][SQL] Fix storage-partitioned join sorting partitions the
wrong way for an ORDER BY over a join-key expression
### Differences from #59189
This is the branch-4.3 backport of #59189. The description below is the
original one. On this branch:
- The change to `EnsureRequirements` does not cherry-pick cleanly, since
the `OrderedDistribution` arm sits one level deeper here. The code is the same
as on master, re-indented, and its comment is re-wrapped.
- The test is the same as on master. Without the change, it fails on
branch-4.3 the same way as on master.
- Ran `KeyGroupedPartitioningSuite`, `EnsureRequirementsSuite`,
`GroupPartitionsExecSuite`, `ProjectedOrderingAndPartitioningSuite` and
`PlannerSuite` on branch-4.3, 360 tests in all, plus `dev/lint-scala`.
### What changes were proposed in this pull request?
When `EnsureRequirements` serves an `ORDER BY` by putting a keyed child's
partitions in key order, it now compares the key rows position by position.
Sort order `i` reads field `i` of the key row. It used to bind the ordering to
the references of the partition expressions instead. The binding takes the
types the key rows hold (`keyDataTypes`), and an assertion pins the
precondition it relies on next to it.
### Why are the changes needed?
A key row holds the values of the partition expressions. Binding the
ordering to their references is right only when each expression is a bare
column, the only thing a scan reports that an `ORDER BY` can name. A join can
report the other side's join key instead. A one-side shuffle does, and so does
an inner broadcast hash join, whose `expandOutputPartitioning` reports the
streamed side's layout over the build side's join key.
With `spark.sql.sources.v2.bucketing.shuffle.enabled` and
`spark.sql.sources.v2.bucketing.sorting.enabled` on, this query returns `b` as
0 to 6 instead of 6 to 0. `ident(id)` is identity-partitioned and holds 94 to
100, and `plain(b)` holds 0 to 6:
```sql
SELECT p.b FROM ident i JOIN plain p ON i.id = 100 - p.b ORDER BY 100 - p.b
```
The same happens with only `spark.sql.sources.v2.bucketing.sorting.enabled`
when `plain` is broadcast.
1. The join shuffles `plain` onto `ident`'s layout and reports `100 - b` as
`plain`'s partition expression. Its key rows hold 94 to 100.
2. `KeyedPartitioning.keysSatisfy` admits that partitioning for `ORDER BY
100 - p.b`, since its expression is the ordering's. So no range shuffle is
planned, and the global sort is dropped as redundant.
3. `EnsureRequirements` binds `100 - b` to `b` and evaluates it on the key
rows, so it computes `100 - (100 - b)`. The partitions end up in the reverse
order.
Two other join keys fail planning instead:
- For a join on `i.id = p.b + p.c`, `ORDER BY p.b + p.c` binds two
references to a key row with one field. Planning fails with an
`ArrayIndexOutOfBoundsException`.
- For a join on `i.id = p.s.a`, `ORDER BY p.s.a` binds the struct `s` to
the long key and reads a field of it. Planning fails with a
`ClassCastException`.
`keysSatisfy` admits a partitioning here only when its expressions are the
ordering's, position by position
(`OrderedDistribution.areAllClusterKeysMatched`). So reading field `i` for sort
order `i` is exact.
### Does this PR introduce _any_ user-facing change?
Yes. Such a query returns its rows in the right order, and the two other
forms no longer fail planning. They all need
`spark.sql.sources.v2.bucketing.sorting.enabled`, which is off by default. A
broadcast join needs nothing else, and a one-side shuffle also needs
`spark.sql.sources.v2.bucketing.shuffle.enabled`. No range shuffle is added.
### How was this patch tested?
New test in `KeyGroupedPartitioningSuite`, "SPARK-59905: ORDER BY a
join-key expression a join reports as a partition expression". It runs five
queries:
- `ORDER BY 100 - p.b`, ascending and descending;
- `ORDER BY p.s.a`;
- `ORDER BY 100 - p.b, p.c DESC` over a two-key identity side, which needs
a mixed-direction `GroupPartitionsExec`;
- `ORDER BY p.b + p.c`.
Each runs through a one-side shuffle and through a broadcast join, with AQE
on and off. The `p.b + p.c` one runs with AQE off only, since AQE's plan
validation fails on it for SPARK-59901, which #59165 fixes. Each run checks the
order of the rows, the number of shuffles, and whether a `GroupPartitionsExec`
sorts the partitions.
All 18 runs fail on master:
- the `100 - p.b` and two-key queries return their rows in the wrong order;
- the `p.s.a` one fails with the `ClassCastException`;
- the `p.b + p.c` one fails with the `ArrayIndexOutOfBoundsException`.
Also ran the `KeyGroupedPartitioning*` suites, `EnsureRequirementsSuite`,
`GroupPartitionsExecSuite`, `ProjectedOrderingAndPartitioningSuite` and
`PlannerSuite`, 427 tests in all, plus `dev/lint-scala`.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 5.5)
Closes #59203 from peter-toth/SPARK-59905-order-by-expression-key-4.3.
Authored-by: Peter Toth <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../sql/catalyst/plans/physical/partitioning.scala | 10 ++--
.../execution/exchange/EnsureRequirements.scala | 28 ++++++++---
.../connector/KeyGroupedPartitioningSuite.scala | 58 ++++++++++++++++++++++
3 files changed, 83 insertions(+), 13 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 632be23cf160..ccb8a4994ec1 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
@@ -805,11 +805,11 @@ case class KeyedPartitioning(
}
case o @ OrderedDistribution(_) if SQLConf.get.v2BucketingAllowSorting =>
- // `EnsureRequirements` orders the key rows by the attributes the
expressions are over, and
- // nothing makes a reducer order-preserving, so that ordering says
nothing about reduced
- // keys. This is the local gate. A reduced position always carries a
transform and an
- // `ORDER BY` cannot name one, so the match below already refuses
every case a query can
- // reach. Do not read the first clause as redundant.
+ // `EnsureRequirements` compares the key rows position by position
under the required
+ // ordering. Nothing makes a reducer order-preserving, so that
comparison says nothing
+ // about reduced keys. This is the local gate. A reduced position
always carries a
+ // transform and an `ORDER BY` cannot name one, so the match below
already refuses every
+ // case a query can reach. Do not read the first clause as redundant.
// An out-of-set key can break the ascending sequence of the declared
keys, so a marked
// layout keeps a global ordering claim only for a single partition
(see the `@param`).
expressionsDescribeKeys && o.areAllClusterKeysMatched(expressions) &&
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala
index acc39d032a16..7d2d1228b63b 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala
@@ -97,10 +97,20 @@ case class EnsureRequirements(
// OrderedDistribution requires grouped KeyedPartitioning
with sorted keys
// according to the distribution's ordering.
val satisfyingKeyedPartitioning = resolution.fold(identity,
_._1)
- // The single-column invariant in
KeyedPartitioning.supportsExpressions guarantees
- // one attribute per partition expression.
- val attrs =
satisfyingKeyedPartitioning.expressions.flatMap(_.references)
- val keyRowOrdering = RowOrdering.create(o.ordering, attrs)
+ // A key row holds the values of the partition expressions.
`keysSatisfy` admits
+ // a partitioning here only when those expressions are the
ordering's, position
+ // by position, so each sort order reads its position of the
key row. The
+ // expression need not be a bare column. A join can report
the other side's join
+ // key, e.g. `100 - b`.
+
assert(o.areAllClusterKeysMatched(satisfyingKeyedPartitioning.expressions),
+ "the partition expressions must be the ordering's,
position by position")
+ val keyRowOrdering = RowOrdering.create(
+
o.ordering.zip(satisfyingKeyedPartitioning.keyDataTypes).zipWithIndex.map {
+ case ((order, dataType), i) =>
+ order.copy(child = BoundReference(i, dataType,
nullable = true),
+ sameOrderExpressions = Seq.empty)
+ },
+ Nil)
val keyOrdering = keyRowOrdering.on((t:
InternalRowComparableWrapper) => t.row)
val keys = satisfyingKeyedPartitioning.partitionKeys
// An empty zip is vacuously sorted, which is the answer for
a single key.
@@ -440,15 +450,17 @@ case class EnsureRequirements(
.orElse(reorderJoinKeysRecursively(
leftKeys, rightKeys, leftPartitioning, None))
case (Some(KeyedPartitioning(clustering, _, _, _, _)), _) =>
- // The single-column invariant in
KeyedPartitioning.supportsExpressions guarantees one
- // attribute per partition expression.
+ // A scan reports one column per partition expression. A side a join
laid out on another
+ // side's keys can report an expression over several, e.g. `b + c`.
`reorder` then sees
+ // more columns than keys and gives up.
val leafExprs = clustering.flatMap(_.references)
reorder(leftKeys.toIndexedSeq, rightKeys.toIndexedSeq, leafExprs,
leftKeys)
.orElse(reorderJoinKeysRecursively(
leftKeys, rightKeys, None, rightPartitioning))
case (_, Some(KeyedPartitioning(clustering, _, _, _, _))) =>
- // The single-column invariant in
KeyedPartitioning.supportsExpressions guarantees one
- // attribute per partition expression.
+ // A scan reports one column per partition expression. A side a join
laid out on another
+ // side's keys can report an expression over several, e.g. `b + c`.
`reorder` then sees
+ // more columns than keys and gives up.
val leafExprs = clustering.flatMap(_.references)
reorder(leftKeys.toIndexedSeq, rightKeys.toIndexedSeq, leafExprs,
rightKeys)
.orElse(reorderJoinKeysRecursively(
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 e5fbdadbd26a..38372fce5775 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
@@ -4559,6 +4559,64 @@ class KeyGroupedPartitioningSuite extends
DistributionAndOrderingSuiteBase with
}
}
+ test("SPARK-59905: ORDER BY a join-key expression a join reports as a
partition expression") {
+ createTable("ident", Array(Column.create("id", LongType)),
Array(identity("id")))
+ sql("INSERT INTO testcat.ns.ident VALUES " + (94 to 100).map(i =>
s"($i)").mkString(", "))
+ createTable("plain", Array(Column.create("b", LongType),
Column.create("c", LongType),
+ Column.create("s", new StructType().add("a", LongType))), Array.empty)
+ sql("INSERT INTO testcat.ns.plain VALUES " +
+ (0 to 6).map(b => s"($b, 94, named_struct('a', ${94 +
b}L))").mkString(", "))
+ createTable("ident2", Array(Column.create("x", LongType),
Column.create("y", LongType)),
+ Array(identity("x"), identity("y")))
+ sql("INSERT INTO testcat.ns.ident2 VALUES (94, 1), (94, 2), (95, 1)")
+ createTable("plain2", Array(Column.create("b", LongType),
Column.create("c", LongType)),
+ Array.empty)
+ sql("INSERT INTO testcat.ns.plain2 VALUES (6, 1), (6, 2), (5, 1)")
+
+ // The join lays `p` out on `i`'s keys and reports the join key, e.g. `100
- b`, as `p`'s
+ // partition expression. A one-side shuffle does that under the shuffle
conf. An inner
+ // broadcast join does it too, by reporting `i`'s layout over `p`'s join
key. The ORDER BY over
+ // that key is served by the key order, with no range shuffle. The keys
come out ascending, so
+ // only another order needs a `GroupPartitionsExec` to sort the partitions.
+ case class Query(left: String, right: String, on: String, order: String,
select: String,
+ expected: Seq[Row], sortingNodes: Int)
+ val ids = (94 to 100).map(i => Row(i.toLong))
+ val singleColumnKeys = Seq(
+ Query("ident", "plain", "i.id = 100 - p.b", "100 - p.b", "p.b",
+ (6 to 0 by -1).map(b => Row(b.toLong)), 0),
+ Query("ident", "plain", "i.id = 100 - p.b", "100 - p.b DESC", "p.b",
+ (0 to 6).map(b => Row(b.toLong)), 1),
+ Query("ident", "plain", "i.id = p.s.a", "p.s.a", "i.id", ids, 0),
+ Query("ident2", "plain2", "i.x = 100 - p.b AND i.y = p.c", "100 - p.b,
p.c DESC", "p.b, p.c",
+ Seq(Row(6L, 2L), Row(6L, 1L), Row(5L, 1L)), 1))
+ // AQE's plan validation fails on `b + c` until SPARK-59901 is fixed, so
it runs with AQE off.
+ val twoColumnKey =
+ Query("ident", "plain", "i.id = p.b + p.c", "p.b + p.c", "i.id", ids, 0)
+
+ for {
+ (query, aqe) <-
+ singleColumnKeys.flatMap(q => Seq(q -> true, q -> false)) :+
(twoColumnKey -> false)
+ broadcast <- Seq(false, true)
+ } {
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> aqe.toString,
+ SQLConf.V2_BUCKETING_SHUFFLE_ENABLED.key -> (!broadcast).toString,
+ SQLConf.V2_BUCKETING_SORTING_ENABLED.key -> "true") {
+ val hint = if (broadcast) "/*+ BROADCAST(p) */ " else ""
+ val df = sql(
+ s"""SELECT $hint${query.select} FROM testcat.ns.${query.left} i
+ |JOIN testcat.ns.${query.right} p ON ${query.on} ORDER BY
${query.order}
+ |""".stripMargin)
+ val clue = s"${query.order}, aqe = $aqe, broadcast = $broadcast"
+ checkAnswer(df, query.expected)
+ val plan = df.queryExecution.executedPlan
+ // A one-side shuffle is the only shuffle, and a broadcast join needs
none.
+ assert(collectAllShuffles(plan).size == (if (broadcast) 0 else 1),
clue)
+ assert(collectAllGroupPartitions(plan).size == query.sortingNodes,
clue)
+ }
+ }
+ }
+
test("SPARK-55992: GroupPartitions string in simple and extended explain") {
val items_partitions = Array(bucket(4, "id"), years("arrive_time"))
createTable(items, itemsColumns, items_partitions)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
