peter-toth commented on code in PR #57815:
URL: https://github.com/apache/spark/pull/57815#discussion_r3755095094
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/WindowResolution.scala:
##########
@@ -163,6 +163,26 @@ object WindowResolution {
errorClass = "INVALID_WINDOW_SPEC_FOR_AGGREGATION_FUNC",
messageParameters = Map("aggFunc" ->
toSQLExpr(agg.aggregateFunction))
)
+ case AggregateExpression(_: PythonUDAF, _, true, _, _) =>
+ windowExpression.failAnalysis(
+ errorClass = "DISTINCT_WINDOW_FUNCTION_UNSUPPORTED",
+ messageParameters = Map("windowExpr" -> toSQLExpr(windowExpression))
+ )
+ case AggregateExpression(function, _, true, _, _)
+ if (windowExpression.windowSpec.frameSpecification match {
+ case SpecifiedWindowFrame(_, UnboundedPreceding, _) =>
+ val distinctExpressions =
function.children.filterNot(_.foldable).map {
+ case sortOrder: SortOrder => sortOrder.child
+ case expression => expression
+ }
+ RowOrdering.isOrderable(distinctExpressions)
+ case _ => false
+ }) => ()
+ case AggregateExpression(_, _, true, _, _) =>
+ windowExpression.failAnalysis(
+ errorClass = "DISTINCT_WINDOW_FUNCTION_UNSUPPORTED",
Review Comment:
**Finding 2.** The message behind this condition is now false.
`common/utils/src/main/resources/error/error-conditions.json:2309` still reads
"Distinct window functions are not supported: `<windowExpr>`.", but after this
PR they are supported whenever the frame's lower bound is `UNBOUNDED PRECEDING`
and the inputs are orderable. Someone who writes `count(DISTINCT x) OVER (ORDER
BY id ROWS BETWEEN 1 PRECEDING AND CURRENT ROW)` is told the feature doesn't
exist, with no hint that widening the frame makes it work.
Three different reasons also land on this one message and can't be told
apart: unsupported frame (this branch), unorderable DISTINCT input (the
`RowOrdering.isOrderable` guard at `:178`), and Python UDAF (`:166`).
Minimum ask: reword the message to name the supported shape, e.g. "DISTINCT
is not supported for this window function: `<windowExpr>`. DISTINCT aggregates
are supported only when the window frame's lower bound is UNBOUNDED PRECEDING
and the DISTINCT inputs are orderable."
Better, and the idiomatic option here, is sub-conditions so each rejection
says which rule it hit:
```json
"DISTINCT_WINDOW_FUNCTION_UNSUPPORTED" : {
"message" : [ "Distinct window functions are not supported: <windowExpr>."
],
"subClass" : {
"FRAME_LOWER_BOUND_NOT_UNBOUNDED_PRECEDING" : {
"message" : [ "DISTINCT aggregates are supported only when the window
frame's lower bound is UNBOUNDED PRECEDING." ]
},
"INPUT_NOT_ORDERABLE" : {
"message" : [ "The DISTINCT input <dataType> is not orderable." ]
},
"PYTHON_UDAF" : {
"message" : [ "DISTINCT is not supported for Python user-defined
aggregate functions." ]
}
},
"sqlState" : "0A000"
}
```
That does mean updating the callers that assert on the bare condition name
(`AnalysisErrorSuite`, `SegmentTreeWindowFunctionSuite:415,419`,
`UnboundedFollowingSegmentTreeSuite:384,388`, and the `listagg.sql` golden
files), so if you'd rather keep the change small the reworded single message is
fine by me.
##########
core/src/main/java/org/apache/spark/util/collection/unsafe/sort/UnsafeExternalSorter.java:
##########
@@ -547,6 +577,8 @@ public void insertKVRecord(Object keyBase, long keyOffset,
int keyLen,
Object valueBase, long valueOffset, int valueLen, long prefix, boolean
prefixIsNull)
throws IOException {
+ spillIfThresholdReached();
Review Comment:
**Finding 3.** `insertKVRecord` did not consult the force-spill thresholds
before this PR; now it does. The new frame needs that (it relies on the sorter
honoring `spark.sql.windowExec.buffer.spill.threshold`), but `insertKVRecord`
is shared, and two aggregation paths reach it through
`UnsafeKVExternalSorter.insertKV`:
- `ObjectAggregationIterator.createExternalSorterForInput`
(`sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/ObjectAggregationIterator.scala:328`)
- `ObjectAggregationMap.dumpToExternalSorter`
(`sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/ObjectAggregationMap.scala:75`)
Both construct the sorter with
`spark.shuffle.spill.numElementsForceSpillThreshold` /
`spark.shuffle.spill.maxSizeInBytesForSpillThreshold` and then insert row by
row, so `ObjectHashAggregateExec` now force-spills where it previously didn't.
Both configs default to "never" (`Integer.MAX_VALUE` / `Long.MAX_VALUE`), so
nothing changes out of the box — but anyone who lowered either to fight OOM
gets a different number of spill files.
I'm not asking you to revert it; the thresholds arguably always meant to
apply here, and the two new `UnsafeKVExternalSorterSuite` cases cover the
mechanism. The ask is that it stops being invisible: mention it under "Does
this PR introduce _any_ user-facing change?", or split it into its own JIRA so
it can be reasoned about (and reverted or backported) independently of the
window feature.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/AnalysisErrorSuite.scala:
##########
@@ -200,12 +200,12 @@ class AnalysisErrorSuite extends AnalysisTest with
DataTypeErrorsBase {
WindowSpecDefinition(
UnresolvedAttribute("a") :: Nil,
SortOrder(UnresolvedAttribute("b"), Ascending) :: Nil,
- UnspecifiedFrame)).as("window")),
+ SpecifiedWindowFrame(RowFrame, CurrentRow,
CurrentRow))).as("window")),
Review Comment:
**Finding 6.** Moving this case to `ROWS BETWEEN CURRENT ROW AND CURRENT
ROW` is the right call — with the default frame it would now pass analysis. But
it leaves this suite covering only one of the three rejection paths
(`WindowResolution.scala:181`, bounded lower bound). Neither branch the PR
*adds* has a test:
- the `RowOrdering.isOrderable` guard at `WindowResolution.scala:178` — e.g.
`count(DISTINCT map_col) OVER (ORDER BY id)` must still fail, and nothing pins
that. This gate is what keeps unorderable keys out of
`UnsafeKVExternalSorter`'s `GenerateOrdering`, so it's worth a regression test;
- the `PythonUDAF` branch at `WindowResolution.scala:166` — without it a
distinct pandas UDAF reaches `AggregateProcessor` and dies with
`INTERNAL_ERROR: Unsupported aggregate function` instead of a user-facing
error. That one probably belongs in a PySpark test rather than here.
Both are cheap and they're the kind of gate that silently rots when someone
widens the frame check later.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/WindowResolution.scala:
##########
@@ -109,11 +112,13 @@ object WindowResolution {
* - Disallows [[FrameLessOffsetWindowFunction]] (e.g. [[Lag]]) without
defined ordering or
* one with a frame which is defined as something other than an offset
frame (e.g.
* `ROWS BETWEEN` is logically incompatible with offset functions).
- * - Disallows distinct aggregate expressions in window functions.
+ * - Allows supported distinct aggregate expressions with orderable inputs
and an unbounded
Review Comment:
**Finding 5.** The rule is documented here in the scaladoc, but the
capability isn't documented anywhere a user will look.
`docs/sql-ref-syntax-qry-select-window.md` describes `window_function` with a
"Ranking / Analytic / Aggregate Functions" breakdown (around `:47`) and never
mentions `DISTINCT`. After this PR `COUNT(DISTINCT x) OVER (...)` works for
some frames and fails for others, and the only way to find out which is to run
the query and read the error.
Two or three lines under the "Aggregate Functions" bullet would cover it:
`DISTINCT` is supported when the frame's lower bound is `UNBOUNDED PRECEDING` —
which includes both the default frame and `OVER ()` — unsupported for bounded
and sliding lower bounds, and the DISTINCT inputs must be orderable.
I don't think a `docs/sql-migration-guide.md` entry is needed, since no
previously-working query changes behavior; shout if you disagree.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/window/WindowEvaluatorFactoryBase.scala:
##########
@@ -141,21 +142,34 @@ trait WindowEvaluatorFactoryBase {
* [[WindowExpression]]s and factory function for the
[[WindowFunctionFrame]].
*/
protected lazy val windowFrameExpressionFactoryPairs = {
- type FrameKey = (String, FrameType, Expression, Expression,
Seq[Expression])
+ type FrameKey =
+ (String, FrameType, Expression, Expression, Seq[Expression],
Option[Expression])
type ExpressionBuffer = mutable.Buffer[Expression]
val framedFunctions = mutable.Map.empty[FrameKey, (ExpressionBuffer,
ExpressionBuffer)]
+ def distinctChildren(ae: AggregateExpression): Seq[Expression] = {
+ ae.aggregateFunction.children.filterNot(_.foldable).map {
Review Comment:
**Finding 7.** `distinctChildren` doesn't dedupe by `canonicalized`, so an
aggregate that mentions the same child twice gets one key column per occurrence.
The realistic case is `listagg(DISTINCT col) WITHIN GROUP (ORDER BY col)
OVER ()`, which this PR newly allows: the `ListAgg` guard at
`WindowResolution.scala:149-153` only fires when `orderSpec.nonEmpty` or the
frame differs from `SpecifiedWindowFrame(RowFrame, UnboundedPreceding,
UnboundedFollowing)`, and `OVER ()` resolves to exactly that frame with an
empty order spec. `ListAgg.children` is `child +: delimiter +:
orderExpressions` (`collect.scala:808`) and `SortOrder` isn't foldable, so
`distinctChildren` returns `[col, col]`. Consequences:
- `distinctKeySchema` becomes `(key0: col, key1: col)`, doubling the dedup
key in the `BytesToBytesMap` and in every sorter record;
- `Utils.toMap` keeps only the last pair for a duplicate key, so
`distinctColumnAttributeLookup` maps `col -> key1` and the rewritten `ListAgg`
reads `key1` only. `key0` is projected and written everywhere and read nowhere.
Deduping by `canonicalized` before `distinctInputAttributes` is built fixes
it, and stays consistent with the frame key since that groups on the same list:
```scala
def distinctChildren(ae: AggregateExpression): Seq[Expression] = {
ae.aggregateFunction.children.filterNot(_.foldable).map {
case sortOrder: SortOrder => sortOrder.child
case expression => expression
}.distinctBy(_.canonicalized)
}
```
While you're here: that `listagg(...) WITHIN GROUP (...) OVER ()` shape is
newly supported and has no test. Worth adding, because `ListAgg`'s eval-time
sort makes it the one full-partition DISTINCT aggregate whose output order is
deterministic — a useful contrast to the `collect_list(DISTINCT ...) OVER ()`
case where it isn't.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/window/DistinctWindowFunctionFrame.scala:
##########
@@ -0,0 +1,548 @@
+/*
+ * 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.{SparkEnv, TaskContext}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.util.UnsafeRowUtils
+import org.apache.spark.sql.execution.{ExternalAppendOnlyUnsafeRowArray,
UnsafeKVExternalSorter}
+import org.apache.spark.sql.execution.metric.SQLMetric
+import org.apache.spark.sql.types.{DataType, IntegerType, StructField,
StructType}
+import org.apache.spark.unsafe.map.BytesToBytesMap
+
+/**
+ * Computes one or more equivalent DISTINCT aggregate expressions for a frame
whose lower bound is
+ * UNBOUNDED PRECEDING.
+ *
+ * Such a frame never removes rows: it either covers the entire partition or
only grows as output
+ * advances. For each qualifying input row this frame finds the first output
row whose upper bound
+ * contains it. A BytesToBytesMap removes duplicates while it remains below
configurable key-count
+ * and memory-size soft limits. When another new key arrives after either
limit has been reached, or
+ * an append fails, the frame permanently falls back to an external sorter for
the rest of the
+ * window partition. For a growing frame, that sorter finds the earliest event
for every key, then
+ * a second external sorter orders those events by (firstVisibleIndex,
inputIndex). The frame feeds
+ * each unique, normalized DISTINCT input to the aggregate processor when its
event becomes visible.
+ * A frame covering the entire partition consumes unique inputs directly
because their order is
+ * undefined.
+ */
+private[window] abstract class DistinctWindowFunctionFrame(
+ target: InternalRow,
+ processor: AggregateProcessor,
+ distinctExpressions: Seq[Expression],
+ filter: Option[Expression],
+ inputSchema: Seq[Attribute],
+ spillSize: SQLMetric,
+ hashFallbackThreshold: Int,
+ spillThreshold: Int,
+ spillSizeThreshold: Long)
+ extends WindowFunctionFrame with AutoCloseable {
+
+ private val distinctFields = distinctExpressions.zipWithIndex.map { case
(expression, index) =>
+ StructField(s"key$index", expression.dataType, expression.nullable)
+ }
+ private val distinctKeySchema = StructType(distinctFields)
+ private val distinctTypes = distinctKeySchema.map(_.dataType)
+ private val positionSchema = StructType(Seq(
+ StructField("firstVisibleIndex", IntegerType, nullable = false),
+ StructField("inputIndex", IntegerType, nullable = false)))
+
+ private val distinctProjection =
UnsafeProjection.create(distinctExpressions, inputSchema)
+ private val distinctOrdering =
RowOrdering.createNaturalAscendingOrdering(distinctTypes)
+ private val canUseHashDedup =
distinctTypes.forall(UnsafeRowUtils.isBinaryStable)
+ private val boundFilter = filter.map(Predicate.create(_, inputSchema))
+ private val positionInput =
+ new SpecificInternalRow(Seq(IntegerType, IntegerType))
+ private val positionProjection =
+ UnsafeProjection.create(Array[DataType](IntegerType, IntegerType))
+
+ private var eventSorter: UnsafeKVExternalSorter = _
+ private var eventIterator: UnsafeKVExternalSorter#KVSorterIterator = _
+ private var nextEventIndex = Int.MaxValue
+
+ Option(TaskContext.get()).foreach(_.addTaskCompletionListener[Unit](_ =>
close()))
+
+ override final def prepare(rows: ExternalAppendOnlyUnsafeRowArray): Unit = {
+ closeEventResources()
+ nextEventIndex = Int.MaxValue
+ processor.initialize(rows.length)
+ val partitionIndex =
Option(TaskContext.get()).map(_.partitionId()).getOrElse(0)
+ boundFilter.foreach(_.initialize(partitionIndex))
+
+ val firstVisibleRows = new FirstVisibleRowsBuilder
+ try {
+ populateFirstVisibleRows(rows, firstVisibleRows)
+ processDistinctRows(firstVisibleRows)
+ spillSize.add(firstVisibleRows.getSpillSize)
+ } finally {
+ firstVisibleRows.close()
+ }
+ prepareFrame(rows)
+ }
+
+ protected def populateFirstVisibleRows(
+ rows: ExternalAppendOnlyUnsafeRowArray,
+ firstVisibleRows: FirstVisibleRowsBuilder): Unit
+
+ protected def processDistinctRows(firstVisibleRows:
FirstVisibleRowsBuilder): Unit
+
+ protected def prepareFrame(rows: ExternalAppendOnlyUnsafeRowArray): Unit
+
+ protected final def prepareOrderedEvents(firstVisibleRows:
FirstVisibleRowsBuilder): Unit = {
+ var newEventSorter: UnsafeKVExternalSorter = null
+ try {
+ newEventSorter = firstVisibleRows.buildEvents()
+ eventSorter = newEventSorter
+ newEventSorter = null
+ eventIterator = eventSorter.sortedIterator()
+ loadNextEvent()
+ } finally {
+ if (newEventSorter != null) {
+ newEventSorter.cleanupResources()
+ }
+ }
+ }
+
+ protected final def addCandidate(
+ row: InternalRow,
+ firstVisibleIndex: Int,
+ inputIndex: Int,
+ firstVisibleRows: FirstVisibleRowsBuilder): Unit = {
+ if (boundFilter.forall(_.eval(row))) {
+ positionInput.setInt(0, firstVisibleIndex)
+ positionInput.setInt(1, inputIndex)
+ val distinctRow = distinctProjection(row)
+ firstVisibleRows.add(distinctRow, positionProjection(positionInput))
+ }
+ }
+
+ /**
+ * Uses a bounded BytesToBytesMap to remove duplicates before they reach the
first external
+ * sorter. The map contains only the earliest row for each distinct key
because candidates arrive
+ * in non-decreasing (firstVisibleIndex, inputIndex) order.
+ *
+ * BytesToBytesMap compares raw UnsafeRow bytes, so binary-unstable keys use
the sorter directly.
+ * Once the map reaches either its key-count or memory-size soft limit and
another new key
+ * arrives, or it cannot append a record, all map entries are transferred to
one external sorter.
+ * The rest of this window partition goes directly to that sorter and never
returns to hash-based
+ * deduplication.
+ */
+ protected final class FirstVisibleRowsBuilder extends AutoCloseable {
+ private val map = if (canUseHashDedup) {
+ val taskMemoryManager = TaskContext.get().taskMemoryManager()
+ new BytesToBytesMap(taskMemoryManager, 64,
taskMemoryManager.pageSizeBytes())
+ } else {
+ null
+ }
+ private var usingMap = map != null
+ private var sorter: UnsafeKVExternalSorter = _
+
+ def add(key: UnsafeRow, value: UnsafeRow): Unit = {
+ if (!usingMap) {
+ insertIntoSorter(key, value)
+ return
+ }
+
+ val location = map.lookup(
+ key.getBaseObject,
+ key.getBaseOffset,
+ key.getSizeInBytes)
+ if (location.isDefined) {
+ return
+ }
+
+ if (map.numKeys() >= hashFallbackThreshold ||
+ map.getTotalMemoryConsumption >= spillSizeThreshold) {
+ switchToSorter()
+ insertIntoSorter(key, value)
+ } else if (!location.append(
+ key.getBaseObject,
+ key.getBaseOffset,
+ key.getSizeInBytes,
+ value.getBaseObject,
+ value.getBaseOffset,
+ value.getSizeInBytes)) {
+ switchToSorter()
+ insertIntoSorter(key, value)
+ }
+ }
+
+ def buildEvents(): UnsafeKVExternalSorter = {
+ var events: UnsafeKVExternalSorter = null
+ try {
+ if (usingMap) {
+ // Make the map spillable before allocating the event sorter. The
destructive iterator
+ // releases the hash array immediately and lets memory pressure
spill remaining pages.
+ val iterator = destructiveMapIterator()
+ events = newSorter(positionSchema, distinctKeySchema)
+ drainMap(iterator, (key, position) => emitEvent(events, position,
key))
+ } else {
+ events = newSorter(positionSchema, distinctKeySchema)
+ if (sorter != null) {
+ consumeDistinctRows(
+ sorter, (key, position) => emitEvent(events, position, key))
+ }
+ }
+ val result = events
+ events = null
+ result
+ } finally {
+ if (events != null) {
+ events.cleanupResources()
+ }
+ }
+ }
+
+ def foreachDistinctRow(consume: (UnsafeRow, UnsafeRow) => Unit): Unit = {
+ if (usingMap) {
+ drainMap(destructiveMapIterator(), consume)
+ } else if (sorter != null) {
+ consumeDistinctRows(sorter, consume)
+ }
+ }
+
+ /**
+ * Returns bytes spilled by the first, distinct-key sorter. The frame
accounts for the second,
+ * event-order sorter separately when closing its event resources. Spill
files created while a
+ * destructive BytesToBytesMap iterator releases the map's data pages are
not included in the
+ * window spill metric.
+ */
+ def getSpillSize: Long = if (sorter == null) 0L else sorter.getSpillSize
+
+ private def insertIntoSorter(key: UnsafeRow, value: UnsafeRow): Unit = {
+ if (sorter == null) {
+ sorter = newSorter(distinctKeySchema, positionSchema)
+ }
+ sorter.insertKV(key, value)
+ }
+
+ private def switchToSorter(): Unit = {
+ assert(sorter == null)
+ val iterator = destructiveMapIterator()
+ sorter = newSorter(distinctKeySchema, positionSchema)
+ drainMap(iterator, (key, position) => sorter.insertKV(key, position))
+ }
+
+ private def destructiveMapIterator(): BytesToBytesMap#MapIterator = {
+ assert(usingMap)
+ val iterator = map.destructiveIterator()
+ usingMap = false
+ iterator
+ }
+
+ private def drainMap(
+ iterator: BytesToBytesMap#MapIterator,
+ consume: (UnsafeRow, UnsafeRow) => Unit): Unit = {
+ val key = new UnsafeRow(distinctKeySchema.length)
+ val position = new UnsafeRow(positionSchema.length)
+ while (iterator.hasNext) {
+ val location = iterator.next()
+ key.pointTo(
+ location.getKeyBase,
+ location.getKeyOffset,
+ location.getKeyLength)
+ position.pointTo(
+ location.getValueBase,
+ location.getValueOffset,
+ location.getValueLength)
+ consume(key, position)
+ }
+ }
+
+ override def close(): Unit = {
+ if (sorter != null) {
+ sorter.cleanupResources()
+ sorter = null
+ }
+ if (map != null) {
+ map.free()
+ usingMap = false
+ }
+ }
+ }
+
+ private def consumeDistinctRows(
+ firstSorter: UnsafeKVExternalSorter,
+ consume: (UnsafeRow, UnsafeRow) => Unit): Unit = {
+ val iterator = firstSorter.sortedIterator()
+ var groupKey: UnsafeRow = null
+ var selectedKey: UnsafeRow = null
+ var selectedPosition: UnsafeRow = null
+ try {
+ while (iterator.next()) {
+ val key = iterator.getKey
+ val position = iterator.getValue
+ if (groupKey == null) {
+ groupKey = key.copy()
+ selectedKey = groupKey
+ selectedPosition = position.copy()
+ } else if (distinctOrdering.compare(groupKey, key) == 0) {
+ if (isEarlier(position, selectedPosition)) {
+ selectedKey = key.copy()
+ selectedPosition = position.copy()
+ }
+ } else {
+ consume(selectedKey, selectedPosition)
+ groupKey = key.copy()
+ selectedKey = groupKey
+ selectedPosition = position.copy()
+ }
+ }
+ if (groupKey != null) {
+ consume(selectedKey, selectedPosition)
+ }
+ } finally {
+ iterator.close()
+ }
+ }
+
+ private def isEarlier(left: InternalRow, right: InternalRow): Boolean = {
+ val leftVisibleIndex = left.getInt(0)
+ val rightVisibleIndex = right.getInt(0)
+ leftVisibleIndex < rightVisibleIndex ||
+ leftVisibleIndex == rightVisibleIndex && left.getInt(1) < right.getInt(1)
+ }
+
+ private def emitEvent(
+ events: UnsafeKVExternalSorter,
+ position: UnsafeRow,
+ distinctValue: UnsafeRow): Unit = {
+ events.insertKV(position, distinctValue)
+ }
+
+ private def newSorter(
+ keySchema: StructType,
+ valueSchema: StructType): UnsafeKVExternalSorter = {
+ val taskContext = TaskContext.get()
+ // The frame owns all of its sorters and registers one task completion
listener at construction.
+ UnsafeKVExternalSorter.createWithCallerOwnedLifecycle(
+ keySchema,
+ valueSchema,
+ SparkEnv.get.blockManager,
+ SparkEnv.get.serializerManager,
+ taskContext.taskMemoryManager().pageSizeBytes,
+ spillThreshold,
+ spillSizeThreshold)
+ }
+
+ private def loadNextEvent(): Unit = {
+ if (eventIterator != null && eventIterator.next()) {
+ nextEventIndex = eventIterator.getKey.getInt(0)
+ } else {
+ nextEventIndex = Int.MaxValue
+ }
+ }
+
+ protected final def updateProcessor(index: Int): Unit = {
+ var bufferUpdated = index == 0
+ while (nextEventIndex <= index) {
+ processor.update(eventIterator.getValue)
Review Comment:
**Finding 1.** The frame hands the `AggregateProcessor` rows that point into
memory it then frees, and `processor.evaluate(target)` runs after the free.
`MutableProjection` does not copy non-primitive values —
`CodeGenerator.setColumn` falls through to `row.update(ordinal, value)` — and
`UnsafeRow.getUTF8String` / `getArray` / `getStruct` return views over the
row's backing memory. So any `DeclarativeAggregate` that keeps an input-derived
value in its buffer (`Max`, `Min`, `First`, `Last` on
`STRING`/`ARRAY`/`STRUCT`) retains a pointer into whichever page the row it was
updated with lives in, and so does the shared `windowFunctionResult` once
`evaluate` copies the reference across.
Two sites free that memory too early:
1. Here. On the iteration that consumes the last event, `loadNextEvent()`
calls `eventIterator.next()`, which returns `false` and — see
`KVSorterIterator.next()` / `close()` in
`sql/core/src/main/java/org/apache/spark/sql/execution/UnsafeKVExternalSorter.java`
— calls `cleanupResources()`, freeing the sorter's pages and deleting its
spill files. `processor.evaluate(target)` then runs on `:359`, and
`WindowEvaluatorFactory.next()` re-reads `target` through
`createResultProjection` for every remaining row of the partition. It's worse
once the event sorter has spilled: `UnsafeSorterSpillReader.loadNext()` reads
into a single reusable array, so consuming two events in one `updateProcessor`
call overwrites the first one's bytes before `evaluate`.
2. `:423`. `foreachDistinctRow` on the hash path routes through `drainMap`
over a **destructive** `BytesToBytesMap` iterator, which frees each data page
as it advances (`BytesToBytesMap.MapIterator.advanceToNextPage`), and
`prepare`'s `finally { firstVisibleRows.close() }` -> `map.free()` frees
whatever is left — both strictly before `prepareFrame` calls
`processor.evaluate(target)` at `:428`. That one is unconditional, not
memory-pressure dependent.
The sorter branch of the same method is already safe precisely because
`consumeDistinctRows` copies (`selectedKey = key.copy()`, `:291`). The map
branch being the odd one out is itself a hint that the copy is load-bearing.
On-heap the freed page goes back into `HeapMemoryAllocator`'s size pool, so
the bytes usually survive and the tests pass — they stop surviving as soon as
another consumer in the task reuses that page. Off-heap,
`TaskMemoryManager.freePage` -> `Platform.freeMemory` really releases it, and
`-Dspark.memory.debugFill=true` overwrites it with `0xa5` right away.
A copy at both sites fixes it:
```suggestion
processor.update(eventIterator.getValue.copy())
```
and at `:423`:
```scala
firstVisibleRows.foreachDistinctRow((key, _) => processor.update(key.copy()))
```
Coverage note: every new test uses `count`/`sum`/`avg` (primitive buffers)
or `collect_list`/`listagg` (`TypedImperativeAggregate`s that copy via
`InternalRow.copyValue`), so nothing exercises the vulnerable shape.
`max(DISTINCT strCol)` and `first(DISTINCT strCol)` over both a growing frame
and `OVER ()` would cover it, and running those with off-heap memory or
`spark.memory.debugFill=true` makes the failure deterministic rather than
luck-dependent.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/window/DistinctWindowFunctionFrame.scala:
##########
@@ -0,0 +1,548 @@
+/*
+ * 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.{SparkEnv, TaskContext}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.util.UnsafeRowUtils
+import org.apache.spark.sql.execution.{ExternalAppendOnlyUnsafeRowArray,
UnsafeKVExternalSorter}
+import org.apache.spark.sql.execution.metric.SQLMetric
+import org.apache.spark.sql.types.{DataType, IntegerType, StructField,
StructType}
+import org.apache.spark.unsafe.map.BytesToBytesMap
+
+/**
+ * Computes one or more equivalent DISTINCT aggregate expressions for a frame
whose lower bound is
+ * UNBOUNDED PRECEDING.
+ *
+ * Such a frame never removes rows: it either covers the entire partition or
only grows as output
+ * advances. For each qualifying input row this frame finds the first output
row whose upper bound
+ * contains it. A BytesToBytesMap removes duplicates while it remains below
configurable key-count
+ * and memory-size soft limits. When another new key arrives after either
limit has been reached, or
+ * an append fails, the frame permanently falls back to an external sorter for
the rest of the
+ * window partition. For a growing frame, that sorter finds the earliest event
for every key, then
+ * a second external sorter orders those events by (firstVisibleIndex,
inputIndex). The frame feeds
+ * each unique, normalized DISTINCT input to the aggregate processor when its
event becomes visible.
+ * A frame covering the entire partition consumes unique inputs directly
because their order is
+ * undefined.
+ */
+private[window] abstract class DistinctWindowFunctionFrame(
+ target: InternalRow,
+ processor: AggregateProcessor,
+ distinctExpressions: Seq[Expression],
+ filter: Option[Expression],
+ inputSchema: Seq[Attribute],
+ spillSize: SQLMetric,
+ hashFallbackThreshold: Int,
+ spillThreshold: Int,
+ spillSizeThreshold: Long)
+ extends WindowFunctionFrame with AutoCloseable {
+
+ private val distinctFields = distinctExpressions.zipWithIndex.map { case
(expression, index) =>
+ StructField(s"key$index", expression.dataType, expression.nullable)
+ }
+ private val distinctKeySchema = StructType(distinctFields)
+ private val distinctTypes = distinctKeySchema.map(_.dataType)
+ private val positionSchema = StructType(Seq(
+ StructField("firstVisibleIndex", IntegerType, nullable = false),
+ StructField("inputIndex", IntegerType, nullable = false)))
+
+ private val distinctProjection =
UnsafeProjection.create(distinctExpressions, inputSchema)
+ private val distinctOrdering =
RowOrdering.createNaturalAscendingOrdering(distinctTypes)
+ private val canUseHashDedup =
distinctTypes.forall(UnsafeRowUtils.isBinaryStable)
+ private val boundFilter = filter.map(Predicate.create(_, inputSchema))
+ private val positionInput =
+ new SpecificInternalRow(Seq(IntegerType, IntegerType))
+ private val positionProjection =
+ UnsafeProjection.create(Array[DataType](IntegerType, IntegerType))
+
+ private var eventSorter: UnsafeKVExternalSorter = _
+ private var eventIterator: UnsafeKVExternalSorter#KVSorterIterator = _
+ private var nextEventIndex = Int.MaxValue
+
+ Option(TaskContext.get()).foreach(_.addTaskCompletionListener[Unit](_ =>
close()))
+
+ override final def prepare(rows: ExternalAppendOnlyUnsafeRowArray): Unit = {
+ closeEventResources()
+ nextEventIndex = Int.MaxValue
+ processor.initialize(rows.length)
+ val partitionIndex =
Option(TaskContext.get()).map(_.partitionId()).getOrElse(0)
+ boundFilter.foreach(_.initialize(partitionIndex))
+
+ val firstVisibleRows = new FirstVisibleRowsBuilder
+ try {
+ populateFirstVisibleRows(rows, firstVisibleRows)
+ processDistinctRows(firstVisibleRows)
+ spillSize.add(firstVisibleRows.getSpillSize)
+ } finally {
+ firstVisibleRows.close()
+ }
+ prepareFrame(rows)
+ }
+
+ protected def populateFirstVisibleRows(
+ rows: ExternalAppendOnlyUnsafeRowArray,
+ firstVisibleRows: FirstVisibleRowsBuilder): Unit
+
+ protected def processDistinctRows(firstVisibleRows:
FirstVisibleRowsBuilder): Unit
+
+ protected def prepareFrame(rows: ExternalAppendOnlyUnsafeRowArray): Unit
+
+ protected final def prepareOrderedEvents(firstVisibleRows:
FirstVisibleRowsBuilder): Unit = {
+ var newEventSorter: UnsafeKVExternalSorter = null
+ try {
+ newEventSorter = firstVisibleRows.buildEvents()
+ eventSorter = newEventSorter
+ newEventSorter = null
+ eventIterator = eventSorter.sortedIterator()
+ loadNextEvent()
+ } finally {
+ if (newEventSorter != null) {
+ newEventSorter.cleanupResources()
+ }
+ }
+ }
+
+ protected final def addCandidate(
+ row: InternalRow,
+ firstVisibleIndex: Int,
+ inputIndex: Int,
+ firstVisibleRows: FirstVisibleRowsBuilder): Unit = {
+ if (boundFilter.forall(_.eval(row))) {
+ positionInput.setInt(0, firstVisibleIndex)
+ positionInput.setInt(1, inputIndex)
+ val distinctRow = distinctProjection(row)
+ firstVisibleRows.add(distinctRow, positionProjection(positionInput))
+ }
+ }
+
+ /**
+ * Uses a bounded BytesToBytesMap to remove duplicates before they reach the
first external
+ * sorter. The map contains only the earliest row for each distinct key
because candidates arrive
+ * in non-decreasing (firstVisibleIndex, inputIndex) order.
+ *
+ * BytesToBytesMap compares raw UnsafeRow bytes, so binary-unstable keys use
the sorter directly.
+ * Once the map reaches either its key-count or memory-size soft limit and
another new key
+ * arrives, or it cannot append a record, all map entries are transferred to
one external sorter.
+ * The rest of this window partition goes directly to that sorter and never
returns to hash-based
+ * deduplication.
+ */
+ protected final class FirstVisibleRowsBuilder extends AutoCloseable {
+ private val map = if (canUseHashDedup) {
+ val taskMemoryManager = TaskContext.get().taskMemoryManager()
+ new BytesToBytesMap(taskMemoryManager, 64,
taskMemoryManager.pageSizeBytes())
+ } else {
+ null
+ }
+ private var usingMap = map != null
+ private var sorter: UnsafeKVExternalSorter = _
+
+ def add(key: UnsafeRow, value: UnsafeRow): Unit = {
+ if (!usingMap) {
+ insertIntoSorter(key, value)
+ return
+ }
+
+ val location = map.lookup(
+ key.getBaseObject,
+ key.getBaseOffset,
+ key.getSizeInBytes)
+ if (location.isDefined) {
+ return
+ }
+
+ if (map.numKeys() >= hashFallbackThreshold ||
+ map.getTotalMemoryConsumption >= spillSizeThreshold) {
+ switchToSorter()
+ insertIntoSorter(key, value)
+ } else if (!location.append(
+ key.getBaseObject,
+ key.getBaseOffset,
+ key.getSizeInBytes,
+ value.getBaseObject,
+ value.getBaseOffset,
+ value.getSizeInBytes)) {
+ switchToSorter()
+ insertIntoSorter(key, value)
+ }
+ }
+
+ def buildEvents(): UnsafeKVExternalSorter = {
+ var events: UnsafeKVExternalSorter = null
+ try {
+ if (usingMap) {
+ // Make the map spillable before allocating the event sorter. The
destructive iterator
+ // releases the hash array immediately and lets memory pressure
spill remaining pages.
+ val iterator = destructiveMapIterator()
+ events = newSorter(positionSchema, distinctKeySchema)
+ drainMap(iterator, (key, position) => emitEvent(events, position,
key))
+ } else {
+ events = newSorter(positionSchema, distinctKeySchema)
+ if (sorter != null) {
+ consumeDistinctRows(
+ sorter, (key, position) => emitEvent(events, position, key))
+ }
+ }
+ val result = events
+ events = null
+ result
+ } finally {
+ if (events != null) {
+ events.cleanupResources()
+ }
+ }
+ }
+
+ def foreachDistinctRow(consume: (UnsafeRow, UnsafeRow) => Unit): Unit = {
+ if (usingMap) {
+ drainMap(destructiveMapIterator(), consume)
+ } else if (sorter != null) {
+ consumeDistinctRows(sorter, consume)
+ }
+ }
+
+ /**
+ * Returns bytes spilled by the first, distinct-key sorter. The frame
accounts for the second,
+ * event-order sorter separately when closing its event resources. Spill
files created while a
+ * destructive BytesToBytesMap iterator releases the map's data pages are
not included in the
+ * window spill metric.
+ */
+ def getSpillSize: Long = if (sorter == null) 0L else sorter.getSpillSize
+
+ private def insertIntoSorter(key: UnsafeRow, value: UnsafeRow): Unit = {
+ if (sorter == null) {
+ sorter = newSorter(distinctKeySchema, positionSchema)
+ }
+ sorter.insertKV(key, value)
+ }
+
+ private def switchToSorter(): Unit = {
+ assert(sorter == null)
+ val iterator = destructiveMapIterator()
+ sorter = newSorter(distinctKeySchema, positionSchema)
+ drainMap(iterator, (key, position) => sorter.insertKV(key, position))
+ }
+
+ private def destructiveMapIterator(): BytesToBytesMap#MapIterator = {
+ assert(usingMap)
+ val iterator = map.destructiveIterator()
+ usingMap = false
+ iterator
+ }
+
+ private def drainMap(
+ iterator: BytesToBytesMap#MapIterator,
+ consume: (UnsafeRow, UnsafeRow) => Unit): Unit = {
+ val key = new UnsafeRow(distinctKeySchema.length)
+ val position = new UnsafeRow(positionSchema.length)
+ while (iterator.hasNext) {
+ val location = iterator.next()
+ key.pointTo(
+ location.getKeyBase,
+ location.getKeyOffset,
+ location.getKeyLength)
+ position.pointTo(
+ location.getValueBase,
+ location.getValueOffset,
+ location.getValueLength)
+ consume(key, position)
+ }
+ }
+
+ override def close(): Unit = {
+ if (sorter != null) {
+ sorter.cleanupResources()
+ sorter = null
+ }
+ if (map != null) {
+ map.free()
+ usingMap = false
+ }
+ }
+ }
+
+ private def consumeDistinctRows(
+ firstSorter: UnsafeKVExternalSorter,
+ consume: (UnsafeRow, UnsafeRow) => Unit): Unit = {
+ val iterator = firstSorter.sortedIterator()
+ var groupKey: UnsafeRow = null
+ var selectedKey: UnsafeRow = null
+ var selectedPosition: UnsafeRow = null
+ try {
+ while (iterator.next()) {
+ val key = iterator.getKey
+ val position = iterator.getValue
+ if (groupKey == null) {
+ groupKey = key.copy()
+ selectedKey = groupKey
+ selectedPosition = position.copy()
+ } else if (distinctOrdering.compare(groupKey, key) == 0) {
+ if (isEarlier(position, selectedPosition)) {
+ selectedKey = key.copy()
+ selectedPosition = position.copy()
+ }
+ } else {
+ consume(selectedKey, selectedPosition)
+ groupKey = key.copy()
+ selectedKey = groupKey
+ selectedPosition = position.copy()
+ }
+ }
+ if (groupKey != null) {
+ consume(selectedKey, selectedPosition)
+ }
+ } finally {
+ iterator.close()
+ }
+ }
+
+ private def isEarlier(left: InternalRow, right: InternalRow): Boolean = {
+ val leftVisibleIndex = left.getInt(0)
+ val rightVisibleIndex = right.getInt(0)
+ leftVisibleIndex < rightVisibleIndex ||
+ leftVisibleIndex == rightVisibleIndex && left.getInt(1) < right.getInt(1)
+ }
+
+ private def emitEvent(
+ events: UnsafeKVExternalSorter,
+ position: UnsafeRow,
+ distinctValue: UnsafeRow): Unit = {
+ events.insertKV(position, distinctValue)
+ }
+
+ private def newSorter(
+ keySchema: StructType,
+ valueSchema: StructType): UnsafeKVExternalSorter = {
+ val taskContext = TaskContext.get()
+ // The frame owns all of its sorters and registers one task completion
listener at construction.
+ UnsafeKVExternalSorter.createWithCallerOwnedLifecycle(
+ keySchema,
+ valueSchema,
+ SparkEnv.get.blockManager,
+ SparkEnv.get.serializerManager,
+ taskContext.taskMemoryManager().pageSizeBytes,
+ spillThreshold,
+ spillSizeThreshold)
+ }
+
+ private def loadNextEvent(): Unit = {
+ if (eventIterator != null && eventIterator.next()) {
+ nextEventIndex = eventIterator.getKey.getInt(0)
+ } else {
+ nextEventIndex = Int.MaxValue
+ }
+ }
+
+ protected final def updateProcessor(index: Int): Unit = {
+ var bufferUpdated = index == 0
+ while (nextEventIndex <= index) {
+ processor.update(eventIterator.getValue)
+ bufferUpdated = true
+ loadNextEvent()
+ }
+ if (bufferUpdated) {
+ processor.evaluate(target)
+ }
+ }
+
+ override final def currentLowerBound(): Int = 0
+
+ private def closeEventResources(): Unit = {
+ if (eventSorter != null) {
+ spillSize.add(eventSorter.getSpillSize)
+ }
+ if (eventIterator != null) {
+ eventIterator.close()
+ eventIterator = null
+ }
+ if (eventSorter != null) {
+ eventSorter.cleanupResources()
+ eventSorter = null
+ }
+ }
+
+ override final def close(): Unit = closeEventResources()
+}
+
+/**
+ * Computes DISTINCT aggregates over the entire window partition. Since every
output row has the
+ * same set of unique inputs, their consumption order is undefined and is not
restored after hash
+ * deduplication or sorting by the DISTINCT key.
+ */
+private[window] final class UnboundedDistinctWindowFunctionFrame(
+ target: InternalRow,
+ processor: AggregateProcessor,
+ distinctExpressions: Seq[Expression],
+ filter: Option[Expression],
+ inputSchema: Seq[Attribute],
+ spillSize: SQLMetric,
+ hashFallbackThreshold: Int,
+ spillThreshold: Int,
+ spillSizeThreshold: Long)
+ extends DistinctWindowFunctionFrame(
+ target,
+ processor,
+ distinctExpressions,
+ filter,
+ inputSchema,
+ spillSize,
+ hashFallbackThreshold,
+ spillThreshold,
+ spillSizeThreshold) {
+
+ private var partitionSize = 0
+
+ override protected def populateFirstVisibleRows(
+ rows: ExternalAppendOnlyUnsafeRowArray,
+ firstVisibleRows: FirstVisibleRowsBuilder): Unit = {
+ val iterator = rows.generateIterator()
+ var inputIndex = 0
+ while (iterator.hasNext) {
+ addCandidate(iterator.next(), 0, inputIndex, firstVisibleRows)
+ inputIndex += 1
+ }
+ }
+
+ override protected def processDistinctRows(
+ firstVisibleRows: FirstVisibleRowsBuilder): Unit = {
+ firstVisibleRows.foreachDistinctRow((key, _) => processor.update(key))
+ }
+
+ override protected def prepareFrame(rows: ExternalAppendOnlyUnsafeRowArray):
Unit = {
+ partitionSize = rows.length
+ processor.evaluate(target)
+ }
+
+ override def write(index: Int, current: InternalRow): Unit = {}
+
+ override def currentUpperBound(): Int = partitionSize
+}
+
+/**
+ * Computes DISTINCT aggregates for a growing frame with an UNBOUNDED
PRECEDING lower bound.
+ */
+private[window] final class UnboundedPrecedingDistinctWindowFunctionFrame(
+ target: InternalRow,
+ processor: AggregateProcessor,
+ distinctExpressions: Seq[Expression],
+ filter: Option[Expression],
+ inputSchema: Seq[Attribute],
+ upperBound: BoundOrdering,
+ spillSize: SQLMetric,
+ hashFallbackThreshold: Int,
+ spillThreshold: Int,
+ spillSizeThreshold: Long)
+ extends DistinctWindowFunctionFrame(
+ target,
+ processor,
+ distinctExpressions,
+ filter,
+ inputSchema,
+ spillSize,
+ hashFallbackThreshold,
+ spillThreshold,
+ spillSizeThreshold) {
+
+ private val rowOffset = upperBound match {
+ case RowBoundOrdering(offset) => Some(offset)
+ case _ => None
+ }
+
+ private var partitionSize = 0
+ private var boundaryIterator: Iterator[UnsafeRow] = Iterator.empty
+ private var nextBoundaryRow: UnsafeRow = _
+ private var boundaryInputIndex = 0
+
+ override protected def populateFirstVisibleRows(
+ rows: ExternalAppendOnlyUnsafeRowArray,
+ firstVisibleRows: FirstVisibleRowsBuilder): Unit = {
+ rowOffset match {
+ case Some(offset) =>
+ // A ROWS bound depends only on row indexes. Calculate the first
visible output row
+ // directly instead of scanning the partition again as output rows.
+ val inputIterator = rows.generateIterator()
+ var inputIndex = 0
+ while (inputIterator.hasNext) {
+ val input = inputIterator.next()
+ val firstVisibleIndex = inputIndex.toLong - offset.toLong
+ if (firstVisibleIndex < rows.length) {
+ addCandidate(
+ input,
+ math.max(firstVisibleIndex, 0L).toInt,
+ inputIndex,
+ firstVisibleRows)
+ }
+ inputIndex += 1
+ }
+
+ case None =>
+ val inputIterator = rows.generateIterator()
+ val outputIterator = rows.generateIterator()
+ var nextInput = WindowFunctionFrame.getNextOrNull(inputIterator)
+ var inputIndex = 0
+ var outputIndex = 0
+
+ while (outputIterator.hasNext && nextInput != null) {
+ val currentOutput = outputIterator.next()
+ while (nextInput != null &&
+ upperBound.compare(nextInput, inputIndex, currentOutput,
outputIndex) <= 0) {
+ addCandidate(nextInput, outputIndex, inputIndex, firstVisibleRows)
+ inputIndex += 1
+ nextInput = WindowFunctionFrame.getNextOrNull(inputIterator)
+ }
+ outputIndex += 1
+ }
+ }
+ }
+
+ override protected def processDistinctRows(
+ firstVisibleRows: FirstVisibleRowsBuilder): Unit = {
+ prepareOrderedEvents(firstVisibleRows)
+ }
+
+ override protected def prepareFrame(rows: ExternalAppendOnlyUnsafeRowArray):
Unit = {
+ partitionSize = rows.length
+ boundaryInputIndex = 0
+ if (rowOffset.isEmpty) {
+ boundaryIterator = rows.generateIterator()
+ nextBoundaryRow = WindowFunctionFrame.getNextOrNull(boundaryIterator)
+ } else {
+ boundaryIterator = Iterator.empty
+ nextBoundaryRow = null
+ }
+ }
+
+ override def write(index: Int, current: InternalRow): Unit = {
+ updateProcessor(index)
+ rowOffset match {
+ case Some(offset) =>
+ // The upper bound is exclusive, hence the extra one after applying
the ROWS offset.
+ val upperBoundIndex = index.toLong + offset.toLong + 1L
+ boundaryInputIndex = math.max(0L, math.min(upperBoundIndex,
partitionSize.toLong)).toInt
+
+ case None =>
+ while (nextBoundaryRow != null &&
Review Comment:
**Finding 4.** This boundary tracking is pure overhead on the RANGE path.
`currentUpperBound()` / `currentLowerBound()` are read in exactly one place in
the tree —
`sql/core/src/main/scala/org/apache/spark/sql/execution/python/ArrowWindowPythonEvaluatorFactory.scala:363-365`
— and a distinct frame can never reach it:
- `SparkStrategies.Window` (`:786-796`) routes a `Window` node to either
`WindowExec` or `ArrowWindowPythonExec` by `WindowFunctionType`, so SQL and
Python window functions are never in the same physical node;
- a distinct `PythonUDAF` is rejected in
`WindowResolution.checkWindowFunction:166`.
So `boundaryIterator` / `nextBoundaryRow` (`:522-523`) exist only to compute
a value nothing reads, and they cost a third live iterator over the partition
plus one extra full pass over `rows` per window partition — re-read from disk
whenever the input buffer spilled. The ROWS branch's arithmetic is free, so
this is specifically the RANGE cost.
Suggest computing it on demand instead, so the scan only happens if someone
actually asks:
```scala
override def currentUpperBound(): Int = {
// Advance lazily: nothing in the SQL window path reads this.
...
}
```
If you'd rather keep the eager version, a one-line comment saying the bound
is maintained only to satisfy the `WindowFunctionFrame` contract would at least
stop the next reader from assuming it's needed.
--
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]
