Implementation:ArroyoSystems Arroyo Sliding Window
| Knowledge Sources | |
|---|---|
| Domains | Streaming, Windowing, Aggregation |
| Last Updated | 2026-02-08 08:00 GMT |
Overview
SlidingAggregatingWindowFunc implements sliding (hopping) window aggregation using a two-phase partial/final aggregation strategy with tiered record batch storage for efficient window overlap computation.
Description
The SlidingAggregatingWindowFunc operator implements the ArrowOperator trait to compute aggregations over overlapping time windows defined by a width (total window size) and a slide (how often a new window starts). The operator uses a two-phase aggregation approach:
Phase 1 -- Partial Aggregation: Incoming batches are binned by their slide-aligned timestamp using a binning_function (a DataFusion PhysicalExpr). Each bin gets a dedicated BinComputingHolder that streams data through an unbounded channel into a partial_aggregation_plan.
Phase 2 -- Final Aggregation: When the watermark advances past a bin boundary, completed partial aggregates are stored in a TieredRecordBatchHolder and the expiring window's final result is computed by gathering all partial results within the window interval and running them through the finish_execution_plan and final_projection.
The TieredRecordBatchHolder is a key optimization structure that stores partial aggregates at multiple time granularities (tier widths), allowing efficient retrieval of batches spanning a window interval by selecting the largest possible tier bins. This minimizes the number of partial batches that must be merged for each window evaluation.
The operator tracks three states via SlidingWindowState: NoData, OnlyBufferedData (data exists in state but not yet in memory), and InMemoryData (tiered record batches hold data ready for windowing).
Usage
Used when a SQL query specifies a HOP or sliding window with both a width and slide parameter. Constructed via SlidingAggregatingWindowConstructor from an api::SlidingWindowAggregateOperator configuration.
Code Reference
Source Location
Signature
pub struct SlidingAggregatingWindowFunc<K: Copy> {
slide: Duration,
width: Duration,
binning_function: Arc<dyn PhysicalExpr>,
partial_aggregation_plan: Arc<dyn ExecutionPlan>,
partial_schema: ArroyoSchema,
finish_execution_plan: Arc<dyn ExecutionPlan>,
receiver: Arc<RwLock<Option<UnboundedReceiver<RecordBatch>>>>,
final_batches_passer: Arc<RwLock<Vec<RecordBatch>>>,
futures: FuturesUnordered<NextBatchFuture<K>>,
execs: BTreeMap<K, BinComputingHolder<K>>,
tiered_record_batches: TieredRecordBatchHolder,
projection_input_schema: SchemaRef,
final_projection: Arc<dyn ExecutionPlan>,
state: SlidingWindowState,
}
pub struct SlidingAggregatingWindowConstructor;
impl OperatorConstructor for SlidingAggregatingWindowConstructor {
type ConfigT = api::SlidingWindowAggregateOperator;
fn with_config(
&self,
config: Self::ConfigT,
registry: Arc<Registry>,
) -> anyhow::Result<ConstructedOperator>;
}
Import
use arroyo_worker::arrow::sliding_aggregating_window::{
SlidingAggregatingWindowFunc, SlidingAggregatingWindowConstructor,
};
I/O Contract
Inputs
| Name | Type | Required | Description |
|---|---|---|---|
| batch | RecordBatch | Yes | Input records partitioned by the binning function into slide-aligned time bins |
| watermark | Watermark | Yes | Event-time watermark triggering window evaluation when a complete window has passed |
| checkpoint | CheckpointBarrier | Yes | Triggers draining of active bin executions and flushing partial aggregates to the state table "t" |
Outputs
| Name | Type | Description |
|---|---|---|
| window_result | RecordBatch | Final aggregation result for each completed sliding window, including window metadata and timestamp |
| watermark | Watermark | Forwarded watermark after all ready windows have been evaluated |
Usage Examples
// Sliding window is created from a SQL query like:
// SELECT key, SUM(value) FROM stream
// GROUP BY key, HOP(event_time, INTERVAL '1' MINUTE, INTERVAL '5' MINUTE)
// where slide = 1 minute and width = 5 minutes
let constructor = SlidingAggregatingWindowConstructor;
let operator = constructor.with_config(sliding_config, registry)?;