This is an automated email from the ASF dual-hosted git repository.
peter-toth 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 b239ef59fdde [SPARK-59948][SQL] Strip aliases from a DSv2 reported
ordering on a nested field
b239ef59fdde is described below
commit b239ef59fddec76af5ecb55c09841544f7ecf8f6
Author: Yahya Kisana <[email protected]>
AuthorDate: Fri Oct 9 12:07:59 2026 +0200
[SPARK-59948][SQL] Strip aliases from a DSv2 reported ordering on a nested
field
### What changes were proposed in this pull request?
`V2ScanPartitioningAndOrdering` converts a scan's `SupportsReportOrdering`
ordering into `SortOrder`s on every application of the rule. For a nested sort
key such as `s.x`, `LogicalPlan.resolve` wraps the result in a new
`Alias(GetStructField(...))` with a fresh `ExprId` each time, so every
application produces a different plan.
This PR strips aliases from the converted sort orders in the ordering pass,
using `trimAliases` from `AliasHelper`. Repeated applications now produce equal
plans, and the reported ordering is expressed directly on `s.x`.
### Why are the changes needed?
1. Breaks the idempotence check of the `Once` batch "Early Filter and
Projection Push-Down" in tests:
`Once strategy's idempotence is broken for batch Early Filter and
Projection Push-Down`
2. Never semantically equals a required `SortOrder` on `s.x`, so Spark adds
a redundant `SortExec` even though the scan already reports that ordering.
3. Makes `PlanMerger` treat two derivations of the same nested ordering as
different, so scan merging can be declined.
### Does this PR introduce _any_ user-facing change?
No. Query results are unchanged. For DSv2 sources that report an ordering
on a nested field:
- When the scan reads the whole struct, the physical plan no longer
contains a redundant `SortExec`. Scans where nested schema pruning narrows the
struct still get the `SortExec`; that gap is tracked in SPARK-60011.
- Scans of such a source can now be merged by `PlanMerger`.
### How was this patch tested?
Added a test to `KeyGroupedPartitioningSuite`: a table that reports an
ordering on `s.x`, queried with `SORT BY s.x`, has no `SortExec` in its plan
and returns the correct answer. Without the fix, the test fails with the `Once`
batch idempotence error.
Added a test to `MergeSubplansSuite`: two scalar subqueries over a table
that reports an ordering on `s.x` are merged into a single scan. Without the
fix, the two scans are not merged.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5.5
Claude Opus 5.5 was used for issue and code review, created the tests, and
comments/PR wording. The fix was written by a human; the AI-created tests were
manually run and reviewed by a human.
peter-toth found a gap, commented below. Will address in a separate PR,
tracked here: https://issues.apache.org/jira/browse/SPARK-60011
Closes #59199 from ykisana/SPARK-59948.
Authored-by: Yahya Kisana <[email protected]>
Signed-off-by: Peter Toth <[email protected]>
(cherry picked from commit c5f15d12c954c37bd711b0ffddb198439d35d523)
Signed-off-by: Peter Toth <[email protected]>
---
.../datasources/v2/V2ScanPartitioningAndOrdering.scala | 9 +++++++--
.../sql/connector/KeyGroupedPartitioningSuite.scala | 13 +++++++++++++
.../sql/execution/planmerging/MergeSubplansSuite.scala | 16 ++++++++++++++++
3 files changed, 36 insertions(+), 2 deletions(-)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanPartitioningAndOrdering.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanPartitioningAndOrdering.scala
index d001b20b5d36..a3342dc83bd7 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanPartitioningAndOrdering.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanPartitioningAndOrdering.scala
@@ -19,7 +19,7 @@ package org.apache.spark.sql.execution.datasources.v2
import org.apache.spark.internal.Logging
import org.apache.spark.internal.LogKeys.{CLASS_NAME, COLUMN_NAMES,
RELATION_NAME}
import org.apache.spark.sql.AnalysisException
-import org.apache.spark.sql.catalyst.expressions.V2ExpressionUtils
+import org.apache.spark.sql.catalyst.expressions.{AliasHelper,
V2ExpressionUtils}
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
import org.apache.spark.sql.catalyst.rules.Rule
import
org.apache.spark.sql.catalyst.trees.TreePattern.DATA_SOURCE_V2_SCAN_RELATION
@@ -35,7 +35,7 @@ import org.apache.spark.util.collection.Utils.sequenceToOption
* and ordering reported by data sources to their catalyst counterparts. Then,
annotates the plan
* with the partitioning and ordering result.
*/
-object V2ScanPartitioningAndOrdering extends Rule[LogicalPlan] with Logging {
+object V2ScanPartitioningAndOrdering extends Rule[LogicalPlan] with
AliasHelper with Logging {
override def apply(plan: LogicalPlan): LogicalPlan = {
val scanRules = Seq[LogicalPlan => LogicalPlan] (partitioning, ordering)
@@ -104,8 +104,13 @@ object V2ScanPartitioningAndOrdering extends
Rule[LogicalPlan] with Logging {
// scan output: truncating it here would also drop the sort orders on
a partition key past a
// pruned column, which still hold.
`DataSourceV2ScanRelation.doCanonicalize` and
// `DataSourceV2ScanExecBase.outputOrdering` restrict it to the scan
output instead.
+ // A nested sort key such as `s.x` resolves to a new
`Alias(GetStructField(...))` on every
+ // run of this rule. Strip the alias so repeated runs give equal plans
and, when the scan
+ // reads the whole struct, the ordering can satisfy a required
ordering on the same field
+ // (see SPARK-60011).
val ordering =
V2ExpressionUtils.toCatalystOrdering(reportedOrdering, relation,
relation.funCatalog)
+ .map(o => o.copy(child = trimAliases(o.child)))
d.copy(ordering = Some(ordering))
}
}
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 5b333989c136..42453b9ed6fb 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
@@ -4095,6 +4095,19 @@ class KeyGroupedPartitioningSuite
}
}
+ test("SPARK-59948: a scan's reported ordering on a nested field") {
+ val nestedItems = "nested_items"
+ createTable(nestedItems, Array(Column.create("id", LongType),
+ Column.create("s", new StructType().add("x", LongType))), Array.empty,
+ ordering = Array(sort(column("s.x"), SortDirection.ASCENDING,
NullOrdering.NULLS_FIRST)))
+ sql(s"INSERT INTO testcat.ns.$nestedItems VALUES " +
+ "(1, named_struct('x', 1L)), (2, named_struct('x', 2L))")
+
+ val df = sql(s"SELECT * FROM testcat.ns.$nestedItems SORT BY s.x")
+ assert(collect(df.queryExecution.executedPlan) { case s: SortExec => s
}.isEmpty)
+ checkAnswer(df, Seq(Row(1L, Row(1L)), Row(2L, Row(2L))))
+ }
+
test("SPARK-47094: SPJ: Support compatible buckets") {
val table1 = "tab1e1"
val table2 = "table2"
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/MergeSubplansSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/MergeSubplansSuite.scala
index 0f94ab43259c..48d1c9afc6ea 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/MergeSubplansSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/MergeSubplansSuite.scala
@@ -2977,6 +2977,22 @@ class MergeSubplansSuite extends PlanTest {
s"the merged scan should re-derive the reported ordering; got
${scans.head.ordering}")
}
+ test("SPARK-59948: merge DSv2 scans reporting an ordering on a nested
field") {
+ // Each scan derives its `s.x ASC` report separately. The nested key
resolves to a fresh
+ // `Alias(GetStructField(...))` per derivation, so unless the alias is
stripped the two
+ // reports never compare equal and the merge is declined.
+ val table = new TestV2Table(StructType(Seq(
+ StructField("s", StructType(Seq(StructField("x", IntegerType)))),
+ StructField("b", IntegerType), StructField("c", StringType))),
+ reportedOrderingCols = Seq("s.x"))
+ val q = testRelation.select(
+ ScalarSubquery(v2ScanReportingOn(table, Seq("s",
"b")).groupBy()(sum($"b").as("sum_b"))),
+ ScalarSubquery(v2ScanReportingOn(table, Seq("s",
"c")).groupBy()(sum($"c").as("sum_c"))))
+ val optimized = Optimize.execute(q.analyze)
+
+ assert(v2Scans(optimized).length == 1, s"the two scans should be fused
into one:\n$optimized")
+ }
+
test("SPARK-58549: do not merge when the rebuilt scan re-derives less than
the combined " +
"ordering") {
// One input reports `a ASC`, the other `a ASC, c ASC`, so the combined
requirement is the
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
