peter-toth commented on code in PR #57346:
URL: https://github.com/apache/spark/pull/57346#discussion_r3860899086
##########
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 2.** This row and the ten other deque rows (`:26`, `:184`, `:196`,
`:208`, `:221`, `:234`, `:247`, `:260`, `:273`, `:286`) were produced at
`bf07c13`. `e02bbe4` then changed `SlidingWindowMinMaxFunctionFrame`, dropping
a per-row `getNextOrNull` and a whole second pass over the row array on every
ROWS frame. Every case in this file is a ROWS frame, since `frameFor` emits
`ROWS BETWEEN $halfW PRECEDING AND $halfW FOLLOWING`, so the change touches
every deque row here.
I measured it rather than asking for a rerun, comparing this head against
the same build with `needsLowerRow` forced back to `true` (5 iterations each,
single window partition):
| shape | with gate | pre-`e02bbe4` |
|---|---|---|
| Section A, MIN, 256K rows, W=1001 | 102 ms | 95 ms |
| Section A, MAX, 256K rows, W=1001 | 75 ms | 78 ms |
| Section H, MIN, 2M rows, W=11 | 334 ms | 346 ms |
| disk-spilling ROWS, MIN, 200K rows, W=1001 | 130 ms | 138 ms |
All within run-to-run noise, in both directions. So every number in this
file stands and no conclusion moves; finding 23 was a dead-work claim, not a
perf claim, and this confirms it.
What is left is provenance: `sql/core/benchmarks/*-results.txt` is meant to
be reproducible from the code at the same commit, and this is the third time
the regeneration has landed ahead of the last code change (R3, R4, now). A full
rerun is not worth it for a within-noise delta - Section B W=4001 naive alone
is ~120 s per iteration. Two cheaper options: make the regeneration the last
commit before merge, or add one line to the body saying which commit generated
the file.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/window/WindowSegmentTreeAllowlistSuite.scala:
##########
@@ -42,7 +43,12 @@ class WindowSegmentTreeAllowlistSuite
private val enableSegTree: Map[String, String] = Map(
SQLConf.WINDOW_SEGMENT_TREE_ENABLED.key -> "true",
- SQLConf.WINDOW_SEGMENT_TREE_MIN_PARTITION_ROWS.key -> "1")
+ SQLConf.WINDOW_SEGMENT_TREE_MIN_PARTITION_ROWS.key -> "1",
+ SQLConf.WINDOW_MONOTONIC_DEQUE_ENABLED.key -> "false")
Review Comment:
**Finding 24.** This pin is right, and it is missing from the sibling suite.
Every case in `SegmentTreeWindowMetricsSuite` is built on `min`/`max`
(`:78-79`, `:94`, `:107`, `:134`, `:154`, `:172`, `:203`), and none of its
seven `withSQLConf` blocks pins the new conf. With the deque on it intercepts
all of them before `eligibleForSegTree` is consulted, so both segment-tree
counters read 0.
Measured: flip `WINDOW_MONOTONIC_DEQUE_ENABLED`'s default to `true` and run
`sql/testOnly org.apache.spark.sql.execution.window.*` on this head. 153
succeeded, 6 failed, and all 6 are in this one suite - it is the only suite in
the package that breaks.
```
- segment-tree path increments numSegmentTreeFrames (one per frame per
partition) *** FAILED ***
0 did not equal 3 expected 3 segtree frames (one per partition), got
metrics =
Map(number of segment-tree frames prepared -> 0, number of segment-tree
fallback frames prepared -> 0)
- fallback path increments numSegmentTreeFallbackFrames *** FAILED ***
- T1 (G1) numPartitions > numTasks, identical length: every partition
counted *** FAILED ***
- T2 (G2) identical-length partitions across keys *** FAILED ***
- T3 (G3) all-length-1 unique keys, fallback path: every partition counted
*** FAILED ***
- T4 (G4) mixed segtree + fallback, non-aliasing order *** FAILED ***
```
Green today only because the conf defaults to `false`, so this is a landmine
for whichever follow-up flips it - the same one you defused here.
`SegmentTreeWindowFunctionSuite` and `UnboundedFollowingSegmentTreeSuite` are
both fine, so it really is just this file.
Fix: add `SQLConf.WINDOW_MONOTONIC_DEQUE_ENABLED.key -> "false"` to each of
that suite's `withSQLConf` blocks (`:72`, `:89`, `:104`, `:127`, `:144`,
`:164`, `:182`), or hoist the repeated prefix into a shared `val` the way you
did here.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/window/MonotonicDequeWindowFunctionSuite.scala:
##########
@@ -0,0 +1,342 @@
+/*
+ * 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)
Review Comment:
**Finding 25.** These four spill tests (`:244`, `:254`, `:266`, `:321`) all
use `rowsBetween`, and after `e02bbe4` a ROWS frame no longer opens a lower
cursor. So they now exercise one `SpillableArrayIterator`, not two, and the
only two RANGE tests (`:200`, `:313`) run in memory. Nothing in the suite
covers the combination any more.
That combination is the subtlest thing in the frame. A
`SpillableArrayIterator` rebinds a single reusable `UnsafeRow` on every
`next()`, and `write` holds `nextRow` across an advance of `lowerIterator`:
```scala
while (nextRow != null && ubound.compare(nextRow, upperBound, current,
index) <= 0) {
if (lbound.compare(nextRow, lowerBound, current, index) < 0) {
lowerBound += 1
if (needsLowerRow) lowerRow =
WindowFunctionFrame.getNextOrNull(lowerIterator) // <-- advances a peer
} else {
... deques(di).admit(nextRow, upperBound)
// <-- still reads nextRow
```
It is safe because each `SpillableArrayIterator` allocates its own
`currentRow`
(`sql/core/src/main/scala/org/apache/spark/sql/execution/ExternalAppendOnlyUnsafeRowArray.scala:243`)
and `UnsafeExternalSorter.getIterator(startIndex)` builds an independent
reader chain rather than reusing a shared one. Both are facts about another
class, and the pre-`e02bbe4` code was the thing that pinned them.
I verified the behaviour rather than just asserting the gap: with
`WINDOW_EXEC_BUFFER_IN_MEMORY_THRESHOLD=4` and
`WINDOW_EXEC_BUFFER_SPILL_THRESHOLD=8`, RANGE ascending, RANGE descending, and
a multi-column `RANGE BETWEEN CURRENT ROW AND CURRENT ROW` all match the naive
baseline, as does a sweep of 10 frame shapes x {ROWS, RANGE} x {spill, no
spill}. So this is coverage, not a bug.
One test closes it:
```scala
test("SPARK-58201: Range-based moving frame: MIN/MAX with spill") {
val df = baseDF.selectExpr("id", "pk", "CAST(id / 2 AS INT) AS ord_val",
"v_int")
withSQLConf(
SQLConf.WINDOW_EXEC_BUFFER_IN_MEMORY_THRESHOLD.key -> "8",
SQLConf.WINDOW_EXEC_BUFFER_SPILL_THRESHOLD.key -> "16") {
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)))
}
}
```
--
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]
