peter-toth commented on code in PR #57346: URL: https://github.com/apache/spark/pull/57346#discussion_r3788479346
########## sql/core/src/main/scala/org/apache/spark/sql/execution/window/SlidingWindowMinMaxFunctionFrame.scala: ########## @@ -0,0 +1,277 @@ +/* + * 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 + +/** + * 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] = _ Review Comment: **Finding 23.** `lowerRow` is only ever passed to `lbound.compare` -- at `:116` and nowhere else. For a `ROWS` frame `lbound` is a `RowBoundOrdering`, whose whole body ignores the row (`BoundOrdering.scala:34-41`): ```scala private[window] final case class RowBoundOrdering(offset: Int) extends BoundOrdering { override def compare( inputRow: InternalRow, inputIndex: Int, outputRow: InternalRow, outputIndex: Int): Int = inputIndex - (outputIndex + offset) } ``` So on every `ROWS BETWEEN ... AND ...` frame -- the case the PR is built for -- `lowerIterator` is walked start to finish per partition and everything it yields is thrown away. That is one extra `getNextOrNull` per row inside the loop this PR exists to shorten, and it may be part of why Section H has the deque at 591 ms against naive's 582 ms at W=1. The spilled case costs more than a wasted call. `ExternalAppendOnlyUnsafeRowArray.generateIterator()` on a spilled array builds a fresh chain of `UnsafeSorterSpillReader`s, each with its own read buffer (`spark.unsafe.sorter.spill.reader.buffer.size`, minimum and default 1 MB). So a spilled partition is read twice and holds twice the reader buffers, per frame, for a cursor that only `RANGE` frames consult. `lbound.isInstanceOf[RowBoundOrdering]` is exactly "the frame is `ROWS`" -- `createBoundOrdering` returns `RowBoundOrdering` for `RowFrame` and `RangeBoundOrdering` for `RangeFrame`, and throws otherwise -- so the gate is a one-liner: ```scala // RowBoundOrdering.compare ignores its inputRow, so the lower cursor is only // needed for RANGE frames, where the comparison reads the order-key value. private[this] val needsLowerRow = !lbound.isInstanceOf[RowBoundOrdering] ``` `prepare`: ```scala if (needsLowerRow) { lowerIterator = input.generateIterator() lowerRow = WindowFunctionFrame.getNextOrNull(lowerIterator) } else { lowerIterator = null lowerRow = null } ``` and both advance sites in `write` (`:118`, `:127`) become `if (needsLowerRow) lowerRow = WindowFunctionFrame.getNextOrNull(lowerIterator)`. Passing a `null` `lowerRow` into `lbound.compare` is safe on that path precisely because `RowBoundOrdering` never dereferences it, which is worth saying in the comment. The spill-safety block just above would then read "both cursors" -> "`nextRow`, and `lowerRow` on RANGE frames". ########## sql/core/benchmarks/WindowBenchmark-results.txt: ########## @@ -2,171 +2,287 @@ Section A - MIN (non-invertible) ================================================================================================ -OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1020-azure +OpenJDK 64-Bit Server VM 17.0.20+8-LTS on Linux 6.17.0-1022-azure AMD EPYC 7763 64-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) 4036 4059 17 0.1 15396.6 1.0X +MIN segtree (default) 377 390 13 0.7 1439.8 10.7X +MIN segtree (blockSize=256) 2220 2226 6 0.1 8470.2 1.8X +MIN monotonic deque (new) 120 122 3 2.2 458.3 33.6X Review Comment: **Finding 1.** Thanks for regenerating this -- it fixes finding 2. It also makes every number in the PR body stale, including the runner attribution, so the body and the artifact now disagree in both directions. | section | description | this file | |---|---|---| | MIN W=1001 | 4,024 / 385 / **116** -> 34.8X, 3.3X vs segtree | 4036 / 377 / **120** -> 33.6X, 3.1X (`:10-12`) | | MAX W=1001 | 4,414 / 409 / **109** -> 40.6X, 3.8X | 4292 / 388 / **105** -> 40.7X, 3.7X (`:24-26`) | | G increasing | 3,938 vs **595** -> 6.6X | 3854 vs **570** -> 6.8X (`:183-184`) | | G decreasing | 3,934 vs **586** -> 6.7X | 3818 vs **554** -> 6.9X (`:195-196`) | | G random | 4,030 vs **625** -> 6.4X | 3911 vs **591** -> 6.6X (`:207-208`) | | H W=1 | ~600 / ~1,570 / ~610 | 582 / 1474 / 591 (`:219-221`) | | H W=3 | ~680 / ~1,690 / ~620 | 663 / 1601 / 593 (`:245-247`) | | H W=11 | ~1,000 / ~2,100 / ~620 -> ~1.6X | 996 / 1987 / 591 -> 1.7X (`:271-273`) | | hardware | "AMD EPYC 9V74" (twice) | `AMD EPYC 7763 64-Core Processor` | Every conclusion survives, so this is a refresh rather than a rethink. One thing to note for next time: the benchmark workflow does not always land on the same runner, so any hardware string or absolute millisecond figure copied into the body will keep going stale on every regeneration. The durable form is to state the ratios you want reviewers to see and point at `sql/core/benchmarks/WindowBenchmark-results.txt` for the raw numbers and the environment header. -- 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]
