[
https://issues.apache.org/jira/browse/SPARK-59658?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
James Xu updated SPARK-59658:
-----------------------------
Description:
Spark executes a bounded moving window frame by re-aggregating every row
currently inside the frame on every output row: {{SlidingWindowFunctionFrame}}
resets the aggregate buffer and re-folds the whole frame for each row produced.
For a window operator with nested sliding frames over one (PARTITION BY, ORDER
BY) key, the cost is the sum of the frame widths (Σw) aggregate-update
evaluations per output row, and the cost is structural — every added horizon
multiplies the work. There is no incremental path: the window aggregate
processor only supports {{{}initialize{}}}/{{{}update{}}}/{{{}evaluate{}}},
with no {{{}remove{}}}, so a moving frame cannot be advanced row-by-row.
The pattern is common in user-facing metric tables that compute several rolling
horizons per measure over a daily partition ordering. A realistic example:
{code:sql}
SELECT
seller_user_id,
sum(cnt) OVER (PARTITION BY seller_user_id ORDER BY receive_date
ROWS BETWEEN 2 PRECEDING AND CURRENT ROW) AS cnt_3d,
sum(cnt) OVER (PARTITION BY seller_user_id ORDER BY receive_date
ROWS BETWEEN 6 PRECEDING AND CURRENT ROW) AS cnt_7d,
sum(cnt) OVER (PARTITION BY seller_user_id ORDER BY receive_date
ROWS BETWEEN 29 PRECEDING AND CURRENT ROW) AS cnt_30d,
sum(cnt) OVER (PARTITION BY seller_user_id ORDER BY receive_date
ROWS BETWEEN 89 PRECEDING AND CURRENT ROW) AS cnt_90d,
sum(cnt) OVER (PARTITION BY seller_user_id ORDER BY receive_date
ROWS BETWEEN 179 PRECEDING AND CURRENT ROW) AS cnt_180d
-- ... repeated for several more measures
FROM daily_aggregates{code}
This shape measures 5 horizons, i.e. Σw = 3+7+30+90+180 = 310 aggregate-update
evaluations per output row per measure family.
In a production workload measured against this shape (a 3,437 CPU-hour
application), 45% of the CPU sits inside the window operator, and the window
stage's own input-sort spill reaches 14.8 TiB on the largest query.
h3. Root cause
{{SlidingWindowFunctionFrame.write}} re-runs the aggregate update for every row
in the frame on every output row. Growing frames are incremental and offset
frames are O(1), so the sliding frame is the only path with no O(1) evaluation.
The work factor is exactly Σw (sum of frame widths) per group per row.
h3. Proposed change
Add an optimizer rule that rewrites each group of qualifying sliding frames as
one running aggregate plus offset differences:
{code:java}
-- Running total, reused as the base for every sliding window
C = sum(x) OVER (PARTITION BY k ORDER BY d
ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)
-- n-row sliding sum == running total minus the running total n+1 rows back
sum(x) OVER (PARTITION BY k ORDER BY d
ROWS BETWEEN n PRECEDING AND CURRENT ROW)
= C - coalesce(lag(C, n + 1) OVER (PARTITION BY k ORDER BY d), 0){code}
For integral accumulators with ANSI off this is provably bit-identical: the
original fold already computes with wrapping addition, and modular reduction
commutes with subtraction, so the running prefix — a magnitude no bounded frame
can reach — overflows harmlessly. Floating-point accumulators (non-associative
addition) and decimal accumulators (overflow becomes a silent NULL) are
excluded, and a nullable measure carries a running count so an all-NULL frame
still returns NULL. The rewrite reuses the window operator's own partition and
order spec, so no new exchange or sort is introduced.
The rule is conservative: only non-distinct, unfiltered {{sum}} over integral
measures with literal {{n PRECEDING AND CURRENT ROW}} frames qualifies, groups
below a minimum total frame width are skipped, and it is gated behind an opt-in
configuration flag, default off.
was:
Spark executes a bounded moving window frame by re\-aggregating every row
currently inside the frame on every output row: {{SlidingWindowFunctionFrame}}
resets the aggregate buffer and re\-folds the whole frame for each row
produced. For a window operator with nested sliding frames over one \(PARTITION
BY, ORDER BY\) key, the cost is the sum of the frame widths \(Σw\)
aggregate\-update evaluations per output row, and the cost is structural —
every added horizon multiplies the work. There is no incremental path: the
window aggregate processor only supports
{{initialize}}/{{update}}/{{evaluate}}, with no {{remove}}, so a moving frame
cannot be advanced row\-by\-row.
The pattern is common in user\-facing metric tables that compute several
rolling horizons per measure over a daily partition ordering. A realistic
example:
{code:sql}
SELECT
seller\_user\_id,
sum\(cnt\) OVER \(PARTITION BY seller\_user\_id ORDER BY receive\_date
ROWS BETWEEN 2 PRECEDING AND CURRENT ROW\) AS cnt\_3d,
sum\(cnt\) OVER \(PARTITION BY seller\_user\_id ORDER BY receive\_date
ROWS BETWEEN 6 PRECEDING AND CURRENT ROW\) AS cnt\_7d,
sum\(cnt\) OVER \(PARTITION BY seller\_user\_id ORDER BY receive\_date
ROWS BETWEEN 29 PRECEDING AND CURRENT ROW\) AS cnt\_30d,
sum\(cnt\) OVER \(PARTITION BY seller\_user\_id ORDER BY receive\_date
ROWS BETWEEN 89 PRECEDING AND CURRENT ROW\) AS cnt\_90d,
sum\(cnt\) OVER \(PARTITION BY seller\_user\_id ORDER BY receive\_date
ROWS BETWEEN 179 PRECEDING AND CURRENT ROW\) AS cnt\_180d
\-\- ...repeated for several more measures
FROM daily\_aggregates
{code}
This shape measures 5 horizons, i.e. Σw = 3\+7\+30\+90\+180 = 310
aggregate\-update evaluations per output row per measure family.
In a production workload measured against this shape \(a 3,437 CPU\-hour
application\), 45% of the CPU sits inside the window operator, and the window
stage's own input\-sort spill reaches 14.8 TiB on the largest query.
h3. Root cause
{{SlidingWindowFunctionFrame.write}} re\-runs the aggregate update for every
row in the frame on every output row. Growing frames are incremental and offset
frames are O\(1\), so the sliding frame is the only path with no O\(1\)
evaluation. The work factor is exactly Σw \(sum of frame widths\) per group per
row.
h3. Proposed change
Add an optimizer rule that rewrites each group of qualifying sliding frames as
one running aggregate plus offset differences:
{code}
C = sum\(x\) OVER \(PARTITION BY k ORDER BY d ROWS BETWEEN
UNBOUNDED PRECEDING AND CURRENT ROW\)
sum\(x\) OVER \(... ROWS BETWEEN n PRECEDING AND CURRENT ROW\)
= C \- coalesce\(lag\(C, n \+ 1\) OVER \(PARTITION BY k ORDER BY d\), 0\)
{code}
For integral accumulators with ANSI off this is provably bit\-identical: the
original fold already computes with wrapping addition, and modular reduction
commutes with subtraction, so the running prefix — a magnitude no bounded frame
can reach — overflows harmlessly. Floating\-point accumulators
\(non\-associative addition\) and decimal accumulators \(overflow becomes a
silent NULL\) are excluded, and a nullable measure carries a running count so
an all\-NULL frame still returns NULL. The rewrite reuses the window operator's
own partition and order spec, so no new exchange or sort is introduced.
The rule is conservative: only non\-distinct, unfiltered {{sum}} over integral
measures with literal {{n PRECEDING AND CURRENT ROW}} frames qualifies, groups
below a minimum total frame width are skipped, and it is gated behind an
opt\-in configuration flag, default off.
h3. Expected impact* Per\-group per\-row work drops from Σw to H \+ 2 aggregate
evaluations \(H = number of distinct widths\): 310 → 7 \(44x\) for the example
shape.
* Microbenchmark on the production shape \(6 measures, widths 2/6/29/89/179, 2M
rows\): 19.7s → 1.69s per window pass \(11.6x\); results verified
bit\-identical against the original plan, including under prefix overflow past
2\^63.
* On the real production query shape, a measured A/B of the rewritten form
shows \~10.5x end\-to\-end and \~20x on the window operator alone.
* Production model for one risk\-model workload family \(21 nodes, window stage
4,424 CPU\-hours\): \~3,481 CPU\-hours recoverable \(projected\).
* No new exchange or sort: both window operators share one sort, and
Exchange/SortExec counts are unchanged after the rewrite.
> Rewrite sliding window frames as prefix differences
> ---------------------------------------------------
>
> Key: SPARK-59658
> URL: https://issues.apache.org/jira/browse/SPARK-59658
> Project: Spark
> Issue Type: Improvement
> Components: SQL
> Affects Versions: 4.4.0
> Reporter: James Xu
> Priority: Major
>
> Spark executes a bounded moving window frame by re-aggregating every row
> currently inside the frame on every output row:
> {{SlidingWindowFunctionFrame}} resets the aggregate buffer and re-folds the
> whole frame for each row produced. For a window operator with nested sliding
> frames over one (PARTITION BY, ORDER BY) key, the cost is the sum of the
> frame widths (Σw) aggregate-update evaluations per output row, and the cost
> is structural — every added horizon multiplies the work. There is no
> incremental path: the window aggregate processor only supports
> {{{}initialize{}}}/{{{}update{}}}/{{{}evaluate{}}}, with no {{{}remove{}}},
> so a moving frame cannot be advanced row-by-row.
> The pattern is common in user-facing metric tables that compute several
> rolling horizons per measure over a daily partition ordering. A realistic
> example:
> {code:sql}
> SELECT
> seller_user_id,
> sum(cnt) OVER (PARTITION BY seller_user_id ORDER BY receive_date
> ROWS BETWEEN 2 PRECEDING AND CURRENT ROW) AS cnt_3d,
> sum(cnt) OVER (PARTITION BY seller_user_id ORDER BY receive_date
> ROWS BETWEEN 6 PRECEDING AND CURRENT ROW) AS cnt_7d,
> sum(cnt) OVER (PARTITION BY seller_user_id ORDER BY receive_date
> ROWS BETWEEN 29 PRECEDING AND CURRENT ROW) AS cnt_30d,
> sum(cnt) OVER (PARTITION BY seller_user_id ORDER BY receive_date
> ROWS BETWEEN 89 PRECEDING AND CURRENT ROW) AS cnt_90d,
> sum(cnt) OVER (PARTITION BY seller_user_id ORDER BY receive_date
> ROWS BETWEEN 179 PRECEDING AND CURRENT ROW) AS cnt_180d
> -- ... repeated for several more measures
> FROM daily_aggregates{code}
> This shape measures 5 horizons, i.e. Σw = 3+7+30+90+180 = 310
> aggregate-update evaluations per output row per measure family.
> In a production workload measured against this shape (a 3,437 CPU-hour
> application), 45% of the CPU sits inside the window operator, and the window
> stage's own input-sort spill reaches 14.8 TiB on the largest query.
> h3. Root cause
> {{SlidingWindowFunctionFrame.write}} re-runs the aggregate update for every
> row in the frame on every output row. Growing frames are incremental and
> offset frames are O(1), so the sliding frame is the only path with no O(1)
> evaluation. The work factor is exactly Σw (sum of frame widths) per group per
> row.
> h3. Proposed change
> Add an optimizer rule that rewrites each group of qualifying sliding frames
> as one running aggregate plus offset differences:
> {code:java}
> -- Running total, reused as the base for every sliding window
> C = sum(x) OVER (PARTITION BY k ORDER BY d
> ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)
> -- n-row sliding sum == running total minus the running total n+1 rows back
> sum(x) OVER (PARTITION BY k ORDER BY d
> ROWS BETWEEN n PRECEDING AND CURRENT ROW)
> = C - coalesce(lag(C, n + 1) OVER (PARTITION BY k ORDER BY d), 0){code}
> For integral accumulators with ANSI off this is provably bit-identical: the
> original fold already computes with wrapping addition, and modular reduction
> commutes with subtraction, so the running prefix — a magnitude no bounded
> frame can reach — overflows harmlessly. Floating-point accumulators
> (non-associative addition) and decimal accumulators (overflow becomes a
> silent NULL) are excluded, and a nullable measure carries a running count so
> an all-NULL frame still returns NULL. The rewrite reuses the window
> operator's own partition and order spec, so no new exchange or sort is
> introduced.
> The rule is conservative: only non-distinct, unfiltered {{sum}} over integral
> measures with literal {{n PRECEDING AND CURRENT ROW}} frames qualifies,
> groups below a minimum total frame width are skipped, and it is gated behind
> an opt-in configuration flag, default off.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
