cloud-fan commented on code in PR #57346:
URL: https://github.com/apache/spark/pull/57346#discussion_r3719175272
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/window/WindowExec.scala:
##########
@@ -29,48 +29,46 @@ import org.apache.spark.sql.execution.metric.{SQLMetric,
SQLMetrics}
* instructions, frames, are used to calculate these aggregates. Frames are
processed in the order
* specified in the window specification (the ORDER BY ... clause). There are
four different frame
* types:
- * - Entire partition: The frame is the entire partition, i.e.
- * UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING. For this case, window
function will take all
- * rows as inputs and be evaluated once.
- * - Growing frame: We only add new rows into the frame, Examples are:
+ * - Entire partition: The frame is the entire partition, i.e. UNBOUNDED
PRECEDING AND UNBOUNDED
+ * FOLLOWING. For this case, window function will take all rows as inputs
and be evaluated
+ * once.
+ * - Growing frame: We only add new rows into the frame, Examples are:
* 1. UNBOUNDED PRECEDING AND 1 PRECEDING
* 2. UNBOUNDED PRECEDING AND CURRENT ROW
- * 3. UNBOUNDED PRECEDING AND 1 FOLLOWING
- * Every time we move to a new row to process, we add some rows to the
frame. We do not remove
- * rows from this frame.
- * - Shrinking frame: We only remove rows from the frame, Examples are:
+ * 3. UNBOUNDED PRECEDING AND 1 FOLLOWING Every time we move to a new row
to process, we add
+ * some rows to the frame. We do not remove rows from this frame.
+ * - Shrinking frame: We only remove rows from the frame, Examples are:
* 1. 1 PRECEDING AND UNBOUNDED FOLLOWING
* 2. CURRENT ROW AND UNBOUNDED FOLLOWING
- * 3. 1 FOLLOWING AND UNBOUNDED FOLLOWING
- * Every time we move to a new row to process, we remove some rows from the
frame. We do not add
- * rows to this frame.
- * - Moving frame: Every time we move to a new row to process, we remove some
rows from the frame
- * and we add some rows to the frame. Examples are:
+ * 3. 1 FOLLOWING AND UNBOUNDED FOLLOWING Every time we move to a new row
to process, we
+ * remove some rows from the frame. We do not add rows to this frame.
+ * - Moving frame: Every time we move to a new row to process, we remove
some rows from the
+ * frame and we add some rows to the frame. Examples are:
* 1. 2 PRECEDING AND 1 PRECEDING
* 2. 1 PRECEDING AND CURRENT ROW
* 3. CURRENT ROW AND 1 FOLLOWING
* 4. 1 PRECEDING AND 1 FOLLOWING
* 5. 1 FOLLOWING AND 2 FOLLOWING
- * - Offset frame: The frame consist of one row, which is an offset number of
rows away from the
- * current row. Only [[OffsetWindowFunction]]s can be processed in an offset
frame. There are
- * three implements of offset frame: [[FrameLessOffsetWindowFunctionFrame]],
- * [[UnboundedOffsetWindowFunctionFrame]] and
[[UnboundedPrecedingOffsetWindowFunctionFrame]].
+ * - Offset frame: The frame consist of one row, which is an offset number
of rows away from the
Review Comment:
Please clean up this rewritten Scaladoc block. This should be `The frame
consists`, the next sentence should say `three implementations`, and the
growing/shrinking examples above currently run into their explanatory sentences
without punctuation.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/window/WindowEvaluatorFactoryBase.scala:
##########
@@ -189,220 +206,242 @@ 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) =>
- target: InternalRow => {
- new UnboundedOffsetWindowFunctionFrame(
+ 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 be OffsetWindowFunction.
Review Comment:
This and the two identical comments below are missing `to`: `OFFSET frame
functions are guaranteed to be OffsetWindowFunction instances.`
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/window/WindowSegmentTreeMemorySuite.scala:
##########
@@ -31,13 +31,13 @@ import org.apache.spark.sql.types.IntegerType
/**
* Memory-manager integration tests for [[WindowSegmentTree]]. Covers:
- * - `SegTreeSpiller` registration with `TaskMemoryManager`
- * - `acquireBlockMemory` grant / partial-grant rollback
- * - `evictUntil` LRU eviction driven by TMM pressure
- * - `spill()` self-trigger short-circuit and rowArray-spilled fall-through
- * - task completion / kill listener releasing all cached blocks
- * T5 (rowArray-spilled priority) and T8 (task-kill listener) are kept as
- * `ignore`d stubs so the matrix stays visible; each documents what it needs.
+ * - `SegTreeSpiller` registration with `TaskMemoryManager`
+ * - `acquireBlockMemory` grant / partial-grant rollback
+ * - `evictUntil` LRU eviction driven by TMM pressure
+ * - `spill()` self-trigger short-circuit and rowArray-spilled fall-through
+ * - task completion / kill listener releasing all cached blocks
+ * T5 (rowArray-spilled priority) and T8 (task-kill listener) are kept as
`ignore`d stubs so the
Review Comment:
Please restore the T5/T8 ignored stubs or remove this sentence. The suite
currently jumps from T4 to T6 and T7 to T9, so this claims direct visibility
for tests that do not exist.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/window/SlidingWindowMinMaxFunctionFrame.scala:
##########
@@ -0,0 +1,299 @@
+/*
+ * 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)
+ deques.foreach(_.clear())
+ 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()
+ deques.foreach(_.admit(row, idx))
Review Comment:
Use indexed `while` loops over `deques` here and at the admit/drop calls in
`write`. These capturing callbacks are created per partition row or repeatedly
as the window advances, adding avoidable allocation to the optimization's hot
path.
--
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]
