Struct WindowOperator 

pub struct WindowOperator {

    pub parent: Arc<Operators>,
    pub node: FlowNodeId,
    pub kind: WindowKind,
    pub group_by: Vec<Expression>,
    pub aggregations: Vec<Expression>,
    pub ts: Option<String>,
    pub compiled_group_by: Vec<CompiledExpr>,
    pub compiled_aggregations: Vec<CompiledExpr>,
    pub layout: RowSchema,
    pub functions: Functions,
    pub row_number_provider: RowNumberProvider,
    pub runtime_context: RuntimeContext,
    pub projected_columns: Vec<String>,
}

The main window operator

Fields

parent: Arc<Operators>

node: FlowNodeId

kind: WindowKind

group_by: Vec<Expression>

aggregations: Vec<Expression>

ts: Option<String>

compiled_group_by: Vec<CompiledExpr>

compiled_aggregations: Vec<CompiledExpr>

layout: RowSchema

functions: Functions

row_number_provider: RowNumberProvider

runtime_context: RuntimeContext

projected_columns: Vec<String>

Column names needed by group_by + aggregations expressions. When empty, no projection is applied (all columns stored).

Implementations

impl WindowOperator

pub fn evict_old_events(&self, state: &mut WindowState, current_timestamp: u64)

Evict old events from rolling window to maintain size limit

pub fn tick_rolling_eviction( &self, txn: &mut FlowTransaction, current_timestamp: u64, ) -> Result<Vec<Diff>>

Tick-based eviction for duration-based rolling windows. Scans all operator state, finds “win:” keys, and evicts old events.

impl WindowOperator

pub fn tick_session_expiration( &self, txn: &mut FlowTransaction, current_timestamp: u64, ) -> Result<Vec<Diff>>

Tick-based session expiration. Scans all operator state, finds “win:” keys with expired sessions.

impl WindowOperator

pub fn get_sliding_window_ids(&self, timestamp_or_row_index: u64) -> Vec<u64>

Determine which windows an event belongs to for sliding windows

pub fn set_sliding_window_start(&self, timestamp: u64, window_id: u64) -> u64

Set window start time for sliding windows (aligned to slide boundaries)

impl WindowOperator

pub fn get_tumbling_window_id(&self, timestamp: u64) -> u64

Determine which window an event belongs to for tumbling windows

pub fn set_tumbling_window_start(&self, timestamp: u64) -> u64

Set window start time for tumbling windows (aligned to window boundaries)

impl WindowOperator

pub fn new( parent: Arc<Operators>, node: FlowNodeId, kind: WindowKind, group_by: Vec<Expression>, aggregations: Vec<Expression>, ts: Option<String>, runtime_context: RuntimeContext, functions: Functions, ) -> Self

pub fn current_timestamp(&self) -> u64

Get the current timestamp in milliseconds

pub fn project_columns(&self, columns: &Columns) -> Columns

Project a single-row Columns down to only the columns needed by window expressions.

pub fn is_count_based(&self) -> bool

Whether this is a count-based window

pub fn size_duration(&self) -> Option<Duration>

Get the window size as duration (if time-based)

pub fn size_count(&self) -> Option<u64>

Get the window size as count (if count-based)

pub fn compute_group_keys(&self, columns: &Columns) -> Result<Vec<Hash128>>

Compute group keys for all rows in Columns

pub fn resolve_event_timestamps( &self, columns: &Columns, row_count: usize, ) -> Result<Vec<u64>>

Resolve event timestamps for all rows. When ts is configured, reads from the named DateTime column. Otherwise falls back to processing time (current clock).

pub fn create_window_key( &self, group_hash: Hash128, window_id: u64, ) -> EncodedKey

Create a window key for storage

pub fn store_row_index( &self, txn: &mut FlowTransaction, group_hash: Hash128, row_number: RowNumber, window_id: u64, ) -> Result<()>

Store a row_number → window_ids mapping. Appends window_id to the existing list (supports sliding windows with multiple windows per event).

pub fn extract_group_values( &self, window_layout: &WindowLayout, events: &[WindowEvent], ) -> Result<(Vec<Value>, Vec<String>)>

Extract group values from window events (all events in a group have the same group values). Evaluates compiled_group_by expressions on the first row of the events.

pub fn events_to_columns( &self, window_layout: &WindowLayout, events: &[WindowEvent], ) -> Result<Columns>

Convert window events to columnar format for aggregation

pub fn apply_aggregations( &self, txn: &mut FlowTransaction, window_key: &EncodedKey, window_layout: &WindowLayout, events: &[WindowEvent], ) -> Result<Option<(Row, bool)>>

Apply aggregations to all events in a window

pub fn process_expired_windows( &self, txn: &mut FlowTransaction, current_timestamp: u64, ) -> Result<Vec<Diff>>

Process expired windows: emit Remove diffs for each, then delete state. Uses the group registry for per-group targeted expiration.

pub fn load_window_state( &self, txn: &mut FlowTransaction, window_key: &EncodedKey, ) -> Result<WindowState>

Load window state from storage

pub fn save_window_state( &self, txn: &mut FlowTransaction, window_key: &EncodedKey, state: &WindowState, ) -> Result<()>

Save window state to storage

pub fn get_and_increment_global_count( &self, txn: &mut FlowTransaction, group_hash: Hash128, ) -> Result<u64>

Get and increment global event count for count-based windows

pub fn create_count_key(&self, group_hash: Hash128) -> EncodedKey

Create a count key for global event counting

pub fn load_group_registry( &self, txn: &mut FlowTransaction, ) -> Result<Vec<Hash128>>

Load the set of active group hashes from the registry.

pub fn register_group( &self, txn: &mut FlowTransaction, group_hash: Hash128, ) -> Result<()>

Register a group hash in the registry if not already present.

pub fn tick_expire_windows( &self, txn: &mut FlowTransaction, current_timestamp: u64, ) -> Result<Vec<Diff>>

Tick-based window expiration for tumbling/sliding windows. Scans all operator state, finds expired “win:” windows, emits Remove and cleans up.

pub fn process_insert( &self, txn: &mut FlowTransaction, columns: &Columns, group_fn: impl Fn(&WindowOperator, &mut FlowTransaction, &Columns, Hash128) -> Result<Vec<Diff>>, ) -> Result<Vec<Diff>>

Shared: partition columns by group keys and call group_fn for each group.

pub fn apply_window_change( &self, txn: &mut FlowTransaction, change: &Change, expire: bool, process_fn: impl Fn(&WindowOperator, &mut FlowTransaction, &Columns) -> Result<Vec<Diff>>, ) -> Result<Vec<Diff>>

Shared: iterate change diffs and process inserts/updates via process_fn. Optionally runs expiration first (all kinds except rolling).

pub fn emit_aggregation_diff( aggregated_row: &Row, is_new: bool, previous_aggregation: Option<(Row, bool)>, ) -> Diff

Shared: emit an Insert or Update diff for an aggregation result. previous_aggregation is the pre-update state (if the window already existed).

Trait Implementations

impl Operator for WindowOperator

fn id(&self) -> FlowNodeId

fn apply(&self, txn: &mut FlowTransaction, change: Change) -> Result<Change>

fn tick( &self, txn: &mut FlowTransaction, timestamp: DateTime, ) -> Result<Option<Change>>

Periodic tick for time-based maintenance (e.g., window eviction). Returns Some(Change) with diffs if maintenance produced changes.

fn pull(&self, txn: &mut FlowTransaction, rows: &[RowNumber]) -> Result<Columns>

impl RawStatefulOperator for WindowOperator

fn state_get( &self, txn: &mut FlowTransaction, key: &EncodedKey, ) -> Result<Option<EncodedRow>>

Get raw bytes for a key

fn state_set( &self, txn: &mut FlowTransaction, key: &EncodedKey, value: EncodedRow, ) -> Result<()>

Set raw bytes for a key

fn state_remove( &self, txn: &mut FlowTransaction, key: &EncodedKey, ) -> Result<()>

Remove a key

fn state_scan(&self, txn: &mut FlowTransaction) -> Result<StateIterator>

Scan all keys for this operator

fn state_range( &self, txn: &mut FlowTransaction, range: EncodedKeyRange, ) -> Result<StateIterator>

Range query between keys

fn state_clear(&self, txn: &mut FlowTransaction) -> Result<()>

Clear all state for this operator

impl WindowStateful for WindowOperator

fn layout(&self) -> RowSchema

Get or create the layout for state rows

fn create_state(&self) -> EncodedRow

Create a new state encoded with default values

fn load_state( &self, txn: &mut FlowTransaction, window_key: &EncodedKey, ) -> Result<EncodedRow>

Load state for a window

fn save_state( &self, txn: &mut FlowTransaction, window_key: &EncodedKey, row: EncodedRow, ) -> Result<()>

Save state for a window

fn scan_keys_in_range( &self, txn: &mut FlowTransaction, range: &EncodedKeyRange, ) -> Result<Vec<EncodedKey>>

Scan keys within a range without removing them (read-only)

fn expire_range( &self, txn: &mut FlowTransaction, range: EncodedKeyRange, ) -> Result<u32>

Expire windows within a given range The range should be constructed by the caller based on their window ordering semantics