Struct StreamJoinOperator 

pub struct StreamJoinOperator { /* private fields */ }

Stream-stream join operator.

Joins events from two streams based on a key column and time bound. Events are matched if they share a key value and their timestamps are within the specified time window.

State Management

Events from both sides are stored in state until they can no longer produce matches (watermark passes timestamp + time_bound). State is automatically cleaned up via timers.

Performance Considerations

• State grows linearly with the number of events within the time window
• For high-cardinality joins, consider using shorter time bounds
• Inner joins use less state than outer joins (no unmatched tracking)

Optimizations (F057)

• CPU-Friendly Encoding: Use JoinRowEncoding::CpuFriendly for 30-50% faster access
• Asymmetric Compaction: Automatically skips compaction on finished/idle sides
• Per-Key Tracking: Aggressive cleanup for sparse key patterns
• Build-Side Pruning: Early pruning based on probe-side watermark progress

Implementations

impl StreamJoinOperator

pub fn new( left_key_column: String, right_key_column: String, time_bound: Duration, join_type: JoinType, ) -> Self

Creates a new stream join operator.

Arguments

• left_key_column - Name of the key column in left stream events
• right_key_column - Name of the key column in right stream events
• time_bound - Maximum time difference for matching events
• join_type - Type of join to perform

pub fn with_id( left_key_column: String, right_key_column: String, time_bound: Duration, join_type: JoinType, operator_id: String, ) -> Self

Creates a new stream join operator with a custom operator ID.

pub fn from_config(config: StreamJoinConfig) -> Self

Creates a new stream join operator from configuration (F057).

This is the recommended constructor for production use, allowing fine-grained control over optimization settings.

Example

use laminar_core::operator::stream_join::{
    StreamJoinOperator, StreamJoinConfig, JoinType, JoinRowEncoding,
};
use std::time::Duration;

let config = StreamJoinConfig::builder()
    .left_key_column("order_id")
    .right_key_column("order_id")
    .time_bound(Duration::from_secs(3600))
    .join_type(JoinType::Inner)
    .row_encoding(JoinRowEncoding::CpuFriendly)
    .build();

let operator = StreamJoinOperator::from_config(config);

pub fn join_type(&self) -> JoinType

Returns the join type.

pub fn time_bound_ms(&self) -> i64

Returns the time bound in milliseconds.

pub fn metrics(&self) -> &JoinMetrics

Returns the metrics.

pub fn reset_metrics(&mut self)

Resets the metrics.

pub fn row_encoding(&self) -> JoinRowEncoding

Returns the row encoding strategy (F057).

pub fn asymmetric_compaction_enabled(&self) -> bool

Returns whether asymmetric compaction is enabled (F057).

pub fn per_key_tracking_enabled(&self) -> bool

Returns whether per-key tracking is enabled (F057).

pub fn left_stats(&self) -> &SideStats

Returns the left-side statistics (F057).

pub fn right_stats(&self) -> &SideStats

Returns the right-side statistics (F057).

pub fn tracked_key_count(&self) -> usize

Returns the number of tracked keys (F057).

pub fn is_side_idle(&self, side: JoinSide, current_time: i64) -> bool

Checks if a side is considered “finished” (idle) (F057).

pub fn effective_build_side(&self) -> JoinSide

Determines the effective build side based on configuration or heuristics (F057).

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

Processes an event from either the left or right side.

This is the main entry point for the join operator. Call this with the appropriate JoinSide to indicate which stream the event came from.

pub fn scan_idle_keys(&mut self, ctx: &mut OperatorContext<'_>)

F057: Scans for idle keys and cleans them up aggressively.

Called periodically (e.g., on timer) to identify keys with no recent activity and remove their state entries.

pub fn should_skip_compaction(&self, side: JoinSide, current_time: i64) -> bool

F057: Checks if compaction should be skipped for a side due to asymmetric optimization.

Trait Implementations

impl Operator for StreamJoinOperator

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