Implementation:ArroyoSystems Arroyo Window Function

Knowledge Sources	ArroyoSystems_Arroyo
Domains	Streaming, Windowing, Arrow_Processing
Last Updated	2026-02-08 08:00 GMT

Overview

WindowFunctionOperator is a streaming operator that applies SQL window functions (e.g., ROW_NUMBER, RANK, LAG) to event-time-partitioned record batches, executing a DataFusion window plan per timestamp bucket and emitting results when the watermark advances.

Description

The WindowFunctionOperator implements the ArrowOperator trait to execute DataFusion window function plans on streaming data. It operates similarly to InstantJoin in its time-bucketed execution model:

Incoming batches are filtered (removing records before the watermark) and split by timestamp using filter_and_split_batches, which sorts, filters, and partitions the batch by unique timestamp values.
For each distinct timestamp, a dedicated InstantComputeHolder is created containing an UnboundedSender and an active NextBatchFuture wrapping a DataFusion execution stream.
Record batches are sent to their timestamp's execution plan via the unbounded channel.
When the watermark advances, all timestamp buckets before the watermark are drained: the sender is dropped (closing the input), and the execution plan's output batches are collected and emitted.

State is checkpointed to an expiring time-keyed table "input" for recovery. On startup, buffered batches from state are replayed into their respective execution plans.

The operator maintains two schema views: input_schema (the full schema including keys) and input_schema_unkeyed (used for sorting and partitioning operations that operate on data columns only).

Usage

Used when a SQL query contains window functions such as ROW_NUMBER() OVER (PARTITION BY key ORDER BY event_time). Constructed via WindowFunctionConstructor from an api::WindowFunctionOperator configuration.

Code Reference

Source Location

Repository: ArroyoSystems_Arroyo
File: crates/arroyo-worker/src/arrow/window_fn.rs

Signature

pub struct WindowFunctionOperator {
    input_schema: ArroyoSchemaRef,
    input_schema_unkeyed: ArroyoSchemaRef,
    execs: BTreeMap<SystemTime, InstantComputeHolder>,
    futures: Arc<Mutex<FuturesUnordered<NextBatchFuture>>>,
    receiver: Arc<RwLock<Option<UnboundedReceiver<RecordBatch>>>>,
    window_exec: Arc<dyn ExecutionPlan>,
}

pub struct WindowFunctionConstructor;
impl OperatorConstructor for WindowFunctionConstructor {
    type ConfigT = api::WindowFunctionOperator;
    fn with_config(
        &self,
        config: Self::ConfigT,
        registry: Arc<Registry>,
    ) -> anyhow::Result<ConstructedOperator>;
}

Import

use arroyo_worker::arrow::window_fn::{WindowFunctionOperator, WindowFunctionConstructor};

I/O Contract

Inputs

Name	Type	Required	Description
batch	RecordBatch	Yes	Input records to be processed by the window function, split by timestamp
watermark	Watermark	Yes	Event-time watermark triggering draining of completed timestamp buckets
checkpoint	CheckpointBarrier	Yes	Triggers flush of the expiring time-key table "input"

Outputs

Name	Type	Description
window_result	RecordBatch	Output of the DataFusion window function plan for each completed timestamp bucket
watermark	Watermark::EventTime	Forwarded watermark after draining all timestamp buckets before it

Usage Examples

// Window function is created from a SQL query like:
// SELECT *, ROW_NUMBER() OVER (PARTITION BY key ORDER BY event_time) as rn
// FROM stream

let constructor = WindowFunctionConstructor;
let operator = constructor.with_config(window_fn_config, registry)?;

Related Pages

Page Connections

Double-click a node to navigate. Hold to expand connections.

Principle

Implementation

Heuristic

Environment