Struct SessionWindowOperator 

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

Creates the standard session output schema. Session window operator.

Groups events by activity periods separated by gaps. Each unique key maintains its own session state independently, supporting multiple concurrent sessions per key.

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:

• six:<key_hash> - SessionIndex (all sessions for a key)
• sac:<session_id> - Per-session 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_max_cached_indices(&mut self, max: usize)

Sets the maximum number of session indices to cache in-memory.

When the cache exceeds this limit, it is cleared and entries are reloaded from the persistent state store on demand. Default: 16,384.

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 window_close_metrics(&self) -> &WindowCloseMetrics

Returns the window close metrics.

pub fn reset_window_close_metrics(&mut self)

Resets the window close 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 across all keys.

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