[EPIC] Improve window function performance (especially for large windows)

@avantgardnerio

Is your feature request related to a problem or challenge?

This ticket tracks various ideas to make evaluating window functions faster for single, often large logical window partitions. This has come up several times recently with @avantgardnerio @2010YOUY01 and @wirybeaver among others.

The core use case is to make evaluating a query like this fast:

SELECT
    -- moving average over the current row and the 5 previous rows
    AVG(cpu_usage) OVER (ORDER BY time ROWS BETWEEN 5 PRECEDING AND CURRENT ROW) as avg_cpu
FROM metrics;

The key property is that the query has one large logical window partition because there is no PARTITION BY clause in the window specification. Even though each frame contains only the current row and 5 preceding rows, the operator must process one global ordered window partition.

A similar problem can happen when there is a PARTITION BY clause but the logical window partitions are imbalanced due to skew or a small number of partition keys.

Today, each logical window partition is executed in a single DataFusion execution partition, which means:

It is limited to a single core
It is limited to a single machine in distributed environments

Background

DataFusion has support for many window functions (see documentation). Window functions are used like this:

SELECT
    customer, time, cpu_usage,
    -- moving average over the current row and the 5 previous rows
    AVG(cpu_usage) OVER (PARTITION BY customer ORDER BY time ROWS BETWEEN 5 PRECEDING AND CURRENT ROW) as avg_cpu
FROM
    metrics;

Which results in something like:

+----------+---------------------+-----------+---------+
| customer | time                | cpu_usage | avg_cpu |
+----------+---------------------+-----------+---------+
| acme     | 2026-01-01T00:00:00 | 10.0      | 10.0    |
| acme     | 2026-01-01T00:01:00 | 20.0      | 15.0    |
...
| globex   | 2026-01-01T00:03:00 | 45.0      | 30.0    |
| globex   | 2026-01-01T00:04:00 | 55.0      | 35.0    |
+----------+---------------------+-----------+---------+

The plan for such a query looks like the following, and typically keeps all cores fully occupied:

BoundedWindowAggExec(...)
  SortExec(customer, time)
    RepartitionExec(Hash(customer))  <-- divides work among partitions and thus cores
      Scan

Here is an example from the tests:

datafusion/datafusion/sqllogictest/test_files/window_topk_pushdown.slt

Lines 112 to 118 in a0e9887

    
           01)ProjectionExec: expr=[depname@0 as depname, running_total@1 as running_total] 
        
           02)--SortPreservingMergeExec: [depname@0 ASC NULLS LAST, empno@2 ASC NULLS LAST], fetch=3 
        
           03)----ProjectionExec: expr=[depname@0 as depname, sum(Int64(1)) PARTITION BY [employees.depname] ORDER BY [employees.empno ASC NULLS LAST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW@2 as running_total, empno@1 as empno] 
        
           04)------BoundedWindowAggExec: wdw=[sum(Int64(1)) PARTITION BY [employees.depname] ORDER BY [employees.empno ASC NULLS LAST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Field { "sum(Int64(1)) PARTITION BY [employees.depname] ORDER BY [employees.empno ASC NULLS LAST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW": nullable Int64 }, frame: ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW], mode=[Sorted] 
        
           05)--------SortExec: TopK(fetch=3), expr=[depname@0 ASC NULLS LAST, empno@1 ASC NULLS LAST], preserve_partitioning=[true] 
        
           06)----------RepartitionExec: partitioning=Hash([depname@0], 4), input_partitions=1 
        
           07)------------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/testing/data/csv/aggregate_test_100_with_dates.csv]]}, projection=[depname, empno], file_type=csv, has_header=true

However, for the case in question, when we remove the PARTITION BY customer from the OVER clause:

SELECT
    customer, time, cpu_usage,
    -- moving average over the current row and the 5 previous rows
    AVG(cpu_usage) OVER (ORDER BY time ROWS BETWEEN 5 PRECEDING AND CURRENT ROW) as avg_cpu
FROM
    metrics;

The plan looks like this (no RepartitionExec, and it executes in a single core):

BoundedWindowAggExec(...)  <-- single partition, single core
  SortExec(time)
    Scan

Here is an example:

datafusion/datafusion/sqllogictest/test_files/window.slt

Lines 4100 to 4103 in a0e9887

    
           01)ProjectionExec: expr=[sn@0 as sn, ts@1 as ts, currency@2 as currency, amount@3 as amount, sum(table_with_pk.amount) ORDER BY [table_with_pk.sn ASC NULLS LAST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW@4 as sum1] 
        
           02)--BoundedWindowAggExec: wdw=[sum(table_with_pk.amount) ORDER BY [table_with_pk.sn ASC NULLS LAST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Field { "sum(table_with_pk.amount) ORDER BY [table_with_pk.sn ASC NULLS LAST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW": nullable Float64 }, frame: ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW], mode=[Sorted] 
        
           03)----SortExec: expr=[sn@0 ASC NULLS LAST], preserve_partitioning=[false] 
        
           04)------DataSourceExec: partitions=1, partition_sizes=[1]

Table definition

CREATE TABLE metrics (
  time TIMESTAMP,
  customer VARCHAR,
  cpu_usage DOUBLE
);

INSERT INTO metrics VALUES
  (TIMESTAMP '2026-01-01 00:00:00', 'acme', 10.0),
  (TIMESTAMP '2026-01-01 00:01:00', 'acme', 20.0),
  (TIMESTAMP '2026-01-01 00:02:00', 'acme', 30.0),
  (TIMESTAMP '2026-01-01 00:03:00', 'acme', 40.0),
  (TIMESTAMP '2026-01-01 00:04:00', 'acme', 50.0),
  (TIMESTAMP '2026-01-01 00:00:00', 'globex', 15.0),
  (TIMESTAMP '2026-01-01 00:01:00', 'globex', 25.0),
  (TIMESTAMP '2026-01-01 00:02:00', 'globex', 35.0),
  (TIMESTAMP '2026-01-01 00:03:00', 'globex', 45.0),
  (TIMESTAMP '2026-01-01 00:04:00', 'globex', 55.0);

Describe the solution you'd like

DataFusion evaluates window functions more quickly
DataFusion can use multiple cores to evaluate window functions for large logical window partitions
Distributed systems such as Ballista can use multiple machines to compute window function results in parallel
DataFusion can complete window function queries even when the logical window partitions are larger than available memory

Describe alternatives you've considered

There are several related approaches, listed below. They are complementary rather than mutually exclusive.

Potential Areas / follow ons

Improve Window Function Performance in general

Prefix Sums / Prefix Scan

Used for cumulative windows such as ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW.

"Halo" Rows / Parallel Bounded Windows

Used for bounded window frames where each output partition needs some rows from neighboring ranges. Note I don't think "halo rows" is a standard database term; it means "overlapping partitions with replicated boundary rows".

WindowAggExec Memory / Spilling

Avoid WindowAggExec OOMs by processing partitions incrementally and spilling when needed.

Support spilling for WindowAggExec: Support spilling for WindowAggExec #22946
PoC from @wirybeaver: Support spilling for WindowAggExec #22947

Intra-Operator Parallelism

Another approach is to keep the logical window as one partition, but parallelize execution inside the operator.

Paper: Efficient Processing of Window Functions in Analytical SQL Queries, PVLDB 2015
General issue from @2010YOUY01: Improve internal worker parallelism support #23174
Window-specific issue from @2010YOUY01: Optimize window function with intra-operator parallelism and vectorization #22355
PoC from @2010YOUY01: POC: Window function intra-operator parallelism and vectorization (up to 50× faster) #22356

Adaptive Query Execution / Runtime-Informed Planning

Relevant if DataFusion wants a general framework for runtime stats, dynamic split points, repartition choices, skew handling, and similar optimizations.

AQE issue from @avantgardnerio: Add AQE to DataFusion #23194
AQE-lite PoC from @avantgardnerio: [PoC/Proposal] AQE-lite: change plan properties at runtime based on stats from pipeline breakers #23167

Related Sort / Merge Parallelism

Relevant because single-partition windows often sit downstream of sort / merge bottlenecks.

feat: parallel sort-preserving merge (PSRS) for >2x merge speedup (demo) #23124 from @Dandandan

Related features

Windows where the size is an expression: Support expression based window function (value based RANGE) #15714
Refactor the code from @2010YOUY01 : Refactor window execution to enable future optimizations #23273

Here is some additional context:

I think we don't have to rely on BoundedWindowAggExec in the future.

The reason is that it does not assume the input is fully sorted by the window partition/order keys, so it performs internal sorting. This adds both implementation complexity and execution overhead.

Here is my guess on why such design decision was made before: because the original window aggregation implementation was fairly naive: it buffered all input before continuing execution. As a result, a downstream implementation was upstreamed to support streaming and out-of-order input, where assuming fully sorted input was not possible for their use cases. Given that goal, the additional complexity made sense because it improved memory efficiency.

However, DataFusion's primary use case is still batched analytics. Under that assumption, we can require the window operator to consume input that has already been repartitioned and sorted by the window partition/order keys. This makes the implementation both simpler and faster, and we can achieve the same memory efficiency given bounded window frames.

So if we continue investing in vectorization and better parallelization, we could potentially evolve WindowAggExec directly. Its current implementation is relatively minimal, making it easier to extend, and eventually we could replace BoundedWindowAggExec.

@akurmustafa

Here is my guess on why such design decision was made before: because the original window aggregation implementation was fairly naive: it buffered all input before continuing execution. As a result, a downstream implementation was upstreamed to support streaming and out-of-order input, where assuming fully sorted input was not possible for their use cases. Given that goal, the additional complexity made sense because it improved memory efficiency.

Yes I think that is mostly accurate (I think @akurmustafa / @mustafasrepo did a lot of the initial work)

I think BoundedWindowExec was designed to support streaming sql engines (e.g. where the ExecutionPlan was started and the incrementally produced results as data was incrementally pushed through it)

However, DataFusion's primary use case is still batched analytics. Under that assumption, we can require the window operator to consume input that has already been repartitioned and sorted by the window partition/order keys. This makes the implementation both simpler and faster, and we can achieve the same memory efficiency given bounded window frames.

I agree it makes the implementation simpler -- I suspect a fully specialized operator could be made faster than spilling / resorting, but I also think we could work on that after we have exhausted all options for making BoundedWindowAggExec faster

So if we continue investing in vectorization and better parallelization, we could potentially evolve WindowAggExec directly. Its current implementation is relatively minimal, making it easier to extend, and eventually we could replace BoundedWindowAggExec.

In general I think focusing efforts on a single execution operator is probably better than trying to optimize both

@alamb

Yes, as @alamb and @2010YOUY01 said the original idea was to have a support for a window operator where given the input is ordered by either or both by PARTITION BY and ORDER BY clauses; Window function didn't buffer all of the batches at its input to save memory and support streaming.

For a window function with following clause PARTITION BY <expr1> ORDER BY <expr2> operator supports input with following orderings:

case1: <expr1>
case2: <exp2>
case3: <exp1> + <exp2>

I think, most of the complexity in the implementation comes from having the support for different use cases. However, I think case 1 and case 2 are mostly for streaming cases. I don't see a benefit for keeping them for non-streaming cases. If we assume input ordering will always be satisfied as in the case 3 (both PARTITION BY AND ORDER BY expression clauses), I think implementation can be simplified and at the end DataFusion can have single Window operator which doesn't expect all input to be buffered.

As far as I remember, WindowAggExec works assumes always case 3 is valid and buffers all of the data at its input.

I agree to @alamb and @2010YOUY01 that focusing on WindowAggExec for improvement is better course and in the future maybe in the planning phase, we can only generate plans that contain WindowAggExec then discontinue the BoundedWindowAggExec.

I agree to @alamb and @2010YOUY01 that focusing on WindowAggExec for improvement is better course and in the future maybe in the planning phase, we can only generate plans that contain WindowAggExec then discontinue the BoundedWindowAggExec.

if we did that we will need to handle lots of other things in the WindowAggExec (like spilling 🤔 ) that the current sort based plans get for free

we will need to handle lots of other things in the WindowAggExec (like spilling 🤔 ) that the current sort based plans get for free

I would press the other way: drop WAG in favor of BWAG (with upper bound being infinity, no partial ordering - full only, ram-only OOM if need be)

AQE-lite's StageBoundaryBuffer is designed to materialize at every pipeline-breaker boundary (spill-or-stream TODO at stage_boundary_buffer.rs:74-87). If BWAG's frame buffer is backed by the StageBoundaryBuffer, BWAG inherits spill behavior for free - small frames stay RAM-resident, large/unbounded frames spill through the buffer.
We could use halo-rows / prefix-scan to create many partitions where there were once one
we can execute each partition sequentially if necessary, based on memory pressure, to materialize all BWAG results onto disk, one partition at a time to avoid OOMs
this keeps spilling as a separate concern, much the way sorting is a separate concern now

If we want, we could develop this functionality within WAG, then deprecate BWAG, but BWAG works now and given the above I hope most window operations would become bounded. I would also argue the degenerate cases that need to hold all rows (potentially spilling to disk) are uncommon and not particularly useful.

Here is my take from my experience after implementing couple of Window Execs that are way more optimized but not for the general case due to the Window API limitation

The requirement for window expressions to get the entire partition in order to output a value is a major thing that prevent from optimizing window expression as it requires both

having all the partition in hand even when the expression might not need that
it requires having it in memory - which takes a lot of memory and you can't do any spilling

It should be similar to GroupsAccumulator that hold state and handle that.

It should be similar to GroupsAccumulator that hold state and handle that.

As in are you are suggesting we need a new API for function implementations (to augment or replace PartitionEvaluator ?

I spent some more time cleaning up this ticket / honing the description to try and capture various approaches

In my opinion if we want to improve window function performance, the most obvious things would be to focus on raw evaluation speed in the core:

This doesn't really address parallelizing single window functions within a window (though SegmentTrees can do so)

I also think this would likely be used widely and should be considered for inclusion in the core:

API: Partitioning::DynamicRange for runtime-discovered split points #23093

@avantgardnerio

@avantgardnerio noted he has a proposed API to help support this work

Expose accumulator state to allow prefix scanning #24035

Navigation Menu

Repository navigation

[EPIC] Improve window function performance (especially for large windows) #23197

Is your feature request related to a problem or challenge?

Background

Describe the solution you'd like

Describe alternatives you've considered

Potential Areas / follow ons

Improve Window Function Performance in general

Prefix Sums / Prefix Scan

"Halo" Rows / Parallel Bounded Windows

WindowAggExec Memory / Spilling

Intra-Operator Parallelism

Adaptive Query Execution / Runtime-Informed Planning

Related Sort / Merge Parallelism

Related features

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

	01)ProjectionExec: expr=[depname@0 as depname, running_total@1 as running_total]
	02)--SortPreservingMergeExec: [depname@0 ASC NULLS LAST, empno@2 ASC NULLS LAST], fetch=3
	03)----ProjectionExec: expr=[depname@0 as depname, sum(Int64(1)) PARTITION BY [employees.depname] ORDER BY [employees.empno ASC NULLS LAST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW@2 as running_total, empno@1 as empno]
	04)------BoundedWindowAggExec: wdw=[sum(Int64(1)) PARTITION BY [employees.depname] ORDER BY [employees.empno ASC NULLS LAST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Field { "sum(Int64(1)) PARTITION BY [employees.depname] ORDER BY [employees.empno ASC NULLS LAST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW": nullable Int64 }, frame: ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW], mode=[Sorted]
	05)--------SortExec: TopK(fetch=3), expr=[depname@0 ASC NULLS LAST, empno@1 ASC NULLS LAST], preserve_partitioning=[true]
	06)----------RepartitionExec: partitioning=Hash([depname@0], 4), input_partitions=1
	07)------------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/testing/data/csv/aggregate_test_100_with_dates.csv]]}, projection=[depname, empno], file_type=csv, has_header=true

	01)ProjectionExec: expr=[sn@0 as sn, ts@1 as ts, currency@2 as currency, amount@3 as amount, sum(table_with_pk.amount) ORDER BY [table_with_pk.sn ASC NULLS LAST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW@4 as sum1]
	02)--BoundedWindowAggExec: wdw=[sum(table_with_pk.amount) ORDER BY [table_with_pk.sn ASC NULLS LAST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Field { "sum(table_with_pk.amount) ORDER BY [table_with_pk.sn ASC NULLS LAST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW": nullable Float64 }, frame: ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW], mode=[Sorted]
	03)----SortExec: expr=[sn@0 ASC NULLS LAST], preserve_partitioning=[false]
	04)------DataSourceExec: partitions=1, partition_sizes=[1]

[EPIC] Improve window function performance (especially for large windows) #23197

Description

Is your feature request related to a problem or challenge?

Background

Describe the solution you'd like

Describe alternatives you've considered

Potential Areas / follow ons

Improve Window Function Performance in general

Prefix Sums / Prefix Scan

"Halo" Rows / Parallel Bounded Windows

WindowAggExec Memory / Spilling

Intra-Operator Parallelism

Adaptive Query Execution / Runtime-Informed Planning

Related Sort / Merge Parallelism

Related features

Activity

2010YOUY01 commented on Jun 26, 2026

alamb commented on Jun 29, 2026

alamb commented on Jun 29, 2026

akurmustafa commented on Jun 29, 2026

alamb commented on Jun 29, 2026

avantgardnerio commented on Jun 29, 2026

rluvaton commented on Jul 6, 2026

alamb commented on Jul 7, 2026

alamb commented on Aug 1, 2026

alamb commented on Aug 5, 2026

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions