From 1691d2d110678026e19eb631dbf4fc03f7f5c764 Mon Sep 17 00:00:00 2001 From: Pavan Date: Fri, 17 Jul 2026 19:45:29 -0500 Subject: [PATCH 1/2] [SPARK-58201][SQL] Optimize sliding window MIN/MAX aggregate functions using monotonic deque --- .../apache/spark/sql/internal/SQLConf.scala | 11 + .../benchmarks/WindowBenchmark-results.txt | 179 ++++++++++----- .../SlidingWindowMinMaxFunctionFrame.scala | 209 ++++++++++++++++++ .../window/WindowEvaluatorFactoryBase.scala | 33 ++- .../execution/benchmark/WindowBenchmark.scala | 114 ++++++++-- 5 files changed, 464 insertions(+), 82 deletions(-) create mode 100644 sql/core/src/main/scala/org/apache/spark/sql/execution/window/SlidingWindowMinMaxFunctionFrame.scala 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 0b3f4693853a7..de9b3ccae0366 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 @@ -4465,6 +4465,15 @@ object SQLConf { .booleanConf .createWithDefault(false) + val WINDOW_MONOTONIC_DEQUE_ENABLED = + buildConf("spark.sql.window.monotonicDeque.enabled") + .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE) + .doc("Use monotonic deques for sliding min/max aggregate window frames. " + + "This provides O(N) complexity instead of O(N * W) or O(N log W).") + .version("5.0.0") + .booleanConf + .createWithDefault(true) + val WINDOW_SEGMENT_TREE_MIN_PARTITION_ROWS = buildConf("spark.sql.window.segmentTree.minPartitionRows") .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE) @@ -8778,6 +8787,8 @@ class SQLConf extends Serializable with Logging with SqlApiConf { def windowSegmentTreeEnabled: Boolean = getConf(WINDOW_SEGMENT_TREE_ENABLED) + def windowMonotonicDequeEnabled: Boolean = getConf(WINDOW_MONOTONIC_DEQUE_ENABLED) + def windowSegmentTreeMinPartitionRows: Int = getConf(WINDOW_SEGMENT_TREE_MIN_PARTITION_ROWS) def windowSegmentTreeBlockSize: Int = getConf(WINDOW_SEGMENT_TREE_BLOCK_SIZE) diff --git a/sql/core/benchmarks/WindowBenchmark-results.txt b/sql/core/benchmarks/WindowBenchmark-results.txt index bb50c20b59213..2e6359af5dc79 100644 --- a/sql/core/benchmarks/WindowBenchmark-results.txt +++ b/sql/core/benchmarks/WindowBenchmark-results.txt @@ -2,171 +2,230 @@ Section A - MIN (non-invertible) ================================================================================================ -OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 6.17.0-1011-azure -Intel(R) Xeon(R) Platinum 8370C CPU @ 2.80GHz +OpenJDK 64-Bit Server VM 23.0.1+11 on Mac OS X 15.1.1 +Apple M1 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) 4350 4380 30 0.1 16592.3 1.0X -MIN segtree (default) 433 442 10 0.6 1652.8 10.0X -MIN segtree (blockSize=256) 2259 2266 9 0.1 8617.9 1.9X +MIN naive (current, baseline) 3311 3407 96 0.1 12628.8 1.0X +MIN segtree (default) 306 318 12 0.9 1165.4 10.8X +MIN monotonic deque (new) 108 116 9 2.4 411.1 30.7X ================================================================================================ Section A - MAX (non-invertible) ================================================================================================ -OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 6.17.0-1011-azure -Intel(R) Xeon(R) Platinum 8370C CPU @ 2.80GHz +OpenJDK 64-Bit Server VM 23.0.1+11 on Mac OS X 15.1.1 +Apple M1 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) 4113 4124 21 0.1 15690.6 1.0X -MAX segtree (default) 415 420 3 0.6 1582.0 9.9X -MAX segtree (blockSize=256) 2279 2288 9 0.1 8692.0 1.8X +MAX naive (current, baseline) 4072 5071 1101 0.1 15533.0 1.0X +MAX segtree (default) 436 1207 1260 0.6 1664.2 9.3X +MAX monotonic deque (new) 110 122 13 2.4 419.9 37.0X ================================================================================================ Section A - SUM (Spark has no inverse; full recompute) ================================================================================================ -OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 6.17.0-1011-azure -Intel(R) Xeon(R) Platinum 8370C CPU @ 2.80GHz +OpenJDK 64-Bit Server VM 23.0.1+11 on Mac OS X 15.1.1 +Apple M1 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) 4828 4870 61 0.1 18418.4 1.0X -SUM segtree (default) 431 436 6 0.6 1643.9 11.2X -SUM segtree (blockSize=256) 2286 2295 13 0.1 8720.2 2.1X +SUM naive (current, baseline) 3335 3421 113 0.1 12723.3 1.0X +SUM segtree (default) 290 299 13 0.9 1105.5 11.5X +SUM monotonic deque (new) 3298 3352 33 0.1 12580.8 1.0X ================================================================================================ Section A - COUNT ================================================================================================ -OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 6.17.0-1011-azure -Intel(R) Xeon(R) Platinum 8370C CPU @ 2.80GHz +OpenJDK 64-Bit Server VM 23.0.1+11 on Mac OS X 15.1.1 +Apple M1 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) 3867 3883 13 0.1 14751.9 1.0X -COUNT segtree (default) 415 417 3 0.6 1582.8 9.3X -COUNT segtree (blockSize=256) 2216 2220 7 0.1 8453.8 1.7X +COUNT naive (current, baseline) 2840 2930 101 0.1 10835.0 1.0X +COUNT segtree (default) 256 266 6 1.0 978.2 11.1X +COUNT monotonic deque (new) 2854 2903 39 0.1 10888.2 1.0X ================================================================================================ Section A - AVG (multi-buffer) ================================================================================================ -OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 6.17.0-1011-azure -Intel(R) Xeon(R) Platinum 8370C CPU @ 2.80GHz +OpenJDK 64-Bit Server VM 23.0.1+11 on Mac OS X 15.1.1 +Apple M1 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) 4483 4495 14 0.0 22804.2 1.0X -AVG segtree (default) 426 435 8 0.5 2165.9 10.5X -AVG segtree (blockSize=256) 1551 1566 18 0.1 7890.5 2.9X +AVG naive (current, baseline) 3612 3635 24 0.1 18373.9 1.0X +AVG segtree (default) 265 280 19 0.7 1345.4 13.7X +AVG monotonic deque (new) 3681 3721 35 0.1 18724.5 1.0X ================================================================================================ Section A - STDDEV_SAMP (multi-buffer, stress) ================================================================================================ -OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 6.17.0-1011-azure -Intel(R) Xeon(R) Platinum 8370C CPU @ 2.80GHz +OpenJDK 64-Bit Server VM 23.0.1+11 on Mac OS X 15.1.1 +Apple M1 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) 97349 97405 59 0.0 48674.4 1.0X -STDDEV_SAMP segtree (default) 5628 5631 3 0.4 2813.9 17.3X -STDDEV_SAMP segtree (blockSize=256) 113685 113987 289 0.0 56842.3 0.9X +STDDEV_SAMP naive (current, baseline) 100017 102789 2613 0.0 50008.7 1.0X +STDDEV_SAMP segtree (default) 3838 3949 96 0.5 1918.9 26.1X +STDDEV_SAMP monotonic deque (new) 105070 106957 3147 0.0 52534.8 1.0X ================================================================================================ Section B - W=10 scaling (stress: Pareto loss zone) ================================================================================================ -OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 6.17.0-1011-azure -Intel(R) Xeon(R) Platinum 8370C CPU @ 2.80GHz +OpenJDK 64-Bit Server VM 23.0.1+11 on Mac OS X 15.1.1 +Apple M1 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 1003 1012 8 2.0 501.6 1.0X -SUM segtree (default) W=11 2378 2384 9 0.8 1188.8 0.4X +SUM naive W=11 841 875 58 2.4 420.6 1.0X +SUM segtree (default) W=11 1675 1720 49 1.2 837.5 0.5X ================================================================================================ Section B - W=50 scaling (stress: Pareto loss zone) ================================================================================================ -OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 6.17.0-1011-azure -Intel(R) Xeon(R) Platinum 8370C CPU @ 2.80GHz +OpenJDK 64-Bit Server VM 23.0.1+11 on Mac OS X 15.1.1 +Apple M1 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 2377 2395 24 0.8 1188.5 1.0X -SUM segtree (default) W=51 2667 2683 17 0.7 1333.7 0.9X +SUM naive W=51 1938 2019 92 1.0 969.2 1.0X +SUM segtree (default) W=51 2124 2146 39 0.9 1061.8 0.9X ================================================================================================ Section B - W=201 scaling ================================================================================================ -OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 6.17.0-1011-azure -Intel(R) Xeon(R) Platinum 8370C CPU @ 2.80GHz +OpenJDK 64-Bit Server VM 23.0.1+11 on Mac OS X 15.1.1 +Apple M1 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 3588 3620 39 0.3 3588.1 1.0X -SUM segtree (default) W=201 1465 1474 10 0.7 1465.1 2.4X +SUM naive W=201 3018 3113 96 0.3 3018.0 1.0X +SUM segtree (default) W=201 1110 1147 31 0.9 1110.3 2.7X ================================================================================================ Section B - W=4001 scaling (stress, + bs=256 cross-block) ================================================================================================ -OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 6.17.0-1011-azure -Intel(R) Xeon(R) Platinum 8370C CPU @ 2.80GHz +OpenJDK 64-Bit Server VM 23.0.1+11 on Mac OS X 15.1.1 +Apple M1 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 129657 129727 67 0.0 64828.5 1.0X -SUM segtree (default) W=4001 3834 3858 24 0.5 1916.8 33.8X -SUM segtree (blockSize=256) W=4001 110788 111483 748 0.0 55393.8 1.2X +SUM naive W=4001 100129 102428 2149 0.0 50064.5 1.0X +SUM segtree (default) W=4001 2690 2728 40 0.7 1345.0 37.2X +SUM segtree (blockSize=256) W=4001 90352 92143 1901 0.0 45175.8 1.1X ================================================================================================ Section F - spill regression guard (String, stress) ================================================================================================ -OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 6.17.0-1011-azure -Intel(R) Xeon(R) Platinum 8370C CPU @ 2.80GHz +OpenJDK 64-Bit Server VM 23.0.1+11 on Mac OS X 15.1.1 +Apple M1 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) 88459 88494 43 0.0 88459.0 1.0X -MAX segtree default (String) 3213 3227 21 0.3 3213.4 27.5X +MAX naive (String) 422 451 27 2.4 422.1 1.0X +MAX segtree default (String) 401 415 12 2.5 400.9 1.1X ================================================================================================ Section C - N-sweep small (stress) ================================================================================================ -OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 6.17.0-1011-azure -Intel(R) Xeon(R) Platinum 8370C CPU @ 2.80GHz +OpenJDK 64-Bit Server VM 23.0.1+11 on Mac OS X 15.1.1 +Apple M1 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 3314 3319 6 0.6 1657.1 1.0X +SUM segtree (default) N=2M 2328 2449 186 0.9 1163.8 1.0X ================================================================================================ Section C - N-sweep mid (stress) ================================================================================================ -OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 6.17.0-1011-azure -Intel(R) Xeon(R) Platinum 8370C CPU @ 2.80GHz +OpenJDK 64-Bit Server VM 23.0.1+11 on Mac OS X 15.1.1 +Apple M1 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 15970 16004 42 0.5 1996.3 1.0X +SUM segtree (default) N=8M 11966 12081 181 0.7 1495.7 1.0X ================================================================================================ Section C - N-sweep large (stress) ================================================================================================ -OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 6.17.0-1011-azure -Intel(R) Xeon(R) Platinum 8370C CPU @ 2.80GHz +OpenJDK 64-Bit Server VM 23.0.1+11 on Mac OS X 15.1.1 +Apple M1 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 38877 38998 121 0.4 2429.8 1.0X +SUM segtree (default) N=16M 30038 30672 574 0.5 1877.4 1.0X +================================================================================================ +Section G - MIN Monotonic Deque vs Segment Tree (Worst-Case: Increasing) +================================================================================================ + +OpenJDK 64-Bit Server VM 23.0.1+11 on Mac OS X 15.1.1 +Apple M1 +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) 3315 3379 65 0.6 1657.4 1.0X +MIN monotonic deque (Increasing) 536 561 27 3.7 268.0 6.2X + + +================================================================================================ +Section G - MIN Monotonic Deque vs Segment Tree (Best-Case: Decreasing) +================================================================================================ + +OpenJDK 64-Bit Server VM 23.0.1+11 on Mac OS X 15.1.1 +Apple M1 +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) 3171 3197 24 0.6 1585.6 1.0X +MIN monotonic deque (Decreasing) 523 528 4 3.8 261.3 6.1X + + +================================================================================================ +Section G - MIN Monotonic Deque vs Segment Tree (Random) +================================================================================================ + +OpenJDK 64-Bit Server VM 23.0.1+11 on Mac OS X 15.1.1 +Apple M1 +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) 3364 3834 574 0.6 1681.8 1.0X +MIN monotonic deque (Random) 553 579 22 3.6 276.3 6.1X + +================================================================================================ +Section H - MIN W=11 scaling (stress, 2M rows) +================================================================================================ + +OpenJDK 64-Bit Server VM 23.0.1+11 on Mac OS X 15.1.1 +Apple M1 +MIN sliding window, W=11, 2M rows: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative +------------------------------------------------------------------------------------------------------------------------ +MIN naive (current, baseline) 935 994 62 2.1 467.5 1.0X +MIN segtree (default) 1716 1747 30 1.2 857.9 0.5X +MIN monotonic deque (new) 555 574 20 3.6 277.7 1.7X + + +================================================================================================ +Section H - MAX W=11 scaling (stress, 2M rows) +================================================================================================ + +OpenJDK 64-Bit Server VM 23.0.1+11 on Mac OS X 15.1.1 +Apple M1 +MAX sliding window, W=11, 2M rows: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative +------------------------------------------------------------------------------------------------------------------------ +MAX naive (current, baseline) 793 804 12 2.5 396.5 1.0X +MAX segtree (default) 1756 2061 387 1.1 878.0 0.5X +MAX monotonic deque (new) 689 706 23 2.9 344.6 1.2X 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 0000000000000..b55d56cf0ac76 --- /dev/null +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/window/SlidingWindowMinMaxFunctionFrame.scala @@ -0,0 +1,209 @@ +/* + * 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.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.types._ + +/** + * 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]]. + */ +private[window] final class SlidingWindowMinMaxFunctionFrame( + target: InternalRow, + processor: AggregateProcessor, + lbound: BoundOrdering, + ubound: Option[BoundOrdering], + functions: Array[Expression], + inputSchema: Seq[Attribute]) + extends WindowFunctionFrame { + + /** Rows of the partition currently being processed. */ + private[this] var input: ExternalAppendOnlyUnsafeRowArray = null + + /** Iterators over the [[input]] */ + private[this] var lowerIterator: Iterator[UnsafeRow] = _ + private[this] var inputIterator: Iterator[UnsafeRow] = _ + + /** The row at lowerBound. */ + 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 + + private[this] val sourceRow = new SpecificInternalRow(functions.map(_.dataType).toIndexedSeq) + + private[this] val deques: Array[MinMaxDeque] = functions.zipWithIndex.map { + case (func, i) => + val isMin = func.isInstanceOf[Min] + val child = func match { + case m: Min => m.child + case m: Max => m.child + } + val boundChild = BindReferences.bindReference(child, inputSchema) + val ordering = TypeUtils.getInterpretedOrdering(child.dataType) + new MinMaxDeque(isMin, boundChild, child.dataType, ordering, i) + } + + override def prepare(rows: ExternalAppendOnlyUnsafeRowArray): Unit = { + input = rows + lowerIterator = input.generateIterator() + lowerRow = WindowFunctionFrame.getNextOrNull(lowerIterator) + deques.foreach(_.deque.clear()) + lowerBound = 0 + + if (ubound.isEmpty) { + val iter = input.generateIterator() + var idx = 0 + while (iter.hasNext) { + val row = iter.next() + deques.foreach(_.admit(row, idx)) + idx += 1 + } + upperBound = input.length + nextRow = null + inputIterator = null + } else { + 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 + 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. + if (ubound.isDefined) { + val ub = ubound.get + while (nextRow != null && ub.compare(nextRow, upperBound, current, index) <= 0) { + if (lbound.compare(nextRow, lowerBound, current, index) < 0) { + lowerBound += 1 + lowerRow = WindowFunctionFrame.getNextOrNull(lowerIterator) + } else { + deques.foreach(_.admit(nextRow, upperBound)) + bufferUpdated = true + } + nextRow = WindowFunctionFrame.getNextOrNull(inputIterator) + upperBound += 1 + } + } + + if (bufferUpdated) { + deques.foreach(_.dropBefore(lowerBound)) + } + + // Write output values to target. + 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 + + private class ValueWithIndex(val value: Any, val index: Int) + + private class MinMaxDeque( + val isMin: Boolean, + val boundChild: Expression, + val dataType: DataType, + val ordering: Ordering[Any], + val bufferIndex: Int) { + val deque = new java.util.ArrayDeque[ValueWithIndex]() + private val tempRow = new SpecificInternalRow(Seq(dataType)) + private val isPrimitive = dataType match { + case BooleanType | ByteType | ShortType | IntegerType | LongType | FloatType | DoubleType | + DateType | TimestampType | TimestampNTZType => true + case _ => false + } + + private def evaluateAndCopy(row: InternalRow): Any = { + val value = boundChild.eval(row) + if (value == null) { + null + } else if (isPrimitive) { + value + } else { + tempRow.update(0, value) + val copiedRow = tempRow.copy() + copiedRow.get(0, dataType) + } + } + + def admit(row: InternalRow, index: Int): Unit = { + val value = evaluateAndCopy(row) + if (value != null) { + if (isMin) { + while (!deque.isEmpty && ordering.compare(deque.peekLast().value, value) >= 0) { + deque.pollLast() + } + } else { + while (!deque.isEmpty && ordering.compare(deque.peekLast().value, value) <= 0) { + deque.pollLast() + } + } + deque.offerLast(new ValueWithIndex(value, index)) + } + } + + def dropBefore(boundary: Int): Unit = { + while (!deque.isEmpty && deque.peekFirst().index < boundary) { + deque.pollFirst() + } + } + + def currentValue(): Any = { + if (deque.isEmpty) { + null + } else { + deque.peekFirst().value + } + } + } +} 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 40cba3d5ceb4c..4caed3166806e 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,9 @@ 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, TimestampAddYMInterval, UnaryMinus, UnboundedFollowing, UnboundedPreceding, UnsafeProjection, WindowExpression} -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.execution.metric.SQLMetric import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.types.{CalendarIntervalType, DateType, DayTimeIntervalType, DecimalType, IntegerType, TimestampNTZType, TimestampType, YearMonthIntervalType} @@ -281,7 +283,19 @@ trait WindowEvaluatorFactoryBase { // Shrinking Frame. case ("AGGREGATE", frameType, lower, UnboundedFollowing, _) => - if (eligibleForSegTree(functions, aggFilters, frameType, conf)) { + val isMinMaxOnly = conf.windowMonotonicDequeEnabled && + functions.nonEmpty && functions.forall { + case _: Min | _: Max => true + case _ => false + } && aggFilters.forall(_.isEmpty) + + if (isMinMaxOnly) { + target: InternalRow => { + val lb = createBoundOrdering(frameType, lower, timeZone) + new SlidingWindowMinMaxFunctionFrame( + target, processor, lb, None, functions, childOutput) + } + } else if (eligibleForSegTree(functions, aggFilters, frameType, conf)) { val segFns = functions.map(_.asInstanceOf[DeclarativeAggregate]) // Shrinking-frame queries `[lower, n)` on `WindowSegmentTree` touch the LRU // for exactly two blocks per query: (1) the lower-edge partial block, and @@ -334,7 +348,20 @@ trait WindowEvaluatorFactoryBase { // Moving Frame. case ("AGGREGATE", frameType, lower, upper, _) => - if (eligibleForSegTree(functions, aggFilters, frameType, conf)) { + val isMinMaxOnly = conf.windowMonotonicDequeEnabled && + functions.nonEmpty && functions.forall { + case _: Min | _: Max => true + case _ => false + } && aggFilters.forall(_.isEmpty) + + if (isMinMaxOnly) { + target: InternalRow => { + val lb = createBoundOrdering(frameType, lower, timeZone) + val ub = createBoundOrdering(frameType, upper, timeZone) + new SlidingWindowMinMaxFunctionFrame( + target, processor, lb, Some(ub), functions, childOutput) + } + } 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/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 2284ff037e839..5491ac389381b 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 @@ -146,15 +146,17 @@ object WindowBenchmark extends SqlBasedBenchmark { def runSectionA( aggFn: String, iters: Int, rows: Long, halfW: Int, stressMark: String): 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") + 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") + val dMonotonic = digest(aggFn, frame, + SQLConf.WINDOW_MONOTONIC_DEQUE_ENABLED.key -> "true") 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") + require(dNaive == dMonotonic, + s"$aggFn monotonic deque digest mismatch: naive=$dNaive monotonic=$dMonotonic") val W = 2 * halfW + 1 val benchmark = new Benchmark( @@ -162,24 +164,26 @@ object WindowBenchmark extends SqlBasedBenchmark { 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) + val nMonotonic = s"$aggFn monotonic deque (new)" + allCaseNames ++= Seq(nNaive, nSeg, nMonotonic) 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") { + 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() } } @@ -274,15 +278,64 @@ 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 B SUM W sweep point") { - runSectionB( - smokeHalfW, stressBs = smokeHalfW >= 2000, smokeRowCount, ITERS_STRESS, "") + runBenchmark("SMOKE: Section A MAX") { + runSectionA("MAX", ITERS_STRESS, smokeRowCount, smokeHalfW, "") } } else { setupIntTable(A_N_INT) @@ -343,6 +396,29 @@ object WindowBenchmark extends SqlBasedBenchmark { runBenchmark("Section C - N-sweep large (stress)") { runSectionC(C_N_LARGE) } + + setupIncreasingTable(2000000L) + runBenchmark("Section G - MIN Monotonic Deque vs Segment Tree (Worst-Case: Increasing)") { + runSectionG("MIN", ITERS_NORMAL, 2000000L, 50000, "Increasing") + } + + setupDecreasingTable(2000000L) + runBenchmark("Section G - MIN Monotonic Deque vs Segment Tree (Best-Case: Decreasing)") { + runSectionG("MIN", ITERS_NORMAL, 2000000L, 50000, "Decreasing") + } + + setupIntTable(2000000L) + runBenchmark("Section G - MIN Monotonic Deque vs Segment Tree (Random)") { + runSectionG("MIN", ITERS_NORMAL, 2000000L, 50000, "Random") + } + + setupIntTable(2000000L) + runBenchmark("Section H - MIN W=11 scaling (stress, 2M rows)") { + runSectionA("MIN", ITERS_STRESS, 2000000L, 5, "") + } + runBenchmark("Section H - MAX W=11 scaling (stress, 2M rows)") { + runSectionA("MAX", ITERS_STRESS, 2000000L, 5, "") + } } // Drain listener before reading metrics. From be2c3e450b5ef00a45ce2b00f81d1aaa25aaf050 Mon Sep 17 00:00:00 2001 From: Pavan Date: Mon, 20 Jul 2026 01:40:01 -0500 Subject: [PATCH 2/2] [SPARK-58201][SQL] Add differential correctness tests and optimize interval primitives --- .../apache/spark/sql/internal/SQLConf.scala | 4 +- .../SlidingWindowMinMaxFunctionFrame.scala | 3 +- .../window/WindowEvaluatorFactoryBase.scala | 17 +- .../MonotonicDequeWindowFunctionSuite.scala | 202 ++++++++++++++++++ .../SegmentTreeWindowFunctionSuite.scala | 5 + .../SegmentTreeWindowMetricsSuite.scala | 5 + .../UnboundedFollowingSegmentTreeSuite.scala | 5 + .../WindowSegmentTreeAllowlistSuite.scala | 3 +- 8 files changed, 228 insertions(+), 16 deletions(-) create mode 100644 sql/core/src/test/scala/org/apache/spark/sql/execution/window/MonotonicDequeWindowFunctionSuite.scala 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 de9b3ccae0366..c11f7c4e7270d 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 @@ -4470,9 +4470,9 @@ object SQLConf { .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE) .doc("Use monotonic deques for sliding min/max aggregate window frames. " + "This provides O(N) complexity instead of O(N * W) or O(N log W).") - .version("5.0.0") + .version("4.3.0") .booleanConf - .createWithDefault(true) + .createWithDefault(false) val WINDOW_SEGMENT_TREE_MIN_PARTITION_ROWS = buildConf("spark.sql.window.segmentTree.minPartitionRows") 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 index b55d56cf0ac76..5c352e2e4d123 100644 --- 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 @@ -159,7 +159,8 @@ private[window] final class SlidingWindowMinMaxFunctionFrame( private val tempRow = new SpecificInternalRow(Seq(dataType)) private val isPrimitive = dataType match { case BooleanType | ByteType | ShortType | IntegerType | LongType | FloatType | DoubleType | - DateType | TimestampType | TimestampNTZType => true + DateType | TimestampType | TimestampNTZType | _: YearMonthIntervalType | + _: DayTimeIntervalType => true case _ => false } 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 4caed3166806e..ad4de66cde0b5 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 @@ -222,6 +222,11 @@ trait WindowEvaluatorFactoryBase { aggFilters) } val conf = SQLConf.get + val isMinMaxOnly = conf.windowMonotonicDequeEnabled && + functions.nonEmpty && functions.forall { + case _: Min | _: Max => true + case _ => false + } && aggFilters.forall(_.isEmpty) val blockSize = conf.windowSegmentTreeBlockSize // Create the factory to produce WindowFunctionFrame. @@ -283,12 +288,6 @@ trait WindowEvaluatorFactoryBase { // Shrinking Frame. case ("AGGREGATE", frameType, lower, UnboundedFollowing, _) => - val isMinMaxOnly = conf.windowMonotonicDequeEnabled && - functions.nonEmpty && functions.forall { - case _: Min | _: Max => true - case _ => false - } && aggFilters.forall(_.isEmpty) - if (isMinMaxOnly) { target: InternalRow => { val lb = createBoundOrdering(frameType, lower, timeZone) @@ -348,12 +347,6 @@ trait WindowEvaluatorFactoryBase { // Moving Frame. case ("AGGREGATE", frameType, lower, upper, _) => - val isMinMaxOnly = conf.windowMonotonicDequeEnabled && - functions.nonEmpty && functions.forall { - case _: Min | _: Max => true - case _ => false - } && aggFilters.forall(_.isEmpty) - if (isMinMaxOnly) { target: InternalRow => { val lb = createBoundOrdering(frameType, lower, timeZone) 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 0000000000000..20dac9c4e6d28 --- /dev/null +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/window/MonotonicDequeWindowFunctionSuite.scala @@ -0,0 +1,202 @@ +/* + * 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._ + + private val enableDeque: Map[String, String] = Map( + SQLConf.WINDOW_MONOTONIC_DEQUE_ENABLED.key -> "true") + + 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): 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: _*) { + build().collect().toSeq + } + + 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("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("Shrinking rows frame: MIN/MAX on primitives (Int/Long/Double)") { + val winSpec = Window.partitionBy($"pk").orderBy($"id") + .rowsBetween(-4, Window.unboundedFollowing) + 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("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("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("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("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) + ) + ) + } + + test("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) + ) + ) + } + + test("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) + ) + ) + } +} diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/window/SegmentTreeWindowFunctionSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/window/SegmentTreeWindowFunctionSuite.scala index 5bf74c351c08e..e8905c0483d68 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/window/SegmentTreeWindowFunctionSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/window/SegmentTreeWindowFunctionSuite.scala @@ -36,6 +36,11 @@ class SegmentTreeWindowFunctionSuite extends SharedSparkSession { import testImplicits._ + override def withSQLConf[T](confs: (String, String)*)(f: => T): T = { + val extraConfs = confs :+ (SQLConf.WINDOW_MONOTONIC_DEQUE_ENABLED.key -> "false") + super.withSQLConf(extraConfs: _*)(f) + } + // Force seg-tree path regardless of partition size (fallback exercised explicitly). private val enableSegTree: Map[String, String] = Map( SQLConf.WINDOW_SEGMENT_TREE_ENABLED.key -> "true", 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 d22c1cdc90da2..93bb9fe4b51d9 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 @@ -35,6 +35,11 @@ class SegmentTreeWindowMetricsSuite import testImplicits._ + override def withSQLConf[T](confs: (String, String)*)(f: => T): T = { + val extraConfs = confs :+ (SQLConf.WINDOW_MONOTONIC_DEQUE_ENABLED.key -> "false") + super.withSQLConf(extraConfs: _*)(f) + } + /** * Run `df` and return the first Window node's seg-tree counter values as * Long. (Other metrics like `spillSize` are UI-pretty-printed strings and diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/window/UnboundedFollowingSegmentTreeSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/window/UnboundedFollowingSegmentTreeSuite.scala index 67316ba7a0489..58a96859902cb 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/window/UnboundedFollowingSegmentTreeSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/window/UnboundedFollowingSegmentTreeSuite.scala @@ -38,6 +38,11 @@ class UnboundedFollowingSegmentTreeSuite extends SharedSparkSession { import testImplicits._ + override def withSQLConf[T](confs: (String, String)*)(f: => T): T = { + val extraConfs = confs :+ (SQLConf.WINDOW_MONOTONIC_DEQUE_ENABLED.key -> "false") + super.withSQLConf(extraConfs: _*)(f) + } + private val enableSegTree: Map[String, String] = Map( SQLConf.WINDOW_SEGMENT_TREE_ENABLED.key -> "true", SQLConf.WINDOW_SEGMENT_TREE_MIN_PARTITION_ROWS.key -> "1") 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 8dbb4ceecf1bb..a644f791dd346 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 @@ -42,7 +42,8 @@ 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 def baseDF = spark.range(0, 120).selectExpr( "id", "(id % 3) AS pk", "CAST(id AS INT) AS v", "CAST(id AS DOUBLE) AS vd")