This is an automated email from the ASF dual-hosted git repository.
peter-toth pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.0 by this push:
new c1431faf1f8a [SPARK-59905][4.0][SQL] Fix storage-partitioned join
sorting partitions the wrong way for an ORDER BY over a join-key expression
c1431faf1f8a is described below
commit c1431faf1f8acf008790161f78779ec1999b2165
Author: Peter Toth <[email protected]>
AuthorDate: Sun Oct 4 09:35:52 2026 +0200
[SPARK-59905][4.0][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.0 backport of #59189. The description below is the
original one. This branch has different code, so the change and the test are
tailored:
- Master fixes the binding in `EnsureRequirements.resolveChild`, which this
branch does not have. Here the same binding lives in
`EnsureRequirements.ensureOrdering`, which sorts a `KeyGroupedPartitioning`'s
partition values for an `ORDER BY`. It bound the ordering to the leaves of the
partition expressions, cast to `Attribute`.
- The change binds sort order `i` to field `i` of the partition value
there, with the same assertion. The type comes from the sort order's own
expression, since this branch has no `keyDataTypes`.
- Only the broadcast route is affected here. The one-side shuffle route
already returns the right order. With only
`spark.sql.sources.v2.bucketing.sorting.enabled`, `SELECT /*+ BROADCAST(p) */
p.b FROM neg i JOIN plain p ON i.id = -p.b ORDER BY -p.b` returns `b` in the
wrong order, where `neg(id)` is identity-partitioned and holds -6 to 0.
- A key with a literal fails planning here instead of returning the wrong
order. For `i.id = 100 - p.b`, the leaf `100` fails the cast to `Attribute`,
and planning fails with a `ClassCastException`.
- The test is the master one with three changes:
- it adds `ORDER BY -p.b`, ascending and descending, over `neg`, the
shape that returns the wrong order here;
- it drops the `GroupPartitionsExec` check, since this branch has no such
node;
- through a one-side shuffle it expects 2 shuffles instead of 1, since a
range shuffle follows for the `ORDER BY` here.
- Without the change, the broadcast runs of the `-p.b` queries return their
rows in the wrong order on branch-4.0, and those of the `100 - p.b` and two-key
queries fail with the `ClassCastException`. The one-side shuffle runs, and the
`p.s.a` and `p.b + p.c` queries, pass on this branch without the change. They
are there to keep the diff to #59189 small.
- Ran `KeyGroupedPartitioningSuite`, `EnsureRequirementsSuite`,
`ProjectedOrderingAndPartitioningSuite` and `PlannerSuite` on branch-4.0, 189
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 #59206 from peter-toth/SPARK-59905-order-by-expression-key-4.0.
Authored-by: Peter Toth <[email protected]>
Signed-off-by: Peter Toth <[email protected]>
---
.../execution/exchange/EnsureRequirements.scala | 16 ++++--
.../connector/KeyGroupedPartitioningSuite.scala | 60 ++++++++++++++++++++++
2 files changed, 72 insertions(+), 4 deletions(-)
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 1f5e22ff2260..e581b78c066d 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
@@ -317,10 +317,18 @@ case class EnsureRequirements(
(plan.outputPartitioning, distribution) match {
case (p @ KeyGroupedPartitioning(expressions, _, partitionValues, _, _,
_),
d @ OrderedDistribution(ordering)) if p.satisfies(d) =>
- val attrs =
expressions.flatMap(_.collectLeaves()).map(_.asInstanceOf[Attribute])
- val partitionOrdering: Ordering[InternalRow] = {
- RowOrdering.create(ordering, attrs)
- }
+ // A partition value holds the values of the partition expressions.
`p.satisfies(d)` holds
+ // here only when those expressions are the ordering's, position by
position, so each sort
+ // order reads its position of the partition value. The expression
need not be a bare
+ // column. A broadcast join can report the other side's join key, e.g.
`100 - b`.
+ assert(d.areAllClusterKeysMatched(expressions),
+ "the partition expressions must be the ordering's, position by
position")
+ val partitionOrdering: Ordering[InternalRow] = RowOrdering.create(
+ ordering.zipWithIndex.map { case (order, i) =>
+ order.copy(child = BoundReference(i, order.child.dataType,
nullable = true),
+ sameOrderExpressions = Seq.empty)
+ },
+ Nil)
// Sort 'commonPartitionValues' and use this mechanism to ensure
BatchScan's
// output partitions are ordered
val sorted = partitionValues.sorted(partitionOrdering)
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 a44bfbe8ef8d..dafb04923058 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
@@ -430,6 +430,66 @@ class KeyGroupedPartitioningSuite extends
DistributionAndOrderingSuiteBase {
}
}
+ 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("neg", Array(Column.create("id", LongType)),
Array(identity("id")))
+ sql("INSERT INTO testcat.ns.neg VALUES " + (0 to 6).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. Over the broadcast
+ // join's layout, the ORDER BY is served by sorting the partition values,
with no range
+ // shuffle. On this branch a one-side shuffle is followed by a range
shuffle for the ORDER BY.
+ case class Query(left: String, right: String, on: String, order: String,
select: String,
+ expected: Seq[Row])
+ val ids = (94 to 100).map(i => Row(i.toLong))
+ val singleColumnKeys = Seq(
+ Query("neg", "plain", "i.id = -p.b", "-p.b", "p.b", (6 to 0 by -1).map(b
=> Row(b.toLong))),
+ Query("neg", "plain", "i.id = -p.b", "-p.b DESC", "p.b", (0 to 6).map(b
=> Row(b.toLong))),
+ Query("ident", "plain", "i.id = 100 - p.b", "100 - p.b", "p.b",
+ (6 to 0 by -1).map(b => Row(b.toLong))),
+ Query("ident", "plain", "i.id = 100 - p.b", "100 - p.b DESC", "p.b",
+ (0 to 6).map(b => Row(b.toLong))),
+ Query("ident", "plain", "i.id = p.s.a", "p.s.a", "i.id", ids),
+ 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))))
+ // As on master, `b + c` runs with AQE off only.
+ val twoColumnKey = Query("ident", "plain", "i.id = p.b + p.c", "p.b +
p.c", "i.id", ids)
+
+ 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)
+ // A one-side shuffle and the range shuffle after it, and none for a
broadcast join.
+ assert(collectAllShuffles(df.queryExecution.executedPlan).size == (if
(broadcast) 0 else 2),
+ clue)
+ }
+ }
+ }
+
test("SPARK-49179: Fix v2 multi bucketed inner joins throw AssertionError") {
val cols = Array(
Column.create("id", LongType),
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
