Window frames over preserved sort order: how DataFusion evaluates PARTITION BY and ORDER BY

TL;DR: A window function is a sorted, grouped fold. DataFusion makes the operator declare what it needs — KeyPartitioned(partition_by) plus an ordering of (partition_by, order_by) — and the planner inserts the repartition and the sort. Then the frame is evaluated with one of two accumulators: a plain one that only adds rows (for UNBOUNDED PRECEDING frames) or a sliding one that adds rows entering the frame and retracts rows leaving it.

The problem: a window function is three things at once

SUM(x) OVER (PARTITION BY dept ORDER BY ts ROWS BETWEEN 2 PRECEDING AND 1 FOLLOWING)

To evaluate this you need:

Grouping — all rows with the same dept must be together.
Ordering — within a group, rows must be in ts order, because the frame is defined relative to “the current row”.
Framing — for each row, a contiguous range of rows (the frame) whose values are aggregated.

Note that (1) and (2) are in tension. Grouping wants rows with equal dept adjacent; ordering wants ts sorted. The resolution is the standard one: sort by (dept, ts). Then equal-dept rows are contiguous and ordered by ts within each group. Partition boundaries are simply the places where dept changes.

The engineering question is: who arranges that? And the answer in DataFusion is that the operator asks, and the planner arranges.

The declaration

rust
// datafusion/physical-plan/src/windows/window_agg_exec.rs:233-263
fn maintains_input_order(&self) -> Vec<bool> {
    vec![true]
}

fn required_input_ordering(&self) -> Vec<Option<OrderingRequirements>> {
    let partition_bys = self.window_expr()[0].partition_by();
    let order_keys = self.window_expr()[0].order_by();
    if self.ordered_partition_by_indices.len() < partition_bys.len() {
        vec![calc_requirements(partition_bys, order_keys)]
    } else {
        let partition_bys = self
            .ordered_partition_by_indices
            .iter()
            .map(|idx| &partition_bys[*idx]);
        vec![calc_requirements(partition_bys, order_keys)]
    }
}

fn input_distribution_requirements(&self) -> InputDistributionRequirements {
    if self.partition_keys().is_empty() {
        InputDistributionRequirements::new(vec![Distribution::SinglePartition])
    } else {
        InputDistributionRequirements::new(vec![Distribution::KeyPartitioned(
            self.partition_keys(),
        )])
    }
}

Read that as: “give me rows grouped by partition_by (or all together if there is no PARTITION BY), and within each group in order_by order.”

The ordering requirement is expressed as alternatives:

rust
// datafusion/physical-plan/src/windows/mod.rs:309-342
pub(crate) fn calc_requirements<
    T: Borrow<Arc<dyn PhysicalExpr>>,
    S: Borrow<PhysicalSortExpr>,
>(
    partition_by_exprs: impl IntoIterator<Item = T>,
    orderby_sort_exprs: impl IntoIterator<Item = S>,
) -> Option<OrderingRequirements> {
    let mut sort_reqs_with_partition = partition_by_exprs
        .into_iter()
        .map(|partition_by| {
            PhysicalSortRequirement::new(Arc::clone(partition_by.borrow()), None)
        })
        .collect::<Vec<_>>();
    let mut sort_reqs = vec![];
    for element in orderby_sort_exprs.into_iter() {
        let PhysicalSortExpr { expr, options } = element.borrow();
        let sort_req = PhysicalSortRequirement::new(Arc::clone(expr), Some(*options));
        if !sort_reqs_with_partition.iter().any(|e| e.expr.eq(expr)) {
            sort_reqs_with_partition.push(sort_req.clone());
        }
        if !sort_reqs.iter().any(|e: &PhysicalSortRequirement| e.expr.eq(expr)) {
            sort_reqs.push(sort_req);
        }
    }

    let mut alternatives = vec![];
    alternatives.extend(LexRequirement::new(sort_reqs_with_partition));
    alternatives.extend(LexRequirement::new(sort_reqs));

    OrderingRequirements::new_alternatives(alternatives, false)
}

Two alternatives: sort by (partition_by, order_by), or sort by order_by alone. The second alternative is the interesting one — it says “if the input is already sorted by order_by only, that’s also acceptable”, and the operator can then find the partition boundaries itself by scanning for changes in the partition_by values. That is exactly what a BoundedWindowAggExec does.

The plan that results

Because the window operator declares both a distribution and an ordering requirement, EnsureRequirements builds this:

text

SortExec: partition_by ASC, order_by ASC          ← ordering requirement
  RepartitionExec: Hash(partition_by)             ← distribution requirement
    Scan

Or, with no PARTITION BY at all, the distribution requirement degrades to SinglePartition and everything is coalesced into one stream.

This is the “PartitionBy / SortPreservingMerge trick” in its simplest form: the repartition makes the groups local, the sort makes the order local, and the window operator then only has to walk contiguous runs.

Avoiding a redundant sort

The window operator doesn’t blindly demand (partition_by, order_by). It inspects the input’s existing ordering and finds the longest permutation of the PARTITION BY columns that is already a prefix of it:

rust
// datafusion/physical-plan/src/windows/mod.rs:344-357
/// This function calculates the indices such that when partition by expressions reordered with the indices
/// resulting expressions define a preset for existing ordering.
/// For instance, if input is ordered by a, b, c and PARTITION BY b, a is used,
/// this vector will be [1, 0]. It means that when we iterate b, a columns with the order [1, 0]
/// resulting vector (a, b) is a preset of the existing ordering (a, b, c).
pub fn get_ordered_partition_by_indices(
    partition_by_exprs: &[Arc<dyn PhysicalExpr>],
    input: &Arc<dyn ExecutionPlan>,
) -> Result<Vec<usize>> {
    let (_, indices) = input
        .equivalence_properties()
        .find_longest_permutation(partition_by_exprs)?;
    Ok(indices)
}

So if the input is already sorted by (a, b, c) and the window says PARTITION BY b, a, the operator discovers that (a, b) is a prefix of the existing ordering, and the required ordering becomes (a, b, c) — not (b, a, ...). The sort is satisfied by the existing order and the WindowAggExec can run with zero sorting.

The ordered_partition_by_indices are then used consistently: in the declared ordering requirement, in partition_by_sort_keys() (the sort expressions used to detect partition boundaries), and in compute_aggregates(), which evaluates only those partition-by columns when scanning for boundaries.

The frame representation

The frame is two bounds plus a unit:

rust
// datafusion/expr/src/window_frame.rs:38-45
#[derive(Clone, PartialEq, Eq, PartialOrd, Hash)]
pub struct WindowFrame {
    /// Frame type - either `ROWS`, `RANGE` or `GROUPS`
    pub units: WindowFrameUnits,
    /// Starting frame boundary
    pub start_bound: WindowFrameBound,
    /// Ending frame boundary
    pub end_bound: WindowFrameBound,

and the bound itself:

rust
// datafusion/expr/src/window_frame.rs:312-351
/// There are five ways to describe starting and ending frame boundaries:
///
/// 1. UNBOUNDED PRECEDING
/// 2. `<expr>` PRECEDING
/// 3. CURRENT ROW
/// 4. `<expr>` FOLLOWING
/// 5. UNBOUNDED FOLLOWING
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Hash)]
pub enum WindowFrameBound {
    /// 1. UNBOUNDED PRECEDING
    ///    The frame boundary is the first row in the partition.
    ///
    /// 2. `<expr>` PRECEDING
    ///    `<expr>` must be a non-negative constant numeric expression. The boundary is a row that
    ///    is `<expr>` "units" prior to the current row.
    Preceding(ScalarValue),
    /// 3. The current row.
    ///
    /// For RANGE and GROUPS frame types, peers of the current row are also
    /// included in the frame, unless specifically excluded by the EXCLUDE clause.
    /// This is true regardless of whether CURRENT ROW is used as the starting or ending frame
    /// boundary.
    CurrentRow,
    /// 4. This is the same as "`<expr>` PRECEDING" except that the boundary is `<expr>` units after the
    ///    current rather than before the current row.
    ///
    /// 5. UNBOUNDED FOLLOWING
    ///    The frame boundary is the last row in the partition.
    Following(ScalarValue),
}

impl WindowFrameBound {
    pub fn is_unbounded(&self) -> bool {
        match self {
            WindowFrameBound::Preceding(elem) => elem.is_null(),
            WindowFrameBound::CurrentRow => false,
            WindowFrameBound::Following(elem) => elem.is_null(),
        }
    }
}

The encoding trick here is that “UNBOUNDED” is a NULL ScalarValue. Preceding(ScalarValue::UInt64(None)) means UNBOUNDED PRECEDING; Preceding(ScalarValue::UInt64(Some(5))) means 5 PRECEDING. One enum variant covers both the unbounded and the bounded case, and is_unbounded() is a null check. This is why the frame arithmetic below can be written as a single match over three cases.

For ROWS frames, the offset is always a UInt64, so the arithmetic is pure integer work:

rust
// datafusion/expr/src/window_state.rs:243-295
fn calculate_range_rows(
    window_frame: &Arc<WindowFrame>,
    length: usize,
    idx: usize,
) -> Result<Range<usize>> {
    let start = match window_frame.start_bound {
        // UNBOUNDED PRECEDING
        WindowFrameBound::Preceding(ScalarValue::UInt64(None)) => 0,
        WindowFrameBound::Preceding(ScalarValue::UInt64(Some(n))) => {
            idx.saturating_sub(n as usize)
        }
        WindowFrameBound::CurrentRow => idx,
        // UNBOUNDED FOLLOWING
        WindowFrameBound::Following(ScalarValue::UInt64(None)) => {
            return internal_err!(
                "Frame start cannot be UNBOUNDED FOLLOWING '{window_frame:?}'"
            );
        }
        WindowFrameBound::Following(ScalarValue::UInt64(Some(n))) => {
            std::cmp::min(idx + n as usize, length)
        }
        ...
    };
    let end = match window_frame.end_bound {
        ...
        WindowFrameBound::CurrentRow => idx + 1,
        // UNBOUNDED FOLLOWING
        WindowFrameBound::Following(ScalarValue::UInt64(None)) => length,
        WindowFrameBound::Following(ScalarValue::UInt64(Some(n))) => {
            std::cmp::min(idx + n as usize + 1, length)
        }
        ...
    };
    Ok(Range { start, end })
}

Note that the ROWS path is stateless while RANGE and GROUPS are stateful:

rust
// datafusion/expr/src/window_state.rs:170-189
/// This object stores the window frame state for use in incremental calculations.
#[derive(Debug, Clone)]
pub enum WindowFrameContext {
    /// ROWS frames are inherently stateless.
    Rows(Arc<WindowFrame>),
    /// RANGE frames are stateful, they store indices specifying where the
    /// previous search left off. This amortizes the overall cost to O(n)
    /// where n denotes the row count.
    Range {
        window_frame: Arc<WindowFrame>,
        state: WindowFrameStateRange,
    },
    /// GROUPS frames are stateful, they store group boundaries and indices
    /// specifying where the previous search left off. This amortizes the
    /// overall cost to O(n) where n denotes the row count.
    Groups { ... },
}

A RANGE BETWEEN INTERVAL '1' HOUR PRECEDING AND CURRENT ROW frame has to binary-search the order-by column for every row; remembering where the last search ended turns that into an amortized linear scan.

The default frames

The defaults are where “no ORDER BY” gets decided:

rust
// datafusion/expr/src/window_frame.rs:145-176
impl WindowFrame {
    /// Creates a new, default window frame (with the meaning of default
    /// depending on whether the frame contains an `ORDER BY` clause and this
    /// ordering is strict (i.e. no ties).
    pub fn new(order_by: Option<bool>) -> Self {
        if let Some(strict) = order_by {
            // This window frame covers the table (or partition if `PARTITION BY`
            // is used) from beginning to the `CURRENT ROW` (with same rank). It
            // is used when the `OVER` clause contains an `ORDER BY` clause but
            // no frame.
            Self {
                units: if strict {
                    WindowFrameUnits::Rows
                } else {
                    WindowFrameUnits::Range
                },
                start_bound: WindowFrameBound::Preceding(ScalarValue::UInt64(None)),
                end_bound: WindowFrameBound::CurrentRow,
                causal: strict,
            }
        } else {
            // This window frame covers the whole table (or partition if `PARTITION BY`
            // is used). It is used when the `OVER` clause does not contain an
            // `ORDER BY` clause and there is no frame.
            Self {
                units: WindowFrameUnits::Rows,
                start_bound: WindowFrameBound::Preceding(ScalarValue::UInt64(None)),
                end_bound: WindowFrameBound::Following(ScalarValue::UInt64(None)),
                causal: false,
            }
        }
    }
}

So:

`OVER` clause	default frame	units
`PARTITION BY d`	`UNBOUNDED PRECEDING → UNBOUNDED FOLLOWING` (whole partition)	ROWS
`PARTITION BY d ORDER BY t` (strict)	`UNBOUNDED PRECEDING → CURRENT ROW`	ROWS
`PARTITION BY d ORDER BY t` (ties)	`UNBOUNDED PRECEDING → CURRENT ROW`	RANGE

The third row is the subtle one: with ties in the ORDER BY column, the default frame is RANGE, so that “current row” means “current row and its peers” — the SQL standard’s RANGE ... CURRENT ROW semantics. With a strict ordering (no ties possible), ROWS and RANGE agree, so ROWS is used because it’s cheaper.

Also note causal. The frame is causal if computing row i doesn’t depend on row i+1:

rust
// datafusion/expr/src/window_frame.rs:208-252
pub fn new_bounds(
    units: WindowFrameUnits,
    start_bound: WindowFrameBound,
    end_bound: WindowFrameBound,
) -> Self {
    let causal = match units {
        WindowFrameUnits::Rows => match &end_bound {
            WindowFrameBound::Following(value) => {
                if value.is_null() {
                    // Unbounded following
                    false
                } else { ... }
            }
            _ => true,
        },
        ...
    };
    Self { units, start_bound, end_bound, causal }
}

Causality is what makes streaming window evaluation possible: if the frame never looks forward, row i can be emitted as soon as row i is seen.

Two accumulators: growing frames vs sliding frames

The choice of accumulator is made at expression-construction time, from the frame’s shape:

rust
// datafusion/physical-plan/src/windows/mod.rs:139-167
/// Creates an appropriate [`WindowExpr`] based on the window frame and
fn window_expr_from_aggregate_expr(
    partition_by: &[Arc<dyn PhysicalExpr>],
    order_by: &[PhysicalSortExpr],
    window_frame: Arc<WindowFrame>,
    aggregate: Arc<AggregateFunctionExpr>,
    filter: Option<Arc<dyn PhysicalExpr>>,
) -> Arc<dyn WindowExpr> {
    // Is there a potentially unlimited sized window frame?
    let unbounded_window = window_frame.is_ever_expanding();

    if !unbounded_window {
        Arc::new(SlidingAggregateWindowExpr::new(
            aggregate, partition_by, order_by, window_frame, filter,
        ))
    } else {
        Arc::new(PlainAggregateWindowExpr::new(
            aggregate, partition_by, order_by, window_frame, filter,
        ))
    }
}

with

rust
// datafusion/expr/src/window_frame.rs:304-309
/// Is the window frame ever-expanding (it always grows in the superset sense).
/// Useful when understanding if set-monotonicity properties of functions can
/// be exploited.
pub fn is_ever_expanding(&self) -> bool {
    self.start_bound.is_unbounded()
}

Ever-expanding means the frame’s start is UNBOUNDED PRECEDING: rows only ever get added to the frame, never removed. SUM(x) OVER (ORDER BY t) is the canonical example. The accumulator is a plain update_batch-only accumulator — no retraction needed.

Sliding frames (ROWS BETWEEN 2 PRECEDING AND 1 FOLLOWING, or anything with a bounded start) need rows to enter and leave. That’s where the sliding accumulator earns its name:

rust
// datafusion/physical-expr/src/window/sliding_aggregate.rs:180-235
/// Given current range and the last range, calculates the accumulator
/// result for the range of interest.
fn get_aggregate_result_inside_range(
    &self,
    last_range: &Range<usize>,
    cur_range: &Range<usize>,
    value_slice: &[ArrayRef],
    accumulator: &mut Box<dyn Accumulator>,
    filter_mask: Option<&BooleanArray>,
) -> Result<ScalarValue> {
    if cur_range.start == cur_range.end {
        // Keep the accumulator synchronized with `last_range`. RANGE frames
        // can become empty between two non-empty frames when the ORDER BY
        // values contain gaps.
        let retract_bound = last_range.end - last_range.start;
        if retract_bound > 0 {
            ... accumulator.retract_batch(&retract)? ...
        }
        self.aggregate
            .default_value(self.aggregate.field().data_type())
    } else {
        // Accumulate any new rows that have entered the window:
        let update_bound = cur_range.end - last_range.end;
        if update_bound > 0 {
            ... accumulator.update_batch(&update)? ...
        }

        // Remove rows that have now left the window:
        let retract_bound = cur_range.start - last_range.start;
        if retract_bound > 0 {
            ... accumulator.retract_batch(&retract)? ...
        }
        ...
    }
}

This is the classic sliding-window trick: instead of re-aggregating the whole frame for every row (which is O(N × W)), you keep the accumulator in sync with the previous frame and only apply the delta:

text

frame(i-1) = [s .. e)
frame(i)   = [s' .. e')
delta      = retract [s .. s'), update [e .. e')

Cost per row: O(1) amortized, versus O(W) for a naive re-aggregation. This requires the aggregate to support retract_batch — hence create_sliding_accumulator vs create_accumulator.

The plain (ever-expanding) accumulator has a corresponding memory trick — it can discard old rows because they’ll never be needed again:

rust
// datafusion/physical-expr/src/window/aggregate.rs:196-220
fn evaluate_stateful(
    &self,
    partition_batches: &PartitionBatches,
    window_agg_state: &mut PartitionWindowAggStates,
    eval_ctx: &WindowEvalContext<'_>,
) -> Result<()> {
    self.aggregate_evaluate_stateful(partition_batches, window_agg_state, eval_ctx)?;

    // Update window frame range for each partition. As we know that
    // non-sliding aggregations will never call `retract_batch`, this value
    // can safely increase, and we can remove "old" parts of the state.
    // This enables us to run queries involving UNBOUNDED PRECEDING frames
    // using bounded memory for suitable aggregations.
    for partition_row in partition_batches.keys() {
        ...
        if self.window_frame.start_bound.is_unbounded() {
            state.window_frame_range.start =
                state.window_frame_range.end.saturating_sub(1);
        }
    }
    Ok(())
}

So SUM(x) OVER (ORDER BY t) — which naively needs the whole prefix summed for every row — can run in bounded memory, because only the last row’s prefix is ever needed again.

The “no ORDER BY” special case

OVER (PARTITION BY d) with no ORDER BY gives the whole-partition frame. The PlainAggregateWindowExpr notices that this frame is the same for every row in the partition and skips all per-row work:

rust
// datafusion/physical-expr/src/window/aggregate.rs:146-168
// Returns true if every row in the partition has the same window frame. This allows
// for preventing bound + function calculation for every row due to the values being the
// same.
//
// This occurs when both bounds fall under either condition below:
//  1. Bound is unbounded (`Preceding` or `Following`)
//  2. Bound is `CurrentRow` while using `Range` units with no order by clause
//  This results in an invalid range specification. Following PostgreSQL's convention,
//  We interpret this as the entire partition being used for the current window frame.
fn is_window_constant_in_partition(
    order_by: &[PhysicalSortExpr],
    window_frame: &WindowFrame,
) -> bool {
    let is_constant_bound = |bound: &WindowFrameBound| match bound {
        WindowFrameBound::CurrentRow => {
            window_frame.units == WindowFrameUnits::Range && order_by.is_empty()
        }
        _ => bound.is_unbounded(),
    };

    is_constant_bound(&window_frame.start_bound)
        && is_constant_bound(&window_frame.end_bound)
}

and the evaluation path checks it first:

rust
// datafusion/physical-expr/src/window/window_expr.rs:333-348
if self.is_constant_in_partition() {
    if not_end {
        let field = self.field()?;
        let out_type = field.data_type();
        return Ok(new_empty_array(out_type));
    }
    let values = if let Some(mask) = filter_mask { ... } else { values };
    accumulator.update_batch(&values)?;
    let value = accumulator.evaluate()?;
    return value.to_array_of_size(record_batch.num_rows());
}

So SUM(x) OVER (PARTITION BY d) is computed once per partition and then broadcast to every row of the partition. That is the difference between O(N) aggregate calls and O(1) per partition.

Two executors: `WindowAggExec` and `BoundedWindowAggExec`

There are two physical window operators, and the difference is memory behaviour.

WindowAggExec is the simple one — and it buffers everything:

rust
// datafusion/physical-plan/src/windows/window_agg_exec.rs:541-582
fn compute_aggregates(&self) -> Result<Option<RecordBatch>> {
    // record compute time on drop
    let _timer = self.baseline_metrics.elapsed_compute().timer();

    let batch = concat_batches(&self.input.schema(), &self.batches)?;
    if batch.num_rows() == 0 {
        return Ok(None);
    }

    let partition_by_sort_keys = self
        .ordered_partition_by_indices
        .iter()
        .map(|idx| self.partition_by_sort_keys[*idx].evaluate_to_sort_column(&batch))
        .collect::<Result<Vec<_>>>()?;
    let partition_points =
        evaluate_partition_ranges(batch.num_rows(), &partition_by_sort_keys)?;

    let mut partition_results = Vec::with_capacity(partition_points.len());
    // Calculate window cols
    for partition_point in partition_points {
        let length = partition_point.end - partition_point.start;
        partition_results.push(compute_window_aggregates(
            &self.window_expr,
            &batch.slice(partition_point.start, length),
        )?)
    }
    let columns = transpose(partition_results)
        .iter()
        .map(|elems| concat(&elems.iter().map(|x| x.as_ref()).collect::<Vec<_>>()))
        .collect::<Result<Vec<ArrayRef>, ArrowError>>()?;
    ...
}

and the stream drains the whole input before computing anything:

rust
// datafusion/physical-plan/src/windows/window_agg_exec.rs:607-628
loop {
    return Poll::Ready(Some(match ready!(self.input.poll_next_unpin(cx)) {
        Some(Ok(batch)) => {
            self.batches.push(batch);
            continue;
        }
        Some(Err(e)) => Err(e),
        None => {
            // Release the input pipeline's resources before computing
            // the final aggregates.
            let input_schema = self.input.schema();
            self.input = Box::pin(EmptyRecordBatchStream::new(input_schema));
            let Some(result) = self.compute_aggregates()? else {
                return Poll::Ready(None);
            };
            self.finished = true;
            ...
            Ok(result)
        }
    }));
}

So WindowAggExec is O(partition) memory. It relies on the repartition to make partitions small, and on the sort to make partition boundaries detectable by simple comparison.

BoundedWindowAggExec is the streaming one. It doesn’t require the input to be sorted by PARTITION BY at all — it hashes the partition keys and groups rows itself:

rust
// datafusion/physical-plan/src/windows/bounded_window_agg_exec.rs:920-955
fn compute_partition_permutation(
    &mut self,
    columns: &[ArrayRef],
    batch: &RecordBatch,
) -> Result<(Vec<PartitionKey>, Vec<u32>, Vec<usize>)> {
    let num_rows = batch.num_rows();
    self.batch_hashes.clear();
    self.batch_hashes.resize(num_rows, 0);
    create_hashes(columns, &self.random_state, &mut self.batch_hashes)?;
    self.input_buffer_hashes.extend(&self.batch_hashes);
    self.row_map_batch.clear();
    let mut keys: Vec<PartitionKey> = vec![];
    // Partition id of each row, in row order:
    let mut row_partition_ids: Vec<usize> = Vec::with_capacity(num_rows);
    for (&hash, row_idx) in self.batch_hashes.iter().zip(0u32..) {
        let entry = self.row_map_batch.find_mut(hash, |(_, group_idx)| {
            let row = get_row_at_idx(columns, row_idx as usize).unwrap();
            // Handle hash collisions with an equality check:
            row == keys[*group_idx]
        });
        ...
    }
    ...
}

Note the hash-collision handling: the hash is only used to find a candidate group; the actual key values are compared for equality. That’s the standard open-addressing-with-equality-check pattern, and it’s why the grouping is correct even with adversarial hash collisions. The rows of each partition are then scattered into contiguous runs (a counting sort by partition id), and the per-partition state is updated incrementally as batches arrive.

The choice between the two executors is made by get_best_fitting_window:

rust
// datafusion/physical-plan/src/windows/mod.rs:657-681
// If all window expressions can run with bounded memory, choose the
// bounded window variant:
if window_expr.iter().all(|e| e.uses_bounded_memory()) {
    Ok(Some(Arc::new(
        BoundedWindowAggExec::try_new(
            window_expr,
            Arc::clone(input),
            input_order_mode,
            !physical_partition_keys.is_empty(),
        )?
        .with_state_observer(state_observer)?,
    ) as _))
} else if input_order_mode != InputOrderMode::Sorted {
    // For `WindowAggExec` to work correctly PARTITION BY columns should be sorted.
    // Hence, if `input_order_mode` is not `Sorted` we should convert
    // input ordering such that it can work with `Sorted` (add `SortExec`).
    // Effectively `WindowAggExec` works only in `Sorted` mode.
    Ok(None)
} else {
    Ok(Some(Arc::new(WindowAggExec::try_new(
        window_expr,
        Arc::clone(input),
        !physical_partition_keys.is_empty(),
    )?) as _))
}

and uses_bounded_memory is decided by the frame’s end bound:

rust
// datafusion/physical-expr/src/window/aggregate.rs:244-246
fn uses_bounded_memory(&self) -> bool {
    !self.window_frame.end_bound.is_unbounded()
}

Here’s the resulting taxonomy, and it’s a nice one:

frame	end bound	executor	memory
`PARTITION BY d` (no ORDER BY)	UNBOUNDED FOLLOWING	`WindowAggExec`	whole partition
`ORDER BY t` (default)	CURRENT ROW	`BoundedWindowAggExec` (if aggregate supports it)	bounded
`ROWS BETWEEN 2 PRECEDING AND 1 FOLLOWING`	FOLLOWING(1)	`BoundedWindowAggExec`	bounded

The “no ORDER BY” case is the odd one out: because the frame extends to the end of the partition, you cannot emit a row until you have seen the whole partition — there’s no way to know the final aggregate value otherwise. That’s a semantic fact, not an implementation limitation, and it’s why the default no-ORDER BY frame forces the buffering executor.

The `functions-window` crate: three evaluation APIs

The built-in window functions don’t implement frame evaluation themselves. They implement PartitionEvaluator, which has three methods, and the choice between them is a big part of the performance story:

rust
// datafusion/expr/src/partition_evaluator.rs:74-88
/// # Implementation Table
///
/// |[`uses_window_frame`]|[`supports_bounded_execution`]|[`include_rank`]|
/// function_to_implement|
/// |---|---|---|---|
/// |false (default)      |false (default)               |false (default)   | [`evaluate_all`]           |
/// |false                |true                          |false             | [`evaluate`]               |
/// |false                |true/false                    |true              | [`evaluate_all_with_rank`] |
/// |true                 |true/false                    |true/false        | [`evaluate`]               |

evaluate_all is the fast path: the whole partition at once, one array out. row_number uses it:

rust
// datafusion/functions-window/src/row_number.rs:140-161
fn evaluate_all(
    &mut self,
    _values: &[ArrayRef],
    num_rows: usize,
) -> Result<ArrayRef> {
    Ok(Arc::new(UInt64Array::from_iter_values(
        1..=(num_rows as u64),
    )))
}

fn evaluate(&mut self,
    _values: &[ArrayRef],
    _range: &Range<usize>,
) -> Result<ScalarValue> {
    self.n_rows += 1;
    Ok(ScalarValue::UInt64(Some(self.n_rows as u64)))
}

row_number over a partition of n rows is literally 1..=n — no per-row state, no frame arithmetic, no accumulator. The per-row evaluate exists only as a fallback for the bounded-execution path.

rank uses the third API, because ranks are peer-based: they need the group boundaries, which the executor computes:

rust
// datafusion/functions-window/src/rank.rs:314-331
fn evaluate_all_with_rank(
    &self,
    num_rows: usize,
    ranks_in_partition: &[Range<usize>],
) -> Result<ArrayRef> {
    let result: ArrayRef = match self.rank_type {
        RankType::Basic => Arc::new(UInt64Array::from_iter_values(
            ranks_in_partition
                .iter()
                .scan(1_u64, |acc, range| {
                    let len = range.end - range.start;
                    let result = iter::repeat_n(*acc, len);
                    *acc += len as u64;
                    Some(result)
                })
                .flatten(),

lead/lag use evaluate_all too, with an explicit shift (functions-window/src/lead_lag.rs:705-722). And cume_dist and percent_rank are rank-based too — they need the total row count of the partition, which is why they cannot be evaluated in a streaming fashion:

rust
// datafusion/functions-window/src/rank.rs:308-310
RankType::Percent => {
    exec_err!("Can not execute PERCENT_RANK in a streaming fashion")
}

Trade-offs and failure modes

The repartition is a real cost, and it’s a full data copy. The RepartitionExec(Hash(partition_by)) that the window’s declaration triggers moves every row. For a window over a table that’s already partitioned by the right key (Parquet row groups sorted by dept), that’s pure waste — and the planner can’t know. preserve_file_partitions and Partitioning::Range exist precisely to let the source declare “I’m already partitioned this way” and skip the shuffle.

Skew in PARTITION BY is fatal. A window with PARTITION BY country where one country is 90% of the rows means one hash partition holds 90% of the data, and WindowAggExec will buffer all of it. The window operator inherits the shuffle’s skew problem, and there’s no partial mitigation: unlike a join, you can’t split a partition’s rows across threads without breaking the ordering guarantee.

WindowAggExec buffers the entire partition. The stream pushes every input batch into self.batches and only computes when the input ends. For a 100 GB partition that’s 100 GB of RAM (or an OOM). BoundedWindowAggExec avoids it, but only when the frame’s end bound is bounded and the aggregate supports retraction/streaming. When neither holds, the only options are to spill or to repartition more finely.

The ORDER BY in the OVER clause is not free — and sometimes isn’t needed. OVER (PARTITION BY d ORDER BY t) sorts every partition. If the window function doesn’t actually need the order — SUM(x) OVER (PARTITION BY d) doesn’t — you still pay for the sort, because the frame semantics depend on the ordering even if the function doesn’t. DataFusion does exploit one case: the is_window_constant_in_partition check above means that for the whole-partition frame, the sort contributes nothing to the result.

Reversing the sort is a legitimate optimization. Some window functions can be evaluated backwards — LAG(x, 1) is LEAD(x, 1) evaluated in reverse order. get_window_mode checks whether the existing input ordering satisfies the requirement with the order-by keys reversed, and if so, the operator runs backwards:

rust
// datafusion/physical-plan/src/windows/mod.rs:695-732
pub fn get_window_mode(
    partitionby_exprs: &[Arc<dyn PhysicalExpr>],
    orderby_keys: &[PhysicalSortExpr],
    input: &Arc<dyn ExecutionPlan>,
) -> Result<Option<(bool, InputOrderMode>)>> {
    ...
    let reverse_orderby_keys =
        orderby_keys.iter().map(|e| e.reverse()).collect::<Vec<_>>();
    for (should_swap, orderbys) in
        [(false, orderby_keys), (true, reverse_orderby_keys.as_ref())]
    {
        ...
        if req.is_empty() || input_eqs.ordering_satisfy_requirement(req)? {
            // Window can be run with existing ordering
            let mode = if indices.len() == partitionby_exprs.len() {
                InputOrderMode::Sorted
            } else if indices.is_empty() {
                InputOrderMode::Linear
            } else {
                InputOrderMode::PartiallySorted(indices)
            };
            return Ok(Some((should_swap, mode)));
        }
    }
    Ok(None)
}

InputOrderMode is the honest description of what the input actually guarantees: Sorted (partition-by columns are a prefix of the ordering), PartiallySorted(indices) (some prefix of the partition-by columns is), or Linear (no ordering at all — only valid when there’s no PARTITION BY). The bounded executor handles all three; the plain one only handles Sorted.

Bottom line

The window operator is where DataFusion’s declarative design pays off most visibly. Three independent requirements — grouping, ordering, and framing — are expressed as metadata, and one optimizer rule turns them into a repartition, a sort, and (sometimes) nothing at all. The evaluation itself is then a two-accumulator choice: grow the frame and only add rows, or slide it and add-and-retract the delta. The frame bounds are a three-variant enum in which “unbounded” is a NULL scalar, and the “no ORDER BY” case is recognized early enough to collapse an O(N) per-row computation into a single per-partition aggregate.