[SPARK-58201][SQL] Optimize sliding window MIN/MAX aggregate function using monotonic deque

[pavan51]

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).

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.
• Zero-Copy Optimization for Primitive Types: Internal rows in Spark (UnsafeRow) recycle memory, meaning reference types (like String, Decimal, Array) must be copied to avoid corruption. However, JVM primitives (like Int, Long, Double, Float, Boolean, Date, Timestamp) are passed by value and are immune to this. We introduce a type check that skips heap allocation/copying entirely for primitive types.
• 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: true) to allow toggling this optimization.

---

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) and a minimal memory footprint.

As a result:

• At standard window scales ($W=1001$), Monotonic Deque is ~30X - 37X faster than the default Spark Naive baseline and ~3X - 4X faster than Segment Tree, and and speedup factor scales with increased window size.
• At large window scales ($W=100,001$), Monotonic Deque achieves ~6X to 6.2X faster then Segment Tree. Because of its $O(1)$ amortized complexity vs the Segment Tree's $O(\log W)$, the speedup factor scales logarithmically with the window size, diverging even further in favor of Monotonic Deque as $W$ grows.
• At small window scales ($W=11$), Default Spark Naive Baseline is 2X faster then Segment Tree, so Monotonic Deque completely resolves the Segment Tree's deoptimization, outperforming Naive and running up to ~3X+ faster than Segment Tree.

---

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: true) 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.

The updated benchmark results under OpenJDK 23.0.1 on Mac OS X 15.1.1 (Apple M1) are:

MIN Sliding Window (W=1001, 256K rows)

• Naive Baseline: 3,450 ms (1.0X)
• Segment Tree: 440 ms (7.8X)
• Monotonic Deque (new): 148 ms 23.3X faster then default Spark and 2.97X faster than Segment Tree

MAX Sliding Window (W=1001, 256K rows)

• Naive Baseline: 4,072 ms (1.0X)
• Segment Tree: 436 ms (9.3X)
• Monotonic Deque (new): 110 ms 37.0X faster then default Spark and 3.96X faster than Segment Tree

MIN Sliding Window Scaling (W=100001, 2M rows)

• Increasing (Worst-Case): Segment Tree = 3,461 ms vs. Monotonic Deque = 675 ms (5.1X faster than Segment Tree)
• Decreasing (Best-Case): Segment Tree = 3,205 ms vs. Monotonic Deque = 518 ms (6.2X faster than Segment Tree)
• Random (Average-Case): Segment Tree = 3,284 ms vs. Monotonic Deque = 546 ms (6.0X faster than Segment Tree)

MIN/MAX Small Window Scaling (W=11, 2M rows - Pareto Loss Zone Check)
In small windows, the naive sliding window scan O(W) is fast, whereas the segment tree has high setup and block query overhead O(log W), causing it to be slower than naive. Monotonic Deque dominates both.

• MIN (2M rows): Naive = 935 ms | Segment Tree = 1,716 ms (0.5X) | Monotonic Deque (new) = 555 ms (1.7X vs Naive, 3.09X vs Segment Tree)
• MAX (2M rows): Naive = 793 ms | Segment Tree = 1,756 ms (0.5X) | Monotonic Deque (new) = 689 ms (1.2X vs Naive, 2.55X vs Segment Tree)

---

Was this patch authored or co-authored using generative AI tooling?

Yes, co-authored by Google DeepMind Coding Assistant

[Ma77Ball]

This adds a new default-on execution path with no correctness test. I think it might be worth considering adding a differential test.

[pavan51]

Thanks for the suggestion! I've added a new differential correctness test suite (MonotonicDequeWindowFunctionSuite ) verifying that the Monotonic Deque matches both the Segment Tree and Naive baselines across primitives, reference types, range/row frames, and null values.

[Ma77Ball]

YearMonthIntervalType (Int) and DayTimeIntervalType (Long) are also primitive-backed and valid for MIN/MAX. Suggested Performance Fix:

Suggested change

	case BooleanType | ByteType | ShortType | IntegerType | LongType | FloatType | DoubleType |
	DateType | TimestampType | TimestampNTZType => true
	case BooleanType | ByteType | ShortType | IntegerType | LongType | FloatType | DoubleType |
	DateType | TimestampType | TimestampNTZType | _: YearMonthIntervalType |
	_: DayTimeIntervalType => true

[pavan51]

Great catch! I've added YearMonthIntervalType and DayTimeIntervalType to the primitive match list to skip the row-copy overhead for interval columns. Thanks!"

[HyukjinKwon]

Can you PR title complete?

[pavan51]

Can you PR title complete?

Updated the title of the PR with complete info

[Ma77Ball]

After a static review, the PR LGTM! I left two suggestions. cc @HyukjinKwon

[Ma77Ball]

This new optimization defaults to on, but the sibling spark.sql.window.segmentTree.enabled (same area, added in 4.2.0) defaults to off. Turning a brand-new execution path on by default for all sliding MIN/MAX window queries on its first release is risky; consider defaulting to false to match and let it bake before flipping on. Fix:

Suggested change

	.createWithDefault(true)
	.createWithDefault(false)

[pavan51]

Unlike the segment tree, the monotonic deque has O(1) amortized cost and does not suffer from tree-building overhead on small windows. However, keeping it disabled by default for its initial release as a safer approach to mitigate risk. I've updated the default to false.

[Ma77Ball]

The isMinMaxOnly guard is duplicated in the Moving Frame case below. Might be better to add it to a small local helper so the two copies don't drift.

[pavan51]

Done. I've refactored this and pulled isMinMaxOnly up to the outer scope of the match block so it is cleanly shared by both the moving and shrinking frame cases.

[pavan51]

Hi @HyukjinKwon

All suggestions from @Ma77Ball 's review have been fully addressed (the feature is now disabled by default for safety, code duplication was refactored, and primitive interval optimization was added).

The new correctness tests are passing locally and CI is green.Could you please take a look when you have a moment? Thanks!

[pavan51]

Hi @peter-toth @Yicong-Huang @HyukjinKwon,

Could any of you please review this PR when you get a chance.

@Ma77Ball has reviewed and approved the changes, so it is ready for your final reviews!