You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by pt...@apache.org on 2026/10/19 07:26:32 UTC
(spark) branch master updated: [SPARK-58201][SQL] Optimize sliding window MIN/MAX aggregate function using monotonic deque
This is an automated email from the ASF dual-hosted git repository.
peter-toth pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 859de7dd4cb6 [SPARK-58201][SQL] Optimize sliding window MIN/MAX aggregate function using monotonic deque
859de7dd4cb6 is described below
commit 859de7dd4cb6ca04c3b5946d5739e557f48e5d15
Author: Pavan51 <pa...@gmail.com>
AuthorDate: Mon Oct 19 09:26:18 2026 +0200
[SPARK-58201][SQL] Optimize sliding window MIN/MAX aggregate function using monotonic deque
### What changes were proposed in this pull request?
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.
---
### Why are the changes needed?
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.
### Benchmarks
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:
### MIN Sliding Window (W=1001, 256K rows)
- 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
### MAX Sliding Window (W=1001, 256K rows)
- 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
### MIN Sliding Window Scaling (W=100001, 2M rows)
- 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
### MIN/MAX Small Window Scaling (W=1, W=3, W=11, 2M rows – Pareto Loss Zone Check)
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.
---
### Does this PR introduce _any_ user-facing change?
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.
---
### How was this patch tested?
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.
---
### Was this patch authored or co-authored using generative AI tooling?
Yes, co-authored by Google DeepMind Coding Assistant
Closes #57346 from pavan51/sliding-window-min-max-monotonic-deque.
Lead-authored-by: Pavan51 <pa...@gmail.com>
Co-authored-by: Pavan <pa...@Lakshmis-MacBook-Air.local>
Co-authored-by: Pavan <pa...@Pavans-MacBook-Pro.local>
Signed-off-by: Peter Toth <pe...@gmail.com>
---
.../org/apache/spark/sql/internal/SQLConf.scala | 13 +
sql/core/benchmarks/WindowBenchmark-results.txt | 212 +++++++---
.../ExternalAppendOnlyUnsafeRowArray.scala | 6 +-
.../window/SegmentTreeWindowFunctionFrame.scala | 16 +-
.../window/SlidingWindowMinMaxFunctionFrame.scala | 327 ++++++++++++++++
.../execution/window/WindowEvaluatorFactory.scala | 6 +-
.../window/WindowEvaluatorFactoryBase.scala | 30 +-
.../spark/sql/execution/window/WindowExec.scala | 7 +-
.../sql/execution/benchmark/WindowBenchmark.scala | 245 ++++++++++--
.../window/MonotonicDequeWindowFunctionSuite.scala | 434 +++++++++++++++++++++
.../window/SegmentTreeWindowMetricsSuite.scala | 7 +
.../window/WindowSegmentTreeAllowlistSuite.scala | 122 ++++--
12 files changed, 1289 insertions(+), 136 deletions(-)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
index 300ccc8ff51f..6af6dfd9cd6c 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
@@ -5032,6 +5032,17 @@ object SQLConf {
.booleanConf
.createWithDefault(false)
+ val WINDOW_MONOTONIC_DEQUE_ENABLED =
+ buildConf("spark.sql.window.monotonicDeque.enabled")
+ .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
+ .doc("Use O(N) monotonic deque for sliding window MIN/MAX. Applies only when the " +
+ "moving frame contains exclusively Min/Max aggregates with no FILTER clauses. " +
+ "It replaces the O(N * W) naive loop and O(N log W) segment tree with an " +
+ "optimized ring-buffer deque.")
+ .version("4.4.0")
+ .booleanConf
+ .createWithDefault(false)
+
val WINDOW_SEGMENT_TREE_ENABLED =
buildConf("spark.sql.window.segmentTree.enabled")
.withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
@@ -9812,6 +9823,8 @@ class SQLConf extends Serializable with Logging with SqlApiConf {
def windowGroupLimitThreshold: Int = getConf(WINDOW_GROUP_LIMIT_THRESHOLD)
+ def windowMonotonicDequeEnabled: Boolean = getConf(WINDOW_MONOTONIC_DEQUE_ENABLED)
+
def windowSegmentTreeEnabled: Boolean = getConf(WINDOW_SEGMENT_TREE_ENABLED)
def windowSegmentTreeMinPartitionRows: Int = getConf(WINDOW_SEGMENT_TREE_MIN_PARTITION_ROWS)
diff --git a/sql/core/benchmarks/WindowBenchmark-results.txt b/sql/core/benchmarks/WindowBenchmark-results.txt
index 291512fd3168..d9c537613307 100644
--- a/sql/core/benchmarks/WindowBenchmark-results.txt
+++ b/sql/core/benchmarks/WindowBenchmark-results.txt
@@ -2,171 +2,287 @@
Section A - MIN (non-invertible)
================================================================================================
-OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1020-azure
+OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1022-azure
AMD EPYC 7763 64-Core Processor
MIN sliding window, W=1001, 256K rows: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-MIN naive (current, baseline) 4014 4035 17 0.1 15313.6 1.0X
-MIN segtree (default) 392 407 16 0.7 1496.8 10.2X
-MIN segtree (blockSize=256) 2199 2214 12 0.1 8388.4 1.8X
+MIN naive (current, baseline) 4031 4062 25 0.1 15377.7 1.0X
+MIN segtree (default) 388 409 13 0.7 1481.7 10.4X
+MIN segtree (blockSize=256) 2208 2229 21 0.1 8422.7 1.8X
+MIN monotonic deque (new) 128 138 7 2.0 488.0 31.5X
================================================================================================
Section A - MAX (non-invertible)
================================================================================================
-OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1020-azure
+OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1022-azure
AMD EPYC 7763 64-Core Processor
MAX sliding window, W=1001, 256K rows: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-MAX naive (current, baseline) 4244 4266 21 0.1 16190.5 1.0X
-MAX segtree (default) 367 373 4 0.7 1401.2 11.6X
-MAX segtree (blockSize=256) 2231 2238 12 0.1 8510.3 1.9X
+MAX naive (current, baseline) 4271 4273 3 0.1 16291.8 1.0X
+MAX segtree (default) 381 392 12 0.7 1455.3 11.2X
+MAX segtree (blockSize=256) 2217 2233 18 0.1 8458.4 1.9X
+MAX monotonic deque (new) 111 136 18 2.4 422.7 38.5X
================================================================================================
Section A - SUM (Spark has no inverse; full recompute)
================================================================================================
-OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1020-azure
+OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1022-azure
AMD EPYC 7763 64-Core Processor
SUM sliding window, W=1001, 256K rows: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-SUM naive (current, baseline) 4111 4131 30 0.1 15683.8 1.0X
-SUM segtree (default) 358 365 8 0.7 1364.2 11.5X
-SUM segtree (blockSize=256) 2228 2245 12 0.1 8498.3 1.8X
+SUM naive (current, baseline) 4168 4292 225 0.1 15901.0 1.0X
+SUM segtree (default) 384 387 6 0.7 1463.2 10.9X
+SUM segtree (blockSize=256) 2221 2310 188 0.1 8472.7 1.9X
================================================================================================
Section A - COUNT
================================================================================================
-OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1020-azure
+OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1022-azure
AMD EPYC 7763 64-Core Processor
COUNT sliding window, W=1001, 256K rows: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-COUNT naive (current, baseline) 3663 3690 22 0.1 13974.5 1.0X
-COUNT segtree (default) 325 335 9 0.8 1238.6 11.3X
-COUNT segtree (blockSize=256) 2161 2165 6 0.1 8242.5 1.7X
+COUNT naive (current, baseline) 3511 3528 14 0.1 13392.8 1.0X
+COUNT segtree (default) 346 360 14 0.8 1318.8 10.2X
+COUNT segtree (blockSize=256) 2184 2192 5 0.1 8332.6 1.6X
================================================================================================
Section A - AVG (multi-buffer)
================================================================================================
-OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1020-azure
+OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1022-azure
AMD EPYC 7763 64-Core Processor
AVG sliding window, W=1001, 192K rows: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-AVG naive (current, baseline) 4441 4463 19 0.0 22588.2 1.0X
-AVG segtree (default) 337 340 4 0.6 1713.3 13.2X
-AVG segtree (blockSize=256) 1398 1415 14 0.1 7111.1 3.2X
+AVG naive (current, baseline) 4618 4631 12 0.0 23490.4 1.0X
+AVG segtree (default) 363 371 7 0.5 1848.2 12.7X
+AVG segtree (blockSize=256) 1417 1435 14 0.1 7209.4 3.3X
================================================================================================
Section A - STDDEV_SAMP (multi-buffer, stress)
================================================================================================
-OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1020-azure
+OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1022-azure
AMD EPYC 7763 64-Core Processor
STDDEV_SAMP sliding window, W=1001, 2M rows (stress): Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------------------
-STDDEV_SAMP naive (current, baseline) 102720 102801 91 0.0 51360.1 1.0X
-STDDEV_SAMP segtree (default) 6107 6132 35 0.3 3053.4 16.8X
-STDDEV_SAMP segtree (blockSize=256) 113831 113863 29 0.0 56915.6 0.9X
+STDDEV_SAMP naive (current, baseline) 101980 102168 165 0.0 50990.0 1.0X
+STDDEV_SAMP segtree (default) 6089 6108 18 0.3 3044.4 16.7X
+STDDEV_SAMP segtree (blockSize=256) 114099 114239 129 0.0 57049.3 0.9X
================================================================================================
-Section B - W=10 scaling (stress: Pareto loss zone)
+Section B - W=11 scaling (stress: Pareto loss zone)
================================================================================================
-OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1020-azure
+OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1022-azure
AMD EPYC 7763 64-Core Processor
SUM scaling, W=11, 2M rows (stress): Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-SUM naive W=11 959 964 4 2.1 479.6 1.0X
-SUM segtree (default) W=11 1926 1933 9 1.0 963.2 0.5X
+SUM naive W=11 962 981 22 2.1 481.0 1.0X
+SUM segtree (default) W=11 1900 1911 9 1.1 950.2 0.5X
================================================================================================
-Section B - W=50 scaling (stress: Pareto loss zone)
+Section B - W=51 scaling (stress: Pareto loss zone)
================================================================================================
-OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1020-azure
+OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1022-azure
AMD EPYC 7763 64-Core Processor
SUM scaling, W=51, 2M rows (stress): Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-SUM naive W=51 2226 2242 27 0.9 1113.0 1.0X
-SUM segtree (default) W=51 2193 2203 12 0.9 1096.5 1.0X
+SUM naive W=51 2172 2186 13 0.9 1085.9 1.0X
+SUM segtree (default) W=51 2161 2175 15 0.9 1080.7 1.0X
================================================================================================
Section B - W=201 scaling
================================================================================================
-OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1020-azure
+OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1022-azure
AMD EPYC 7763 64-Core Processor
SUM scaling, W=201, 1M rows: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-SUM naive W=201 3466 3556 187 0.3 3466.0 1.0X
-SUM segtree (default) W=201 1224 1232 7 0.8 1224.1 2.8X
+SUM naive W=201 3371 3378 7 0.3 3370.8 1.0X
+SUM segtree (default) W=201 1237 1250 10 0.8 1237.2 2.7X
================================================================================================
Section B - W=4001 scaling (stress, + bs=256 cross-block)
================================================================================================
-OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1020-azure
+OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1022-azure
AMD EPYC 7763 64-Core Processor
SUM scaling, W=4001, 2M rows (stress): Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-SUM naive W=4001 125108 125307 192 0.0 62553.9 1.0X
-SUM segtree (default) W=4001 3308 3320 21 0.6 1653.9 37.8X
-SUM segtree (blockSize=256) W=4001 110622 111593 1667 0.0 55311.1 1.1X
+SUM naive W=4001 121642 121898 229 0.0 60820.9 1.0X
+SUM segtree (default) W=4001 3295 3315 29 0.6 1647.7 36.9X
+SUM segtree (blockSize=256) W=4001 110661 110774 98 0.0 55330.6 1.1X
================================================================================================
Section F - spill regression guard (String, stress)
================================================================================================
-OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1020-azure
+OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1022-azure
AMD EPYC 7763 64-Core Processor
MAX String spill guard, W=1001, 1M rows (stress): Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
--------------------------------------------------------------------------------------------------------------------------------
-MAX naive (String) 59293 59370 67 0.0 59292.7 1.0X
-MAX segtree default (String) 2796 2808 16 0.4 2796.2 21.2X
+MAX naive (String) 53281 53371 85 0.0 53280.8 1.0X
+MAX segtree default (String) 2704 2717 14 0.4 2703.7 19.7X
================================================================================================
Section C - N-sweep small (stress)
================================================================================================
-OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1020-azure
+OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1022-azure
AMD EPYC 7763 64-Core Processor
SUM N-sweep (segtree-only), W=1001, 2M rows (stress): Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------------------
-SUM segtree (default) N=2M 2804 2815 16 0.7 1402.1 1.0X
+SUM segtree (default) N=2M 2786 2819 39 0.7 1392.9 1.0X
================================================================================================
Section C - N-sweep mid (stress)
================================================================================================
-OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1020-azure
+OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1022-azure
AMD EPYC 7763 64-Core Processor
SUM N-sweep (segtree-only), W=1001, 8M rows (stress): Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------------------
-SUM segtree (default) N=8M 13613 13648 32 0.6 1701.6 1.0X
+SUM segtree (default) N=8M 13494 13530 33 0.6 1686.7 1.0X
================================================================================================
Section C - N-sweep large (stress)
================================================================================================
-OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1020-azure
+OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1022-azure
AMD EPYC 7763 64-Core Processor
SUM N-sweep (segtree-only), W=1001, 16M rows (stress): Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
-------------------------------------------------------------------------------------------------------------------------------------
-SUM segtree (default) N=16M 33919 33948 49 0.5 2119.9 1.0X
+SUM segtree (default) N=16M 33713 37293 6146 0.5 2107.0 1.0X
+
+
+================================================================================================
+Section G - MIN Monotonic Deque vs Segment Tree (Worst-Case: Increasing)
+================================================================================================
+
+OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 7763 64-Core Processor
+MIN sliding window (Increasing), W=100001, 2M rows: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
+----------------------------------------------------------------------------------------------------------------------------------
+MIN segtree (Increasing) 3820 3852 31 0.5 1909.8 1.0X
+MIN monotonic deque (Increasing) 539 545 4 3.7 269.4 7.1X
+
+
+================================================================================================
+Section G - MIN Monotonic Deque vs Segment Tree (Best-Case: Decreasing)
+================================================================================================
+
+OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 7763 64-Core Processor
+MIN sliding window (Decreasing), W=100001, 2M rows: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
+----------------------------------------------------------------------------------------------------------------------------------
+MIN segtree (Decreasing) 3798 3842 44 0.5 1899.2 1.0X
+MIN monotonic deque (Decreasing) 530 535 6 3.8 265.0 7.2X
+
+
+================================================================================================
+Section G - MIN Monotonic Deque vs Segment Tree (Random)
+================================================================================================
+
+OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 7763 64-Core Processor
+MIN sliding window (Random), W=100001, 2M rows: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
+------------------------------------------------------------------------------------------------------------------------------
+MIN segtree (Random) 3876 3896 21 0.5 1938.0 1.0X
+MIN monotonic deque (Random) 558 564 6 3.6 278.9 6.9X
+
+
+================================================================================================
+Section H - MIN W=1 scaling (2M rows)
+================================================================================================
+
+OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 7763 64-Core Processor
+MIN sliding window, W=1, 2M rows: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
+------------------------------------------------------------------------------------------------------------------------
+MIN naive (current, baseline) [H W=1] 596 604 9 3.4 298.1 1.0X
+MIN segtree (default) [H W=1] 1481 1484 3 1.4 740.3 0.4X
+MIN monotonic deque (new) [H W=1] 549 566 26 3.6 274.4 1.1X
+
+
+================================================================================================
+Section H - MAX W=1 scaling (2M rows)
+================================================================================================
+
+OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 7763 64-Core Processor
+MAX sliding window, W=1, 2M rows: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
+------------------------------------------------------------------------------------------------------------------------
+MAX naive (current, baseline) [H W=1] 583 588 4 3.4 291.5 1.0X
+MAX segtree (default) [H W=1] 1466 1470 7 1.4 732.9 0.4X
+MAX monotonic deque (new) [H W=1] 549 553 4 3.6 274.4 1.1X
+
+
+================================================================================================
+Section H - MIN W=3 scaling (2M rows)
+================================================================================================
+
+OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 7763 64-Core Processor
+MIN sliding window, W=3, 2M rows: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
+------------------------------------------------------------------------------------------------------------------------
+MIN naive (current, baseline) [H W=3] 665 669 4 3.0 332.4 1.0X
+MIN segtree (default) [H W=3] 1589 1621 30 1.3 794.6 0.4X
+MIN monotonic deque (new) [H W=3] 556 557 1 3.6 278.2 1.2X
+
+
+================================================================================================
+Section H - MAX W=3 scaling (2M rows)
+================================================================================================
+
+OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 7763 64-Core Processor
+MAX sliding window, W=3, 2M rows: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
+------------------------------------------------------------------------------------------------------------------------
+MAX naive (current, baseline) [H W=3] 657 663 8 3.0 328.3 1.0X
+MAX segtree (default) [H W=3] 1588 1601 23 1.3 793.8 0.4X
+MAX monotonic deque (new) [H W=3] 550 555 5 3.6 274.9 1.2X
+
+
+================================================================================================
+Section H - MIN W=11 scaling (stress, 2M rows)
+================================================================================================
+
+OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 7763 64-Core Processor
+MIN sliding window, W=11, 2M rows (stress): Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
+--------------------------------------------------------------------------------------------------------------------------
+MIN naive (current, baseline) [H W=11] 981 984 3 2.0 490.3 1.0X
+MIN segtree (default) [H W=11] 1991 1997 6 1.0 995.7 0.5X
+MIN monotonic deque (new) [H W=11] 554 557 4 3.6 277.0 1.8X
+
+
+================================================================================================
+Section H - MAX W=11 scaling (stress, 2M rows)
+================================================================================================
+
+OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1022-azure
+AMD EPYC 7763 64-Core Processor
+MAX sliding window, W=11, 2M rows (stress): Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
+--------------------------------------------------------------------------------------------------------------------------
+MAX naive (current, baseline) [H W=11] 975 978 5 2.1 487.5 1.0X
+MAX segtree (default) [H W=11] 1978 1978 0 1.0 988.8 0.5X
+MAX monotonic deque (new) [H W=11] 555 558 4 3.6 277.5 1.8X
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/ExternalAppendOnlyUnsafeRowArray.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/ExternalAppendOnlyUnsafeRowArray.scala
index 9fb2a915ac2a..3791d2068f06 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/ExternalAppendOnlyUnsafeRowArray.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/ExternalAppendOnlyUnsafeRowArray.scala
@@ -256,7 +256,7 @@ class ExternalAppendOnlyUnsafeRowArray(
private[this] class SpillableArrayIterator(
iterator: UnsafeSorterIterator,
numFieldPerRow: Int)
- extends ExternalAppendOnlyUnsafeRowArrayIterator {
+ extends ExternalAppendOnlyUnsafeRowArrayIterator with Closeable {
private val currentRow = new UnsafeRow(numFieldPerRow)
@@ -269,7 +269,9 @@ class ExternalAppendOnlyUnsafeRowArray(
currentRow
}
- override protected def closeIfNeeded(): Unit = iterator match {
+ override protected def closeIfNeeded(): Unit = close()
+
+ override def close(): Unit = iterator match {
case c: Closeable => c.close()
case _ => // do nothing
}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/window/SegmentTreeWindowFunctionFrame.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/window/SegmentTreeWindowFunctionFrame.scala
index 496126150f54..70ba16ee1c56 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/window/SegmentTreeWindowFunctionFrame.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/window/SegmentTreeWindowFunctionFrame.scala
@@ -314,19 +314,25 @@ private[window] final class SegmentTreeWindowFunctionFrame(
if (fallbackUsed) fallback.currentUpperBound() else upperBound
/**
- * Drop references to open rowArray iterators. Idempotent.
+ * Close and drop references to open rowArray iterators. Idempotent.
*
- * Note: `ExternalAppendOnlyUnsafeRowArrayIterator.closeIfNeeded()` is
- * `protected`, so we cannot invoke it. Spark's own `SlidingWindowFunctionFrame`
- * also does not close its iterator; the backing `UnsafeExternalSorter` is
- * released by the enclosing `WindowExec`'s `TaskCompletionListener`.
+ * `SpillableArrayIterator` implements `Closeable`, so spill readers are
+ * released at partition boundary rather than deferred to task completion.
*/
private def closeIters(): Unit = {
+ closeIterator(boundIter)
+ closeIterator(lowerIter)
+ closeIterator(upperIter)
boundIter = null
lowerIter = null
upperIter = null
}
+ private def closeIterator(it: Iterator[UnsafeRow]): Unit = it match {
+ case c: java.io.Closeable => c.close()
+ case _ => // in-memory iterator: no-op
+ }
+
override def close(): Unit = {
if (tree != null) {
tree.close()
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/window/SlidingWindowMinMaxFunctionFrame.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/window/SlidingWindowMinMaxFunctionFrame.scala
new file mode 100644
index 000000000000..8fb4737094bf
--- /dev/null
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/window/SlidingWindowMinMaxFunctionFrame.scala
@@ -0,0 +1,327 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.window
+
+import java.io.Closeable
+
+import org.apache.spark.TaskContext
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.aggregate._
+import org.apache.spark.sql.catalyst.util.TypeUtils
+import org.apache.spark.sql.execution.ExternalAppendOnlyUnsafeRowArray
+import org.apache.spark.sql.execution.metric.SQLMetric
+
+/**
+ * An optimized sliding window frame that calculates min and/or max aggregate functions using
+ * monotonic deques. This provides O(N) time complexity instead of O(N * W) of
+ * [[SlidingWindowFunctionFrame]] or O(N log W) of [[SegmentTreeWindowFunctionFrame]].
+ *
+ * This frame is only instantiated when `isMinMaxOnly` is true (all window functions are Min or
+ * Max and no FILTER clause is used), enforced upstream in [[WindowEvaluatorFactoryBase]].
+ */
+private[window] final class SlidingWindowMinMaxFunctionFrame(
+ target: InternalRow,
+ processor: AggregateProcessor,
+ lbound: BoundOrdering,
+ ubound: BoundOrdering,
+ functions: Array[Expression],
+ inputSchema: Seq[Attribute],
+ numMonotonicDequeFrames: Option[SQLMetric] = None)
+ extends WindowFunctionFrame with AutoCloseable {
+
+ /** Rows of the partition currently being processed. */
+ private[this] var input: ExternalAppendOnlyUnsafeRowArray = null
+
+ // Spill-safety: when `input` (ExternalAppendOnlyUnsafeRowArray) spills, its
+ // iterator reuses a single UnsafeRow whose pointer is rebound on each next().
+ // This is safe because nextRow (and lowerRow on RANGE frames) are used for
+ // comparison *before* calling getNextOrNull. Values are extracted from the row
+ // via `evaluateAndCopy` before advancing. DO NOT cache a historical row without
+ // an explicit .copy(); the shared reusable UnsafeRow would silently mutate.
+ private[this] var lowerIterator: Iterator[UnsafeRow] = _
+ private[this] var inputIterator: Iterator[UnsafeRow] = _
+
+ // RowBoundOrdering.compare ignores its inputRow (it only uses the index), so
+ // the lower cursor is only needed for RANGE frames. Passing a null lowerRow to
+ // lbound.compare is safe on the ROWS path for exactly this reason.
+ private[this] val needsLowerRow = !lbound.isInstanceOf[RowBoundOrdering]
+
+ /** The row at lowerBound. Only valid when needsLowerRow is true. */
+ private[this] var lowerRow: UnsafeRow = null
+
+ /** The next row from `input`. */
+ private[this] var nextRow: InternalRow = null
+
+ /**
+ * Index of the first input row with a value equal to or greater than the lower bound of the
+ * current output row.
+ */
+ private[this] var lowerBound = 0
+
+ /**
+ * Index of the first input row with a value greater than the upper bound of the current output
+ * row.
+ */
+ private[this] var upperBound = 0
+
+ // `sourceRow` is used as the `source` argument to `processor.evaluate(source, target)`.
+ // Layout compatibility is guaranteed because Min/Max each contribute exactly one
+ // `aggBufferAttributes` entry typed `child.dataType`, which equals `Min/Max.dataType`.
+ // Neither is a `SizeBasedWindowFunction`, so no extra slot is prepended.
+ // `isMinMaxOnly` (enforced in WindowEvaluatorFactoryBase) ensures this invariant holds.
+ private[this] val sourceRow = new SpecificInternalRow(functions.map(_.dataType).toIndexedSeq)
+
+ // Each deque is addressed by its position in this array (one entry per Min/Max function),
+ // so no separate per-deque ordinal is needed.
+ private[this] val deques: Array[MinMaxDeque] = functions.map { func =>
+ val isMin = func.isInstanceOf[Min]
+ val child = func match {
+ case m: Min => m.child
+ case m: Max => m.child
+ }
+ new MinMaxDeque(
+ isMin,
+ BindReferences.bindReference(child, inputSchema),
+ TypeUtils.getInterpretedOrdering(child.dataType))
+ }
+
+ private def closeIterator(it: Iterator[UnsafeRow]): Unit = it match {
+ case c: Closeable => c.close()
+ case _ => // in-memory iterator: no-op
+ }
+
+ override def prepare(rows: ExternalAppendOnlyUnsafeRowArray): Unit = {
+ // Close prior-partition cursors before replacing them so spill readers are
+ // released at partition boundary, not deferred to task completion.
+ if (needsLowerRow && lowerIterator != null) {
+ closeIterator(lowerIterator)
+ lowerIterator = null
+ }
+ if (inputIterator != null) {
+ closeIterator(inputIterator)
+ inputIterator = null
+ }
+
+ numMonotonicDequeFrames.foreach(_ += 1)
+ input = rows
+ if (needsLowerRow) {
+ lowerIterator = input.generateIterator()
+ lowerRow = WindowFunctionFrame.getNextOrNull(lowerIterator)
+ } else {
+ lowerIterator = null
+ lowerRow = null
+ }
+ var di = 0
+ while (di < deques.length) { deques(di).clear(); di += 1 }
+ lowerBound = 0
+
+ inputIterator = input.generateIterator()
+ nextRow = WindowFunctionFrame.getNextOrNull(inputIterator)
+ upperBound = 0
+ }
+
+ override def write(index: Int, current: InternalRow): Unit = {
+ var bufferUpdated = index == 0
+
+ // Drop all rows from the buffer for which the input row value is smaller than
+ // the output row lower bound.
+ while (lowerBound < upperBound && lbound.compare(lowerRow, lowerBound, current, index) < 0) {
+ lowerBound += 1
+ if (needsLowerRow) lowerRow = WindowFunctionFrame.getNextOrNull(lowerIterator)
+ bufferUpdated = true
+ }
+
+ // Add all rows to the buffer for which the input row value is equal to or less than
+ // the output row upper bound.
+ while (nextRow != null && ubound.compare(nextRow, upperBound, current, index) <= 0) {
+ if (lbound.compare(nextRow, lowerBound, current, index) < 0) {
+ lowerBound += 1
+ if (needsLowerRow) lowerRow = WindowFunctionFrame.getNextOrNull(lowerIterator)
+ } else {
+ var di = 0
+ while (di < deques.length) { deques(di).admit(nextRow, upperBound); di += 1 }
+ bufferUpdated = true
+ }
+ nextRow = WindowFunctionFrame.getNextOrNull(inputIterator)
+ upperBound += 1
+ }
+
+ if (bufferUpdated) {
+ var di = 0
+ while (di < deques.length) { deques(di).dropBefore(lowerBound); di += 1 }
+ }
+
+ // Write output values to target.
+ // See sourceRow comment above for why evaluate(sourceRow, target) is safe here.
+ if (processor != null && bufferUpdated) {
+ var i = 0
+ while (i < deques.length) {
+ sourceRow.update(i, deques(i).currentValue())
+ i += 1
+ }
+ processor.evaluate(sourceRow, target)
+ }
+ }
+
+ override def currentLowerBound(): Int = lowerBound
+
+ override def currentUpperBound(): Int = upperBound
+
+ // Register close() to release any open spill readers when the task completes.
+ // Covers exceptional exits and cancellation where prepare() is never called again.
+ {
+ val tc = TaskContext.get()
+ if (tc != null) tc.addTaskCompletionListener[Unit](_ => close())
+ }
+
+ override def close(): Unit = {
+ if (needsLowerRow && lowerIterator != null) {
+ closeIterator(lowerIterator)
+ lowerIterator = null
+ }
+ if (inputIterator != null) {
+ closeIterator(inputIterator)
+ inputIterator = null
+ }
+ }
+
+ // MinMaxDeque fields are plain constructor params (not vals) since this is a private inner
+ // class and nothing outside reads them.
+ private class MinMaxDeque(
+ isMin: Boolean,
+ boundChild: Expression,
+ ordering: Ordering[Any]) {
+
+ private var capacity = 16
+ private var values = new Array[Any](capacity)
+ private var indices = new Array[Int](capacity)
+ private var head = 0
+ private var tail = 0
+ private var size = 0
+
+ def clear(): Unit = {
+ var i = 0
+ while (i < size) {
+ values((head + i) & (capacity - 1)) = null
+ i += 1
+ }
+ head = 0
+ tail = 0
+ size = 0
+ }
+
+ private def expand(): Unit = {
+ val newCapacity = capacity * 2
+ val newValues = new Array[Any](newCapacity)
+ val newIndices = new Array[Int](newCapacity)
+
+ var i = 0
+ while (i < size) {
+ val idx = (head + i) & (capacity - 1)
+ newValues(i) = values(idx)
+ newIndices(i) = indices(idx)
+ i += 1
+ }
+
+ values = newValues
+ indices = newIndices
+ head = 0
+ tail = size
+ capacity = newCapacity
+ }
+
+ private def isEmpty: Boolean = size == 0
+
+ private def peekLastValue(): Any = {
+ values((tail - 1) & (capacity - 1))
+ }
+
+ private def pollLast(): Unit = {
+ tail = (tail - 1) & (capacity - 1)
+ values(tail) = null
+ size -= 1
+ }
+
+ private def peekFirstIndex(): Int = {
+ indices(head)
+ }
+
+ private def pollFirst(): Unit = {
+ values(head) = null
+ head = (head + 1) & (capacity - 1)
+ size -= 1
+ }
+
+ private def offerLast(value: Any, index: Int): Unit = {
+ if (size == capacity) {
+ expand()
+ }
+ values(tail) = value
+ indices(tail) = index
+ tail = (tail + 1) & (capacity - 1)
+ size += 1
+ }
+
+ // ExtractWindowExpressions hoists window aggregate arguments into the Project below,
+ // so `boundChild` is a BoundReference over an UnsafeRow. `InternalRow.copyValue` explicitly
+ // copies UTF8String, BinaryView, InternalRow (Struct), ArrayData, and MapData.
+ // Binary (byte[]) and Decimal fall through to the identity case but are still safe:
+ // UnsafeRow.getBinary allocates a fresh byte[] (see Platform.copyMemory in UnsafeRow),
+ // and UnsafeRow.getDecimal constructs a detached Decimal from a Long or a new BigDecimal.
+ // Neither retains a pointer into the UnsafeRow's page memory, so the deque never holds a
+ // stale reference when the source UnsafeRow is recycled on spill.
+ private def evaluateAndCopy(row: InternalRow): Any = {
+ val value = boundChild.eval(row)
+ if (value == null) null else InternalRow.copyValue(value)
+ }
+
+ def admit(row: InternalRow, index: Int): Unit = {
+ val value = evaluateAndCopy(row)
+ if (value != null) {
+ // Use strict inequality (> for min-deque, < for max-deque) so that among equal-valued
+ // elements we retain the first one, matching the behavior of the naive and segment-tree
+ // paths (both of which keep the earliest equal value in the window).
+ if (isMin) {
+ while (!isEmpty && ordering.compare(peekLastValue(), value) > 0) {
+ pollLast()
+ }
+ } else {
+ while (!isEmpty && ordering.compare(peekLastValue(), value) < 0) {
+ pollLast()
+ }
+ }
+ offerLast(value, index)
+ }
+ }
+
+ def dropBefore(boundary: Int): Unit = {
+ while (!isEmpty && peekFirstIndex() < boundary) {
+ pollFirst()
+ }
+ }
+
+ def currentValue(): Any = {
+ if (isEmpty) {
+ null
+ } else {
+ values(head)
+ }
+ }
+ }
+}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/window/WindowEvaluatorFactory.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/window/WindowEvaluatorFactory.scala
index 8526f670b406..c804c2991abc 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/window/WindowEvaluatorFactory.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/window/WindowEvaluatorFactory.scala
@@ -32,12 +32,14 @@ class WindowEvaluatorFactory(
val childOutput: Seq[Attribute],
val spillSize: SQLMetric,
segmentTreeFrames: SQLMetric,
- segmentTreeFallbackFrames: SQLMetric)
+ segmentTreeFallbackFrames: SQLMetric,
+ monotonicDequeFrames: SQLMetric)
extends PartitionEvaluatorFactory[InternalRow, InternalRow] with WindowEvaluatorFactoryBase {
override def numSegmentTreeFrames: Option[SQLMetric] = Some(segmentTreeFrames)
override def numSegmentTreeFallbackFrames: Option[SQLMetric] =
Some(segmentTreeFallbackFrames)
+ override def numMonotonicDequeFrames: Option[SQLMetric] = Some(monotonicDequeFrames)
override def createEvaluator(): PartitionEvaluator[InternalRow, InternalRow] = {
new WindowPartitionEvaluator()
@@ -69,7 +71,7 @@ class WindowEvaluatorFactory(
val types = partitionSpec.map(_.dataType)
val ordering = InterpretedOrdering.forSchema(types)
(key1: UnsafeRow, key2: UnsafeRow) => ordering.compare(key1, key2) == 0
- }
+ }
// Manage the stream and the grouping.
var nextRow: UnsafeRow = null
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/window/WindowEvaluatorFactoryBase.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/window/WindowEvaluatorFactoryBase.scala
index 4a238a7ab1c8..0a10a3809c15 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/window/WindowEvaluatorFactoryBase.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/window/WindowEvaluatorFactoryBase.scala
@@ -23,7 +23,7 @@ import scala.collection.mutable.ArrayBuffer
import org.apache.spark.{SparkException, TaskContext}
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.{Add, AggregateWindowFunction, Ascending, Attribute, BoundReference, CurrentRow, DateAdd, DateAddYMInterval, DecimalAddNoOverflowCheck, Descending, Expression, ExtractANSIIntervalDays, FrameLessOffsetWindowFunction, FrameType, IdentityProjection, IntegerLiteral, MutableProjection, NamedExpression, OffsetWindowFunction, PythonFuncExpression, RangeFrame, RowFrame, RowOrdering, SortOrder, SpecifiedWindowFrame, TimestampAddInterval, TimestampA [...]
-import org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression, DeclarativeAggregate}
+import org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression, DeclarativeAggregate, Max, Min}
import org.apache.spark.sql.errors.QueryExecutionErrors
import org.apache.spark.sql.execution.metric.SQLMetric
import org.apache.spark.sql.internal.SQLConf
@@ -62,6 +62,13 @@ trait WindowEvaluatorFactoryBase {
}
}
+ /**
+ * Counter for [[SlidingWindowMinMaxFunctionFrame]] observability. Incremented each time a
+ * monotonic deque frame is prepared. Default `None` means the subclass does not integrate with
+ * the deque path.
+ */
+ def numMonotonicDequeFrames: Option[SQLMetric] = None
+
/**
* Create the resulting projection.
*
@@ -351,7 +358,26 @@ trait WindowEvaluatorFactoryBase {
// Moving Frame.
case ("AGGREGATE", frameType, lower, upper, _) =>
- if (eligibleForSegTree(functions, aggFilters, frameType, conf)) {
+ val isMinMaxOnly = functions.nonEmpty && functions.forall {
+ case _: Min => true
+ case _: Max => true
+ case _ => false
+ } && aggFilters.forall(_.isEmpty) &&
+ conf.getConf(SQLConf.WINDOW_MONOTONIC_DEQUE_ENABLED)
+ if (isMinMaxOnly) {
+ target: InternalRow => {
+ val lb = createBoundOrdering(frameType, lower, timeZone)
+ val ub = createBoundOrdering(frameType, upper, timeZone)
+ new SlidingWindowMinMaxFunctionFrame(
+ target,
+ processor,
+ lb,
+ ub,
+ functions,
+ childOutput,
+ numMonotonicDequeFrames)
+ }
+ } else if (eligibleForSegTree(functions, aggFilters, frameType, conf)) {
val segFns = functions.map(_.asInstanceOf[DeclarativeAggregate])
val cacheHint = estimateMaxCachedBlocks(lower, upper, frameType, blockSize)
target: InternalRow => {
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/window/WindowExec.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/window/WindowExec.scala
index e80565c96139..c5f784113058 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/window/WindowExec.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/window/WindowExec.scala
@@ -96,7 +96,9 @@ case class WindowExec(
SQLMetrics.createMetric(sparkContext, "number of segment-tree frames prepared"),
"numSegmentTreeFallbackFrames" ->
SQLMetrics.createMetric(sparkContext,
- "number of segment-tree fallback frames prepared")
+ "number of segment-tree fallback frames prepared"),
+ "numMonotonicDequeFrames" ->
+ SQLMetrics.createMetric(sparkContext, "number of monotonic-deque frames prepared")
)
protected override def doExecute(): RDD[InternalRow] = {
@@ -108,7 +110,8 @@ case class WindowExec(
child.output,
longMetric("spillSize"),
longMetric("numSegmentTreeFrames"),
- longMetric("numSegmentTreeFallbackFrames"))
+ longMetric("numSegmentTreeFallbackFrames"),
+ longMetric("numMonotonicDequeFrames"))
// Start processing.
if (conf.usePartitionEvaluator) {
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/benchmark/WindowBenchmark.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/benchmark/WindowBenchmark.scala
index 2284ff037e83..19455911d08a 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/benchmark/WindowBenchmark.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/benchmark/WindowBenchmark.scala
@@ -27,15 +27,18 @@ import org.apache.spark.sql.internal.SQLConf
* Benchmark for window functions with bounded ROWS frames.
*
* Matrix (see PR description for rationale):
- * - A: 5 aggregates x 3 cells (naive / segtree default / segtree bs=256) @ W=1001.
- * Per-case N so naive ~3-5s/iter; STDDEV_SAMP pinned @ N=2M (multi-buffer stress).
- * - B: SUM-over-INT, W sweep {10, 50, 201, 4001}; W=10/50 Pareto-loss stress,
+ * - A: 6 aggregates (MIN/MAX: naive/segtree-default/segtree-bs256/deque; others:
+ * naive/segtree-default/segtree-bs256) @ W=1001. Per-case N so naive ~3-5s/iter;
+ * STDDEV_SAMP pinned @ N=2M (multi-buffer stress).
+ * - B: SUM-over-INT, W sweep {W=11, W=51, W=201, W=4001}; W=11/W=51 Pareto-loss stress,
* W=4001 also runs bs=256.
* - F: Spill guard, 1M String x MAX x W=1001 (stress).
* - C: N-sweep {2M, 8M, 16M} segtree-only @ W=1001 (memory-pressure invariance).
+ * - G: MIN deque vs segtree, W=100001, 2M rows; Increasing/Decreasing/Random.
+ * - H: MIN/MAX deque vs naive/segtree; W=1/3/11, 2M rows (narrow-window coverage).
*
- * Compare Per Row(ns) column for O(log W) scaling. Dev smoke via positional
- * mainArgs: (0)=rowCount, (1)=halfWindow (default 100); do NOT combine with
+ * Compare Per Row(ns) column for O(log W) scaling. Dev smoke via positional mainArgs:
+ * (0)=rowCount, (1)=halfWindow (default 100); do NOT combine with
* SPARK_GENERATE_BENCHMARK_FILES=1.
*/
object WindowBenchmark extends SqlBasedBenchmark {
@@ -45,9 +48,9 @@ object WindowBenchmark extends SqlBasedBenchmark {
private val A_N_AVG: Long = 192L * 1024 // AVG @ W=1001
private val A_N_STDDEV: Long = 2L * 1000L * 1000L // STDDEV stress
- // Section B: W-sweep (W=10/50 stress: Pareto loss zone; W=4001 stress: O(W) cliff).
- private val B_N_W10: Long = 2L * 1000L * 1000L
- private val B_N_W50: Long = 2L * 1000L * 1000L
+ // Section B: W-sweep (W=11/51 stress: Pareto loss zone; W=4001 stress: O(W) cliff).
+ private val B_N_W11: Long = 2L * 1000L * 1000L
+ private val B_N_W51: Long = 2L * 1000L * 1000L
private val B_N_W201: Long = 1L * 1000L * 1000L
private val B_N_W4001: Long = 2L * 1000L * 1000L
@@ -62,6 +65,13 @@ object WindowBenchmark extends SqlBasedBenchmark {
private val MAIN_HALF_W: Int = 500 // Section A W=1001
+ // Section G: MIN deque vs segtree, 2M rows x W=100001.
+ private val G_N: Long = 2L * 1000L * 1000L
+ private val G_HALF_W: Int = 50000 // W=100001
+
+ // Section H: narrow-window scaling; 2M rows x W=1/3/11.
+ private val H_N: Long = 2L * 1000L * 1000L
+
private val ITERS_NORMAL: Int = 5
private val ITERS_STRESS: Int = 3
@@ -144,43 +154,89 @@ object WindowBenchmark extends SqlBasedBenchmark {
}
def runSectionA(
- aggFn: String, iters: Int, rows: Long, halfW: Int, stressMark: String): Unit = {
+ aggFn: String,
+ iters: Int,
+ rows: Long,
+ halfW: Int,
+ stressMark: String,
+ labelSuffix: String = "",
+ withSegBs: Boolean = true): Unit = {
val frame = frameFor(halfW)
- val dNaive = digest(aggFn, frame)
- val dSeg = digest(aggFn, frame, SQLConf.WINDOW_SEGMENT_TREE_ENABLED.key -> "true")
- val dSegBs = digest(aggFn, frame,
- SQLConf.WINDOW_SEGMENT_TREE_ENABLED.key -> "true",
- SQLConf.WINDOW_SEGMENT_TREE_BLOCK_SIZE.key -> "256")
- require(dNaive == dSeg,
- s"$aggFn segtree digest mismatch: naive=$dNaive seg=$dSeg")
- require(dNaive == dSegBs,
- s"$aggFn segtree (bs=256) digest mismatch: naive=$dNaive seg=$dSegBs")
+ val dNaive = digest(aggFn, frame, SQLConf.WINDOW_MONOTONIC_DEQUE_ENABLED.key -> "false")
+ val dSeg = digest(
+ aggFn,
+ frame,
+ SQLConf.WINDOW_MONOTONIC_DEQUE_ENABLED.key -> "false",
+ SQLConf.WINDOW_SEGMENT_TREE_ENABLED.key -> "true")
+ require(dNaive == dSeg, s"$aggFn segtree digest mismatch: naive=$dNaive seg=$dSeg")
+ val isMinOrMax = aggFn == "MIN" || aggFn == "MAX"
+ if (isMinOrMax) {
+ val dMonotonic =
+ digest(aggFn, frame, SQLConf.WINDOW_MONOTONIC_DEQUE_ENABLED.key -> "true")
+ require(dNaive == dMonotonic,
+ s"$aggFn monotonic deque digest mismatch: naive=$dNaive monotonic=$dMonotonic")
+ }
+
+ // Section A (W=1001) also verifies non-default block size at this window width.
+ if (withSegBs) {
+ val dSegBs = digest(
+ aggFn,
+ frame,
+ SQLConf.WINDOW_MONOTONIC_DEQUE_ENABLED.key -> "false",
+ SQLConf.WINDOW_SEGMENT_TREE_ENABLED.key -> "true",
+ SQLConf.WINDOW_SEGMENT_TREE_BLOCK_SIZE.key -> "256")
+ require(dNaive == dSegBs,
+ s"$aggFn segtree bs=256 digest mismatch: naive=$dNaive segBs=$dSegBs")
+ }
val W = 2 * halfW + 1
val benchmark = new Benchmark(
s"$aggFn sliding window, W=$W, ${rowsLabel(rows)} rows$stressMark",
rows, output = output)
- val nNaive = s"$aggFn naive (current, baseline)"
- val nSeg = s"$aggFn segtree (default)"
- val nSegBs = s"$aggFn segtree (blockSize=256)"
- allCaseNames ++= Seq(nNaive, nSeg, nSegBs)
+ // Use labelSuffix to disambiguate case names when runSectionA is called from
+ // multiple sections (e.g., Section A and Section H) to avoid duplicate entries
+ // in allCaseNames and incorrect metric merging in the Memory/Spill trailer.
+ val nNaive = s"$aggFn naive (current, baseline)$labelSuffix"
+ val nSeg = s"$aggFn segtree (default)$labelSuffix"
+ allCaseNames ++= Seq(nNaive, nSeg)
+ val nSegBs = s"$aggFn segtree (blockSize=256)$labelSuffix"
+ if (withSegBs) {
+ allCaseNames += nSegBs
+ }
benchmark.addCase(nNaive, numIters = iters) { _ =>
currentCase = nNaive
- spark.sql(s"SELECT $aggFn(v) $frame FROM t").noop()
+ withSQLConf(SQLConf.WINDOW_MONOTONIC_DEQUE_ENABLED.key -> "false") {
+ spark.sql(s"SELECT $aggFn(v) $frame FROM t").noop()
+ }
}
benchmark.addCase(nSeg, numIters = iters) { _ =>
currentCase = nSeg
- withSQLConf(SQLConf.WINDOW_SEGMENT_TREE_ENABLED.key -> "true") {
+ withSQLConf(
+ SQLConf.WINDOW_MONOTONIC_DEQUE_ENABLED.key -> "false",
+ SQLConf.WINDOW_SEGMENT_TREE_ENABLED.key -> "true") {
spark.sql(s"SELECT $aggFn(v) $frame FROM t").noop()
}
}
- benchmark.addCase(nSegBs, numIters = iters) { _ =>
- currentCase = nSegBs
- withSQLConf(
- SQLConf.WINDOW_SEGMENT_TREE_ENABLED.key -> "true",
- SQLConf.WINDOW_SEGMENT_TREE_BLOCK_SIZE.key -> "256") {
- spark.sql(s"SELECT $aggFn(v) $frame FROM t").noop()
+ if (withSegBs) {
+ benchmark.addCase(nSegBs, numIters = iters) { _ =>
+ currentCase = nSegBs
+ withSQLConf(
+ SQLConf.WINDOW_MONOTONIC_DEQUE_ENABLED.key -> "false",
+ SQLConf.WINDOW_SEGMENT_TREE_ENABLED.key -> "true",
+ SQLConf.WINDOW_SEGMENT_TREE_BLOCK_SIZE.key -> "256") {
+ spark.sql(s"SELECT $aggFn(v) $frame FROM t").noop()
+ }
+ }
+ }
+ if (isMinOrMax) {
+ val nMonotonic = s"$aggFn monotonic deque (new)$labelSuffix"
+ allCaseNames += nMonotonic
+ benchmark.addCase(nMonotonic, numIters = iters) { _ =>
+ currentCase = nMonotonic
+ withSQLConf(SQLConf.WINDOW_MONOTONIC_DEQUE_ENABLED.key -> "true") {
+ spark.sql(s"SELECT $aggFn(v) $frame FROM t").noop()
+ }
}
}
benchmark.run()
@@ -202,16 +258,22 @@ object WindowBenchmark extends SqlBasedBenchmark {
allCaseNames ++= Seq(nNaive, nSeg)
benchmark.addCase(nNaive, numIters = iters) { _ =>
currentCase = nNaive
- spark.sql(s"SELECT $aggFn(v) $frame FROM t").noop()
+ withSQLConf(SQLConf.WINDOW_MONOTONIC_DEQUE_ENABLED.key -> "false") {
+ spark.sql(s"SELECT $aggFn(v) $frame FROM t").noop()
+ }
}
benchmark.addCase(nSeg, numIters = iters) { _ =>
currentCase = nSeg
- withSQLConf(SQLConf.WINDOW_SEGMENT_TREE_ENABLED.key -> "true") {
+ withSQLConf(
+ SQLConf.WINDOW_MONOTONIC_DEQUE_ENABLED.key -> "false",
+ SQLConf.WINDOW_SEGMENT_TREE_ENABLED.key -> "true") {
spark.sql(s"SELECT $aggFn(v) $frame FROM t").noop()
}
}
if (stressBs) {
- val dSegBs = digest(aggFn, frame,
+ val dSegBs = digest(
+ aggFn,
+ frame,
SQLConf.WINDOW_SEGMENT_TREE_ENABLED.key -> "true",
SQLConf.WINDOW_SEGMENT_TREE_BLOCK_SIZE.key -> "256")
require(dNaive == dSegBs,
@@ -242,11 +304,15 @@ object WindowBenchmark extends SqlBasedBenchmark {
allCaseNames ++= Seq(nNaive, nSeg)
benchmark.addCase(nNaive, numIters = ITERS_STRESS) { _ =>
currentCase = nNaive
- spark.sql(s"SELECT MAX(v) $frame FROM t").noop()
+ withSQLConf(SQLConf.WINDOW_MONOTONIC_DEQUE_ENABLED.key -> "false") {
+ spark.sql(s"SELECT MAX(v) $frame FROM t").noop()
+ }
}
benchmark.addCase(nSeg, numIters = ITERS_STRESS) { _ =>
currentCase = nSeg
- withSQLConf(SQLConf.WINDOW_SEGMENT_TREE_ENABLED.key -> "true") {
+ withSQLConf(
+ SQLConf.WINDOW_MONOTONIC_DEQUE_ENABLED.key -> "false",
+ SQLConf.WINDOW_SEGMENT_TREE_ENABLED.key -> "true") {
spark.sql(s"SELECT MAX(v) $frame FROM t").noop()
}
}
@@ -274,15 +340,69 @@ object WindowBenchmark extends SqlBasedBenchmark {
benchmark.run()
}
+ def setupIncreasingTable(n: Long): Unit = {
+ spark
+ .range(n)
+ .selectExpr("id", "cast(id as int) as v")
+ .coalesce(1)
+ .createOrReplaceTempView("t")
+ }
+
+ def setupDecreasingTable(n: Long): Unit = {
+ spark
+ .range(n)
+ .selectExpr("id", s"cast(($n - id) as int) as v")
+ .coalesce(1)
+ .createOrReplaceTempView("t")
+ }
+
+ def runSectionG(aggFn: String, iters: Int, rows: Long, halfW: Int, pattern: String): Unit = {
+ val frame = frameFor(halfW)
+ val dSeg = digest(
+ aggFn,
+ frame,
+ SQLConf.WINDOW_MONOTONIC_DEQUE_ENABLED.key -> "false",
+ SQLConf.WINDOW_SEGMENT_TREE_ENABLED.key -> "true")
+ val dMonotonic = digest(aggFn, frame, SQLConf.WINDOW_MONOTONIC_DEQUE_ENABLED.key -> "true")
+ require(dSeg == dMonotonic,
+ s"$aggFn $pattern digest mismatch: seg=$dSeg monotonic=$dMonotonic")
+
+ val W = 2 * halfW + 1
+ val benchmark = new Benchmark(
+ s"$aggFn sliding window ($pattern), W=$W, ${rowsLabel(rows)} rows",
+ rows, output = output)
+ val nSeg = s"$aggFn segtree ($pattern)"
+ val nMonotonic = s"$aggFn monotonic deque ($pattern)"
+ allCaseNames ++= Seq(nSeg, nMonotonic)
+
+ benchmark.addCase(nSeg, numIters = iters) { _ =>
+ currentCase = nSeg
+ withSQLConf(
+ SQLConf.WINDOW_MONOTONIC_DEQUE_ENABLED.key -> "false",
+ SQLConf.WINDOW_SEGMENT_TREE_ENABLED.key -> "true") {
+ spark.sql(s"SELECT $aggFn(v) $frame FROM t").noop()
+ }
+ }
+ benchmark.addCase(nMonotonic, numIters = iters) { _ =>
+ currentCase = nMonotonic
+ withSQLConf(SQLConf.WINDOW_MONOTONIC_DEQUE_ENABLED.key -> "true") {
+ spark.sql(s"SELECT $aggFn(v) $frame FROM t").noop()
+ }
+ }
+ benchmark.run()
+ }
+
try {
if (smokeMode) {
setupIntTable(smokeRowCount)
runBenchmark("SMOKE: Section A MIN") {
runSectionA("MIN", ITERS_STRESS, smokeRowCount, smokeHalfW, "")
}
+ runBenchmark("SMOKE: Section A MAX") {
+ runSectionA("MAX", ITERS_STRESS, smokeRowCount, smokeHalfW, "")
+ }
runBenchmark("SMOKE: Section B SUM W sweep point") {
- runSectionB(
- smokeHalfW, stressBs = smokeHalfW >= 2000, smokeRowCount, ITERS_STRESS, "")
+ runSectionB(smokeHalfW, stressBs = smokeHalfW >= 2000, smokeRowCount, ITERS_STRESS, "")
}
} else {
setupIntTable(A_N_INT)
@@ -309,13 +429,13 @@ object WindowBenchmark extends SqlBasedBenchmark {
runSectionA("STDDEV_SAMP", ITERS_STRESS, A_N_STDDEV, MAIN_HALF_W, " (stress)")
}
- setupIntTable(B_N_W10)
- runBenchmark("Section B - W=10 scaling (stress: Pareto loss zone)") {
- runSectionB(5, stressBs = false, B_N_W10, ITERS_STRESS, " (stress)")
+ setupIntTable(B_N_W11)
+ runBenchmark("Section B - W=11 scaling (stress: Pareto loss zone)") {
+ runSectionB(5, stressBs = false, B_N_W11, ITERS_STRESS, " (stress)")
}
- setupIntTable(B_N_W50)
- runBenchmark("Section B - W=50 scaling (stress: Pareto loss zone)") {
- runSectionB(25, stressBs = false, B_N_W50, ITERS_STRESS, " (stress)")
+ setupIntTable(B_N_W51)
+ runBenchmark("Section B - W=51 scaling (stress: Pareto loss zone)") {
+ runSectionB(25, stressBs = false, B_N_W51, ITERS_STRESS, " (stress)")
}
setupIntTable(B_N_W201)
runBenchmark("Section B - W=201 scaling") {
@@ -343,6 +463,45 @@ object WindowBenchmark extends SqlBasedBenchmark {
runBenchmark("Section C - N-sweep large (stress)") {
runSectionC(C_N_LARGE)
}
+
+ setupIncreasingTable(G_N)
+ runBenchmark("Section G - MIN Monotonic Deque vs Segment Tree (Worst-Case: Increasing)") {
+ runSectionG("MIN", ITERS_NORMAL, G_N, G_HALF_W, "Increasing")
+ }
+
+ setupDecreasingTable(G_N)
+ runBenchmark("Section G - MIN Monotonic Deque vs Segment Tree (Best-Case: Decreasing)") {
+ runSectionG("MIN", ITERS_NORMAL, G_N, G_HALF_W, "Decreasing")
+ }
+
+ setupIntTable(G_N)
+ runBenchmark("Section G - MIN Monotonic Deque vs Segment Tree (Random)") {
+ runSectionG("MIN", ITERS_NORMAL, G_N, G_HALF_W, "Random")
+ }
+
+ // Section H: narrow-window scaling. W=1/3/11 covers the full range where the
+ // naive path's codegen'd per-row scan may compete with the deque's interpreted
+ // BoundReference.eval + Ordering[Any].compare. Results show whether a width
+ // gate is needed before enabling the conf by default.
+ setupIntTable(H_N)
+ runBenchmark("Section H - MIN W=1 scaling (2M rows)") {
+ runSectionA("MIN", ITERS_STRESS, H_N, 0, "", " [H W=1]", withSegBs = false)
+ }
+ runBenchmark("Section H - MAX W=1 scaling (2M rows)") {
+ runSectionA("MAX", ITERS_STRESS, H_N, 0, "", " [H W=1]", withSegBs = false)
+ }
+ runBenchmark("Section H - MIN W=3 scaling (2M rows)") {
+ runSectionA("MIN", ITERS_STRESS, H_N, 1, "", " [H W=3]", withSegBs = false)
+ }
+ runBenchmark("Section H - MAX W=3 scaling (2M rows)") {
+ runSectionA("MAX", ITERS_STRESS, H_N, 1, "", " [H W=3]", withSegBs = false)
+ }
+ runBenchmark("Section H - MIN W=11 scaling (stress, 2M rows)") {
+ runSectionA("MIN", ITERS_STRESS, H_N, 5, " (stress)", " [H W=11]", withSegBs = false)
+ }
+ runBenchmark("Section H - MAX W=11 scaling (stress, 2M rows)") {
+ runSectionA("MAX", ITERS_STRESS, H_N, 5, " (stress)", " [H W=11]", withSegBs = false)
+ }
}
// Drain listener before reading metrics.
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/window/MonotonicDequeWindowFunctionSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/window/MonotonicDequeWindowFunctionSuite.scala
new file mode 100644
index 000000000000..490675839981
--- /dev/null
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/window/MonotonicDequeWindowFunctionSuite.scala
@@ -0,0 +1,434 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.window
+
+import org.apache.spark.sql.{DataFrame, QueryTest, Row}
+import org.apache.spark.sql.expressions.Window
+import org.apache.spark.sql.functions._
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+
+
+/**
+ * Correctness tests verifying the monotonic deque-based sliding window frame optimization. Runs
+ * differential testing to ensure equivalence between:
+ * 1. Monotonic Deque (Enabled)
+ * 2. Segment Tree (Deque disabled, SegTree enabled)
+ * 3. Naive Baseline (Both disabled)
+ */
+class MonotonicDequeWindowFunctionSuite extends QueryTest with SharedSparkSession {
+
+ import testImplicits._
+
+ // Disable AQE so executedPlan.collect can descend into WindowExec without being
+ // blocked by AdaptiveSparkPlanExec (a LeafExecNode). This matches SegmentTreeWindowMetricsSuite.
+ private val enableDeque: Map[String, String] = Map(
+ SQLConf.WINDOW_MONOTONIC_DEQUE_ENABLED.key -> "true",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false")
+
+ private val disableDequeSegTree: Map[String, String] = Map(
+ SQLConf.WINDOW_MONOTONIC_DEQUE_ENABLED.key -> "false",
+ SQLConf.WINDOW_SEGMENT_TREE_ENABLED.key -> "true",
+ SQLConf.WINDOW_SEGMENT_TREE_MIN_PARTITION_ROWS.key -> "1")
+
+ private val disableDequeNaive: Map[String, String] = Map(
+ SQLConf.WINDOW_MONOTONIC_DEQUE_ENABLED.key -> "false",
+ SQLConf.WINDOW_SEGMENT_TREE_ENABLED.key -> "false")
+
+ /** Build `df` thrice (Deque, SegTree, Naive) and assert equal results. */
+ private def checkEquivalence(
+ build: () => DataFrame,
+ expectDeque: Boolean = true,
+ expectedDequeCount: Int = 3): Unit = {
+ val naiveResult: Seq[Row] = withSQLConf(disableDequeNaive.toSeq: _*) {
+ build().collect().toSeq
+ }
+ val segTreeResult: Seq[Row] = withSQLConf(disableDequeSegTree.toSeq: _*) {
+ build().collect().toSeq
+ }
+ val dequeResult: Seq[Row] = withSQLConf(enableDeque.toSeq: _*) {
+ val df = build()
+ val res = df.collect().toSeq
+
+ // Verify routing actually hit (or didn't hit) the deque.
+ // Use the registered metric key "numMonotonicDequeFrames" (not the display name).
+ val windowNodes = df.queryExecution.executedPlan.collect {
+ case w: WindowExec => w
+ }
+ assert(windowNodes.nonEmpty, "No WindowExec found in the query plan")
+ val dequeCount =
+ windowNodes.flatMap(_.metrics.get("numMonotonicDequeFrames").map(_.value)).sum
+
+ if (expectDeque) {
+ assert(dequeCount > 0, "Monotonic deque was enabled but no frames were routed to it")
+ if (expectedDequeCount >= 0) {
+ assert(dequeCount == expectedDequeCount,
+ s"Expected $expectedDequeCount deque frames, got $dequeCount")
+ }
+ } else {
+ assert(dequeCount == 0, "Monotonic deque was used but expected to fallback")
+ }
+ res
+ }
+
+ QueryTest.sameRows(naiveResult, dequeResult, isSorted = false).foreach { err =>
+ fail(s"Monotonic Deque output differs from Naive baseline.\n$err")
+ }
+ QueryTest.sameRows(segTreeResult, dequeResult, isSorted = false).foreach { err =>
+ fail(s"Monotonic Deque output differs from Segment Tree baseline.\n$err")
+ }
+ }
+
+ private def baseDF: DataFrame = {
+ spark
+ .range(0, 100)
+ .selectExpr(
+ "id",
+ "(id % 3) AS pk",
+ "CAST(id AS INT) AS v_int",
+ "CAST(id AS LONG) AS v_long",
+ "CAST(id AS DOUBLE) AS v_double",
+ "CAST(id AS STRING) AS v_str")
+ }
+
+ test("SPARK-58201: Moving rows frame: MIN/MAX on primitives (Int/Long/Double)") {
+ val winSpec = Window.partitionBy($"pk").orderBy($"id").rowsBetween(-3, 2)
+ checkEquivalence(() =>
+ baseDF.select(
+ $"id",
+ min($"v_int").over(winSpec),
+ max($"v_int").over(winSpec),
+ min($"v_long").over(winSpec),
+ max($"v_long").over(winSpec),
+ min($"v_double").over(winSpec),
+ max($"v_double").over(winSpec)))
+ }
+
+ test("SPARK-58201: Fallback for mixed aggregates (SUM + MIN/MAX)") {
+ val winSpec = Window.partitionBy($"pk").orderBy($"id").rowsBetween(-4, 2)
+ checkEquivalence(() =>
+ baseDF.select(
+ $"id",
+ min($"v_int").over(winSpec),
+ sum($"v_int").over(winSpec)),
+ expectDeque = false)
+ }
+
+ test("SPARK-58201: Fallback for FILTER clauses") {
+ val df = spark.sql("""SELECT id,
+ | MIN(id) FILTER (WHERE id % 2 = 0) OVER (
+ | PARTITION BY (id % 3) ORDER BY id ROWS BETWEEN 2 PRECEDING AND 2 FOLLOWING
+ | ) AS v
+ |FROM RANGE(0, 20)""".stripMargin)
+ // Deque shouldn't be used since FILTER is not supported. We can't use checkEquivalence
+ // because checkEquivalence builds DF inside, so we'll just check metrics.
+ withSQLConf(
+ SQLConf.WINDOW_MONOTONIC_DEQUE_ENABLED.key -> "true",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ val res = df.collect()
+ val windowNodes = df.queryExecution.executedPlan.collect {
+ case w: WindowExec => w
+ }
+ val dequeCount =
+ windowNodes.flatMap(_.metrics.get("numMonotonicDequeFrames").map(_.value)).sum
+ assert(dequeCount == 0, "Monotonic deque was used for FILTER clause")
+ }
+ }
+
+ // Deque takes precedence over segment tree when both flags are enabled.
+ // baseDF has pk in {0, 1, 2} => 3 window partitions => exactly 3 deque frames,
+ // pinning the one-frame-per-partition metric contract.
+ test("SPARK-58201: deque takes precedence over seg-tree; metric pins frame-per-partition") {
+ withSQLConf(
+ SQLConf.WINDOW_MONOTONIC_DEQUE_ENABLED.key -> "true",
+ SQLConf.WINDOW_SEGMENT_TREE_ENABLED.key -> "true",
+ SQLConf.WINDOW_SEGMENT_TREE_MIN_PARTITION_ROWS.key -> "1",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ val winSpec = Window.partitionBy($"pk").orderBy($"id").rowsBetween(-3, 2)
+ val df = baseDF.select($"id", min($"v_int").over(winSpec))
+ val naiveResult = withSQLConf(
+ SQLConf.WINDOW_MONOTONIC_DEQUE_ENABLED.key -> "false",
+ SQLConf.WINDOW_SEGMENT_TREE_ENABLED.key -> "false") {
+ baseDF.select($"id", min($"v_int").over(winSpec)).collect().toSeq
+ }
+ QueryTest.sameRows(naiveResult, df.collect().toSeq, isSorted = false).foreach { err =>
+ fail(s"Deque output differs from naive baseline.\n$err")
+ }
+ val windowNodes = df.queryExecution.executedPlan.collect { case w: WindowExec => w }
+ assert(windowNodes.nonEmpty, "No WindowExec found in plan")
+ val dequeCount =
+ windowNodes.flatMap(_.metrics.get("numMonotonicDequeFrames").map(_.value)).sum
+ val segCount =
+ windowNodes.flatMap(_.metrics.get("numSegmentTreeFrames").map(_.value)).sum
+ // One SlidingWindowMinMaxFunctionFrame per window partition (pk=0, pk=1, pk=2).
+ assert(dequeCount == 3,
+ s"Expected 3 deque frames (one per window partition), got $dequeCount")
+ assert(segCount == 0,
+ s"Deque should take precedence over seg-tree; got segCount=$segCount")
+ }
+ }
+
+ test("SPARK-58201: Moving rows frame: MIN/MAX on reference types (String)") {
+ val winSpec = Window.partitionBy($"pk").orderBy($"id").rowsBetween(-2, 3)
+ checkEquivalence(() =>
+ baseDF.select($"id", min($"v_str").over(winSpec), max($"v_str").over(winSpec)))
+ }
+
+ test("SPARK-58201: MIN/MAX on Date and Timestamp types") {
+ val df = baseDF.selectExpr(
+ "id",
+ "pk",
+ "CAST(id * 24 * 3600 AS TIMESTAMP) AS v_ts",
+ "date_add(to_date('1970-01-01'), CAST(id AS INT)) AS v_date")
+ val winSpec = Window.partitionBy($"pk").orderBy($"id").rowsBetween(-2, 2)
+ checkEquivalence(() =>
+ df.select(
+ $"id",
+ min($"v_ts").over(winSpec),
+ max($"v_ts").over(winSpec),
+ min($"v_date").over(winSpec),
+ max($"v_date").over(winSpec)))
+ }
+
+ test("SPARK-58201: MIN/MAX on Interval types (YearMonthIntervalType and DayTimeIntervalType)") {
+ val df = baseDF.selectExpr(
+ "id",
+ "pk",
+ "make_ym_interval(0, CAST(id AS INT)) AS v_ym",
+ "make_dt_interval(CAST(id AS INT), 0, 0, 0) AS v_dt")
+ val winSpec = Window.partitionBy($"pk").orderBy($"id").rowsBetween(-3, 3)
+ checkEquivalence(() =>
+ df.select(
+ $"id",
+ min($"v_ym").over(winSpec),
+ max($"v_ym").over(winSpec),
+ min($"v_dt").over(winSpec),
+ max($"v_dt").over(winSpec)))
+ }
+
+ test("SPARK-58201: MIN/MAX with null values in partition") {
+ val df = spark
+ .range(0, 50)
+ .selectExpr("id", "(id % 2) AS pk", "IF(id % 5 == 0, null, CAST(id AS INT)) AS v")
+ val winSpec = Window.partitionBy($"pk").orderBy($"id").rowsBetween(-2, 2)
+ checkEquivalence(
+ () => df.select($"id", min($"v").over(winSpec), max($"v").over(winSpec)),
+ expectedDequeCount = 2) // pk = id % 2 => 2 partitions
+ }
+
+ test("SPARK-58201: MIN/MAX on all-null partition") {
+ val df = spark.range(0, 20).selectExpr("id", "1 AS pk", "CAST(null AS INT) AS v")
+ val winSpec = Window.partitionBy($"pk").orderBy($"id").rowsBetween(-2, 2)
+ checkEquivalence(
+ () => df.select($"id", min($"v").over(winSpec), max($"v").over(winSpec)),
+ expectedDequeCount = 1)
+ }
+
+ test("SPARK-58201: Range-based moving frame: MIN/MAX on primitive types") {
+ val df = baseDF.selectExpr("id", "pk", "CAST(id / 2 AS INT) AS ord_val", "v_int")
+ val winSpec = Window.partitionBy($"pk").orderBy($"ord_val").rangeBetween(-2, 2)
+ checkEquivalence(() =>
+ df.select($"id", min($"v_int").over(winSpec), max($"v_int").over(winSpec)))
+ }
+
+ // Verify strict inequality preserves first-of-equals behavior
+ // under collated strings and signed zero.
+
+ test("SPARK-58201: MIN/MAX on collated strings (UTF8_LCASE) preserves first-of-equals") {
+ // Under UTF8_LCASE, 'Bob' and 'bob' compare equal. MIN must keep the
+ // first occurrence (lowest index), matching naive/segment-tree semantics.
+ val df = spark.sql("""SELECT id, 1 AS pk,
+ | CASE WHEN id = 0 THEN COLLATE('Bob', 'UTF8_LCASE')
+ | WHEN id = 1 THEN COLLATE('bob', 'UTF8_LCASE')
+ | WHEN id = 2 THEN COLLATE('alice', 'UTF8_LCASE')
+ | WHEN id = 3 THEN COLLATE('BOB', 'UTF8_LCASE')
+ | ELSE COLLATE(CAST(id AS STRING), 'UTF8_LCASE')
+ | END AS v
+ |FROM RANGE(0, 20)""".stripMargin)
+ val winSpec = Window.partitionBy($"pk").orderBy($"id").rowsBetween(-2, 2)
+ checkEquivalence(
+ () => df.select($"id", min($"v").over(winSpec), max($"v").over(winSpec)),
+ expectedDequeCount = 1) // 1 AS pk => single partition
+ }
+
+ test("SPARK-58201: MIN/MAX on Double with signed zero (+0.0 / -0.0)") {
+ // SQLOrderingUtil.compareDoubles treats -0.0 == +0.0, so they compare
+ // equal. MIN must keep the first occurrence, matching naive semantics.
+ val df = spark
+ .range(0, 20)
+ .selectExpr(
+ "id",
+ "1 AS pk",
+ """CASE
+ WHEN id % 4 = 0 THEN CAST('-0.0' AS DOUBLE)
+ WHEN id % 4 = 1 THEN CAST(0.0 AS DOUBLE)
+ WHEN id % 4 = 2 THEN CAST(id AS DOUBLE)
+ ELSE CAST(-id AS DOUBLE)
+ END AS v""")
+ val winSpec = Window.partitionBy($"pk").orderBy($"id").rowsBetween(-2, 2)
+ checkEquivalence(
+ () => df.select($"id", min($"v").over(winSpec), max($"v").over(winSpec)),
+ expectedDequeCount = 1) // 1 AS pk => single partition
+ }
+
+ // Spill coverage: lower thresholds to force ExternalAppendOnlyUnsafeRowArray
+ // to use its SpillableArrayIterator, which recycles a single UnsafeRow.
+
+ test("SPARK-58201: Moving rows frame: MIN/MAX on reference types (String) with spill") {
+ withSQLConf(
+ SQLConf.WINDOW_EXEC_BUFFER_IN_MEMORY_THRESHOLD.key -> "8",
+ SQLConf.WINDOW_EXEC_BUFFER_SPILL_THRESHOLD.key -> "16") {
+ val winSpec = Window.partitionBy($"pk").orderBy($"id").rowsBetween(-2, 3)
+ checkEquivalence(() =>
+ baseDF.select($"id", min($"v_str").over(winSpec), max($"v_str").over(winSpec)))
+ }
+ }
+
+ test("SPARK-58201: MIN/MAX on ArrayType with spill") {
+ val df = spark
+ .range(0, 60)
+ .selectExpr("id", "(id % 3) AS pk", "array(CAST(id AS INT), CAST(id * 2 AS INT)) AS v")
+ withSQLConf(
+ SQLConf.WINDOW_EXEC_BUFFER_IN_MEMORY_THRESHOLD.key -> "8",
+ SQLConf.WINDOW_EXEC_BUFFER_SPILL_THRESHOLD.key -> "16") {
+ val winSpec = Window.partitionBy($"pk").orderBy($"id").rowsBetween(-2, 2)
+ checkEquivalence(() => df.select($"id", min($"v").over(winSpec), max($"v").over(winSpec)))
+ }
+ }
+
+ test("SPARK-58201: MIN/MAX on StructType with spill") {
+ val df = spark
+ .range(0, 60)
+ .selectExpr(
+ "id",
+ "(id % 3) AS pk",
+ "named_struct('a', CAST(id AS INT), 'b', CAST(id * 3 AS INT)) AS v")
+ withSQLConf(
+ SQLConf.WINDOW_EXEC_BUFFER_IN_MEMORY_THRESHOLD.key -> "8",
+ SQLConf.WINDOW_EXEC_BUFFER_SPILL_THRESHOLD.key -> "16") {
+ val winSpec = Window.partitionBy($"pk").orderBy($"id").rowsBetween(-2, 2)
+ checkEquivalence(() => df.select($"id", min($"v").over(winSpec), max($"v").over(winSpec)))
+ }
+ }
+
+ // Both-FOLLOWING frame: lowerBound-advances-without-admitting branch
+ // (both bounds on the FOLLOWING side).
+ test("SPARK-58201: both-FOLLOWING rows frame exercises lowerBound-advances branch") {
+ val winSpec = Window.partitionBy($"pk").orderBy($"id").rowsBetween(2, 4)
+ checkEquivalence(() =>
+ baseDF.select($"id", min($"v_int").over(winSpec), max($"v_int").over(winSpec)))
+ }
+
+ // Wide ascending data forces MinMaxDeque.expand() (ring-buffer grow path).
+ test("SPARK-58201: wide window on ascending data forces ring-buffer expand") {
+ val df = spark.range(0, 300).selectExpr("id", "1 AS pk", "CAST(id AS INT) AS v")
+ val winSpec = Window.partitionBy($"pk").orderBy($"id").rowsBetween(-70, 0)
+ checkEquivalence(
+ () => df.select($"id", min($"v").over(winSpec), max($"v").over(winSpec)),
+ expectedDequeCount = 1)
+ }
+
+ // Both-PRECEDING frame: first rows have an empty window.
+ test("SPARK-58201: both-PRECEDING rows frame (empty window for first rows)") {
+ val winSpec = Window.partitionBy($"pk").orderBy($"id").rowsBetween(-4, -2)
+ checkEquivalence(() =>
+ baseDF.select($"id", min($"v_int").over(winSpec), max($"v_int").over(winSpec)))
+ }
+
+ // Wide random data: exercises non-monotonic admission/eviction on the sliding path.
+ // rand(42) produces non-monotone values so both admit and evict paths are exercised
+ // at each step, unlike the strictly-increasing default data.
+ test("SPARK-58201: wide random rows frame") {
+ val df = spark.range(0, 300)
+ .selectExpr("id", "(id % 3) AS pk", "CAST(rand(42) * 10000 AS INT) AS v_int")
+ val winSpec = Window.partitionBy($"pk").orderBy($"id").rowsBetween(-60, 40)
+ checkEquivalence(() =>
+ df.select($"id", min($"v_int").over(winSpec), max($"v_int").over(winSpec)))
+ }
+
+ // RANGE frame with tied order keys: peer rows share the same ord value, so the
+ // frame boundary spans the full peer group rather than splitting it.
+ test("SPARK-58201: range frame on tied order key") {
+ // Within each pk group, id/6 repeats every 2 rows creating genuine peer groups.
+ val df = spark.range(0, 90)
+ .selectExpr(
+ "id", "(id % 3) AS pk", "CAST(id / 6 AS INT) AS ord", "CAST(id AS INT) AS v_int")
+ val winSpec = Window.partitionBy($"pk").orderBy($"ord").rangeBetween(-1, 1)
+ checkEquivalence(() =>
+ df.select($"id", min($"v_int").over(winSpec), max($"v_int").over(winSpec)))
+ }
+
+ // DECIMAL and BINARY types: UnsafeRow.getDecimal/getBinary allocate fresh objects
+ // per call, so InternalRow.copyValue is safe for all reference types. Low spill
+ // thresholds verify correctness when the UnsafeRow backing store is recycled.
+ test("SPARK-58201: MIN/MAX on DECIMAL and BINARY types with spill") {
+ val df = spark
+ .range(0, 60)
+ .selectExpr(
+ "id",
+ "(id % 3) AS pk",
+ "CAST(id AS DECIMAL(38, 10)) AS v_dec",
+ "CAST(CAST(id AS STRING) AS BINARY) AS v_bin")
+ withSQLConf(
+ SQLConf.WINDOW_EXEC_BUFFER_IN_MEMORY_THRESHOLD.key -> "8",
+ SQLConf.WINDOW_EXEC_BUFFER_SPILL_THRESHOLD.key -> "16") {
+ val winSpec = Window.partitionBy($"pk").orderBy($"id").rowsBetween(-2, 2)
+ checkEquivalence(() =>
+ df.select(
+ $"id",
+ min($"v_dec").over(winSpec),
+ max($"v_dec").over(winSpec),
+ min($"v_bin").over(winSpec),
+ max($"v_bin").over(winSpec)))
+ }
+ }
+
+ // RANGE + spill: the lower cursor is gated on RANGE, so RANGE alone opens a
+ // second SpillableArrayIterator (lowerIterator) concurrently with
+ // inputIterator. Both iterators have independent UnsafeRow cursors (each
+ // SpillableArrayIterator allocates its own currentRow on construction), so
+ // advancing lowerIterator while inputIterator still holds nextRow is safe.
+ // Spill thresholds are set low enough to force ExternalAppendOnlyUnsafeRowArray
+ // into its SpillableArrayIterator path, pinning this two-iterator invariant.
+ test("SPARK-58201: Range-based moving frame: MIN/MAX with spill") {
+ val df = baseDF.selectExpr("id", "pk", "CAST(id / 2 AS INT) AS ord_val", "v_int")
+ withSQLConf(
+ SQLConf.WINDOW_EXEC_BUFFER_IN_MEMORY_THRESHOLD.key -> "8",
+ SQLConf.WINDOW_EXEC_BUFFER_SPILL_THRESHOLD.key -> "16") {
+ val winSpec = Window.partitionBy($"pk").orderBy($"ord_val").rangeBetween(-2, 2)
+ checkEquivalence(() =>
+ df.select($"id", min($"v_int").over(winSpec), max($"v_int").over(winSpec)))
+ }
+ }
+
+ // Multi-partition RANGE with forced spill.
+ test("SPARK-58201: multi-partition RANGE with spill") {
+ val df = spark.range(0, 200)
+ .selectExpr(
+ "id", "(id % 9) AS pk", "CAST(id / 3 AS INT) AS ord_val", "CAST(id AS INT) AS v")
+ withSQLConf(
+ SQLConf.WINDOW_EXEC_BUFFER_IN_MEMORY_THRESHOLD.key -> "4",
+ SQLConf.WINDOW_EXEC_BUFFER_SPILL_THRESHOLD.key -> "8") {
+ val winSpec = Window.partitionBy($"pk").orderBy($"ord_val").rangeBetween(-1, 1)
+ checkEquivalence(
+ () =>
+ df.select($"id", min($"v").over(winSpec), max($"v").over(winSpec)),
+ expectedDequeCount = 9)
+ }
+ }
+}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/window/SegmentTreeWindowMetricsSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/window/SegmentTreeWindowMetricsSuite.scala
index d22c1cdc90da..a71314ff283c 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/window/SegmentTreeWindowMetricsSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/window/SegmentTreeWindowMetricsSuite.scala
@@ -72,6 +72,7 @@ class SegmentTreeWindowMetricsSuite
withSQLConf(
SQLConf.WINDOW_SEGMENT_TREE_ENABLED.key -> "true",
SQLConf.WINDOW_SEGMENT_TREE_MIN_PARTITION_ROWS.key -> "1",
+ SQLConf.WINDOW_MONOTONIC_DEQUE_ENABLED.key -> "false",
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
// Two aggregates share one Window -> one frame; 3 partitions => 3 frames.
val df = baseDF.select($"id",
@@ -90,6 +91,7 @@ class SegmentTreeWindowMetricsSuite
SQLConf.WINDOW_SEGMENT_TREE_ENABLED.key -> "true",
// Threshold > partition size (40 rows/partition) forces fallback on every partition.
SQLConf.WINDOW_SEGMENT_TREE_MIN_PARTITION_ROWS.key -> "1000",
+ SQLConf.WINDOW_MONOTONIC_DEQUE_ENABLED.key -> "false",
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
val df = baseDF.select($"id", min($"v").over(winSpec).as("mn"))
val m = windowMetricValues(df)
@@ -103,6 +105,7 @@ class SegmentTreeWindowMetricsSuite
test("feature flag off: both segment-tree counters stay at zero") {
withSQLConf(
SQLConf.WINDOW_SEGMENT_TREE_ENABLED.key -> "false",
+ SQLConf.WINDOW_MONOTONIC_DEQUE_ENABLED.key -> "false",
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
val df = baseDF.select($"id", min($"v").over(winSpec).as("mn"))
val m = windowMetricValues(df)
@@ -127,6 +130,7 @@ class SegmentTreeWindowMetricsSuite
withSQLConf(
SQLConf.WINDOW_SEGMENT_TREE_ENABLED.key -> "true",
SQLConf.WINDOW_SEGMENT_TREE_MIN_PARTITION_ROWS.key -> "64",
+ SQLConf.WINDOW_MONOTONIC_DEQUE_ENABLED.key -> "false",
SQLConf.SHUFFLE_PARTITIONS.key -> "1",
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
// 3 window partitions x 100 rows, all >= 64 => segtree, one task.
@@ -144,6 +148,7 @@ class SegmentTreeWindowMetricsSuite
withSQLConf(
SQLConf.WINDOW_SEGMENT_TREE_ENABLED.key -> "true",
SQLConf.WINDOW_SEGMENT_TREE_MIN_PARTITION_ROWS.key -> "64",
+ SQLConf.WINDOW_MONOTONIC_DEQUE_ENABLED.key -> "false",
SQLConf.SHUFFLE_PARTITIONS.key -> "1",
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
// Two distinct keys, each exactly 200 rows (> 64 => segtree).
@@ -164,6 +169,7 @@ class SegmentTreeWindowMetricsSuite
withSQLConf(
SQLConf.WINDOW_SEGMENT_TREE_ENABLED.key -> "true",
SQLConf.WINDOW_SEGMENT_TREE_MIN_PARTITION_ROWS.key -> "64",
+ SQLConf.WINDOW_MONOTONIC_DEQUE_ENABLED.key -> "false",
SQLConf.SHUFFLE_PARTITIONS.key -> "1",
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
// 100 unique keys, 1 row each, all < 64 => fallback. Old guard
@@ -182,6 +188,7 @@ class SegmentTreeWindowMetricsSuite
withSQLConf(
SQLConf.WINDOW_SEGMENT_TREE_ENABLED.key -> "true",
SQLConf.WINDOW_SEGMENT_TREE_MIN_PARTITION_ROWS.key -> "64",
+ SQLConf.WINDOW_MONOTONIC_DEQUE_ENABLED.key -> "false",
SQLConf.SHUFFLE_PARTITIONS.key -> "1",
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
// 4 window partitions with lengths (50, 50, 100, 100) in sort order.
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/window/WindowSegmentTreeAllowlistSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/window/WindowSegmentTreeAllowlistSuite.scala
index 8dbb4ceecf1b..5fe17f736df9 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/window/WindowSegmentTreeAllowlistSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/window/WindowSegmentTreeAllowlistSuite.scala
@@ -17,7 +17,7 @@
package org.apache.spark.sql.execution.window
-import org.apache.spark.sql.DataFrame
+import org.apache.spark.sql.{DataFrame, QueryTest}
import org.apache.spark.sql.execution.SparkPlanInfo
import org.apache.spark.sql.execution.metric.SQLMetricsTestUtils
import org.apache.spark.sql.execution.ui.SparkPlanGraph
@@ -42,7 +42,12 @@ class WindowSegmentTreeAllowlistSuite
private val enableSegTree: Map[String, String] = Map(
SQLConf.WINDOW_SEGMENT_TREE_ENABLED.key -> "true",
- SQLConf.WINDOW_SEGMENT_TREE_MIN_PARTITION_ROWS.key -> "1")
+ SQLConf.WINDOW_SEGMENT_TREE_MIN_PARTITION_ROWS.key -> "1",
+ SQLConf.WINDOW_MONOTONIC_DEQUE_ENABLED.key -> "false")
+
+ private val enableDeque: Map[String, String] = Map(
+ SQLConf.WINDOW_MONOTONIC_DEQUE_ENABLED.key -> "true",
+ SQLConf.WINDOW_SEGMENT_TREE_ENABLED.key -> "false")
private def baseDF = spark.range(0, 120).selectExpr(
"id", "(id % 3) AS pk", "CAST(id AS INT) AS v", "CAST(id AS DOUBLE) AS vd")
@@ -70,6 +75,26 @@ class WindowSegmentTreeAllowlistSuite
total("number of segment-tree fallback frames prepared"))
}
+ private def dequeCounters(df: DataFrame): Long = {
+ val previousExecutionIds = currentExecutionIds()
+ df.collect()
+ sparkContext.listenerBus.waitUntilEmpty(10000)
+ val executionId = currentExecutionIds().diff(previousExecutionIds).head
+ val metricValues = statusStore.executionMetrics(executionId)
+ val graph = SparkPlanGraph(SparkPlanInfo.fromSparkPlan(df.queryExecution.executedPlan))
+ val windowNode = graph.allNodes.find(_.name == "Window").getOrElse {
+ fail(s"No Window node in plan:\n${df.queryExecution.executedPlan}")
+ }
+ def total(name: String): Long = {
+ val mOpt = windowNode.metrics.find(_.name == name)
+ mOpt.map { m =>
+ val raw = metricValues(m.accumulatorId)
+ "-?\\d+".r.findFirstIn(raw).map(_.toLong).getOrElse(0L)
+ }.getOrElse(0L)
+ }
+ total("number of monotonic-deque frames prepared")
+ }
+
// Positive: allowlisted aggregates route to segtree
Seq(
@@ -102,38 +127,40 @@ class WindowSegmentTreeAllowlistSuite
// Negative: non-allowlisted aggregates fall through
- test("collect_list falls through (unbounded buffer)") {
- withSQLConf(enableSegTree.toSeq: _*) {
- val df = baseDF.withColumn("agg", collect_list($"v").over(winSpec))
+ private def checkFallbackEquivalence(buildDf: () => DataFrame): Unit = {
+ val segTreeResult = withSQLConf(enableSegTree.toSeq: _*) {
+ val df = buildDf()
val (seg, _) = segTreeCounters(df)
- assert(seg == 0, s"collect_list should not use segment tree (got $seg frames)")
+ assert(seg == 0, s"Should not use segment tree (got $seg frames)")
+ df.collect().toSeq
+ }
+ val naiveResult = withSQLConf(
+ SQLConf.WINDOW_SEGMENT_TREE_ENABLED.key -> "false",
+ SQLConf.WINDOW_MONOTONIC_DEQUE_ENABLED.key -> "false") {
+ buildDf().collect().toSeq
+ }
+ QueryTest.sameRows(naiveResult, segTreeResult, isSorted = false).foreach { err =>
+ fail(s"Output differs from baseline.\n$err")
}
}
+ test("collect_list falls through (unbounded buffer)") {
+ checkFallbackEquivalence(() => baseDF.withColumn("agg", collect_list($"v").over(winSpec)))
+ }
+
test("collect_set falls through (unbounded buffer)") {
- withSQLConf(enableSegTree.toSeq: _*) {
- val df = baseDF.withColumn("agg", collect_set($"v").over(winSpec))
- val (seg, _) = segTreeCounters(df)
- assert(seg == 0, s"collect_set should not use segment tree (got $seg frames)")
- }
+ checkFallbackEquivalence(() => baseDF.withColumn("agg", collect_set($"v").over(winSpec)))
}
test("approx_count_distinct (HyperLogLog++) falls through (fail-closed)") {
- withSQLConf(enableSegTree.toSeq: _*) {
- val df = baseDF.withColumn("agg", approx_count_distinct($"v").over(winSpec))
- val (seg, _) = segTreeCounters(df)
- assert(seg == 0,
- s"approx_count_distinct is intentionally not on the allowlist (got $seg frames)")
- }
+ checkFallbackEquivalence(() =>
+ baseDF.withColumn("agg", approx_count_distinct($"v").over(winSpec)))
}
test("percentile_approx falls through (sketch buffer not auditable)") {
- withSQLConf(enableSegTree.toSeq: _*) {
- val df = baseDF.withColumn(
- "agg", percentile_approx($"vd", lit(0.5), lit(100)).over(winSpec))
- val (seg, _) = segTreeCounters(df)
- assert(seg == 0, s"percentile_approx should not use segment tree (got $seg frames)")
- }
+ checkFallbackEquivalence(() =>
+ baseDF.withColumn("agg",
+ percentile_approx($"vd", lit(0.5), lit(100)).over(winSpec)))
}
// Gate: aggregates carrying a FILTER (WHERE ...) clause fall through.
@@ -144,19 +171,15 @@ class WindowSegmentTreeAllowlistSuite
// (e.g., pushing the predicate into the aggregate function), this test
// fails and forces an explicit eligibility review.
test("FILTER (WHERE ...) disables segment-tree path") {
- withSQLConf(enableSegTree.toSeq: _*) {
- withTempView("t") {
- baseDF.createOrReplaceTempView("t")
- val df = spark.sql(
- """SELECT id, pk, v,
+ withTempView("t") {
+ baseDF.createOrReplaceTempView("t")
+ checkFallbackEquivalence(() => {
+ spark.sql("""SELECT id, pk, v,
| sum(v) FILTER (WHERE v % 2 = 0)
| OVER (PARTITION BY pk ORDER BY id ROWS BETWEEN 3 PRECEDING AND 3 FOLLOWING)
| AS filtered_sum
|FROM t""".stripMargin)
- val (seg, _) = segTreeCounters(df)
- assert(seg == 0,
- s"filtered aggregate must not take segment-tree path (got $seg segtree frames)")
- }
+ })
}
}
@@ -174,4 +197,39 @@ class WindowSegmentTreeAllowlistSuite
s"Window group containing a non-allowlisted agg must fall through (got $seg)")
}
}
+
+ test("MIN/MAX routes to the monotonic-deque path") {
+ withSQLConf(enableDeque.toSeq: _*) {
+ val df = baseDF.withColumn("agg", min($"vd").over(winSpec))
+ val dequeFrames = dequeCounters(df)
+ assert(dequeFrames > 0, s"MIN should bump numMonotonicDequeFrames (got $dequeFrames)")
+ }
+ }
+
+ test("mix of MIN/MAX + other allowlisted aggregates falls through deque") {
+ withSQLConf(enableDeque.toSeq: _*) {
+ val df = baseDF.withColumn("m", min($"v").over(winSpec))
+ .withColumn("s", sum($"v").over(winSpec))
+ val dequeFrames = dequeCounters(df)
+ assert(dequeFrames == 0,
+ s"Window group containing a non-MIN/MAX agg must fall through (got $dequeFrames)")
+ }
+ }
+
+ test("FILTER (WHERE ...) disables monotonic-deque path") {
+ withSQLConf(enableDeque.toSeq: _*) {
+ withTempView("t") {
+ baseDF.createOrReplaceTempView("t")
+ val df = spark.sql("""SELECT id, pk, v,
+ | min(v) FILTER (WHERE v % 2 = 0)
+ | OVER (PARTITION BY pk ORDER BY id ROWS BETWEEN 3 PRECEDING AND 3 FOLLOWING)
+ | AS filtered_min
+ |FROM t""".stripMargin)
+ val dequeFrames = dequeCounters(df)
+ assert(
+ dequeFrames == 0,
+ s"filtered MIN must not take monotonic-deque path (got $dequeFrames)")
+ }
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org