[EPIC] Improve window function performance for large windows

[alamb]

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:

1. It is limited to a single core
2. 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 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.

Additional context

Prefix Sums / Prefix Scan

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

• Paper: Prefix Sums and Their Applications
• PoC from @avantgardnerio: [PoC] Parallel prefix scan for cumulative-aggregate window functions coralogix/arrow-datafusion#426

"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".

• Main PoC from @avantgardnerio: Parallel bounded RANGE-frame window functions without PARTITION BY (draft) #23026
• API: runtime partition extrema for range-aware operators #23089
• feat: PartitionExtrema + SortExec observer + BoundedWindowAggExec passthrough #23090
• API: Partitioning::DynamicRange for runtime-discovered split points #23093
• feat(physical-expr): add Partitioning::DynamicRange variant #23094

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

Window Frame Evaluation / Vectorization

Improve CPU and memory efficiency for window frame boundary calculations.

• Vectorize window functions (GroupsAccumulator-style treatment): Vectorize window functions (GroupsAccumulator-style) #15607
• Improved performance of RANGE preceding window functions #4904

Intra-Operator Parallelism

Alternative/complementary approach: 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

[2010YOUY01]

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.