ulysses-you opened a new pull request, #57396:
URL: https://github.com/apache/spark/pull/57396
### What changes were proposed in this pull request?
Adds a new physical rule `PushDownLocalSort`, wired into both the AQE
`queryStagePreparationRules` and the non-AQE `QueryExecution` preparations
(right after `EnsureRequirements`, before `CombineAdjacentAggregation`), gated
by a new internal config `spark.sql.execution.pushDownLocalSort` (default
`true`).
`EnsureRequirements` inserts one local `SortExec(global = false)` above
every operator whose `requiredChildOrdering` is not satisfied. When two such
requirements are in a prefix-cover relationship, a stage computes multiple
local sorts that only differ in width. The rule pushes the wider local sort
down through order-preserving operators onto the narrower one below, widening
it so a single sort serves both operators and the redundant upper sort is
dropped.
For example, a sort aggregate stacked on a window over the same clustering
keys:
```
SortAggregate(key = [a, b, c])
Sort([a, b, c], global = false) <- upper, wider
Window([a], [b])
Sort([a, b], global = false) <- lower, narrower
Exchange(hashpartitioning([a]))
```
becomes:
```
SortAggregate(key = [a, b, c])
Window([a], [b]) requiredChildOrdering [a, b] satisfied
by [a, b, c]
Sort([a, b, c], global = false) <- single sort now serves both operators
Exchange(hashpartitioning([a]))
```
Order-preserving operators traversed: `ProjectExec`, `FilterExec`,
`SortAggregateExec`, `WindowExecBase`, `WindowGroupLimitExec`. When one renames
an ordering column in its output (`b AS x`), the ordering is rewritten from the
operator's output space back to its child's space as it is pushed through
(plain renames only). The rule never crosses a shuffle or a
non-order-preserving operator.
`CollectMetricsExec` is intentionally excluded: it observes rows in input
order, so widening a sort under it would silently change order-sensitive
observed metrics (`first`/`last`/`collect_list`).
This is complementary to `RemoveRedundantSorts`, which only removes a sort
already satisfied by its child; it never moves a wider sort down to absorb a
narrower one.
### Why are the changes needed?
Stacked windows, or a sort aggregate over a window with prefix-compatible
ordering requirements, compute several local sorts that a single wider sort
could satisfy. Reusing the widest sort saves the redundant sorting work in the
stage.
### Does this PR introduce _any_ user-facing change?
No. It only removes redundant local sorts from physical plans; query results
are unchanged. Guarded by `spark.sql.execution.pushDownLocalSort`.
### How was this patch tested?
New `PushDownLocalSortSuite` (AQE on/off variants) with:
- positive SQL cases: stacked windows, filter above the sorts, window to
sort aggregate, renaming project;
- negative SQL cases: disjoint orderings, a shuffle between the sorts,
sort-direction mismatch, a project that computes the ordering column;
- plan-level cases: multi-operator chain, renaming-project rewrite, three
stacked sorts in one pass.
Verified the TPCDS/TPCH `PlanStability` golden plans are unchanged.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
