pavan51 opened a new pull request, #59779: URL: https://github.com/apache/spark/pull/59779
… using monotonic deque This PR proposes to optimize sliding window `MIN`/`MAX` aggregate functions using a monotonic deque implementation ([SlidingWindowMinMaxFunctionFrame](https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/execution/window/SlidingWindowMinMaxFunctionFrame.scala)). Key improvements: - **Monotonic Deque Optimization**: When all window functions are `MIN` and/or `MAX`, we maintain a double-ended queue (deque) of `(value, index)` pairs for each function. New values are admitted and elements that fall outside the sliding window boundary are dropped from the front in amortized $O(1)$ time per row. - **Wired in Factory**: We modify `WindowEvaluatorFactoryBase` to detect when the query uses only `MIN`/`MAX` and automatically instantiate `SlidingWindowMinMaxFunctionFrame`. - **SQL Configuration**: Added `spark.sql.window.monotonicDeque.enabled` (default: `false`) to allow toggling this optimization. - **SpillableArrayIterator now implements Closeable**: `ExternalAppendOnlyUnsafeRowArray.SpillableArrayIterator` now mixes in `java.io.Closeable` with a public `close()` method that delegates to the underlying `UnsafeSorterIterator`. This allows `closeIterator` pattern-matches in window frames to successfully release spill readers at partition boundaries, instead of falling through to the no-op branch. - **SegmentTreeWindowFunctionFrame closes spill readers at partition boundary**: `SegmentTreeWindowFunctionFrame.closeIters()` now closes its three iterators (`boundIter`, `lowerIter`, `upperIter`) via the same `Closeable` pattern before nulling them, ensuring spill readers are released eagerly at partition boundaries rather than deferred to task completion. --- Currently, sliding window aggregates in Spark are executed using: 1. **Default Spark (Naive) Sliding Window** (`SlidingWindowFunctionFrame`): Re-scans all rows in the sliding window buffer on every row, resulting in **O(N x W)** time complexity. 2. **Segment Tree** (`SegmentTreeWindowFunctionFrame`): Built over partition blocks, offering **O(N x log W)** complexity but with significant block-caching and tree-construction overhead. While the Segment Tree is an improvement over Naive for large windows, it has three major deficiencies: * Logarithmic Complexity Overhead: As the window size $W$ grows, the query overhead increases logarithmically (O(\log W)). * Pareto Loss Zone at **Small Windows**: At small window sizes (e.g. $W=11$), the Segment Tree's setup, block allocation, and pointer-chasing traversal overhead make it significantly slower than even the Naive baseline **(Segment tree running ~2X slower then default Spark Naive**). * High Heap Allocation & GC Pressure: The Segment Tree divides rows into blocks and builds a tree structure over the partition data. To build and store leaf and branch nodes, it allocates a large number of node objects and tree arrays on the JVM heap. This leads to high heap allocation and garbage collection (GC) pressure, even for primitive types. By implementing a **Monotonic deque**, we achieve true linear **O(N)** time complexity (amortized **O(1)** operations per row regardless of window size). For moving frames, the space complexity is bounded by $O(W)$, maintaining a minimal memory footprint. The raw numbers and environment header are in [sql/core/benchmarks/WindowBenchmark-results.txt](https://github.com/apache/spark/blob/master/sql/core/benchmarks/WindowBenchmark-results.txt). Key ratios from CI runs typically show: - Segment Tree: **~9.5X - 11X faster** than Naive Baseline - Monotonic Deque (new): **~27X - 35X faster** than the current Spark implementation (Naive) and **~3X faster** than Segment Tree - Segment Tree: **~10X - 11X faster** than Naive Baseline - Monotonic Deque (new): **~30X - 40X faster** than the current Spark implementation (Naive) and **~3.2X - 3.6X faster** than Segment Tree - Increasing (Worst-Case): Monotonic Deque is **~6X - 7.5X faster** than Segment Tree - Decreasing (Best-Case): Monotonic Deque is **~6X - 7.5X faster** than Segment Tree - Random (Average-Case): Monotonic Deque is **~5.5X - 7X faster** than Segment Tree In small windows, the naive sliding window scan **O(W)** is fast, whereas the segment tree incurs **O(log W)** setup and query overhead, making it slower than naive. The Monotonic Deque removes this overhead while maintaining linear **O(N)** complexity. - **W=1:** Monotonic Deque matches Naive speed (Segment tree is **~2.5X slower**) - **W=3:** Monotonic Deque is **~1.1X - 1.2X faster** than Naive (Segment tree is > 2X slower) - **W=11:** Monotonic Deque is **~1.6X - 1.8X faster** than Naive (Segment tree is ~2X slower) > **Note:** Absolute timings and exact multipliers vary depending on the CI runner hardware. See the linked `WindowBenchmark-results.txt` for the exact environment header and raw numbers from the latest run. --- No user-facing behavior changes. It is a purely physical plan performance optimization. There is a new configuration: - `spark.sql.window.monotonicDeque.enabled` (Default: `false`) to enable/disable the monotonic deque optimization. --- 1. **Correctness Unit Tests**: Ran `SegmentTreeWindowFunctionSuite` and `DataFrameWindowFramesSuite` (all 80 tests succeeded), and added a new differential correctness suite `MonotonicDequeWindowFunctionSuite` verifying monotonic deque outputs against Naive and Segment Tree baselines across multiple data types (including date/timestamp/interval types), range/row frames, and null values. 2. **Correctness Digests**: Verified that output digests generated by `WindowBenchmark` match the baseline exactly. 3. **Benchmarks**: Added monotonic deque cases and small/large window scaling comparison runs to the `WindowBenchmark` suite. --- Yes, co-authored by Google DeepMind Coding Assistant Closes #57346 from pavan51/sliding-window-min-max-monotonic-deque. Lead-authored-by: Pavan51 <[email protected]> (cherry picked from commit 859de7dd4cb6ca04c3b5946d5739e557f48e5d15) <!-- Thanks for sending a pull request! Here are some tips for you: 1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html 2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html 3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'. 4. Be sure to keep the PR description updated to reflect all changes. 5. Please write your PR title to summarize what this PR proposes. 6. If possible, provide a concise example to reproduce the issue for a faster review. 7. If you want to add a new configuration, please read the guideline first for naming configurations in 'common/utils/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'. 8. If you want to add or modify an error type or message, please read the guideline first in 'common/utils/src/main/resources/error/README.md'. --> ### What changes were proposed in this pull request? <!-- Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue. If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below. 1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers. 2. If you fix some SQL features, you can provide some references of other DBMSes. 3. If there is design documentation, please add the link. 4. If there is a discussion in the mailing list, please add the link. --> ### Why are the changes needed? <!-- Please clarify why the changes are needed. For instance, 1. If you propose a new API, clarify the use case for a new API. 2. If you fix a bug, you can clarify why it is a bug. --> ### Does this PR introduce _any_ user-facing change? <!-- Note that it means *any* user-facing change including all aspects such as new features, bug fixes, or other behavior changes. Documentation-only updates are not considered user-facing changes. If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible. If possible, please also clarify if this is a user-facing change compared to the released Spark versions or within the unreleased branches such as master. If no, write 'No'. --> ### How was this patch tested? <!-- If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible. If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future. If tests were not added, please describe why they were not added and/or why it was difficult to add. If benchmark tests were added, please run the benchmarks in GitHub Actions for the consistent environment, and the instructions could accord to: https://spark.apache.org/developer-tools.html#github-workflow-benchmarks. --> ### Was this patch authored or co-authored using generative AI tooling? <!-- If generative AI tooling has been used in the process of authoring this patch, please include the phrase: 'Generated-by: ' followed by the name of the tool and its version. If no, write 'No'. Please refer to the [ASF Generative Tooling Guidance](https://www.apache.org/legal/generative-tooling.html) for details. --> -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
