The DISTINCT rejection check is ineffective in this context: in the moving-frame branch the functions passed into AggregateProcessor(...) are the inner aggregate functions (e.g., Count, Sum) and not AggregateExpression, so the case ae: AggregateExpression => ae.isDistinct branch will never match. If you want to keep the “no DISTINCT” contract local to eligibility, derive the isDistinct flag(s) from the original expressions (the WindowExpression(AggregateExpression, _) nodes) and pass that information into eligibleForSegTree, or perform the distinct check while building aggFilters (same place you pattern match on AggregateExpression).

The comment “keeping listener registration idempotent” looks inaccurate here: AggregateProcessor construction doesn’t register task listeners. Please update the comment to reflect the actual reason for using a val (e.g., avoiding per-factory invocation allocation) without referencing listener registration.

This file contains merge-blocking placeholders and nonstandard markers: backticked “the class documentation”, multiple TODO(SPARK-XXXXX) stubs, and a FIXME(name) line. Before merging, please replace placeholders with concrete references (doc path/URL or class name) and actual SPARK JIRA IDs, and convert the FIXME(...) to a TODO(SPARK-xxxxx) (without personal attribution) so it fits Spark’s tracking conventions.

This file contains merge-blocking placeholders and nonstandard markers: backticked “the class documentation”, multiple TODO(SPARK-XXXXX) stubs, and a FIXME(name) line. Before merging, please replace placeholders with concrete references (doc path/URL or class name) and actual SPARK JIRA IDs, and convert the FIXME(...) to a TODO(SPARK-xxxxx) (without personal attribution) so it fits Spark’s tracking conventions.

This file contains merge-blocking placeholders and nonstandard markers: backticked “the class documentation”, multiple TODO(SPARK-XXXXX) stubs, and a FIXME(name) line. Before merging, please replace placeholders with concrete references (doc path/URL or class name) and actual SPARK JIRA IDs, and convert the FIXME(...) to a TODO(SPARK-xxxxx) (without personal attribution) so it fits Spark’s tracking conventions.

This file contains merge-blocking placeholders and nonstandard markers: backticked “the class documentation”, multiple TODO(SPARK-XXXXX) stubs, and a FIXME(name) line. Before merging, please replace placeholders with concrete references (doc path/URL or class name) and actual SPARK JIRA IDs, and convert the FIXME(...) to a TODO(SPARK-xxxxx) (without personal attribution) so it fits Spark’s tracking conventions.

This file contains merge-blocking placeholders and nonstandard markers: backticked “the class documentation”, multiple TODO(SPARK-XXXXX) stubs, and a FIXME(name) line. Before merging, please replace placeholders with concrete references (doc path/URL or class name) and actual SPARK JIRA IDs, and convert the FIXME(...) to a TODO(SPARK-xxxxx) (without personal attribution) so it fits Spark’s tracking conventions.

This file contains merge-blocking placeholders and nonstandard markers: backticked “the class documentation”, multiple TODO(SPARK-XXXXX) stubs, and a FIXME(name) line. Before merging, please replace placeholders with concrete references (doc path/URL or class name) and actual SPARK JIRA IDs, and convert the FIXME(...) to a TODO(SPARK-xxxxx) (without personal attribution) so it fits Spark’s tracking conventions.

Remove the duplicated word in the comment (‘the the’).

The reference to “spark-floating-point-digest-trap skill” isn’t a stable/public pointer for future maintainers. Please replace it with an in-repo doc link, a SPARK JIRA, or a brief self-contained explanation of the failure mode and why the chosen digest avoids it.

The fallback frame is allocated in every SegmentTreeWindowFunctionFrame ctor call, even for partitions that will take the segtree path exclusively. For a task with many frames this is avoidable cost. Consider constructing it lazily inside prepare() only when rows.length < minPartitionRows.

prepare resets all segtree state at the top (tree, iters, cursors, bounds) before the fallback-path branch calls fallback.prepare(rows). If that fallback.prepare throws (e.g., OOM during listener init), the frame is left reset-but-unprepared; a subsequent write would dispatch with fallbackUsed = true against an unprepared fallback. Safer: defer the reset until after the size-check branch, or mark the frame BROKEN on failure so write fails loudly.

writeRow and writeRange reimplement the monotone-cursor admit/drop loop from SlidingWindowFunctionFrame.write (WindowFunctionFrame.scala:466-499), swapped from drop-then-admit to admit-then-drop and without the buffer maintenance. I verified the two orderings converge to the same (lowerBound, upperBound) at each step for RowFrame, so correctness holds — but two parallel implementations of the same invariant are a maintenance hazard: a future fix to sliding could silently diverge here. Consider extracting the admit/drop cursor advance as a shared helper; only the queryInto vs. processor.initialize+update+evaluate should differ.

Using taskMemoryManager.getTungstenMemoryMode() registers the spiller as ON_HEAP or OFF_HEAP depending on Tungsten mode, but the actual cached state (SpecificInternalRow buffers, Array[Array[InternalRow]]) is always on the JVM heap. Under OFF_HEAP Tungsten these on-heap allocations get charged to the off-heap pool — phantom pressure on the off-heap budget and no accounting of real heap pressure. The T9 OFF_HEAP path test codifies this behavior but may be enshrining a bug. Consider hardcoding MemoryMode.ON_HEAP since these allocations are unambiguously on-heap.

The tree and the frame each register their own TaskCompletionListener to call close() (frame: SegmentTreeWindowFunctionFrame.scala:113-118). In the production frame path this is two listeners per tree — the frame's listener already drives tree.close(). Safe because close is idempotent, but noise in the listener chain. Consider dropping the tree's own listener in the frame-driven path (keep it only behind a registerSelfClose: Boolean ctor flag for standalone test use), and document that the frame owns lifetime.

The comment claims "each Frame calls processor.initialize(...) in prepare", but SegmentTreeWindowFunctionFrame.prepare does NOT — only the embedded fallback does. The segtree path relies on processor.evaluate(source, target) against an external buffer and never touches the internal one. (This is a different concern from the Copilot comment on the "listener registration idempotent" parenthetical below.) Please also note that cross-partition reuse is safe for segtree because it never reads the internal buffer — that's the load-bearing invariant, not that initialize is always called.

The PR description advertises a 9-function allowlist (MIN/MAX/SUM/COUNT/AVG/STDDEV_POP/STDDEV_SAMP/VAR_POP/VAR_SAMP), but this filter accepts any non-window DeclarativeAggregate. First, Last, AnyValue, BitAndAgg/BitOrAgg/BitXorAgg, BoolAnd, BoolOr, the Regr* family, and others all route through the segtree path silently, yet the test suites exercise only the 9 advertised aggregates. Either tighten eligibility to an explicit allowlist that matches the description, or add coverage for every eligible DeclarativeAggregate and update the description. The safer choice for a new opt-in feature is the allowlist.

The dedicated cannotAcquireMemoryForWindowAggregateError factory now pins the error class to UNABLE_TO_ACQUIRE_MEMORY; the || "unable" disjunct predates that and is now strictly less precise.

Walking the cause chain for a target type is already available via org.apache.commons.lang3.exception.ExceptionUtils.getRootCause, which Spark tests use extensively. Optional simplification — something like Option(ExceptionUtils.getRootCause(e)).collect { case a: ArithmeticException => a } — just less bespoke test plumbing.

Two of the four new confs are public (spark.sql.window.segmentTree.enabled and .blockSize), both pinned at version=4.2.0, while the PR description labels the feature "WIP / Draft." Public SQLConfs become a long-tail deprecation commitment once they ship. Is the public surface intentional for a WIP feature? Demoting both to .internal() for the initial landing and promoting once the design stabilizes would give you more freedom to evolve defaults / names.

The class doc explains what the structure is but not why segment tree over alternatives — and that omission matters because alternative optimizations for the same problem exist (monotonic deque for MIN/MAX; add/remove with inverse for invertible aggregates). The load-bearing correctness argument for tree merges is twofold and currently implicit:

1. No inverse required — any aggregate whose mergeExpressions is associative works, including MIN/MAX and other non-invertible cases.
2. Bounded FP error per query — tree merges do O(log W) additions per query with no cumulative state, so floating-point error is bounded per output row. Any add/remove / inverse-based scheme (cf. SPARK-36844) accumulates error across the whole partition and drifts over long partitions.

The second point in particular is the reason to prefer segtree for STDDEV/VAR/FP-SUM even though an inverse-based O(1) approach looks cheaper on paper. Worth adding a short paragraph to the class doc so future authors considering alternative optimizations understand what property was preserved here.