[
https://issues.apache.org/jira/browse/SPARK-58891?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated SPARK-58891:
-----------------------------------
Labels: pull-request-available (was: )
> Reorder stacked window operators to minimize shuffle exchanges
> --------------------------------------------------------------
>
> Key: SPARK-58891
> URL: https://issues.apache.org/jira/browse/SPARK-58891
> Project: Spark
> Issue Type: Improvement
> Components: SQL
> Affects Versions: 5.0.0
> Reporter: James Xu
> Priority: Major
> Labels: pull-request-available
>
> Problem:
> When a query applies many window functions over the same rows, Spark emits a
> stack of Window operators whose order follows the select list. Walking the
> stack bottom\-up, each Window whose partition spec is not a superset of the
> partitioning key immediately below it gets its own Exchange. If the partition
> specs are interleaved in the select list, windows that could share a single
> shuffle never become adjacent, so every distinct spec layer pays a full
> exchange.
> Scenario 1: Interleaved, subset\-related partition specs.
> A typical ranking/LAG batch \(e.g. a per\-user ETL\) interleaves three
> partition
> specs over one base key:{code}
> 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
> {code}
> The specifications are three: {{\(user\_id, is\_primary\)}}, {{\(user\_id,
> tier, is\_primary\)}}
> and {{\(user\_id, model, priority\)}}, each appearing in an ASC and a DESC
> layer \(plus a
> LAG on the middle spec\), so the seven are distinct window operators.
> {{\(user\_id, is\_primary\)}} is a subset of {{\(user\_id, tier,
> is\_primary\)}}, so a
> {{HashPartitioning\(user\_id, is\_primary\)}} exchange could satisfy the
> {{ClusteredDistribution}} of the {{\(user\_id, tier, is\_primary\)}} windows.
> But the
> interleaving prevents this riding from ever happening, and every distinct spec
> layer pays one exchange even though only the two incomparable minimal specs,
> {{\(user\_id, is\_primary\)}} and {{\(user\_id, model, priority\)}}, strictly
> need one.
> In a production ETL with 13 stacked windows over two base keys, 7 of the 8
> window exchanges are avoidable: the window\-chain shuffle write is \~3.0 TiB,
> about 60% of the job's 5.0 TiB total, of which \~1.7 TiB comes from the
> avoidable exchanges.
> Scenario 2: Many window layers over the same base key.
> When a wide query accumulates dozens of ranking/cumulative window expressions
> over one grouping key, the number of exchanges scales with the number of
> distinct spec layers \(up to one per layer\) instead of with the number of
> truly
> independent partitionings. Reordering collapses repeated and subset\-related
> specs onto shared exchanges, regardless of stack size.
> Root Cause:
> Window stack order is determined by select\-list / addWindow insertion order.
> EnsureRequirements creates an exchange keyed exactly on a window's partition
> spec, so a window can only "ride" an existing exchange when its spec is a
> superset of that key. The only reordering rule, TransposeWindow, swaps
> adjacent pairs and only when the upper spec is a strict semantic subset of the
> lower one, so it cannot fix an interleaving where a superset window appears
> above \(not below\) its subset. The inefficiency is therefore structural, not
> query\-specific: N interleaved subset\-related specs cost N exchanges instead
> of
> the number of minimal spec classes.
> Solution:
> Reorder a stack of adjacent Window operators so that windows with the same or
> a
> superset partition spec become contiguous and share one exchange. Concretely,
> group the windows by the minimal partition spec their own spec contains, emit
> the minimal spec first in each group, and let every other member ride its
> exchange, so the result needs one exchange per minimal spec \(the provable
> lower
> bound\). The rewrite is a pure logical transform: it only permutes windows and
> preserves the chain's output attributes; it declines to touch nondeterministic
> windows, chains with inter\-window references or an empty partition spec, or a
> top window pinned directly under a rank Filter \(so WindowGroupLimit inference
> still fires\). It is gated behind an internal flag defaulting to off.
> Expected Impact:
> Scenario 1 \(interleaved specs\): total window exchanges drop from 8 to 3 on
> the
> reference production shape \(7 buyer\-side to 2, seller side already at 1 and
> left as a no\-op\). This removes \~1.7 TiB of shuffle per run at current data
> volumes, together with the CPU and spill costs that ride on it \(one window
> stage alone spills \~1.4\-1.7 TB\). Attribute\-dedup \(SPARK\-41805, in\-tree
> since
> 3.4\) alone brings the same shape from 8 to 6 exchanges; this rule is what
> reaches 3.
> Benchmark: a committed microbenchmark over \~16M rows with the interleaved
> 7\-window shape measures baseline 8,575 ms vs optimized 6,466 ms \(\~1.3X,
> per\-row
> 511 ns \-> 385 ns\) on a 4\-core machine with AQE off.
> Scenario 2 \(large chains\): exchange count collapses to the number of minimal
> partition specs regardless of the number of window layers; gains grow with the
> number of repeated subset\-related specs.
> Sort reduction is out of scope here: windows with equal partition specs but
> different ORDER BY still need separate local sorts, and are addressed by a
> separate logical rewrite that folds ASC/DESC pairs onto one sort.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
