Struct WindowOperator 

pub struct WindowOperator {

    pub parent: Arc<Operators>,
    pub node: FlowNodeId,
    pub window_type: WindowType,
    pub size: WindowSize,
    pub slide: Option<WindowSlide>,
    pub group_by: Vec<Expression>,
    pub aggregations: Vec<Expression>,
    pub compiled_group_by: Vec<CompiledExpr>,
    pub compiled_aggregations: Vec<CompiledExpr>,
    pub layout: Schema,
    pub functions: Functions,
    pub row_number_provider: RowNumberProvider,
    pub min_events: usize,
    pub max_window_count: Option<usize>,
    pub max_window_age: Option<Duration>,
    pub clock: Clock,
}

The main window operator

Fields

parent: Arc<Operators>

node: FlowNodeId

window_type: WindowType

size: WindowSize

slide: Option<WindowSlide>

group_by: Vec<Expression>

aggregations: Vec<Expression>

compiled_group_by: Vec<CompiledExpr>

compiled_aggregations: Vec<CompiledExpr>

layout: Schema

functions: Functions

row_number_provider: RowNumberProvider

min_events: usize

max_window_count: Option<usize>

max_window_age: Option<Duration>

clock: Clock

Implementations

impl WindowOperator

pub fn should_evict_rolling_window( &self, state: &WindowState, current_timestamp: u64, ) -> bool

Check if rolling window should evict old events

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

Evict old events from rolling window to maintain size limit

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 For time-based windows, pass the event timestamp For count-based windows, pass the row index (0-based)

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)

pub fn should_start_new_tumbling_window( &self, current_window_start: u64, event_timestamp: u64, ) -> bool

Check if tumbling window should be moved to a new window due to time boundaries

pub fn should_expire_tumbling_window( &self, state: &WindowState, current_timestamp: u64, ) -> bool

Check if a tumbling window should be expired (closed)

impl WindowOperator

pub fn new( parent: Arc<Operators>, node: FlowNodeId, window_type: WindowType, size: WindowSize, slide: Option<WindowSlide>, group_by: Vec<Expression>, aggregations: Vec<Expression>, min_events: usize, max_window_count: Option<usize>, max_window_age: Option<Duration>, clock: Clock, functions: Functions, ) -> Self

pub fn current_timestamp(&self) -> u64

Get the current timestamp in milliseconds

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

Compute group keys for all rows in Columns

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

Extract timestamps for all rows in Columns

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

Create a window key for storage

pub fn extract_timestamp_from_row(&self, row: &Row) -> Result<u64>

Extract timestamp from row data

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

Extract group values from window events (all events in a group have the same group values) TODO: Refactor to use column-based evaluation when window operator is needed

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

Convert window events to columnar format for aggregation

pub fn apply_aggregations( &self, txn: &mut FlowTransaction, window_key: &EncodedKey, 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 and clean up state

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

impl WindowOperator

Additional helper methods for window triggering

pub fn should_trigger_window( &self, state: &WindowState, current_timestamp: u64, ) -> bool

Check if a window should be triggered (emitted)

Trait Implementations

impl Operator for WindowOperator

fn id(&self) -> FlowNodeId

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

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<EncodedValues>>

Get raw bytes for a key

fn state_set( &self, txn: &mut FlowTransaction, key: &EncodedKey, value: EncodedValues, ) -> 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) -> Schema

Get or create the layout for state rows

fn create_state(&self) -> EncodedValues

Create a new state encoded with default values

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

Load state for a window

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

Save state for a window

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