peter-toth commented on code in PR #57346: URL: https://github.com/apache/spark/pull/57346#discussion_r3780030872
########## sql/core/src/main/scala/org/apache/spark/sql/execution/window/SlidingWindowMinMaxFunctionFrame.scala: ########## @@ -0,0 +1,280 @@ +/* + * 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.execution.metric.SQLMetric +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]]. + * + * This frame is only instantiated when `isMinMaxOnly` is true (all window functions are Min or + * Max and no FILTER clause is used), enforced upstream in [[WindowEvaluatorFactoryBase]]. + */ +private[window] final class SlidingWindowMinMaxFunctionFrame( + target: InternalRow, + processor: AggregateProcessor, + lbound: BoundOrdering, + ubound: BoundOrdering, + functions: Array[Expression], + inputSchema: Seq[Attribute], + numMonotonicDequeFrames: Option[SQLMetric] = None) + extends WindowFunctionFrame { + + /** Rows of the partition currently being processed. */ + private[this] var input: ExternalAppendOnlyUnsafeRowArray = null + + // Spill-safety: when `input` (ExternalAppendOnlyUnsafeRowArray) spills, its + // iterator reuses a single UnsafeRow whose pointer is rebound on each next(). + // This is safe because both cursors follow a read-before-advance pattern: + // `lowerRow`/`nextRow` are used for comparison *before* calling getNextOrNull. + // Values are extracted from the row via `evaluateAndCopy` before advancing. + // DO NOT cache a historical row without an explicit .copy(); the shared + // reusable UnsafeRow would silently mutate. + private[this] var lowerIterator: Iterator[UnsafeRow] = _ + private[this] var inputIterator: Iterator[UnsafeRow] = _ + + /** 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 + + // `sourceRow` is used as the `source` argument to `processor.evaluate(source, target)`. + // Layout compatibility is guaranteed because Min/Max each contribute exactly one + // `aggBufferAttributes` entry typed `child.dataType`, which equals `Min/Max.dataType`. + // Neither is a `SizeBasedWindowFunction`, so no extra slot is prepended. + // `isMinMaxOnly` (enforced in WindowEvaluatorFactoryBase) ensures this invariant holds. + private[this] val sourceRow = new SpecificInternalRow(functions.map(_.dataType).toIndexedSeq) + + // Each deque is addressed by its position in this array (one entry per Min/Max function), + // so no separate per-deque ordinal is needed. + private[this] val deques: Array[MinMaxDeque] = functions.map { func => + val isMin = func.isInstanceOf[Min] + val child = func match { + case m: Min => m.child + case m: Max => m.child + } + new MinMaxDeque( + isMin, + BindReferences.bindReference(child, inputSchema), + child.dataType, + TypeUtils.getInterpretedOrdering(child.dataType)) + } + + override def prepare(rows: ExternalAppendOnlyUnsafeRowArray): Unit = { + numMonotonicDequeFrames.foreach(_ += 1) + input = rows + lowerIterator = input.generateIterator() + lowerRow = WindowFunctionFrame.getNextOrNull(lowerIterator) + var di = 0 + while (di < deques.length) { deques(di).clear(); di += 1 } + lowerBound = 0 + + inputIterator = input.generateIterator() + nextRow = WindowFunctionFrame.getNextOrNull(inputIterator) + upperBound = 0 + } + + override def write(index: Int, current: InternalRow): Unit = { + var bufferUpdated = index == 0 + + // Drop all rows from the buffer for which the input row value is smaller than + // the output row lower bound. + while (lowerBound < upperBound && lbound.compare(lowerRow, lowerBound, current, index) < 0) { + lowerBound += 1 + lowerRow = WindowFunctionFrame.getNextOrNull(lowerIterator) + bufferUpdated = true + } + + // Add all rows to the buffer for which the input row value is equal to or less than + // the output row upper bound. + while (nextRow != null && ubound.compare(nextRow, upperBound, current, index) <= 0) { + if (lbound.compare(nextRow, lowerBound, current, index) < 0) { + lowerBound += 1 + lowerRow = WindowFunctionFrame.getNextOrNull(lowerIterator) + } else { + var di = 0 + while (di < deques.length) { deques(di).admit(nextRow, upperBound); di += 1 } + bufferUpdated = true + } + nextRow = WindowFunctionFrame.getNextOrNull(inputIterator) + upperBound += 1 + } + + if (bufferUpdated) { + var di = 0 + while (di < deques.length) { deques(di).dropBefore(lowerBound); di += 1 } + } + + // Write output values to target. + // See sourceRow comment above for why evaluate(sourceRow, target) is safe here. + if (processor != null && bufferUpdated) { + var i = 0 + while (i < deques.length) { + sourceRow.update(i, deques(i).currentValue()) + i += 1 + } + processor.evaluate(sourceRow, target) + } + } + + override def currentLowerBound(): Int = lowerBound + + override def currentUpperBound(): Int = upperBound + + // MinMaxDeque fields are plain constructor params (not vals) since this is a private inner + // class and nothing outside reads them. + private class MinMaxDeque( + isMin: Boolean, + boundChild: Expression, + dataType: DataType, Review Comment: **Finding 7.** `dataType` is never read anywhere in `MinMaxDeque` -- it is the last piece of the `isPrimitive` allowlist you removed in `5489b88b9c0`. Dropping the parameter also makes the `org.apache.spark.sql.types._` import unused, since `DataType` is the only name it supplies in this file. The reason I still have this Blocking is the other half of the finding, which lives in the PR body rather than the code. "Key improvements" still says: > **Zero-Copy Optimization for Primitive Types**: Internal rows in Spark (`UnsafeRow`) recycle memory [...] We introduce a type check that skips heap allocation/copying entirely for primitive types. There is no such type check any more. `evaluateAndCopy` (`:242`) calls `InternalRow.copyValue` unconditionally, and `values` is an `Array[Any]`, so every primitive is boxed on the way in regardless of any branch. Removing it was the right call -- the branch only skipped a `copyValue` that returns primitives as-is -- but a reader who takes the description at face value will go hunting for code that isn't there. Please drop the bullet. `MonotonicDequeWindowFunctionSuite:317` has the same leftover: "regardless of the isPrimitive branch" names a branch that no longer exists. ########## sql/core/benchmarks/WindowBenchmark-results.txt: ########## @@ -3,170 +3,292 @@ 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 +AMD EPYC 9V74 80-Core Processor MIN sliding window, W=1001, 256K rows: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative ------------------------------------------------------------------------------------------------------------------------ -MIN naive (current, baseline) 4014 4035 17 0.1 15313.6 1.0X -MIN segtree (default) 392 407 16 0.7 1496.8 10.2X -MIN segtree (blockSize=256) 2199 2214 12 0.1 8388.4 1.8X +MIN naive (current, baseline) 4024 4041 15 0.1 15349.9 1.0X +MIN segtree (default) 385 405 12 0.7 1469.8 10.4X +MIN segtree (blockSize=256) 2350 2358 5 0.1 8965.1 1.7X +MIN monotonic deque (new) 116 122 10 2.3 440.9 34.8X ================================================================================================ 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 +AMD EPYC 9V74 80-Core Processor MAX sliding window, W=1001, 256K rows: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative ------------------------------------------------------------------------------------------------------------------------ -MAX naive (current, baseline) 4244 4266 21 0.1 16190.5 1.0X -MAX segtree (default) 367 373 4 0.7 1401.2 11.6X -MAX segtree (blockSize=256) 2231 2238 12 0.1 8510.3 1.9X +MAX naive (current, baseline) 4414 4426 16 0.1 16838.4 1.0X +MAX segtree (default) 409 421 9 0.6 1558.5 10.8X +MAX segtree (blockSize=256) 2350 2362 21 0.1 8965.6 1.9X +MAX monotonic deque (new) 109 115 7 2.4 415.0 40.6X ================================================================================================ 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 +AMD EPYC 9V74 80-Core Processor SUM sliding window, W=1001, 256K rows: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative ------------------------------------------------------------------------------------------------------------------------ -SUM naive (current, baseline) 4111 4131 30 0.1 15683.8 1.0X -SUM segtree (default) 358 365 8 0.7 1364.2 11.5X -SUM segtree (blockSize=256) 2228 2245 12 0.1 8498.3 1.8X +SUM naive (current, baseline) 4195 4239 87 0.1 16003.1 1.0X +SUM segtree (default) 382 384 2 0.7 1455.3 11.0X +SUM segtree (blockSize=256) 2344 2358 11 0.1 8939.8 1.8X ================================================================================================ 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 +AMD EPYC 9V74 80-Core Processor COUNT sliding window, W=1001, 256K rows: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative ------------------------------------------------------------------------------------------------------------------------ -COUNT naive (current, baseline) 3663 3690 22 0.1 13974.5 1.0X -COUNT segtree (default) 325 335 9 0.8 1238.6 11.3X -COUNT segtree (blockSize=256) 2161 2165 6 0.1 8242.5 1.7X +COUNT naive (current, baseline) 3793 3820 37 0.1 14468.3 1.0X +COUNT segtree (default) 343 353 14 0.8 1307.2 11.1X +COUNT segtree (blockSize=256) 2301 2310 7 0.1 8777.8 1.6X ================================================================================================ 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 +AMD EPYC 9V74 80-Core Processor AVG sliding window, W=1001, 192K rows: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative ------------------------------------------------------------------------------------------------------------------------ -AVG naive (current, baseline) 4441 4463 19 0.0 22588.2 1.0X -AVG segtree (default) 337 340 4 0.6 1713.3 13.2X -AVG segtree (blockSize=256) 1398 1415 14 0.1 7111.1 3.2X +AVG naive (current, baseline) 4853 4867 11 0.0 24685.6 1.0X +AVG segtree (default) 347 351 5 0.6 1764.2 14.0X +AVG segtree (blockSize=256) 1480 1490 11 0.1 7525.9 3.3X ================================================================================================ Section A - STDDEV_SAMP (multi-buffer, stress) ================================================================================================ OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1020-azure -AMD EPYC 7763 64-Core Processor +AMD EPYC 9V74 80-Core Processor STDDEV_SAMP sliding window, W=1001, 2M rows (stress): Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative ------------------------------------------------------------------------------------------------------------------------------------ -STDDEV_SAMP naive (current, baseline) 102720 102801 91 0.0 51360.1 1.0X -STDDEV_SAMP segtree (default) 6107 6132 35 0.3 3053.4 16.8X -STDDEV_SAMP segtree (blockSize=256) 113831 113863 29 0.0 56915.6 0.9X +STDDEV_SAMP naive (current, baseline) 93930 94004 65 0.0 46965.0 1.0X +STDDEV_SAMP segtree (default) 6079 6127 64 0.3 3039.4 15.5X +STDDEV_SAMP segtree (blockSize=256) 119939 120538 870 0.0 59969.4 0.8X ================================================================================================ 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 +AMD EPYC 9V74 80-Core Processor SUM scaling, W=11, 2M rows (stress): Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative ------------------------------------------------------------------------------------------------------------------------ -SUM naive W=11 959 964 4 2.1 479.6 1.0X -SUM segtree (default) W=11 1926 1933 9 1.0 963.2 0.5X +SUM naive W=11 932 938 7 2.1 466.0 1.0X +SUM segtree (default) W=11 2007 2018 14 1.0 1003.3 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 +AMD EPYC 9V74 80-Core Processor SUM scaling, W=51, 2M rows (stress): Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative ------------------------------------------------------------------------------------------------------------------------ -SUM naive W=51 2226 2242 27 0.9 1113.0 1.0X -SUM segtree (default) W=51 2193 2203 12 0.9 1096.5 1.0X +SUM naive W=51 2229 2248 23 0.9 1114.6 1.0X +SUM segtree (default) W=51 2282 2298 17 0.9 1141.0 1.0X ================================================================================================ 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 +AMD EPYC 9V74 80-Core Processor SUM scaling, W=201, 1M rows: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative ------------------------------------------------------------------------------------------------------------------------ -SUM naive W=201 3466 3556 187 0.3 3466.0 1.0X -SUM segtree (default) W=201 1224 1232 7 0.8 1224.1 2.8X +SUM naive W=201 3545 3563 18 0.3 3545.3 1.0X +SUM segtree (default) W=201 1289 1304 10 0.8 1289.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 +AMD EPYC 9V74 80-Core Processor SUM scaling, W=4001, 2M rows (stress): Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative ------------------------------------------------------------------------------------------------------------------------ -SUM naive W=4001 125108 125307 192 0.0 62553.9 1.0X -SUM segtree (default) W=4001 3308 3320 21 0.6 1653.9 37.8X -SUM segtree (blockSize=256) W=4001 110622 111593 1667 0.0 55311.1 1.1X +SUM naive W=4001 127772 127823 46 0.0 63886.1 1.0X +SUM segtree (default) W=4001 3428 3532 174 0.6 1713.9 37.3X +SUM segtree (blockSize=256) W=4001 116806 116911 91 0.0 58403.2 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 +AMD EPYC 9V74 80-Core Processor MAX String spill guard, W=1001, 1M rows (stress): Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative -------------------------------------------------------------------------------------------------------------------------------- -MAX naive (String) 59293 59370 67 0.0 59292.7 1.0X -MAX segtree default (String) 2796 2808 16 0.4 2796.2 21.2X +MAX naive (String) 50895 50921 39 0.0 50895.3 1.0X +MAX segtree default (String) 2731 2740 9 0.4 2730.8 18.6X ================================================================================================ Section C - N-sweep small (stress) ================================================================================================ OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1020-azure -AMD EPYC 7763 64-Core Processor +AMD EPYC 9V74 80-Core Processor SUM N-sweep (segtree-only), W=1001, 2M rows (stress): Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative ------------------------------------------------------------------------------------------------------------------------------------ -SUM segtree (default) N=2M 2804 2815 16 0.7 1402.1 1.0X +SUM segtree (default) N=2M 2916 2934 23 0.7 1458.1 1.0X ================================================================================================ Section C - N-sweep mid (stress) ================================================================================================ OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1020-azure -AMD EPYC 7763 64-Core Processor +AMD EPYC 9V74 80-Core Processor SUM N-sweep (segtree-only), W=1001, 8M rows (stress): Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative ------------------------------------------------------------------------------------------------------------------------------------ -SUM segtree (default) N=8M 13613 13648 32 0.6 1701.6 1.0X +SUM segtree (default) N=8M 14217 14288 63 0.6 1777.2 1.0X ================================================================================================ Section C - N-sweep large (stress) ================================================================================================ OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1020-azure -AMD EPYC 7763 64-Core Processor +AMD EPYC 9V74 80-Core Processor SUM N-sweep (segtree-only), W=1001, 16M rows (stress): Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative ------------------------------------------------------------------------------------------------------------------------------------- -SUM segtree (default) N=16M 33919 33948 49 0.5 2119.9 1.0X +SUM segtree (default) N=16M 35566 35607 36 0.4 2222.9 1.0X + + +================================================================================================ +Section G - MIN Monotonic Deque vs Segment Tree (Worst-Case: Increasing) +================================================================================================ + +OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1020-azure +AMD EPYC 9V74 80-Core Processor +MIN sliding window (Increasing), W=100001, 2M rows: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative +---------------------------------------------------------------------------------------------------------------------------------- +MIN segtree (Increasing) 3938 3956 12 0.5 1968.9 1.0X +MIN monotonic deque (Increasing) 595 607 12 3.4 297.7 6.6X + + +================================================================================================ +Section G - MIN Monotonic Deque vs Segment Tree (Best-Case: Decreasing) +================================================================================================ + +OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1020-azure +AMD EPYC 9V74 80-Core Processor +MIN sliding window (Decreasing), W=100001, 2M rows: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative +---------------------------------------------------------------------------------------------------------------------------------- +MIN segtree (Decreasing) 3934 3973 25 0.5 1967.0 1.0X +MIN monotonic deque (Decreasing) 586 592 7 3.4 292.9 6.7X + + +================================================================================================ +Section G - MIN Monotonic Deque vs Segment Tree (Random) +================================================================================================ + +OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1020-azure +AMD EPYC 9V74 80-Core Processor +MIN sliding window (Random), W=100001, 2M rows: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative +------------------------------------------------------------------------------------------------------------------------------ +MIN segtree (Random) 4030 4033 4 0.5 2015.2 1.0X +MIN monotonic deque (Random) 625 642 15 3.2 312.6 6.4X + + +================================================================================================ +Section H - MIN W=1 scaling (2M rows) +================================================================================================ + +OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1020-azure +AMD EPYC 9V74 80-Core Processor +MIN sliding window, W=1, 2M rows: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative +------------------------------------------------------------------------------------------------------------------------ +MIN naive (current, baseline) [H W=1] 602 617 23 3.3 301.2 1.0X +MIN segtree (default) [H W=1] 1575 1587 10 1.3 787.6 0.4X +MIN segtree (blockSize=256) [H W=1] 58079 58108 25 0.0 29039.7 0.0X Review Comment: **Finding 2.** This row and the five like it (`MIN`/`MAX` x `[H W=1]`/`[H W=3]`/`[H W=11]` -- lines 221, 235, 249, 263, 277, 291) cannot be produced by the benchmark code at this head. The file was generated at `7dffef1882e`, where Section H still called `runSectionA` with the `blockSize=256` cell always on -- which is exactly what finding 19 asked you to stop doing. The fix landed one commit later in `5489b88b9c0`, adding the `withSegBs` parameter and passing `withSegBs = false` for all six Section H runs. With that, `benchmark.addCase(nSegBs, ...)` is never reached and `nSegBs` never enters `allCaseNames`, so regenerating drops these six rows. Worth saying explicitly: the *previous* cause of this finding is genuinely fixed. Section F now reads 50,895ms naive vs 2,731ms segtree, which is the shape I expected once the conf was pinned. This is a fresh regeneration on top of a correct fix, not a re-fix. ########## sql/core/src/test/scala/org/apache/spark/sql/execution/window/MonotonicDequeWindowFunctionSuite.scala: ########## @@ -0,0 +1,341 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.sql.execution.window + +import org.apache.spark.sql.{DataFrame, QueryTest, Row} +import org.apache.spark.sql.expressions.Window +import org.apache.spark.sql.functions._ +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.test.SharedSparkSession + + +/** + * Correctness tests verifying the monotonic deque-based sliding window frame optimization. Runs + * differential testing to ensure equivalence between: + * 1. Monotonic Deque (Enabled) + * 2. Segment Tree (Deque disabled, SegTree enabled) + * 3. Naive Baseline (Both disabled) + */ +class MonotonicDequeWindowFunctionSuite extends QueryTest with SharedSparkSession { + + import testImplicits._ + + // Disable AQE so executedPlan.collect can descend into WindowExec without being + // blocked by AdaptiveSparkPlanExec (a LeafExecNode). This matches SegmentTreeWindowMetricsSuite. + private val enableDeque: Map[String, String] = Map( + SQLConf.WINDOW_MONOTONIC_DEQUE_ENABLED.key -> "true", + SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") + + private val disableDequeSegTree: Map[String, String] = Map( + SQLConf.WINDOW_MONOTONIC_DEQUE_ENABLED.key -> "false", + SQLConf.WINDOW_SEGMENT_TREE_ENABLED.key -> "true", + SQLConf.WINDOW_SEGMENT_TREE_MIN_PARTITION_ROWS.key -> "1") + + private val disableDequeNaive: Map[String, String] = Map( + SQLConf.WINDOW_MONOTONIC_DEQUE_ENABLED.key -> "false", + SQLConf.WINDOW_SEGMENT_TREE_ENABLED.key -> "false") + + /** Build `df` thrice (Deque, SegTree, Naive) and assert equal results. */ + private def checkEquivalence(build: () => DataFrame, expectDeque: Boolean = true): Unit = { + val naiveResult: Seq[Row] = withSQLConf(disableDequeNaive.toSeq: _*) { + build().collect().toSeq + } + val segTreeResult: Seq[Row] = withSQLConf(disableDequeSegTree.toSeq: _*) { + build().collect().toSeq + } + val dequeResult: Seq[Row] = withSQLConf(enableDeque.toSeq: _*) { + val df = build() + val res = df.collect().toSeq + + // Verify routing actually hit (or didn't hit) the deque. + // Use the registered metric key "numMonotonicDequeFrames" (not the display name). + val windowNodes = df.queryExecution.executedPlan.collect { + case w: WindowExec => w + } + assert(windowNodes.nonEmpty, "No WindowExec found in the query plan") + val dequeCount = + windowNodes.flatMap(_.metrics.get("numMonotonicDequeFrames").map(_.value)).sum + + if (expectDeque) { + assert(dequeCount > 0, "Monotonic deque was enabled but no frames were routed to it") + } else { + assert(dequeCount == 0, "Monotonic deque was used but expected to fallback") + } + res + } + + QueryTest.sameRows(naiveResult, dequeResult, isSorted = false).foreach { err => + fail(s"Monotonic Deque output differs from Naive baseline.\n$err") + } + QueryTest.sameRows(segTreeResult, dequeResult, isSorted = false).foreach { err => + fail(s"Monotonic Deque output differs from Segment Tree baseline.\n$err") + } + } + + private def baseDF: DataFrame = { + spark + .range(0, 100) + .selectExpr( + "id", + "(id % 3) AS pk", + "CAST(id AS INT) AS v_int", + "CAST(id AS LONG) AS v_long", + "CAST(id AS DOUBLE) AS v_double", + "CAST(id AS STRING) AS v_str") + } + + test("SPARK-58201: Moving rows frame: MIN/MAX on primitives (Int/Long/Double)") { + val winSpec = Window.partitionBy($"pk").orderBy($"id").rowsBetween(-3, 2) + checkEquivalence(() => + baseDF.select( + $"id", + min($"v_int").over(winSpec), + max($"v_int").over(winSpec), + min($"v_long").over(winSpec), + max($"v_long").over(winSpec), + min($"v_double").over(winSpec), + max($"v_double").over(winSpec))) + } + + test("SPARK-58201: Fallback for mixed aggregates (SUM + MIN/MAX)") { + val winSpec = Window.partitionBy($"pk").orderBy($"id").rowsBetween(-4, 2) + checkEquivalence(() => + baseDF.select( + $"id", + min($"v_int").over(winSpec), + sum($"v_int").over(winSpec)), + expectDeque = false) + } + + test("SPARK-58201: Fallback for FILTER clauses") { + val df = spark.sql("""SELECT id, + | MIN(id) FILTER (WHERE id % 2 = 0) OVER ( + | PARTITION BY (id % 3) ORDER BY id ROWS BETWEEN 2 PRECEDING AND 2 FOLLOWING + | ) AS v + |FROM RANGE(0, 20)""".stripMargin) + // Deque shouldn't be used since FILTER is not supported. We can't use checkEquivalence + // because checkEquivalence builds DF inside, so we'll just check metrics. + withSQLConf( + SQLConf.WINDOW_MONOTONIC_DEQUE_ENABLED.key -> "true", + SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") { + val res = df.collect() + val windowNodes = df.queryExecution.executedPlan.collect { + case w: WindowExec => w + } + val dequeCount = + windowNodes.flatMap(_.metrics.get("numMonotonicDequeFrames").map(_.value)).sum + assert(dequeCount == 0, "Monotonic deque was used for FILTER clause") + } + } + + test("SPARK-58201: Moving rows frame: MIN/MAX on reference types (String)") { + val winSpec = Window.partitionBy($"pk").orderBy($"id").rowsBetween(-2, 3) + checkEquivalence(() => + baseDF.select($"id", min($"v_str").over(winSpec), max($"v_str").over(winSpec))) + } + + test("SPARK-58201: MIN/MAX on Date and Timestamp types") { + val df = baseDF.selectExpr( + "id", + "pk", + "CAST(id * 24 * 3600 AS TIMESTAMP) AS v_ts", + "date_add(to_date('1970-01-01'), CAST(id AS INT)) AS v_date") + val winSpec = Window.partitionBy($"pk").orderBy($"id").rowsBetween(-2, 2) + checkEquivalence(() => + df.select( + $"id", + min($"v_ts").over(winSpec), + max($"v_ts").over(winSpec), + min($"v_date").over(winSpec), + max($"v_date").over(winSpec))) + } + + test("SPARK-58201: MIN/MAX on Interval types (YearMonthIntervalType and DayTimeIntervalType)") { + val df = baseDF.selectExpr( + "id", + "pk", + "make_ym_interval(0, CAST(id AS INT)) AS v_ym", + "make_dt_interval(CAST(id AS INT), 0, 0, 0) AS v_dt") + val winSpec = Window.partitionBy($"pk").orderBy($"id").rowsBetween(-3, 3) + checkEquivalence(() => + df.select( + $"id", + min($"v_ym").over(winSpec), + max($"v_ym").over(winSpec), + min($"v_dt").over(winSpec), + max($"v_dt").over(winSpec))) + } + + test("SPARK-58201: MIN/MAX with null values in partition") { + val df = spark + .range(0, 50) + .selectExpr("id", "(id % 2) AS pk", "IF(id % 5 == 0, null, CAST(id AS INT)) AS v") + val winSpec = Window.partitionBy($"pk").orderBy($"id").rowsBetween(-2, 2) + checkEquivalence(() => df.select($"id", min($"v").over(winSpec), max($"v").over(winSpec))) + } + + test("SPARK-58201: MIN/MAX on all-null partition") { + val df = spark.range(0, 20).selectExpr("id", "1 AS pk", "CAST(null AS INT) AS v") + val winSpec = Window.partitionBy($"pk").orderBy($"id").rowsBetween(-2, 2) + checkEquivalence(() => df.select($"id", min($"v").over(winSpec), max($"v").over(winSpec))) + } + + test("SPARK-58201: Range-based moving frame: MIN/MAX on primitive types") { + val df = baseDF.selectExpr("id", "pk", "CAST(id / 2 AS INT) AS ord_val", "v_int") + val winSpec = Window.partitionBy($"pk").orderBy($"ord_val").rangeBetween(-2, 2) + checkEquivalence(() => + df.select($"id", min($"v_int").over(winSpec), max($"v_int").over(winSpec))) + } + + // Finding 3: verify strict inequality preserves first-of-equals behavior Review Comment: **Finding 22.** These comments cite review-finding numbers: `Finding 3` here, `Finding 6` at `:241`, `Finding 18` at `:281`, `:288`, `:295`, `:302`, `:309`, and `Finding 7 + 18` at `:317`. The numbers only mean something inside this PR's review thread -- a year from now they read as a dangling reference. The explanations after them are genuinely useful and worth keeping; just drop the `Finding N:` prefix, or point at `SPARK-58201` instead. ########## sql/core/src/main/scala/org/apache/spark/sql/execution/window/WindowEvaluatorFactory.scala: ########## @@ -32,12 +32,15 @@ class WindowEvaluatorFactory( val childOutput: Seq[Attribute], val spillSize: SQLMetric, segmentTreeFrames: SQLMetric, - segmentTreeFallbackFrames: SQLMetric) - extends PartitionEvaluatorFactory[InternalRow, InternalRow] with WindowEvaluatorFactoryBase { + segmentTreeFallbackFrames: SQLMetric, + monotonicDequeFrames: SQLMetric) + extends PartitionEvaluatorFactory[InternalRow, InternalRow] Review Comment: **Finding 16.** Thanks for reverting the four segment-tree suites -- that was the bulk of it. What is left is much smaller but still reformat-only, and this line is the clearest case: the previous ` extends PartitionEvaluatorFactory[InternalRow, InternalRow] with WindowEvaluatorFactoryBase {` was 95 characters, so nothing forced a split, and both the 4-space `extends` and the separate `with` line go against the 2-space continuation every sibling in this package uses (`SlidingWindowFunctionFrame`, `SegmentTreeWindowFunctionFrame`, `WindowExec`). All this file actually needs is the new constructor param and the `override def`. The rest, none of it touching deque behaviour: - `WindowBenchmark.scala` -- `runSectionB`'s signature exploded to one param per line with no signature change (`:253`); `spark.range(n)` -> `spark` + `.range(n)` in the four table setups; `new Benchmark(...)` split across three lines in four places; three `require(...)` calls split; and the aligned trailing comments on `A_N_INT` / `A_N_AVG` / `A_N_STDDEV` / `C_HALF_W` / `MAIN_HALF_W` collapsed to a single space. - `WindowSegmentTreeAllowlistSuite.scala` -- Scaladoc re-wrapped, `baseDF` and `total(...)` turned into leading-dot chains, the 13-element `Seq(...).foreach` re-indented wholesale, two `assert(x == 0, msg)` calls split. Separately, `test("FILTER (WHERE ...) disables segment-tree path")` traded `withTempView("t")` for a manual `try`/`finally` + `dropTempView`; `checkFallbackEquivalence` works fine inside `withTempView`, so that can go back too. (The `}` re-indent at `:75` in this file I'd keep -- that one was genuinely mis-indented before.) At this size I don't think it blocks merge any more, but it will sit in these files forever, so I'd still rather it went with the PR that introduced it. -- 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]
