xumingming opened a new pull request, #58144:
URL: https://github.com/apache/spark/pull/58144
### What changes were proposed in this pull request?
Turn `TransposeWindow` into a whole-chain reordering rule so that a stack of
adjacent `Window` operators connected by transparent projections is reordered
to insert the fewest possible exchanges. Windows are grouped by the minimal
partition spec their partition spec contains (smaller spec first within a
group), so every member of a group rides the group leader's exchange; the
result needs one exchange per minimal partition spec, which is the provable
lower bound. When a `Filter` pins the top window (an `InferWindowGroupLimit`
rank pattern), only the windows below it are reordered, preserving the strict
`Filter`-over-`Window` shape.
The behavior is gated behind the internal
`spark.sql.optimizer.windowReorder.enabled` config (default `false`), and the
partition-spec subset reasoning honors
`spark.sql.requireAllClusterKeysForDistribution`.
### Why are the changes needed?
A stacked-window query with interleaved partition specs pays one `Exchange`
per distinct spec layer, because a window can only ride an existing exchange
when its spec is a superset of that key, and the adjacent-pair-only
`TransposeWindow` cannot fix an interleaving where a superset window appears
above (not below) its subset. For example, the three specs below all share
the
user key, yet the ASC/DESC interleaving forces one exchange per layer:
```sql
SELECT user_id,
ROW_NUMBER() OVER (PARTITION BY user_id, is_primary ORDER BY
create_ts) AS rn_user,
ROW_NUMBER() OVER (PARTITION BY user_id, tier, is_primary ORDER BY
create_ts) AS rn_user_tier,
ROW_NUMBER() OVER (PARTITION BY user_id, model, priority ORDER BY
create_ts) AS rn_user_model,
ROW_NUMBER() OVER (PARTITION BY user_id, is_primary ORDER BY
create_ts DESC) AS rn_user_desc,
ROW_NUMBER() OVER (PARTITION BY user_id, tier, is_primary ORDER BY
create_ts DESC) AS rn_user_tier_desc,
ROW_NUMBER() OVER (PARTITION BY user_id, model, priority ORDER BY
create_ts DESC) AS rn_user_model_desc,
LAG (create_ts) OVER (PARTITION BY user_id, tier, is_primary ORDER BY
create_ts) AS prev_create_ts
FROM serving_log
```
Reordering collapses repeated and subset-related specs onto shared exchanges
(`HashPartitioning(user_id, is_primary)` rides every `(user_id, tier,
is_primary)` window, so two exchanges suffice). In a production ETL with 13
stacked windows, this cuts window exchanges from 8 to 3 and avoids ~1.7 TiB
of
shuffle per run; a committed microbenchmark measures ~1.3X on the 16M-row
interleaved shape above. See SPARK-58891.
### Does this PR introduce _any_ user-facing change?
No. The reordering is an internal, opt-in optimization (default `false`), so
existing plans are unchanged unless the config is enabled.
### How was this patch tested?
- Extended `TransposeWindowSuite` with stack-reordering cases (minimal-spec
grouping, idempotence, transparent links, pin-top with rank filters,
determinism/reference/empty-spec guards,
`requireAllClusterKeysForDistribution`
interaction).
- Added `WindowStackExchangeSuite` (physical) asserting the resulting
exchange counts, and `WindowReorderBenchmark` with a committed results
snapshot.
- Ran the window optimizer suites, `DataFrameWindowFunctionsSuite`,
`SQLWindowFunctionSuite`, window-frames and `EnsureRequirementsSuite`;
scalastyle clean for both main and test sources.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Pi
--
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]
