cloud-fan commented on code in PR #57346:
URL: https://github.com/apache/spark/pull/57346#discussion_r3966477110
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/benchmark/WindowBenchmark.scala:
##########
@@ -27,15 +27,18 @@ import org.apache.spark.sql.internal.SQLConf
* Benchmark for window functions with bounded ROWS frames.
*
* Matrix (see PR description for rationale):
- * - A: 5 aggregates x 3 cells (naive / segtree default / segtree bs=256) @
W=1001.
- * Per-case N so naive ~3-5s/iter; STDDEV_SAMP pinned @ N=2M (multi-buffer
stress).
- * - B: SUM-over-INT, W sweep {10, 50, 201, 4001}; W=10/50 Pareto-loss
stress,
- * W=4001 also runs bs=256.
+ * - A: 5 aggregates x 4 cells (naive / segtree default / segtree bs=256 /
monotonic deque
Review Comment:
**Non-blocking (P2):** This matrix does not match the cases registered
below: normal mode invokes six aggregates, and only MIN/MAX get the fourth
monotonic-deque cell, so Section A is not a `5 aggregates x 4 cells` matrix.
Section B also labels half-widths 5 and 25 as W=10 and W=50 even though the
inclusive frames contain 11 and 51 rows away from partition edges. Please
correct the matrix description and rename those cases/constants to W=11/W=51,
or change the frame construction if even widths were intended.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -4840,6 +4840,15 @@ object SQLConf {
.booleanConf
.createWithDefault(false)
+ 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) " +
Review Comment:
**Non-blocking (P2):** This help text promises the O(N) deque for sliding
MIN/MAX, but the factory deliberately falls back when the same frame contains
any non-MIN/MAX aggregate or any aggregate has a `FILTER`. With this flag true
and segment tree false, supported `MIN + SUM` or filtered `MIN` windows can
therefore remain on the O(N * W) path despite matching the current description.
Please qualify the text to say the moving frame must contain only MIN/MAX
aggregates and no FILTER clauses.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/window/MonotonicDequeWindowFunctionSuite.scala:
##########
@@ -0,0 +1,360 @@
+/*
+ * 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)))
+ }
+
+ // Verify strict inequality preserves first-of-equals behavior
+ // under collated strings and signed zero.
+
+ test("SPARK-58201: MIN/MAX on collated strings (UTF8_LCASE) preserves
first-of-equals") {
+ // Under UTF8_LCASE, 'Bob' and 'bob' compare equal. MIN must keep the
+ // first occurrence (lowest index), matching naive/segment-tree semantics.
+ val df = spark.sql("""SELECT id, 1 AS pk,
+ | CASE WHEN id = 0 THEN COLLATE('Bob', 'UTF8_LCASE')
+ | WHEN id = 1 THEN COLLATE('bob', 'UTF8_LCASE')
+ | WHEN id = 2 THEN COLLATE('alice', 'UTF8_LCASE')
+ | WHEN id = 3 THEN COLLATE('BOB', 'UTF8_LCASE')
+ | ELSE COLLATE(CAST(id AS STRING), 'UTF8_LCASE')
+ | END AS v
+ |FROM RANGE(0, 20)""".stripMargin)
+ 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 Double with signed zero (+0.0 / -0.0)") {
+ // SQLOrderingUtil.compareDoubles treats -0.0 == +0.0, so they compare
+ // equal. MIN must keep the first occurrence, matching naive semantics.
+ val df = spark
+ .range(0, 20)
+ .selectExpr(
+ "id",
+ "1 AS pk",
+ """CASE
+ WHEN id % 4 = 0 THEN CAST(-0.0 AS DOUBLE)
+ WHEN id % 4 = 1 THEN CAST(0.0 AS DOUBLE)
+ WHEN id % 4 = 2 THEN CAST(id AS DOUBLE)
+ ELSE CAST(-id AS DOUBLE)
+ END AS v""")
+ val winSpec = Window.partitionBy($"pk").orderBy($"id").rowsBetween(-2, 2)
+ checkEquivalence(() => df.select($"id", min($"v").over(winSpec),
max($"v").over(winSpec)))
+ }
+
+ // Spill coverage: lower thresholds to force ExternalAppendOnlyUnsafeRowArray
+ // to use its SpillableArrayIterator, which recycles a single UnsafeRow.
+
+ test("SPARK-58201: Moving rows frame: MIN/MAX on reference types (String)
with spill") {
+ withSQLConf(
+ SQLConf.WINDOW_EXEC_BUFFER_IN_MEMORY_THRESHOLD.key -> "8",
+ SQLConf.WINDOW_EXEC_BUFFER_SPILL_THRESHOLD.key -> "16") {
+ val winSpec = Window.partitionBy($"pk").orderBy($"id").rowsBetween(-2, 3)
+ checkEquivalence(() =>
+ baseDF.select($"id", min($"v_str").over(winSpec),
max($"v_str").over(winSpec)))
+ }
+ }
+
+ test("SPARK-58201: MIN/MAX on ArrayType with spill") {
+ val df = spark
+ .range(0, 60)
+ .selectExpr("id", "(id % 3) AS pk", "array(CAST(id AS INT), CAST(id * 2
AS INT)) AS v")
+ withSQLConf(
+ SQLConf.WINDOW_EXEC_BUFFER_IN_MEMORY_THRESHOLD.key -> "8",
+ SQLConf.WINDOW_EXEC_BUFFER_SPILL_THRESHOLD.key -> "16") {
+ val winSpec = Window.partitionBy($"pk").orderBy($"id").rowsBetween(-2, 2)
+ checkEquivalence(() => df.select($"id", min($"v").over(winSpec),
max($"v").over(winSpec)))
+ }
+ }
+
+ test("SPARK-58201: MIN/MAX on StructType with spill") {
+ val df = spark
+ .range(0, 60)
+ .selectExpr(
+ "id",
+ "(id % 3) AS pk",
+ "named_struct('a', CAST(id AS INT), 'b', CAST(id * 3 AS INT)) AS v")
+ withSQLConf(
+ SQLConf.WINDOW_EXEC_BUFFER_IN_MEMORY_THRESHOLD.key -> "8",
+ SQLConf.WINDOW_EXEC_BUFFER_SPILL_THRESHOLD.key -> "16") {
+ val winSpec = Window.partitionBy($"pk").orderBy($"id").rowsBetween(-2, 2)
+ checkEquivalence(() => df.select($"id", min($"v").over(winSpec),
max($"v").over(winSpec)))
+ }
+ }
+
+ // Both-FOLLOWING frame: lowerBound-advances-without-admitting branch
+ // (both bounds on the FOLLOWING side).
+ test("SPARK-58201: both-FOLLOWING rows frame exercises lowerBound-advances
branch") {
+ val winSpec = Window.partitionBy($"pk").orderBy($"id").rowsBetween(2, 4)
+ checkEquivalence(() =>
+ baseDF.select($"id", min($"v_int").over(winSpec),
max($"v_int").over(winSpec)))
+ }
+
+ // Wide ascending data forces MinMaxDeque.expand() (ring-buffer grow path).
+ 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)))
+ }
+
+ // Both-PRECEDING frame: first rows have an empty window.
+ test("SPARK-58201: both-PRECEDING rows frame (empty window for first rows)")
{
+ val winSpec = Window.partitionBy($"pk").orderBy($"id").rowsBetween(-4, -2)
+ checkEquivalence(() =>
+ baseDF.select($"id", min($"v_int").over(winSpec),
max($"v_int").over(winSpec)))
+ }
+
+ // Wide random data: exercises the normal sliding path at scale.
Review Comment:
**Non-blocking (P2):** These two fixtures do not contain the shapes their
names advertise. Within each `pk`, `v_int = id` is strictly increasing, and `id
/ 3` is unique after partitioning by `id % 3`, so the RANGE case has no peer
ties. A regression in non-monotonic admission/eviction or peer-group RANGE
bounds would leave both tests green. Please generate deterministic
pseudo-random values for the wide case and repeated order-key values within
each partition for the RANGE case, while retaining the baseline-equivalence
checks.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/window/MonotonicDequeWindowFunctionSuite.scala:
##########
@@ -0,0 +1,360 @@
+/*
+ * 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")
Review Comment:
**Non-blocking (P2):** The current `> 0` assertion does not pin the metric's
advertised frame-per-window-partition contract, and every routing test keeps
the two optimization flags mutually exclusive. On `baseDF`, one shared MIN/MAX
frame across three window partitions should report exactly 3; an increment per
task, expression, or row still passes today. Please assert that exact count and
add an eligible both-flags-enabled case that reports deque=3, segment-tree=0,
and remains output-equivalent to the naive baseline.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/window/SlidingWindowMinMaxFunctionFrame.scala:
##########
@@ -0,0 +1,286 @@
+/*
+ * 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 nextRow (and lowerRow on RANGE frames) 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] = _
+
+ // RowBoundOrdering.compare ignores its inputRow (it only uses the index), so
+ // the lower cursor is only needed for RANGE frames. Passing a null lowerRow
to
+ // lbound.compare is safe on the ROWS path for exactly this reason.
+ private[this] val needsLowerRow = !lbound.isInstanceOf[RowBoundOrdering]
+
+ /** The row at lowerBound. Only valid when needsLowerRow is true. */
+ 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),
+ TypeUtils.getInterpretedOrdering(child.dataType))
+ }
+
+ override def prepare(rows: ExternalAppendOnlyUnsafeRowArray): Unit = {
+ numMonotonicDequeFrames.foreach(_ += 1)
+ input = rows
+ if (needsLowerRow) {
+ lowerIterator = input.generateIterator()
Review Comment:
**Blocking (P1):** `lowerIterator` owns a second spill reader for RANGE
frames, and a bounded frame does not necessarily consume it to EOF. On the next
window partition, `prepare` replaces this cursor after the shared row array has
been cleared, but `UnsafeSorterSpillReader` closes its stream only on
exhaustion or explicit close. With many forced-spill keys in one task, this can
retain one reader and deleted-but-open spill file per key until task
completion, eventually exhausting file descriptors or local disk. Please give
both frame cursors explicit per-partition close ownership before clearing or
replacing them, and add a multi-key forced-spill RANGE test that verifies
readers close at each boundary.
**Recommended change:** Introduce an explicit closeable cursor boundary and
close every active frame cursor before replacing or clearing the prior
partition state.
**Why this works:** Expose an idempotent close operation that reaches the
underlying spill reader without draining unread rows, then invoke it from the
frame's partition-reset path for both lowerIterator and inputIterator while
retaining task-completion cleanup as a final fallback.
**Scope:** SlidingWindowMinMaxFunctionFrame cursor lifecycle, the minimal
ExternalAppendOnlyUnsafeRowArray iterator close plumbing, and focused
multi-partition RANGE spill coverage.
**Compatibility:** SQL results and routing remain unchanged; in-memory
cursor closure is a no-op and spilled resources are released earlier at their
partition ownership boundary.
**Risks:** A non-idempotent implementation could double-close readers during
partition reset and task completion. Closing a cursor before the frame finishes
consuming its current row could invalidate live row state.
**Constraints:** Close prior-partition cursors before input.clear deletes
spill files and before new iterators are assigned. Do not obtain closure by
draining the unread remainder of a wide RANGE partition. Keep task-completion
cleanup as a fallback for cancellation and exceptional exits.
**Success:** Across many forced-spill RANGE partitions in one task, every
prior spill reader closes before the next partition begins and output remains
equivalent to the naive 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]
