[SPARK-58201][SQL] Optimize sliding window MIN/MAX aggregate function…

[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. Naive Sliding Window (SlidingWindowFunctionFrame): Re-scans all rows in the sliding window buffer on every row, resulting in $O(N \cdot W)$ time complexity.
2. Segment Tree (SegmentTreeWindowFunctionFrame): Built over partition blocks, offering $O(N \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 (running at ~0.5X the speed of 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 ~3X - 4X faster than Segment Tree, and ~30X - 37X faster than the Naive baseline.
• At large window scales ($W=100,001$), Monotonic Deque achieves 6X+ speedup over 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$), 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).
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) (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) (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?

Generated-by: Antigravity (Google DeepMind Advanced Agentic 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.

[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

[HyukjinKwon]

Can you PR title complete?