[
https://issues.apache.org/jira/browse/SPARK-59658?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated SPARK-59658:
-----------------------------------
Labels: pull-request-available (was: )
> 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
> Labels: pull-request-available
>
> 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]
