peter-toth commented on code in PR #57346: URL: https://github.com/apache/spark/pull/57346#discussion_r3726648942
########## sql/core/src/test/scala/org/apache/spark/sql/execution/window/MonotonicDequeWindowFunctionSuite.scala: ########## @@ -0,0 +1,271 @@ +/* + * 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, 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 + val windowNodes = df.queryExecution.executedPlan.collect { Review Comment: **Finding 5.** The routing assertion I asked for is here, but it can never pass, and it takes the whole differential suite with it: 13 of these 14 tests fail. CI already shows it -- `70b7b0f` has 2 failing checks ("Report test results", "Build modules: sql - other tests") with 13 annotations, every one `List() was empty No WindowExec found in the query plan`. The current head only adds `WindowBenchmark-results.txt` on top of that commit, so it fails identically. Two independent causes: 1. **AQE.** `AdaptiveSparkPlanExec extends LeafExecNode` (`AdaptiveSparkPlanExec.scala:76`), so `executedPlan.collect { case w: WindowExec => ... }` never descends into it and `windowNodes` is always empty. This is exactly why the three sibling suites route through `SparkPlanInfo.fromSparkPlan` instead -- it has a `case a: AdaptiveSparkPlanExec => a.executedPlan :: Nil` branch -- and why `SegmentTreeWindowMetricsSuite` additionally pins `ADAPTIVE_EXECUTION_ENABLED -> "false"`. 2. **Metric key.** `WindowExec.metrics` registers `"numMonotonicDequeFrames"` (`WindowExec.scala:97`); this reads `"monotonicDequeFrames"`, so `metrics.get` returns `None` and `dequeCount` is always `0`. `WindowSegmentTreeAllowlistSuite.dequeCounters` gets this right only because it looks up the *display* name (`"number of monotonic-deque frames prepared"`), not the map key. The second is masked by the first -- line 65 fails before line 66 runs -- so fixing one alone just moves the failure to `assert(dequeCount > 0)`. ```scala // Verify routing actually hit (or didn't hit) the deque val windowNodes = stripAQEPlan(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 ``` `stripAQEPlan` needs `with AdaptiveSparkPlanHelper` on the suite; adding `ADAPTIVE_EXECUTION_ENABLED -> "false"` to `enableDeque` works just as well and matches `SegmentTreeWindowMetricsSuite`. Both lines are duplicated in the FILTER test at `:129`/`:132`, which is why that one test passes -- `dequeCount == 0` holds vacuously, so the negative assertions in this suite currently prove nothing either. I applied both fixes locally on `7dffef1` and all 14 tests pass, so this is harness-only; the frame itself is fine. ########## sql/core/src/test/scala/org/apache/spark/sql/execution/window/SegmentTreeWindowMetricsSuite.scala: ########## @@ -26,19 +26,18 @@ import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.test.SharedSparkSession /** - * SQLMetrics visibility coverage for [[SegmentTreeWindowFunctionFrame]]: - * segtree path bumps `numSegmentTreeFrames`; fallback path bumps - * `numSegmentTreeFallbackFrames`; feature-flag off leaves both at 0. + * SQLMetrics visibility coverage for [[SegmentTreeWindowFunctionFrame]]: segtree path bumps + * `numSegmentTreeFrames`; fallback path bumps `numSegmentTreeFallbackFrames`; feature-flag off + * leaves both at 0. */ -class SegmentTreeWindowMetricsSuite - extends SharedSparkSession with SQLMetricsTestUtils { +class SegmentTreeWindowMetricsSuite extends SharedSparkSession with SQLMetricsTestUtils { Review Comment: **Finding 16.** This file is reformatted end to end with no functional change, and it isn't the only one. `dev/lint-scala` runs scalafmt only over `sql/api` and `sql/connect/{common,server,shims,client/jvm}` -- `sql/core` is scalastyle-only -- so reformatting it here is churn rather than compliance. Measured against the merge-base (`781cc206ec7`), whitespace-insensitively (`git diff -w --stat`): | file | non-whitespace lines changed | deque-related content | |---|---|---| | `SegmentTreeWindowFunctionSuite.scala` | 263 | none | | `UnboundedFollowingSegmentTreeSuite.scala` | 186 | none | | `WindowSegmentTreeMemorySuite.scala` | 122 | none | | `SegmentTreeWindowMetricsSuite.scala` | 102 | none | | `WindowEvaluatorFactoryBase.scala` | 136 of 468 | ~30 lines | | `WindowExec.scala` | 53 of 69 | 4 lines | `git diff 781cc206ec7 7dffef1882e -- <file> | grep -iE 'monotonic|deque'` is empty for all four suites, so those 673 lines carry nothing this PR needs. Reverting them plus the two main-file reflows takes the PR from 1906/781 to roughly 700/100. It costs more than review time: `git blame` on six files now points here for lines this PR didn't change, and the `branch-4.x` backport will conflict on hunks that carry no change. Reverting the `WindowExec.scala` Scaladoc block also closes @cloud-fan's [thread at `:52`](https://github.com/apache/spark/pull/57346#discussion_r3719175272) -- the growing-frame item there still reads `3. UNBOUNDED PRECEDING AND 1 FOLLOWING Every time we move to a new row to process, we add some rows to the frame.` with no punctuation between the two sentences, which is an artifact of the reflow rather than something worth fixing in place. ########## sql/core/src/main/scala/org/apache/spark/sql/execution/window/SlidingWindowMinMaxFunctionFrame.scala: ########## @@ -0,0 +1,303 @@ +/* + * 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]]. + * + * This frame is only instantiated when `isMinMaxOnly` is true (all window functions are Min or + * Max), enforced upstream in [[WindowEvaluatorFactoryBase]]. + */ +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 + + // 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) + + // `bufferIndex` is not stored; deques are accessed by position in the array. + 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 = { + input = rows + lowerIterator = input.generateIterator() + lowerRow = WindowFunctionFrame.getNextOrNull(lowerIterator) + var di = 0 + while (di < deques.length) { deques(di).clear(); di += 1 } + lowerBound = 0 + + if (ubound.isEmpty) { Review Comment: **Finding 15.** This branch is now unreachable. After the routing change, `isMinMaxOnly` is consulted in exactly one place, and that place always passes `Some(ub)` (`WindowEvaluatorFactoryBase.scala:354`): ```scala case ("AGGREGATE", frameType, lower, upper, _) => if (isMinMaxOnly) { target: InternalRow => ... new SlidingWindowMinMaxFunctionFrame( target, processor, lb, Some(ub), functions, childOutput) ``` `grep -n 'isMinMaxOnly\|new SlidingWindowMinMaxFunctionFrame'` over the module returns only the definition at `:238` and that use at `:354`, so `ubound` is never `None`: this branch never runs, and `write`'s `if (ubound.isDefined)` guard at `:139` is always true. Better removed than kept as a hook. `ubound` becomes a plain `BoundOrdering`, `write` loses a nesting level, and -- the part that matters -- nobody can re-enable the O(N)-per-aggregate path by passing `None` without also re-reading finding 4. Two pieces of prose describe the removed behaviour and should go with it: - the class Scaladoc at `:32`, "only instantiated when `isMinMaxOnly` is true (all window functions are Min or Max)": it's bounded *moving* frames only, and `isMinMaxOnly` also requires the conf to be on and no `FILTER`; - the PR description's "Note that for shrinking frames, it requires buffering the entire partition in memory/disk", which now describes something the code cannot do. ########## sql/core/src/test/scala/org/apache/spark/sql/execution/window/MonotonicDequeWindowFunctionSuite.scala: ########## @@ -0,0 +1,271 @@ +/* + * 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, 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 + val windowNodes = df.queryExecution.executedPlan.collect { + case w: org.apache.spark.sql.execution.window.WindowExec => w + } + assert(windowNodes.nonEmpty, "No WindowExec found in the query plan") + val dequeCount = windowNodes.flatMap(_.metrics.get("monotonicDequeFrames").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) Review Comment: **Finding 18.** Every frame in this suite straddles the current row -- `(-3,2)`, `(-4,2)`, `(-2,3)`, `(-2,2)`, `(-3,3)` -- and none is wider than 7 rows. Two paths therefore never execute: 1. **The `lowerBound`-advances-without-admitting branch** (`SlidingWindowMinMaxFunctionFrame.scala:142-144`). It only fires when a row enters the admit loop already below the lower bound, which needs both bounds on the same side of the current row -- `ROWS BETWEEN 2 FOLLOWING AND 4 FOLLOWING`. It is the one place where `lowerBound` and `lowerRow` move without setting `bufferUpdated`, so it's also where a desync between the two cursors would go unnoticed. 2. **`MinMaxDeque.expand()`** (`:203`). `capacity` starts at 16 and only grows once 16 candidates are live at the same time, which needs a window wider than 16 over data that doesn't pop -- ascending values for a min-deque. The ring-buffer copy `newValues(i) = values((head + i) % capacity)` is a classic off-by-one site and no unit test reaches it (the benchmarks do at W=1001+, but they only compare digests). I ran these five on top of `7dffef1` and they all pass, so this is coverage, not a bug: ```scala test("SPARK-58201: both-FOLLOWING rows frame") { val winSpec = Window.partitionBy($"pk").orderBy($"id").rowsBetween(2, 4) checkEquivalence(() => baseDF.select($"id", min($"v_int").over(winSpec), max($"v_int").over(winSpec))) } test("SPARK-58201: wide window on ascending data forces ring-buffer expand") { val df = spark.range(0, 300).selectExpr("id", "1 AS pk", "CAST(id AS INT) AS v") val winSpec = Window.partitionBy($"pk").orderBy($"id").rowsBetween(-70, 0) checkEquivalence(() => df.select($"id", min($"v").over(winSpec), max($"v").over(winSpec))) } ``` plus `rowsBetween(-4, -2)` (both preceding, so the frame is empty for the first rows), `rowsBetween(-60, 40)` over `rand(42)`, and `rangeBetween(1, 3)` on a tied order key. A `DECIMAL(38,10)` + `BINARY` pair under the lowered spill thresholds also passes and would cover the two types finding 7 mentions. ########## sql/core/src/test/scala/org/apache/spark/sql/execution/benchmark/WindowBenchmark.scala: ########## @@ -274,15 +350,72 @@ object WindowBenchmark extends SqlBasedBenchmark { benchmark.run() } + def setupIncreasingTable(n: Long): Unit = { + spark + .range(n) + .selectExpr("id", "cast(id as int) as v") + .coalesce(1) + .createOrReplaceTempView("t") + } + + def setupDecreasingTable(n: Long): Unit = { + spark + .range(n) + .selectExpr("id", s"cast(($n - id) as int) as v") + .coalesce(1) + .createOrReplaceTempView("t") + } + + def runSectionG(aggFn: String, iters: Int, rows: Long, halfW: Int, pattern: String): Unit = { + val frame = frameFor(halfW) + val dSeg = digest( + aggFn, + frame, + SQLConf.WINDOW_MONOTONIC_DEQUE_ENABLED.key -> "false", + SQLConf.WINDOW_SEGMENT_TREE_ENABLED.key -> "true") + val dMonotonic = digest(aggFn, frame, SQLConf.WINDOW_MONOTONIC_DEQUE_ENABLED.key -> "true") + require( + dSeg == dMonotonic, + s"$aggFn $pattern digest mismatch: seg=$dSeg monotonic=$dMonotonic") + + val W = 2 * halfW + 1 + val benchmark = new Benchmark( + s"$aggFn sliding window ($pattern), W=$W, ${rowsLabel(rows)} rows", + rows, + output = output) + val nSeg = s"$aggFn segtree ($pattern)" + val nMonotonic = s"$aggFn monotonic deque ($pattern)" + allCaseNames ++= Seq(nSeg, nMonotonic) + + benchmark.addCase(nSeg, numIters = iters) { _ => + currentCase = nSeg + withSQLConf( + SQLConf.WINDOW_MONOTONIC_DEQUE_ENABLED.key -> "false", + SQLConf.WINDOW_SEGMENT_TREE_ENABLED.key -> "true") { + spark.sql(s"SELECT $aggFn(v) $frame FROM t").noop() + } + } + benchmark.addCase(nMonotonic, numIters = iters) { _ => + currentCase = nMonotonic + withSQLConf(SQLConf.WINDOW_MONOTONIC_DEQUE_ENABLED.key -> "true") { + spark.sql(s"SELECT $aggFn(v) $frame FROM t").noop() + } + } + benchmark.run() + } + try { if (smokeMode) { setupIntTable(smokeRowCount) runBenchmark("SMOKE: Section A MIN") { runSectionA("MIN", ITERS_STRESS, smokeRowCount, smokeHalfW, "") } + runBenchmark("SMOKE: Section A MAX") { + runSectionA("MAX", ITERS_STRESS, smokeRowCount, smokeHalfW, "") + } + setupIntTable(B_N_W10) runBenchmark("SMOKE: Section B SUM W sweep point") { - runSectionB( - smokeHalfW, stressBs = smokeHalfW >= 2000, smokeRowCount, ITERS_STRESS, "") + runSectionB(5, stressBs = false, smokeRowCount, ITERS_STRESS, " (stress)") Review Comment: **Finding 11.** Section B is back in smoke mode, but no longer on the smoke arguments. `setupIntTable(B_N_W10)` on the line above builds the full 2,000,000-row table and `halfW` is hardcoded to `5`, so neither `mainArgs(0)` nor `mainArgs(1)` controls this section -- while the class doc still documents them (`:36`: "Dev smoke via positional mainArgs: (0)=rowCount, (1)=halfWindow (default 100)"). `rows` is still passed as `smokeRowCount`, so `Benchmark` divides a 2M-row run by the smoke row count and every `Per Row(ns)` in that section is off by `B_N_W10 / smokeRowCount`. The `" (stress)"` mark is hardcoded too, so the header claims a stress run the arguments didn't ask for. It reads like a leftover from local debugging. ```suggestion runSectionB(smokeHalfW, stressBs = smokeHalfW >= 2000, smokeRowCount, ITERS_STRESS, "") ``` and drop the `setupIntTable(B_N_W10)` line above it -- `setupIntTable(smokeRowCount)` at `:409` already covers the smoke path. ########## sql/core/src/main/scala/org/apache/spark/sql/execution/window/WindowEvaluatorFactoryBase.scala: ########## @@ -189,220 +206,230 @@ trait WindowEvaluatorFactoryBase { // Map the groups to a (unbound) expression and frame factory pair. var numExpressions = 0 val timeZone = SQLConf.get.sessionLocalTimeZone - framedFunctions.toSeq.map { - case (key, (expressions, functionSeq)) => - val ordinal = numExpressions - val functions = functionSeq.toArray - - // Construct an aggregate processor if we need one. - // Currently we don't allow mixing of Pandas UDF and SQL aggregation functions - // in a single Window physical node. Therefore, we can assume no SQL aggregation - // functions if Pandas UDF exists. In the future, we might mix Pandas UDF and SQL - // aggregation function in a single physical node. - val aggFilters: Array[Option[Expression]] = expressions.map { - case WindowExpression(ae: AggregateExpression, _) => ae.filter - case _ => None - }.toArray - // Keep as `def` (lazy / per-call): the FRAME_LESS_OFFSET / - // UNBOUNDED_OFFSET / UNBOUNDED_PRECEDING_OFFSET branches do not read - // `processor`. Eager `val` construction would invoke - // `AggregateProcessor.apply` on Lag / Lead / NthValue and throw - // `INTERNAL_ERROR: Unsupported aggregate function`. - def processor = if (functions.exists(_.isInstanceOf[PythonFuncExpression])) { - null - } else { - AggregateProcessor( - functions, - ordinal, - childOutput, - (expressions, schema) => - MutableProjection.create(expressions, schema), - aggFilters) - } - val conf = SQLConf.get - val blockSize = conf.windowSegmentTreeBlockSize - - // Create the factory to produce WindowFunctionFrame. - val factory = key match { - // Frameless offset Frame - case ("FRAME_LESS_OFFSET", _, IntegerLiteral(offset), _, expr) => - target: InternalRow => - new FrameLessOffsetWindowFunctionFrame( - target, - ordinal, - // OFFSET frame functions are guaranteed be OffsetWindowFunction. - functions.map(_.asInstanceOf[OffsetWindowFunction]), - childOutput, - (expressions, schema) => - MutableProjection.create(expressions, schema), - offset, - expr.nonEmpty) - case ("UNBOUNDED_OFFSET", _, IntegerLiteral(offset), _, expr) => + framedFunctions.toSeq.map { case (key, (expressions, functionSeq)) => + val ordinal = numExpressions + val functions = functionSeq.toArray + + // Construct an aggregate processor if we need one. + // Currently we don't allow mixing of Pandas UDF and SQL aggregation functions + // in a single Window physical node. Therefore, we can assume no SQL aggregation + // functions if Pandas UDF exists. In the future, we might mix Pandas UDF and SQL + // aggregation function in a single physical node. + val aggFilters: Array[Option[Expression]] = expressions.map { + case WindowExpression(ae: AggregateExpression, _) => ae.filter + case _ => None + }.toArray + // Keep as `def` (lazy / per-call): the FRAME_LESS_OFFSET / + // UNBOUNDED_OFFSET / UNBOUNDED_PRECEDING_OFFSET branches do not read + // `processor`. Eager `val` construction would invoke + // `AggregateProcessor.apply` on Lag / Lead / NthValue and throw + // `INTERNAL_ERROR: Unsupported aggregate function`. + def processor = if (functions.exists(_.isInstanceOf[PythonFuncExpression])) { + null + } else { + AggregateProcessor( + functions, + ordinal, + childOutput, + (expressions, schema) => MutableProjection.create(expressions, schema), + aggFilters) + } + val conf = SQLConf.get + val isMinMaxOnly = conf.windowMonotonicDequeEnabled && + functions.nonEmpty && functions.forall { + case _: Min | _: Max => true + case _ => false + } && aggFilters.forall(_.isEmpty) + val blockSize = conf.windowSegmentTreeBlockSize + + // Create the factory to produce WindowFunctionFrame. + val factory = key match { + // Frameless offset Frame + case ("FRAME_LESS_OFFSET", _, IntegerLiteral(offset), _, expr) => + target: InternalRow => + new FrameLessOffsetWindowFunctionFrame( + target, + ordinal, + // OFFSET frame functions are guaranteed to be OffsetWindowFunction. + functions.map(_.asInstanceOf[OffsetWindowFunction]), + childOutput, + (expressions, schema) => MutableProjection.create(expressions, schema), + offset, + expr.nonEmpty) + case ("UNBOUNDED_OFFSET", _, IntegerLiteral(offset), _, expr) => + target: InternalRow => { + new UnboundedOffsetWindowFunctionFrame( + target, + ordinal, + // OFFSET frame functions are guaranteed to be OffsetWindowFunction. + functions.map(_.asInstanceOf[OffsetWindowFunction]), + childOutput, + (expressions, schema) => MutableProjection.create(expressions, schema), + offset, + expr.nonEmpty) + } + case ("UNBOUNDED_PRECEDING_OFFSET", _, IntegerLiteral(offset), _, expr) => + target: InternalRow => { + new UnboundedPrecedingOffsetWindowFunctionFrame( + target, + ordinal, + // OFFSET frame functions are guaranteed to be OffsetWindowFunction. + functions.map(_.asInstanceOf[OffsetWindowFunction]), + childOutput, + (expressions, schema) => MutableProjection.create(expressions, schema), + offset, + expr.nonEmpty) + } + + // Entire Partition Frame. + case ("AGGREGATE", _, UnboundedPreceding, UnboundedFollowing, _) => + target: InternalRow => { + new UnboundedWindowFunctionFrame(target, processor) + } + + // Growing Frame. + case ("AGGREGATE", frameType, UnboundedPreceding, upper, _) => + target: InternalRow => { + new UnboundedPrecedingWindowFunctionFrame( + target, + processor, + createBoundOrdering(frameType, upper, timeZone)) + } + + // Shrinking Frame. + case ("AGGREGATE", frameType, lower, UnboundedFollowing, _) => + if (eligibleForSegTree(functions, aggFilters, frameType, conf)) { + val segFns = functions.map(_.asInstanceOf[DeclarativeAggregate]) + // Shrinking-frame queries `[lower, n)` on `WindowSegmentTree` touch the LRU + // for exactly two blocks per query: (1) the lower-edge partial block, and + // (2) the partition's last block (the right-partial `mergeBlockRange(bhi, 0, + // ...)` calls `ensureBlockLevels(bhi)` on every multi-block query). Middle + // blocks of `[lower, n)` are answered directly from `blockAggregates` and + // never go through the LRU. The lower-edge block advances monotonically with + // the output row, so once the cursor crosses a boundary the previous block + // is never revisited; the last block stays hot because every query touches + // it. Hint = 2 keeps both resident; routing through `estimateMaxCachedBlocks` + // would produce 8 by default (no `IntegerLiteral` upper match) -- correct + // numerically but misleading about what the shrinking path actually needs. + // Note: tuning this down to 1 would thrash, evicting the last block on every + // query and forcing it to be rebuilt. + val cacheHint = Some(2) target: InternalRow => { - new UnboundedOffsetWindowFunctionFrame( + val tc = TaskContext.get() + if (tc == null) { + throw SparkException.internalError( + "WindowEvaluatorFactoryBase.shrinkingSegTreeFrameFactory requires " + + "an active TaskContext") + } + val tmm = tc.taskMemoryManager() + val lb = createBoundOrdering(frameType, lower, timeZone) + new SegmentTreeWindowFunctionFrame( target, - ordinal, - // OFFSET frame functions are guaranteed be OffsetWindowFunction. - functions.map(_.asInstanceOf[OffsetWindowFunction]), + processor, + segFns, childOutput, - (expressions, schema) => - MutableProjection.create(expressions, schema), - offset, - expr.nonEmpty) + frameType, + lb, + ubound = None, + fallbackFactory = + () => new UnboundedFollowingWindowFunctionFrame(target, processor, lb), + (e, s) => MutableProjection.create(e, s), + conf, + cacheHint, + tmm, + numSegmentTreeFrames, + numSegmentTreeFallbackFrames) } - case ("UNBOUNDED_PRECEDING_OFFSET", _, IntegerLiteral(offset), _, expr) => - target: InternalRow => { - new UnboundedPrecedingOffsetWindowFunctionFrame( + } else { target: InternalRow => + { + new UnboundedFollowingWindowFunctionFrame( target, - ordinal, - // OFFSET frame functions are guaranteed be OffsetWindowFunction. - functions.map(_.asInstanceOf[OffsetWindowFunction]), - childOutput, - (expressions, schema) => - MutableProjection.create(expressions, schema), - offset, - expr.nonEmpty) - } - - // Entire Partition Frame. - case ("AGGREGATE", _, UnboundedPreceding, UnboundedFollowing, _) => - target: InternalRow => { - new UnboundedWindowFunctionFrame(target, processor) + processor, + createBoundOrdering(frameType, lower, timeZone)) } + } - // Growing Frame. - case ("AGGREGATE", frameType, UnboundedPreceding, upper, _) => - target: InternalRow => { - new UnboundedPrecedingWindowFunctionFrame( + // Moving Frame. + case ("AGGREGATE", frameType, lower, upper, _) => + if (isMinMaxOnly) { target: InternalRow => + { + val lb = createBoundOrdering(frameType, lower, timeZone) + val ub = createBoundOrdering(frameType, upper, timeZone) + numMonotonicDequeFrames.foreach(_ += 1) Review Comment: **Finding 17.** This counts frame *constructions*, not preparations. `WindowEvaluatorFactory` builds its frames once per `eval()` -- once per RDD partition -- in `val frames = factories.map(_(windowFunctionResult))` (`WindowEvaluatorFactory.scala:106`), and only then calls `prepare` once per *window* partition inside `fetchNextPartition` (`:124`). So a task holding three window partitions bumps this once. The sibling counter is bumped in `prepare`: `numSegmentTreeFrames.foreach(_ += 1)` at `SegmentTreeWindowFunctionFrame.scala:170`, deliberately after `tree.build(rows)` succeeds. `SegmentTreeWindowMetricsSuite`'s first test is named "one per frame per partition" and asserts `=== 3L` for three partitions. So the two are not comparable, although their labels read as a matched pair ("number of segment-tree frames prepared" / "number of monotonic-deque frames prepared"), and it contradicts this metric's own Scaladoc at `:50` ("Incremented each time a monotonic deque frame is prepared"). Passing the metric into `SlidingWindowMinMaxFunctionFrame` and bumping it in `prepare()` matches the segtree path and makes the three counters sum to the number of prepared frames. ########## sql/core/src/main/scala/org/apache/spark/sql/execution/window/SlidingWindowMinMaxFunctionFrame.scala: ########## @@ -0,0 +1,303 @@ +/* + * 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]]. + * + * This frame is only instantiated when `isMinMaxOnly` is true (all window functions are Min or + * Max), enforced upstream in [[WindowEvaluatorFactoryBase]]. + */ +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 + + // 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) + + // `bufferIndex` is not stored; deques are accessed by position in the array. + 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 = { + input = rows + lowerIterator = input.generateIterator() + lowerRow = WindowFunctionFrame.getNextOrNull(lowerIterator) + var di = 0 + while (di < deques.length) { deques(di).clear(); di += 1 } + lowerBound = 0 + + if (ubound.isEmpty) { + // Shrinking frame (BETWEEN <lower> AND UNBOUNDED FOLLOWING): admit the entire partition + // up front. Note: on monotone input the deque may retain O(N) elements per aggregate, + // which is acceptable because the partition itself is already O(N). + val iter = input.generateIterator() + var idx = 0 + while (iter.hasNext) { + val row = iter.next() + var dj = 0 + while (dj < deques.length) { deques(dj).admit(row, idx); dj += 1 } + 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 { + 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, + ordering: Ordering[Any]) { + + 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 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 + } + + // For primitive types the value is already by-value safe. For reference types we use + // InternalRow.copyValue which handles the deep copy in a single allocation. + private def evaluateAndCopy(row: InternalRow): Any = { Review Comment: **Finding 7.** The substance of this is fixed -- `tempRow` and its `copy()` are gone. Two smaller parts of the same finding are still open, and your summary comment says one of them was removed: > Removed the custom copy logic, per-deque `tempRow`, and `isPrimitive` allowlist. `isPrimitive` is still here at `:191`, still carrying @Ma77Ball's interval types. It buys nothing: `InternalRow.copyValue` already passes anything it doesn't recognise straight through (`InternalRow.scala:138`, `case _ => value`), so the branch skips a five-case pattern match, not an allocation. And "skips heap allocation/copying entirely for primitive types" in the description isn't true either way -- `values` is `Array[Any]`, so every `Int`/`Long`/`Double` is boxed on `offerLast`. Second, the comment above this method overstates what `copyValue` does. Its cases are `UTF8String`, `BinaryView`, `InternalRow`, `ArrayData`, `MapData` -- and nothing else. `BinaryType` (a `byte[]`) and `Decimal` are both accepted by `MIN`/`MAX` and both fall through uncopied, so for those two types this method neither takes the primitive shortcut nor deep-copies. It is safe today, but for a reason the comment doesn't give: after `ExtractWindowExpressions`, a window aggregate's arguments are always hoisted into the `Project` below (`Analyzer.scala:3588`, `function.children.map(extractExpr)`), so `boundChild` is a `BoundReference` over an `UnsafeRow`, and `UnsafeRow.getBinary`/`getDecimal` allocate a fresh object per call. That's the invariant the whole copy strategy rests on -- and it's what makes `isPrimitive` safe too -- so it's worth stating instead of the current sentence. I confirmed `DECIMAL(38,10)` and `BINARY` are correct under the lowered spill thresholds; see finding 18 for the test. ########## sql/core/src/test/scala/org/apache/spark/sql/execution/benchmark/WindowBenchmark.scala: ########## @@ -343,6 +476,45 @@ object WindowBenchmark extends SqlBasedBenchmark { runBenchmark("Section C - N-sweep large (stress)") { runSectionC(C_N_LARGE) } + + setupIncreasingTable(G_N) + runBenchmark("Section G - MIN Monotonic Deque vs Segment Tree (Worst-Case: Increasing)") { + runSectionG("MIN", ITERS_NORMAL, G_N, G_HALF_W, "Increasing") + } + + setupDecreasingTable(G_N) + runBenchmark("Section G - MIN Monotonic Deque vs Segment Tree (Best-Case: Decreasing)") { + runSectionG("MIN", ITERS_NORMAL, G_N, G_HALF_W, "Decreasing") + } + + setupIntTable(G_N) + runBenchmark("Section G - MIN Monotonic Deque vs Segment Tree (Random)") { + runSectionG("MIN", ITERS_NORMAL, G_N, G_HALF_W, "Random") + } + + // Section H: narrow-window scaling. W=1/3/11 covers the full range where the + // naive path's codegen'd per-row scan may compete with the deque's interpreted + // BoundReference.eval + Ordering[Any].compare. Results show whether a width + // gate is needed before enabling the conf by default. + setupIntTable(H_N) + runBenchmark("Section H - MIN W=1 scaling (2M rows)") { Review Comment: **Finding 19.** Thanks for adding this -- W=1/3/11 is exactly the data finding 10 was missing, and it answers the question (deque at parity with naive at W=1, 1.1X at W=3, 1.6X at W=11). One cost worth removing: Section H reuses `runSectionA`, which always emits the `segtree (blockSize=256)` cell, and at these widths that cell is the single most expensive thing in the whole benchmark: ``` MIN segtree (blockSize=256) [H W=1] 58079 58108 25 0.0 29039.7 0.0X MAX segtree (blockSize=256) [H W=11] 58598 58657 60 0.0 29298.8 0.0X ``` Six Section H runs x `ITERS_STRESS` x ~58 s is roughly 17 minutes of the benchmark job, plus six `dSegBs` digest pre-checks over 2M rows, to publish a column Section A already covers at W=1001 and that nobody would configure at W=1. A `withSegBs: Boolean = true` parameter on `runSectionA`, passed `false` from Section H, keeps the narrow-window data without it. ########## sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala: ########## @@ -4689,6 +4689,15 @@ object SQLConf { "The threshold of window group limit must be -1, 0 or positive integer.") .createWithDefault(1000) + val WINDOW_MONOTONIC_DEQUE_ENABLED = + buildConf("spark.sql.window.monotonicDeque.enabled") + .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE) + .doc("Use O(N) monotonic deque for sliding window MIN/MAX. It replaces the O(N * W) " + + "naive loop and O(N log W) segment tree with an optimized ring-buffer deque.") + .version("4.2.0") Review Comment: **Finding 14.** `4.2.0` is the version the *segment tree* shipped in -- `WINDOW_SEGMENT_TREE_ENABLED` ten lines below says the same thing, correctly. Both `branch-4.2` and `branch-4.3` are cut: `master` is `5.0.0-SNAPSHOT` and the latest release branch is on `4.4.0-SNAPSHOT`, so as written this config claims to exist in a release that shipped without it. Unless you mean this to be master-only (it doesn't look like it -- new default-off optimization, no binary incompatibility), the first version that can carry it is the `branch-4.x` one: ```suggestion .version("4.4.0") ``` -- 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]
