Support spilling for WindowAggExec

[wirybeaver]

Is your feature request related to a problem or challenge?

WindowAggExec can require buffering an entire window partition before it can evaluate some window functions. Today that buffering is memory-only, so queries over large or skewed partitions can exhaust the configured memory pool even when the runtime has a spill-capable disk manager.

This is surprising because other memory-intensive physical operators in DataFusion already participate in spill paths, but window aggregation does not. A query with a large PARTITION BY group, or a query without PARTITION BY, can therefore fail with Resources exhausted rather than using the configured spill storage.

Motivation sequence

The desired behavior is for WindowAggExec to keep the existing partition-at-a-time evaluation model, but use disk spill as a fallback when a complete window partition no longer fits in the memory reservation.

sequenceDiagram
    autonumber
    participant Input as Sorted input stream
    participant Window as WindowAggExec
    participant Memory as MemoryReservation
    participant Spill as SpillManager / DiskManager
    participant Expr as Window expression evaluator
    participant Output as Downstream consumer

    Input->>Window: RecordBatch with ordered window partitions
    loop For each logical window partition
        Window->>Memory: Reserve memory for buffered partition rows
        alt Reservation succeeds
            Window->>Window: Buffer partition rows in memory
        else Reservation fails and disk spill is enabled
            Window->>Spill: Write buffered rows to spill file
            Window->>Memory: Release memory for spilled rows
            Input->>Window: Continue same partition
            Window->>Spill: Append remaining partition rows
        else Reservation fails and spill is unavailable
            Window-->>Output: Return ResourcesExhausted
        end

        Window->>Spill: Read spilled rows, if partition spilled
        Window->>Expr: Evaluate window functions for complete partition
        Expr-->>Window: Window result columns
        Window-->>Output: Emit original columns plus window columns
    end

Loading

Describe the solution you'd like

Add spill support to WindowAggExec:

• Track buffered partition batches with a MemoryReservation.
• Preserve the current sorted-input partition semantics and finish one partition at a time.
• When the active partition cannot grow its reservation and a disk manager is available, write the buffered partition batches to spill files through the existing SpillManager.
• Read spilled partitions back when the partition is ready for window expression evaluation.
• Report spill metrics such as spill count, spilled rows, and spilled bytes.
• Keep the non-spill path unchanged when memory is sufficient.

Describe alternatives you've considered

A more advanced alternative would be implementing streaming evaluation for more window frame/function combinations. That can reduce memory further for specific functions, but it is a larger semantic change and does not cover all window functions. Operator-level spill support is still useful as a general fallback for large partitions.

Additional context

This would make WindowAggExec behave more consistently with DataFusion's other spill-aware physical operators and make memory-limited window queries fail less often when spill storage is configured.

[2010YOUY01]

A more advanced alternative would be implementing streaming evaluation for more window frame/function combinations.

I think this is doable and would be a more efficient, for bounded window frames.

I have a PoC for that, and I can split that into smaller PRs also with a better API design, what's in the PoC was a bit rough. I'm wondering are you interested in reviewing that?

• POC: Window function intra-operator parallelism and vectorization (up to 50× faster) #22356

but it is a larger semantic change and does not cover all window functions.

I'm not sure if there is any efficient spilling algorithm for not bounded windows like ROWS BETWEEN expr1 PRECEDING AND expr2 FOLLOWING? Otherwise in such cases we have to give up I suppose.

[rluvaton]

WindowAggExec can require buffering an entire window partition before it can evaluate some window functions. Today that buffering is memory-only, so queries over large or skewed partitions can exhaust the configured memory pool even when the runtime has a spill-capable disk manager.

Actually the current situation is worse (and I had that in the back of my mind for a long time)

the WindowAggExec buffer the entire input even if every window partition is 1 row, it still buffer all of it, then concat, then evaluate

[wirybeaver]

Calling out an important clarification from the discussion: the current WindowAggExec memory behavior is worse than only "large window partition may OOM".

Current upstream behavior is:

buffer all input -> concat all input -> compute all partitions -> emit once

That means memory usage can scale with the full child input, even if every logical window partition is small.

The proposed direction changes the execution model to:

buffer one active partition -> spill it if needed -> compute completed partition -> emit partition output

So the feature is not just adding a spill file fallback to the existing whole-input buffer. It also makes WindowAggExec process completed partitions incrementally, with spill used when the active partition cannot grow its memory reservation.

[wirybeaver]

One tradeoff worth making explicit for this feature request: the previous whole-input WindowAggExec model may still be the fastest path for small/medium inputs when memory is sufficient, because it concatenates once, evaluates over larger contiguous batches, and emits fewer output batches.

The proposed spill-oriented approach optimizes for a different failure mode: large/skewed inputs and memory-limited execution.

Current upstream model:
  buffer all input -> concat all input -> compute all partitions -> emit once

Spill-oriented model:
  buffer one active partition -> spill it if needed -> compute completed partition -> emit partition output

If we want to preserve the existing fully in-memory fast path, one possible design is to keep the current WindowAggExec behavior and introduce a separate operator such as SpillingWindowAggExec, selected by the planner/config when spill support is desired.

I am also open to exploring whether the spill/streaming work should be integrated with BoundedWindowAggExec, especially for bounded frames as mentioned above. My hesitation is that BoundedWindowAggExec already has a specialized in-memory state/pruning model, so disk-backed state there likely deserves a separate focused design rather than being mixed into the initial spill PR.

[2010YOUY01]

I am also open to exploring whether the spill/streaming work should be integrated with BoundedWindowAggExec, especially for bounded frames as mentioned above. My hesitation is that BoundedWindowAggExec already has a specialized in-memory state/pruning model, so disk-backed state there likely deserves a separate focused design rather than being mixed into the initial spill PR.

Here are some quick ideas. I may not have explained everything clearly yet, but I’ll put together an epic issue for improving window functions to better explain the direction.

I think BoundedWindowAggExec should eventually be deprecated in favor of a new streaming implementation. My concern is that it assumes the input may not be fully ordered by group key and partition key, and that assumption gets in the way of a more efficient implementation.

So my preference would be to move directly toward a better streaming implementation, rather than adding an intermediate spilling-based step.

The workloads that a streaming approach cannot fully solve are:

• a single partition that does not fit in memory
• and, a window frame that moves unpredictably from row to row

Those cases likely need an LRU-like algorithm, but I don’t think that should be the current priority. Since the window operator is still fairly basic at the moment, I think we should make the in-memory cases better first.