Struct AsofJoinOperator 

pub struct AsofJoinOperator { /* private fields */ }

ASOF join operator.

Performs temporal proximity joins between two event streams. Left events probe the right-side state for the closest matching timestamp.

State Management

Right-side events are stored in memory in per-key BTreeMap structures. State is persisted to the state store on checkpoint and cleaned up based on watermark progress.

Performance Characteristics

• Matching: O(log n) per left event (BTreeMap range query)
• State size: Bounded by right-side events within tolerance + watermark lag
• Memory: Linear in right-side event count per key

Implementations

impl AsofJoinOperator

pub fn new(config: AsofJoinConfig) -> Self

Creates a new ASOF join operator.

pub fn with_id(config: AsofJoinConfig, operator_id: String) -> Self

Creates a new ASOF join operator with explicit ID.

pub fn config(&self) -> &AsofJoinConfig

Returns the configuration.

pub fn metrics(&self) -> &AsofJoinMetrics

Returns the metrics.

pub fn reset_metrics(&mut self)

Resets the metrics.

pub fn watermark(&self) -> i64

Returns the current watermark.

pub fn state_size(&self) -> usize

Returns the total number of right-side events in state.

pub fn process_left( &mut self, event: &Event, _ctx: &mut OperatorContext<'_>, ) -> OutputVec

Processes a left-side event (probe side).

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

Processes a right-side event (build side).

pub fn on_watermark( &mut self, watermark: i64, _ctx: &mut OperatorContext<'_>, ) -> OutputVec

Handles watermark updates and triggers state cleanup.

Trait Implementations

impl Operator for AsofJoinOperator

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