James Xu created SPARK-58891:
--------------------------------
Summary: 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
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]
