peter-toth commented on code in PR #57346: URL: https://github.com/apache/spark/pull/57346#discussion_r3702525149
########## sql/core/src/main/scala/org/apache/spark/sql/execution/window/SlidingWindowMinMaxFunctionFrame.scala: ########## @@ -0,0 +1,278 @@ +/* + * 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(_.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 MinMaxDeque( + val isMin: Boolean, + val boundChild: Expression, + val dataType: DataType, + val ordering: Ordering[Any], + val bufferIndex: Int) { + + 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 + + private val tempRow = new SpecificInternalRow(Seq(dataType)) + private val isPrimitive = dataType match { + case BooleanType | ByteType | ShortType | IntegerType | LongType | FloatType | DoubleType | + DateType | TimestampType | TimestampNTZType | _: YearMonthIntervalType | + _: DayTimeIntervalType => true + case _ => false + } + + def clear(): Unit = { + var i = 0 + while (i < size) { + values((head + i) % capacity) = 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 + 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) % capacity) + } + + private def pollLast(): Unit = { + tail = (tail - 1 + capacity) % capacity + values(tail) = null + size -= 1 + } + + private def peekFirstIndex(): Int = { + indices(head) + } + + private def pollFirst(): Unit = { + values(head) = null + head = (head + 1) % capacity + size -= 1 + } + + private def offerLast(value: Any, index: Int): Unit = { + if (size == capacity) { + expand() + } + values(tail) = value + indices(tail) = index + tail = (tail + 1) % capacity + size += 1 + } + + 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 (!isEmpty && ordering.compare(peekLastValue(), value) >= 0) { Review Comment: **Finding 3.** Popping on `>=` (and `<=` for max) makes the deque keep the **last** of several equal-comparing values, while both existing paths keep the **first**. `Min.updateExpressions` is `least(min, child)`, and `Least.eval` is (`arithmetic.scala:1354`): ```scala if (r == null || ordering.lt(evalc, r)) evalc else r ``` so on a tie it keeps `r`, the already-accumulated value. `mergeExpressions` is `least(min.left, min.right)`, which likewise keeps the earlier block, so the segment tree agrees with naive. The deque disagrees. That only matters where `TypeUtils.getInterpretedOrdering` treats distinguishable values as equal, and there are two such cases in reach: - **Collated strings.** `PhysicalStringType.ordering` is `CollationFactory.fetchCollation(collationId).comparator` (`PhysicalDataType.scala:370`), so under `UTF8_LCASE` `'Bob'` and `'bob'` compare equal. `MIN(name)` over a window containing both returns `Bob` on the naive/segtree paths and `bob` here. Window aggregates over collated columns are explicitly supported (SPARK-47443). - **Signed zero.** `SQLOrderingUtil.compareDoubles` is `if (x == y) 0 else Double.compare(x, y)`, so `-0.0` and `0.0` tie — and `NormalizeFloatingNumbers` explicitly leaves window aggregate children alone ("we don't need to normalize the `windowExpressions`, as they are executed per input row and should take the input row as it is", `NormalizeFloatingNumbers.scala:78-80`). So `MIN(v)` over `{-0.0, 0.0}` flips sign depending on the config. Strict comparisons make the deque keep the lowest index among ties, which is exactly `Least`/`Greatest` semantics, and cost nothing asymptotically (each element is still pushed and popped once): ```scala if (isMin) { while (!isEmpty && ordering.compare(peekLastValue(), value) > 0) { pollLast() } } else { while (!isEmpty && ordering.compare(peekLastValue(), value) < 0) { pollLast() } } ``` `MonotonicDequeWindowFunctionSuite` can't see this today: `v_str` is default-collated (binary), so no ties, and there is no float column with signed zero. Please add a `UTF8_LCASE` string case and a `±0.0` double case to `checkEquivalence` — both fail on the current code. ########## sql/core/benchmarks/WindowBenchmark-results.txt: ########## @@ -2,171 +2,230 @@ Section A - MIN (non-invertible) ================================================================================================ -OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1020-azure -AMD EPYC 7763 64-Core Processor +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) 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) 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.20+8-LTS on Linux 6.17.0-1020-azure -AMD EPYC 7763 64-Core Processor +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) 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) 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.20+8-LTS on Linux 6.17.0-1020-azure -AMD EPYC 7763 64-Core Processor +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) 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) 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.20+8-LTS on Linux 6.17.0-1020-azure -AMD EPYC 7763 64-Core Processor +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) 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) 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.20+8-LTS on Linux 6.17.0-1020-azure -AMD EPYC 7763 64-Core Processor +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) 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) 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.20+8-LTS on Linux 6.17.0-1020-azure -AMD EPYC 7763 64-Core Processor +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) 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) 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.20+8-LTS on Linux 6.17.0-1020-azure -AMD EPYC 7763 64-Core Processor +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 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 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.20+8-LTS on Linux 6.17.0-1020-azure -AMD EPYC 7763 64-Core Processor +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 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 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.20+8-LTS on Linux 6.17.0-1020-azure -AMD EPYC 7763 64-Core Processor +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 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 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.20+8-LTS on Linux 6.17.0-1020-azure -AMD EPYC 7763 64-Core Processor +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 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 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.20+8-LTS on Linux 6.17.0-1020-azure -AMD EPYC 7763 64-Core Processor +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) 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) 422 451 27 2.4 422.1 1.0X Review Comment: **Finding 2.** These two Section F rows are measuring the monotonic deque, not the paths they are labelled with — the file was generated while the config still defaulted to `true`. Against the base file: | case | base | this PR | |---|---|---| | `MAX naive (String)` | 59,293 ms | 422 ms | | `MAX segtree default (String)` | 2,796 ms | 401 ms | 422 ms for 1M rows x W=1001 works out to ~0.4 ns per `UTF8String` comparison, which no hardware change explains; and the two cases previously differed 21x but now land within 5% of each other, which is what you get when both run the same code. The mechanism is that `runSpillGuard` never pins the new conf: ```scala benchmark.addCase(nNaive, numIters = ITERS_STRESS) { _ => currentCase = nNaive spark.sql(s"SELECT MAX(v) $frame FROM t").noop() // no WINDOW_MONOTONIC_DEQUE_ENABLED } ``` so under the old default it took `isMinMaxOnly`; and the segtree case took it too, because `WindowEvaluatorFactoryBase` checks `isMinMaxOnly` *before* `eligibleForSegTree`. Two asks: 1. Regenerate the file with the shipped default (`false`). Section A/B/C are unaffected because their naive cases either pin the conf or use SUM, so Section F is the only contaminated section — but it's the one the file calls the "spill regression guard". 2. Regenerate on the standard runner rather than locally. The base file is `OpenJDK 17 on Linux 6.17.0-1020-azure / AMD EPYC 7763` (the `benchmark.yml` workflow); switching every section to `OpenJDK 23 on Mac OS X / Apple M1` makes the sections this PR doesn't touch incomparable to what they replaced. Worth pinning the conf explicitly in `runSpillGuard` and `runSectionB` too, so the file can't silently drift again if the default ever flips. ########## sql/core/src/test/scala/org/apache/spark/sql/execution/benchmark/WindowBenchmark.scala: ########## @@ -146,40 +146,44 @@ 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( 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) + 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) { _ => Review Comment: **Finding 9.** Two problems from the deque case *replacing* the `blockSize=256` case rather than being added alongside it. First, coverage was deleted. The `dSegBs` digest check and the `segtree (blockSize=256)` case are gone from Section A, so nothing there exercises a non-default block size any more — while the class doc still promises it (`WindowBenchmark.scala:34`): ``` * - A: 5 aggregates x 3 cells (naive / segtree default / segtree bs=256) @ W=1001. ``` Section B only runs bs=256 at W=4001 (`stressBs = true`), so Section A was the only W=1001 data point for it. Either keep it as a fourth cell or update the doc to say it moved. Second, `runSectionA` also drives SUM, COUNT, AVG and STDDEV_SAMP, and for those `isMinMaxOnly` is `false` — so the results file now publishes four `monotonic deque (new)` rows that are just a second naive run under an irrelevant conf: ``` SUM naive (current, baseline) 3335 ... 1.0X SUM monotonic deque (new) 3298 ... 1.0X COUNT monotonic deque (new) 2854 ... 1.0X (naive: 2840) AVG monotonic deque (new) 3681 ... 1.0X (naive: 3612) STDDEV_SAMP monotonic deque (new) 105070 ... 1.0X (naive: 100017) ``` A reader can't tell those from "the deque ran and didn't help". Gating the case on `Set("MIN", "MAX").contains(aggFn)` keeps Section A honest and cuts four redundant multi-second runs (~110s of the STDDEV row alone). ########## sql/core/src/main/scala/org/apache/spark/sql/execution/window/SlidingWindowMinMaxFunctionFrame.scala: ########## @@ -0,0 +1,278 @@ +/* + * 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(_.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 MinMaxDeque( + val isMin: Boolean, + val boundChild: Expression, + val dataType: DataType, + val ordering: Ordering[Any], + val bufferIndex: Int) { + + 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 + + private val tempRow = new SpecificInternalRow(Seq(dataType)) + private val isPrimitive = dataType match { + case BooleanType | ByteType | ShortType | IntegerType | LongType | FloatType | DoubleType | + DateType | TimestampType | TimestampNTZType | _: YearMonthIntervalType | + _: DayTimeIntervalType => true + case _ => false + } + + def clear(): Unit = { + var i = 0 + while (i < size) { + values((head + i) % capacity) = 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 + 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) % capacity) + } + + private def pollLast(): Unit = { + tail = (tail - 1 + capacity) % capacity + values(tail) = null + size -= 1 + } + + private def peekFirstIndex(): Int = { + indices(head) + } + + private def pollFirst(): Unit = { + values(head) = null + head = (head + 1) % capacity + size -= 1 + } + + private def offerLast(value: Any, index: Int): Unit = { + if (size == capacity) { + expand() + } + values(tail) = value + indices(tail) = index + tail = (tail + 1) % capacity + size += 1 + } + + private def evaluateAndCopy(row: InternalRow): Any = { Review Comment: **Finding 7.** This is `InternalRow.copyValue` with two extra allocations in front of it. `tempRow.copy()` dispatches to `BaseGenericInternalRow.copy()`, whose entire body is: ```scala newValues(i) = InternalRow.copyValue(genericGet(i)) ... new GenericInternalRow(newValues) ``` so every admitted non-primitive value allocates a `GenericInternalRow` plus an `Array[Any]` just to call a function that's directly callable — on a hot per-row path. And `copyValue` already passes primitives through untouched (`case _ => value`), so the `isPrimitive` allowlist and the per-deque `tempRow` field are both doing nothing that `copyValue` doesn't: ```scala private def evaluateAndCopy(row: InternalRow): Any = InternalRow.copyValue(boundChild.eval(row)) ``` That drops `tempRow`, drops `isPrimitive`, and makes the list @Ma77Ball asked you to extend unnecessary rather than something to keep in sync — it currently omits primitive-backed types that would silently take the slow branch (`TimeType`, and any future one), and `copyValue` gets them right by construction. One note on the description while this code is in view: "skips heap allocation/copying entirely for primitive types" isn't quite what happens. `values` is `Array[Any]` and `boundChild.eval` returns `Any`, so every primitive is boxed on admit regardless of which branch it takes; what the branch saves is the `GenericInternalRow` above, not the boxing. ########## sql/core/src/test/scala/org/apache/spark/sql/execution/benchmark/WindowBenchmark.scala: ########## @@ -343,6 +396,29 @@ object WindowBenchmark extends SqlBasedBenchmark { runBenchmark("Section C - N-sweep large (stress)") { runSectionC(C_N_LARGE) } + + setupIncreasingTable(2000000L) Review Comment: **Finding 11.** Three small things in the new sections: 1. Section H calls `runSectionA`, which names its cases `s"$aggFn naive (current, baseline)"` etc. — byte-identical to Section A's. So `allCaseNames` ends up with duplicate entries, and the Memory/Spill trailer looks each name up in a single `metrics` map, merging Section A's 256K-row numbers with Section H's 2M-row numbers under one label and printing the row twice. Threading the `stressMark` argument (already a parameter, currently passed `""`) into the case names, or giving `runSectionA` a label suffix, fixes it. 2. `2000000L` and `50000` are inlined here and in the three `runSectionG` calls, while every other section uses a named constant at the top of the object (`A_N_INT`, `B_N_W10`, `C_HALF_W`, `MAIN_HALF_W`). Worth adding `G_N` / `G_HALF_W` / `H_N` / `H_HALF_W` to match. 3. Smoke mode swapped `SMOKE: Section B SUM W sweep point` for `SMOKE: Section A MAX`, so the smoke run no longer touches `runSectionB` at all. Adding the MAX smoke case is good; removing the Section B one leaves that helper unsmoked. ########## sql/core/src/main/scala/org/apache/spark/sql/execution/window/SlidingWindowMinMaxFunctionFrame.scala: ########## @@ -0,0 +1,278 @@ +/* + * 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(_.clear()) + lowerBound = 0 + + if (ubound.isEmpty) { + val iter = input.generateIterator() Review Comment: **Finding 4.** For the shrinking frame (`... BETWEEN <lower> AND UNBOUNDED FOLLOWING`) this admits the entire partition up front, and a min-deque over a non-decreasing column retains every single element: `admit` only pops while `peekLastValue() >= value`, so on non-decreasing input nothing is ever popped. `values`/`indices` therefore grow to the full partition length — one reference slot, one boxed value and one int per row **per aggregate** (~24 bytes, ~48 counting the `expand()` doubling slack, transiently ~3x while both arrays are live during the copy). That's a regression against *both* baselines, not just one: | | shrinking-frame memory | spillable | counted in `peakExecutionMemory` | |---|---|---|---| | `UnboundedFollowingWindowFunctionFrame` | none — re-scans with a fresh iterator per `write` | n/a | n/a | | `SegmentTreeWindowFunctionFrame` | one agg buffer per `blockSize` rows (default 65536) + LRU-bounded level arrays | yes | yes | | `SlidingWindowMinMaxFunctionFrame` | one boxed value + int per row, per aggregate | **no** | **no** | Worth noting which row is the real baseline: `spark.sql.window.segmentTree.enabled` is itself default-off, so for anyone on defaults this config switches away from the naive frame — the one that buffers nothing at all. `MIN(v) OVER (ORDER BY id ROWS BETWEEN 4 PRECEDING AND UNBOUNDED FOLLOWING)` over a 50M-row partition ordered ascending, an entirely ordinary shape, goes from O(1) to roughly 1.2 GB, ~2.4 GB with array slack, and doubles again if a `MAX` over a descending column shares the frame. Partitions that size are routine — `ExternalAppendOnlyUnsafeRowArray` spills precisely so they can be. And it's invisible while it happens: there's no `MemoryConsumer`, so nothing counts against the task budget and nothing can be reclaimed under pressure. This factory branch doesn't even call `TaskContext.get()`, unlike both segtree branches, which do and throw if it's absent (`WindowEvaluatorFactoryBase.scala:314-319`). For reference on the bar here: SPARK-56546 (#55422) shipped the segment tree with a `SegTreeSpiller extends MemoryConsumer` (documented invariants I1/I2/I8), a `minPartitionRows` gate, a runtime `fallbackFactory`, and two routing metrics — all in its **first** commit, all behind a default-off flag. Two ways to close this, either is fine: 1. **Preferred — decide per partition in `prepare`, and register the memory.** `prepare` already knows `input.length` before it admits anything, so the shrinking path can bail out to the fallback frame before allocating, mirroring `minPartitionRows` in the other direction. Pair that with a `MemoryConsumer` whose `spill()` returns `0L` (there's precedent for a non-spilling consumer in `WindowSegmentTree`) so the footprint at least becomes visible to the TMM and to `peakExecutionMemory`, and pressures other consumers instead of silently OOMing. 2. **Minimum scope — route only the moving frame through the deque.** There the deque is bounded by W and is strictly *less* memory than `SlidingWindowFunctionFrame`, which buffers W results of `nextRow.copy()` — a full copy of every column, not just the aggregated one. This drops the shrinking-frame win, though, and that's the bigger one given the naive path there is O(N^2), so I'd rather see option 1. Real spilling seems like a fair follow-up rather than a blocker for this PR, and it should be more tractable here than it was for the segment tree: `dropBefore` only removes from the head and `admit` only touches the tail, so the deque's interior is never read. Chunking `values`/`indices` into segments would let interior segments spill and be faulted back in strict order as `dropBefore` advances — a sequential read, no random access. Worth a separate JIRA either way. Last thing, and it's the part I'd most want fixed regardless of which option you pick: the docs currently claim the opposite of all this. The description criticises the segment tree for "High Heap Allocation & GC Pressure ... allocates a large number of node objects and tree arrays on the JVM heap" and then claims the deque delivers "a minimal memory footprint" — for the shrinking frame that inverts the actual trade-off. And the config `.doc(...)` says only "This provides O(N) complexity instead of O(N * W) or O(N log W)", which reads as a time bound, when in the shrinking case O(N) is also the *space* bound and unspillable. Both need a sentence naming the memory cost so a user can make an informed decision about turning this on. ########## 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 = { Review Comment: **Finding 5.** Differential comparison alone can't tell "the deque agrees with the baselines" from "the deque never ran". If `isMinMaxOnly` ever evaluates to `false` — a type gate added later, a refactor of the `functions.forall` match, a bad conf read — all three config combinations collapse onto the same frame and every test in this suite still passes, silently. The module already has the pattern for this. `WindowEvaluatorFactoryBase` exposes `numSegmentTreeFrames` / `numSegmentTreeFallbackFrames` (defaulting to `None` so the Arrow factory can skip them), `WindowEvaluatorFactory` wires them, and `WindowSegmentTreeAllowlistSuite` asserts routing off them: ```scala val (seg, fallback) = segTreeCounters(df) assert(seg > 0, s"$name should bump numSegmentTreeFrames (got $seg)") ``` Please add a `numMonotonicDequeFrames` metric the same way and assert `> 0` in at least one positive case here. That also closes the other half of the gap: neither routing gate has a test today, and both are one-liners against `WindowSegmentTreeAllowlistSuite`'s equivalents — - `functions.forall { case _: Min | _: Max => ... }`: `MIN(v)` and `SUM(v)` over the same window spec must land in one frame group and *not* take the deque (mirrors "mix of allowlisted + non-allowlisted aggregates falls through entirely"). - `aggFilters.forall(_.isEmpty)`: `MIN(v) FILTER (WHERE v % 2 = 0) OVER (...)` must not take it (mirrors "FILTER (WHERE ...) disables segment-tree path"). The metric is worth having for its own sake too — the segment tree reports "number of segment-tree frames prepared" in the UI, and right now there is no way for a user to tell whether this optimization fired. ########## 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), Review Comment: **Finding 6.** This is the only test that reaches `evaluateAndCopy`'s copy branch, and it can't reach the condition that branch exists for. `ExternalAppendOnlyUnsafeRowArray` hands out two different kinds of iterator. `InMemoryBufferIterator` yields rows out of `inMemoryBuffer`, which holds `row.copy()` — nothing is recycled, so a `UTF8String` read from one stays valid forever. Only `SpillableArrayIterator` recycles: ```scala private val currentRow = new UnsafeRow(numFieldPerRow) ... currentRow.pointTo(iterator.getBaseObject, iterator.getBaseOffset, iterator.getRecordLength) ``` That is the case where an uncopied `UTF8String` in the deque would dangle. Every partition in this suite is 100 rows over 3 keys (and 50 / 20 in the null tests), so nothing gets near `spark.sql.windowExec.buffer.in.memory.threshold` (4096) and the spilled iterator is never constructed. To be clear about what I'm claiming: I traced the invariant and the code is **correct** today. `SegmentTreeWindowFunctionFrame:95` spells out the contract that makes it safe — > Spill-safety invariant: when `rowArray` spills, its iterator reuses a single `UnsafeRow` whose pointer is rebound on each `next()`. Tolerated here because the cursor is **read-before-advance** [...] DO NOT cache a historical row into a separate field without an explicit `.copy()`; the shared reusable UnsafeRow would silently mutate. and this frame satisfies it: both cursors read `lowerRow` / `nextRow` for comparison before calling `getNextOrNull`, and `admit` copies the value out rather than retaining the row. So this is a coverage-and-documentation gap, not a live bug. But it's an unguarded one — the sibling frame documents that contract in capitals because it's easy to break, and nothing here records that it was considered or fails if someone breaks it. Two small asks. First, a comment on the `lowerIterator` / `inputIterator` declarations pointing at the same invariant. Second, lowering the thresholds on this test, which covers it without a big table: ```scala 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))) } } ``` Worth extending to the complex types if you want them in scope — `ArrayType`/`StructType` are orderable and so valid for `MIN`/`MAX`, they take the same copy branch, and neither is covered. (Unrelated but while you're here: the existing `test("...")` names in this suite should carry the `SPARK-58201` prefix, per the convention in the surrounding window suites.) ########## sql/core/src/main/scala/org/apache/spark/sql/execution/window/SlidingWindowMinMaxFunctionFrame.scala: ########## @@ -0,0 +1,278 @@ +/* + * 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(_.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) Review Comment: **Finding 8.** This is the second caller of `AggregateProcessor.evaluate(source, target)`, and its Scaladoc now documents only the first one: > **Contract**: `source` must share this processor's internal `aggBufferAttributes` layout [...] The segment-tree path enforces this upstream in `WindowEvaluatorFactoryBase.eligibleForSegTree`, which restricts eligible functions to `WindowSegmentTree.EligibleAggregates` [...] The contract is invisible at the call site and easy to break from either end. The contract does hold here, but only because of a coincidence worth writing down: `Min`/`Max` each contribute exactly one `aggBufferAttributes` entry, typed `child.dataType`, which is also `Min.dataType`/`Max.dataType` — so `new SpecificInternalRow(functions.map(_.dataType))` happens to reproduce the buffer layout exactly, and neither function is a `SizeBasedWindowFunction`, so nothing is prepended. Change either side of that (a `Min` variant whose `dataType` differs from its buffer type, a second buffer slot) and the `assert` in `evaluate` is all that catches it. Please extend that doc to name `isMinMaxOnly` as the second gate, and add a line here recording why `sourceRow`'s schema is layout-compatible. That's precisely the drift the original comment was written to guard against. ########## 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 = { Review Comment: **Finding 13.** With the shipped default of `false` this override (and the identical ones in `SegmentTreeWindowMetricsSuite:38` and `UnboundedFollowingSegmentTreeSuite:41`, plus the entry added to `WindowSegmentTreeAllowlistSuite.enableSegTree`) is a no-op — it forces a value the conf already has. It also has a cost: because the extra pair is *appended*, it wins over anything the caller passes, so no test in these three suites can enable the deque any more. If a future test wants to check that a segtree-eligible query prefers the deque, it will silently measure the wrong thing. Two ways out, either is fine — drop the overrides and rely on the default, or keep them but prepend instead of append so a caller can still opt in: ```scala override def withSQLConf[T](confs: (String, String)*)(f: => T): T = super.withSQLConf((SQLConf.WINDOW_MONOTONIC_DEQUE_ENABLED.key -> "false") +: confs: _*)(f) ``` Worth noting these overrides only earn their keep if the default is meant to be `true` — which, together with the description and the Section F benchmark rows (finding 2), is the third artifact in this PR that still assumes the pre-review default. ########## sql/core/src/test/scala/org/apache/spark/sql/execution/benchmark/WindowBenchmark.scala: ########## @@ -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)") { Review Comment: **Finding 10.** W=11 is the narrowest window measured, and the margin is already thinning there: 1.7X over naive for MIN and 1.2X for MAX. The two most common bounded frames in real queries are narrower than that — `ROWS BETWEEN 1 PRECEDING AND CURRENT ROW` (W=2) and `ROWS BETWEEN CURRENT ROW AND CURRENT ROW` (W=1) — and neither is measured. That matters because `isMinMaxOnly` has no width gate at all: with the conf on, *every* `MIN`/`MAX` moving frame takes the deque regardless of W. And the per-row work isn't obviously cheaper at small W — the deque does an interpreted `BoundReference.eval`, boxes the result, and calls `Ordering[Any].compare` through a megamorphic call site, against the naive path's codegen'd `MutableProjection` over W buffered rows. At W=1 the naive path does one codegen'd update per row. The segment tree guards exactly this shape with `spark.sql.window.segmentTree.minPartitionRows`, and the description leans on "Monotonic Deque completely resolves the Segment Tree's deoptimization" — a W=1/W=3 row in Section H would either back that up across the whole range or show where a width gate is needed. Given the conf is default-off this doesn't block merging, but it's the data needed before it can ever be flipped on. ########## sql/core/src/main/scala/org/apache/spark/sql/execution/window/SlidingWindowMinMaxFunctionFrame.scala: ########## @@ -0,0 +1,278 @@ +/* + * 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(_.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 MinMaxDeque( + val isMin: Boolean, + val boundChild: Expression, + val dataType: DataType, + val ordering: Ordering[Any], + val bufferIndex: Int) { Review Comment: **Finding 12.** `bufferIndex` is never read — the deques are indexed by their position in the `deques` array at the two use sites (`write` walks `i < deques.length`, and `prepare`/`write` use `foreach`), so the constructor parameter and the `.zipWithIndex` that feeds it can both go: ```scala 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), child.dataType, TypeUtils.getInterpretedOrdering(child.dataType)) } ``` While editing the signature: none of these need to be `val`s. `MinMaxDeque` is a private inner class and nothing outside reads `isMin` / `boundChild` / `dataType` / `ordering`, so plain constructor params keep them off the class as fields. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
