Struct SessionWindowOperator 

pub struct SessionWindowOperator<A: Aggregator> { /* private fields */ }

Session window operator.

Groups events by activity periods separated by gaps. Each unique key maintains its own session state independently.

Session Lifecycle

1. Start: First event for a key creates a new session
2. Extend: Events within gap period extend the session
3. Close: Timer fires when gap expires, emitting results
4. Merge: Late data may merge previously separate sessions

State Management

Session state is stored using prefixed keys:

• ses:<key_hash> - Session metadata (start, end, key)
• sac:<key_hash> - Accumulator state

Emit Strategies

• OnWatermark: Emit when watermark passes session end
• OnUpdate: Emit after every state update
• OnWindowClose: Only emit on final closure
• Changelog: Emit CDC records with Z-set weights
• Final: Suppress all intermediate, drop late data

Implementations

impl<A: Aggregator> SessionWindowOperator<A>

where A::Acc: Archive + for<'a> RkyvSerialize<HighSerializer<AlignedVec, ArenaHandle<'a>, Error>>, <A::Acc as Archive>::Archived: for<'a> CheckBytes<HighValidator<'a, Error>> + RkyvDeserialize<A::Acc, HighDeserializer<Error>>,

pub fn new(gap: Duration, aggregator: A, allowed_lateness: Duration) -> Self

Creates a new session window operator.

Arguments

• gap - The inactivity gap that closes a session
• aggregator - Aggregation function to apply within sessions
• allowed_lateness - Grace period for late data after session close

Panics

Panics if gap or allowed lateness does not fit in i64.

pub fn with_id( gap: Duration, aggregator: A, allowed_lateness: Duration, operator_id: String, ) -> Self

Creates a new session window operator with a custom operator ID.

Panics

Panics if gap or allowed lateness does not fit in i64.

pub fn set_key_column(&mut self, column_index: usize)

Sets the key column for per-key session tracking.

If not set, a single global session is maintained.

pub fn key_column(&self) -> Option<usize>

Returns the key column index if set.

pub fn set_emit_strategy(&mut self, strategy: EmitStrategy)

Sets the emit strategy for this operator.

pub fn emit_strategy(&self) -> &EmitStrategy

Returns the current emit strategy.

pub fn set_late_data_config(&mut self, config: LateDataConfig)

Sets the late data handling configuration.

pub fn late_data_config(&self) -> &LateDataConfig

Returns the current late data configuration.

pub fn late_data_metrics(&self) -> &LateDataMetrics

Returns the late data metrics.

pub fn reset_late_data_metrics(&mut self)

Resets the late data metrics counters.

pub fn gap_ms(&self) -> i64

Returns the gap timeout in milliseconds.

pub fn allowed_lateness_ms(&self) -> i64

Returns the allowed lateness in milliseconds.

pub fn active_session_count(&self) -> usize

Returns the number of active sessions.

Trait Implementations

impl<A: Aggregator> Operator for SessionWindowOperator<A>

where A::Acc: 'static + Archive + for<'a> RkyvSerialize<HighSerializer<AlignedVec, ArenaHandle<'a>, Error>>, <A::Acc as Archive>::Archived: for<'a> CheckBytes<HighValidator<'a, Error>> + RkyvDeserialize<A::Acc, HighDeserializer<Error>>,

fn process(&mut self, event: &Event, ctx: &mut OperatorContext<'_>) -> OutputVec

Process an incoming event

fn on_timer(&mut self, timer: Timer, ctx: &mut OperatorContext<'_>) -> OutputVec

Handle timer expiration

fn checkpoint(&self) -> OperatorState

Checkpoint the operator’s state

fn restore(&mut self, state: OperatorState) -> Result<(), OperatorError>

Restore from a checkpoint