Implementation:ArroyoSystems Arroyo Session Window
| Knowledge Sources | |
|---|---|
| Domains | Streaming, Windowing, Aggregation |
| Last Updated | 2026-02-08 08:00 GMT |
Overview
SessionAggregatingWindowFunc implements session window aggregation, grouping events into sessions separated by a configurable inactivity gap and computing aggregates over each session.
Description
The SessionAggregatingWindowFunc operator implements the ArrowOperator trait to perform session-based windowed aggregation. Sessions are defined by a gap duration: events within the gap of the most recent event in a session are merged into that session. When the watermark advances past the session's end (last event time + gap), the session is finalized and the aggregation result is emitted.
Key internal structures:
- SessionWindowConfig -- Holds the gap duration, input schema, window field metadata, and the final aggregation ExecutionPlan.
- KeyComputingHolder -- Per-key state machine that manages an ActiveSession (streaming aggregation via DataFusion) and buffered batches ordered by start time.
- ActiveSession -- Wraps a running DataFusion aggregation plan fed by an unbounded channel; produces a SessionWindowResult when finished.
The operator tracks keys by their next watermark action time (keys_by_next_watermark_action) and by their earliest data time (keys_by_start_time) using BTreeMap structures for efficient watermark-driven advancement. State is checkpointed to an expiring time-keyed table "s" and a global keyed state "e" that stores the earliest batch time per subtask.
Usage
Used when a SQL query specifies a SESSION window with a gap parameter. The Arroyo planner emits a SessionWindowAggregateOperator configuration which is deserialized by SessionAggregatingWindowConstructor.
Code Reference
Source Location
Signature
pub struct SessionAggregatingWindowFunc {
config: Arc<SessionWindowConfig>,
keys_by_next_watermark_action: BTreeMap<SystemTime, HashSet<Vec<u8>>>,
key_computations: HashMap<Vec<u8>, KeyComputingHolder>,
keys_by_start_time: BTreeMap<SystemTime, HashSet<Vec<u8>>>,
row_converter: Converter,
}
pub struct SessionAggregatingWindowConstructor;
impl OperatorConstructor for SessionAggregatingWindowConstructor {
type ConfigT = api::SessionWindowAggregateOperator;
fn with_config(
&self,
config: Self::ConfigT,
registry: Arc<Registry>,
) -> anyhow::Result<ConstructedOperator>;
}
Import
use arroyo_worker::arrow::session_aggregating_window::{
SessionAggregatingWindowFunc, SessionAggregatingWindowConstructor,
};
I/O Contract
Inputs
| Name | Type | Required | Description |
|---|---|---|---|
| batch | RecordBatch | Yes | Input records with key columns, data columns, and a timestamp column |
| watermark | Watermark | Yes | Event-time watermark triggering session finalization for sessions ending before the watermark |
| checkpoint | CheckpointBarrier | Yes | Triggers flush of the expiring time-key table and global keyed state |
Outputs
| Name | Type | Description |
|---|---|---|
| session_result | RecordBatch | Aggregated result per session containing key columns, window struct (start, end), aggregate columns, and timestamp |
| watermark | Watermark | Forwarded watermark after session advancement |
Usage Examples
// Session window is created from a SQL query like:
// SELECT key, COUNT(*) FROM stream
// GROUP BY key, SESSION(event_time, INTERVAL '5' MINUTE)
// The planner generates a SessionWindowAggregateOperator config
// with gap_micros = 300_000_000 (5 minutes in microseconds)
let constructor = SessionAggregatingWindowConstructor;
let operator = constructor.with_config(session_config, registry)?;