[jira] [Updated] (SPARK-56546) [SQL] Optimize sliding window aggregation with segment tree

["Kent Yao (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/SPARK-56546?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Kent Yao updated SPARK-56546:
-----------------------------
    Description: 
h2. Motivation

Apache Spark's current sliding-window aggregate evaluation recomputes the full frame for every output row when no inverse exists (MIN/MAX, or when the inverse is numerically unstable — see SPARK-36844). This makes W-large sliding frames O(N×W) even though O(N log W) algorithms are well-known in the streaming-analytics literature (Leis et al. 2015, Arasu & Widom 2004).

Common symptoms: sliding {{MIN/MAX/SUM/AVG/COUNT/STDDEV}} queries over W ≳ 1000 rows become CPU-bound, spill-sensitive (String/binary types), and scale poorly with W. At W = 4001 / 2M rows, the current engine takes ~100 s per sliding pass on modern hardware.

h2. Proposal

Introduce a block-chunked segment-tree window function frame that evaluates sliding aggregates in O(log(W)) per output row, gated behind a new SQLConf {{spark.sql.window.segmentTree.enabled}} (default *false* — opt-in) and available from Spark 4.2.0.

The implementation is internal-only (no public API change, no new SQL syntax); a sibling {{SegmentTreeWindowFunctionFrame}} is selected by {{WindowFunctionFrameFactory}} whenever (1) the frame is a non-invertible sliding range/rows frame, (2) the aggregate is segment-tree-eligible (MIN/MAX/SUM/COUNT/AVG/STDDEV_POP/STDDEV_SAMP/VAR_POP/VAR_SAMP), and (3) the SQLConf is enabled. All other frames fall back to the existing path unchanged.

Trade-offs are documented inline: for very small W (≤ ~50) the segment-tree's block-chunked build overhead exceeds naive recomputation — those cases are a Pareto-loss zone (see "Caveats" below).

h2. Expected benefit

3-JDK GHA benchmark (JDK 17 / 21 / 25 on Azure EPYC 9V74 80-core, Linux 6.17, OpenJDK LTS),
WindowBenchmark rev4, all runs G1-G5 green, baseline per-iter 3–5 s target (§10.5 SPIP):

|| Aggregate || W / N || JDK 17 || JDK 21 || JDK 25 ||
| MIN (non-invertible) | 1001 / 256K | 8.9× | 9.0× | 9.0× |
| MAX (non-invertible) | 1001 / 256K | 10.2× | 13.1× | 14.2× |
| SUM (full recompute) | 1001 / 256K | 9.6× | 9.5× | 9.9× |
| COUNT | 1001 / 256K | 9.4× | 9.3× | 9.3× |
| AVG (multi-buffer) | 1001 / 192K | 10.6× | 10.6× | 10.2× |
| STDDEV_SAMP (multi-buffer, stress) | 1001 / 2M | 18.7× | 19.3× | 18.5× |
| SUM W-sweep (stress, cross-block) | 4001 / 2M | 32.4× | 32.6× | 29.0× |
| MAX String spill guard (stress) | 1001 / 1M | 15.7× | 16.4× | 17.6× |

N-sweep (segtree-only, W=1001, stress, JDK 17 per-row nanoseconds): N=2M → 1358 ns/row, N=8M → 1374 ns/row, N=16M → 1338 ns/row — *sub-linear / near-constant*, refuting any cache-resident hypothesis and matching the O(log W) cost model.

h3. Caveats (documented, not blockers)

* W ≤ 50 *Pareto-loss zone*: naive is ~2–3× faster than segtree (expected — segtree build overhead dominates). Covered by explicit {{(stress)}} benchmark cases W=11 / W=51. The feature is opt-in precisely so users/tuners can stay on naive for small-W workloads.
* W=10 pathological (0.4×): same cause, called out in §10.5 of the design doc.
* Segtree is a secondary path only; *the naive frame is untouched* and remains the default.

h2. Scope

* Component: SQL
* Affects: 4.2.0 (master)
* No public API / SQL grammar change
* New SQLConf: {{spark.sql.window.segmentTree.enabled}} (boolean, default false, since 4.2.0)
* New SQLMetrics on window operator: segment-tree frame construction counts
* All existing window tests remain green; new unit suite {{WindowSegmentTreeSuite}}, PBT suite, and benchmark {{WindowBenchmark}} (3-JDK GHA) cover the new path.

h2. Design document

Internal design + batch history tracked under yaooqinn/todos {{features/window-segment-tree/}}. Upstream PR will link the consolidated design doc once R4 review closes.

h2. Batch status (internal tracking)

* Batches 1 / 1.5 / 2 / 2.1 / 2.5 / 3a–3c ✅ merged (green)
* R1a block memory accounting ✅
* R2c M6 SQLMetrics ✅
* C2 benchmark rev4 (3–5 s baseline + §10.5 stress coverage) ✅
* R4 community review + upstream PR: *pending*


  was:
h3. Background

Spark's {{SlidingWindowFunctionFrame}} (sql/core, {{WindowFunctionFrame.scala}}) recomputes the aggregate over the entire frame for every output row in O(n*W) time, where n is the partition size and W is the frame width. This is quadratic in W for sliding-window workloads (e.g., rolling MAX, AVG, STDDEV over sensor time-series or financial tick data) and dominates wall-clock on windows wider than a few hundred rows.

Prior art: Leis et al., "Efficient Processing of Window Functions in Analytical SQL Queries" (VLDB 2015) -- a segment tree per partition gives O(n*log(W)) for any mergeable (associative) aggregate, including non-invertible ones (MIN, MAX, generic {{DeclarativeAggregate}} with {{mergeExpressions}}). DuckDB and Umbra use the same technique.

h3. Proposal

Introduce a segment-tree-backed {{WindowFunctionFrame}} path for bounded sliding ROWS and RANGE frames over {{DeclarativeAggregate}} functions (non-invertible and invertible alike). The new path is gated by a default-off SQLConf so it is opt-in and fully reversible.

Key elements:
* New {{WindowSegmentTree}} data structure (block-chunked; leaves = cached {{UnsafeRow}} blocks; internal nodes = pre-merged aggregate buffers).
* Integration via {{SegmentTreeWindowFunctionFrame}}, selected by {{WindowEvaluatorFactoryBase}} when the feature flag is on and the frame is sliding / mergeable.
* Memory integration with {{TaskMemoryManager}} (block-chunked LRU cache, spill via {{ExternalAppendOnlyUnsafeRowArray}}), so it does not regress low-memory operators.
* Observability: two new SQLMetrics ({{numSegmentTreeFrames}}, {{numSegmentTreeFallbackFrames}}) and EXPLAIN visibility.
* Fallback: any unsupported case (ImperativeAggregate, UDAF via ScalaUDAF, RANGE with multi-column ORDER BY, etc.) routes to the existing {{SlidingWindowFunctionFrame}} with no behavioral change.

h3. Expected benefit

Preliminary in-tree {{WindowBenchmark}} numbers (GHA fork run, JDK 17/21/25, 2M rows unless noted):
* MIN / MAX / SUM / AVG / COUNT: ~9-10x at W=1001
* STDDEV_SAMP: ~18x at W=1001 (stress case)
* W sweep at W=4001: ~31x (scales with W as expected: O(W/log W))
* String / spill path: ~17x

Final numbers will be refreshed from the committed {{sql/core/benchmarks/WindowBenchmark-results.txt}} once the current GHA matrix lands.

h3. Scope

Phase 1 (this umbrella's first wave of PRs):
* Bounded ROWS and RANGE sliding frames
* {{DeclarativeAggregate}} with {{mergeExpressions}}
* Default-off SQLConf, fallback on any unsupported shape

Phase 2 (separate JIRAs, not blocking):
* FAME prefix/suffix within-block cumulative
* Adaptive block size, fanout tuning
* Invertible fast-path for SUM/COUNT/AVG

h3. Safety / rollback

* SQLConf {{spark.sql.window.segmentTree.enabled}} defaults to {{false}}. Setting it back to {{false}} restores the exact pre-feature execution path.
* All existing window tests must stay green with the flag on and off; a new {{SegmentTreeWindowFunctionSuite}} covers edge cases (NULL, NaN, Decimal overflow, Binary, UDAF fallback, RANGE with ties / NULL keys, small-partition fallback, spill).

h3. Related work

* Leis et al., VLDB 2015: "Efficient Processing of Window Functions in Analytical SQL Queries".
* DuckDB segment-tree window implementation.
* Velox PR #9049 (partial FAME attempt; documented pitfalls informed this design).

h3. Subtasks

Subtasks will be filed as the implementation splits into PR-sized units (data structure, frame integration, memory/spill, metrics, RANGE support, benchmark). A split-strategy document will accompany the first subtask.

h3. Design doc

A public design doc summary will be linked here before the first subtask PR is opened.


> [SQL] Optimize sliding window aggregation with segment tree
> -----------------------------------------------------------
>
>                 Key: SPARK-56546
>                 URL: https://issues.apache.org/jira/browse/SPARK-56546
>             Project: Spark
>          Issue Type: Improvement
>          Components: SQL
>    Affects Versions: 4.2.0
>            Reporter: Kent Yao
>            Assignee: Kent Yao
>            Priority: Major
>
> h2. Motivation
> Apache Spark's current sliding-window aggregate evaluation recomputes the full frame for every output row when no inverse exists (MIN/MAX, or when the inverse is numerically unstable — see SPARK-36844). This makes W-large sliding frames O(N×W) even though O(N log W) algorithms are well-known in the streaming-analytics literature (Leis et al. 2015, Arasu & Widom 2004).
> Common symptoms: sliding {{MIN/MAX/SUM/AVG/COUNT/STDDEV}} queries over W ≳ 1000 rows become CPU-bound, spill-sensitive (String/binary types), and scale poorly with W. At W = 4001 / 2M rows, the current engine takes ~100 s per sliding pass on modern hardware.
> h2. Proposal
> Introduce a block-chunked segment-tree window function frame that evaluates sliding aggregates in O(log(W)) per output row, gated behind a new SQLConf {{spark.sql.window.segmentTree.enabled}} (default *false* — opt-in) and available from Spark 4.2.0.
> The implementation is internal-only (no public API change, no new SQL syntax); a sibling {{SegmentTreeWindowFunctionFrame}} is selected by {{WindowFunctionFrameFactory}} whenever (1) the frame is a non-invertible sliding range/rows frame, (2) the aggregate is segment-tree-eligible (MIN/MAX/SUM/COUNT/AVG/STDDEV_POP/STDDEV_SAMP/VAR_POP/VAR_SAMP), and (3) the SQLConf is enabled. All other frames fall back to the existing path unchanged.
> Trade-offs are documented inline: for very small W (≤ ~50) the segment-tree's block-chunked build overhead exceeds naive recomputation — those cases are a Pareto-loss zone (see "Caveats" below).
> h2. Expected benefit
> 3-JDK GHA benchmark (JDK 17 / 21 / 25 on Azure EPYC 9V74 80-core, Linux 6.17, OpenJDK LTS),
> WindowBenchmark rev4, all runs G1-G5 green, baseline per-iter 3–5 s target (§10.5 SPIP):
> || Aggregate || W / N || JDK 17 || JDK 21 || JDK 25 ||
> | MIN (non-invertible) | 1001 / 256K | 8.9× | 9.0× | 9.0× |
> | MAX (non-invertible) | 1001 / 256K | 10.2× | 13.1× | 14.2× |
> | SUM (full recompute) | 1001 / 256K | 9.6× | 9.5× | 9.9× |
> | COUNT | 1001 / 256K | 9.4× | 9.3× | 9.3× |
> | AVG (multi-buffer) | 1001 / 192K | 10.6× | 10.6× | 10.2× |
> | STDDEV_SAMP (multi-buffer, stress) | 1001 / 2M | 18.7× | 19.3× | 18.5× |
> | SUM W-sweep (stress, cross-block) | 4001 / 2M | 32.4× | 32.6× | 29.0× |
> | MAX String spill guard (stress) | 1001 / 1M | 15.7× | 16.4× | 17.6× |
> N-sweep (segtree-only, W=1001, stress, JDK 17 per-row nanoseconds): N=2M → 1358 ns/row, N=8M → 1374 ns/row, N=16M → 1338 ns/row — *sub-linear / near-constant*, refuting any cache-resident hypothesis and matching the O(log W) cost model.
> h3. Caveats (documented, not blockers)
> * W ≤ 50 *Pareto-loss zone*: naive is ~2–3× faster than segtree (expected — segtree build overhead dominates). Covered by explicit {{(stress)}} benchmark cases W=11 / W=51. The feature is opt-in precisely so users/tuners can stay on naive for small-W workloads.
> * W=10 pathological (0.4×): same cause, called out in §10.5 of the design doc.
> * Segtree is a secondary path only; *the naive frame is untouched* and remains the default.
> h2. Scope
> * Component: SQL
> * Affects: 4.2.0 (master)
> * No public API / SQL grammar change
> * New SQLConf: {{spark.sql.window.segmentTree.enabled}} (boolean, default false, since 4.2.0)
> * New SQLMetrics on window operator: segment-tree frame construction counts
> * All existing window tests remain green; new unit suite {{WindowSegmentTreeSuite}}, PBT suite, and benchmark {{WindowBenchmark}} (3-JDK GHA) cover the new path.
> h2. Design document
> Internal design + batch history tracked under yaooqinn/todos {{features/window-segment-tree/}}. Upstream PR will link the consolidated design doc once R4 review closes.
> h2. Batch status (internal tracking)
> * Batches 1 / 1.5 / 2 / 2.1 / 2.5 / 3a–3c ✅ merged (green)
> * R1a block memory accounting ✅
> * R2c M6 SQLMetrics ✅
> * C2 benchmark rev4 (3–5 s baseline + §10.5 stress coverage) ✅
> * R4 community review + upstream PR: *pending*



--
This message was sent by Atlassian Jira
(v8.20.10#820010)

---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org