Struct TumblingWindowOperator 

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

Tumbling window operator.

Processes events through non-overlapping, fixed-size time windows. Events are assigned to windows based on their timestamps, aggregated, and results are emitted based on the configured EmitStrategy.

Emit Strategies

• OnWatermark (default): Emit when watermark passes window end
• Periodic: Emit intermediate results at intervals, final on watermark
• OnUpdate: Emit after every state update

Late Data Handling

Events that arrive after window_end + allowed_lateness are considered late. Their behavior is controlled by LateDataConfig:

• Drop the event (default)
• Route to a named side output for separate processing

State Management

Window state is stored in the operator context’s state store using prefixed keys:

• win:<window_id> - Accumulator state
• meta:<window_id> - Window metadata (registration status, etc.)

Watermark Triggering

Windows are triggered when the watermark advances past window_end + allowed_lateness. This ensures late data within the grace period is still processed.

Implementations

impl<A: Aggregator> TumblingWindowOperator<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( assigner: TumblingWindowAssigner, aggregator: A, allowed_lateness: Duration, ) -> Self

Creates a new tumbling window operator.

Arguments

• assigner - Window assigner for determining window boundaries
• aggregator - Aggregation function to apply within windows
• allowed_lateness - Grace period for late data after window close

Panics

Panics if allowed lateness does not fit in i64.

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

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

Panics

Panics if allowed lateness does not fit in i64.

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

Sets the emit strategy for this window operator.

Arguments

• strategy - The emit strategy to use

Example

use laminar_core::operator::window::{
    TumblingWindowAssigner, TumblingWindowOperator, CountAggregator, EmitStrategy,
};
use std::time::Duration;

let assigner = TumblingWindowAssigner::new(Duration::from_secs(60));
let mut operator = TumblingWindowOperator::new(
    assigner,
    CountAggregator::new(),
    Duration::from_secs(5),
);

// Emit every 10 seconds instead of waiting for watermark
operator.set_emit_strategy(EmitStrategy::Periodic(Duration::from_secs(10)));

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.

Arguments

• config - The late data configuration to use

Example

use laminar_core::operator::window::{
    TumblingWindowAssigner, TumblingWindowOperator, CountAggregator, LateDataConfig,
};
use std::time::Duration;

let assigner = TumblingWindowAssigner::new(Duration::from_secs(60));
let mut operator = TumblingWindowOperator::new(
    assigner,
    CountAggregator::new(),
    Duration::from_secs(5),
);

// Route late events to a side output
operator.set_late_data_config(LateDataConfig::with_side_output("late_events".to_string()));

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.

Use this to monitor late data behavior and set up alerts.

pub fn reset_late_data_metrics(&mut self)

Resets the late data metrics counters.

pub fn assigner(&self) -> &TumblingWindowAssigner

Returns the window assigner.

pub fn allowed_lateness_ms(&self) -> i64

Returns the allowed lateness in milliseconds.

Trait Implementations

impl<A: Aggregator> Operator for TumblingWindowOperator<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