Struct AsOfJoinExecutor 

pub struct AsOfJoinExecutor<S: StateStore, const T: u8, E: AsOfRowEncoding> {

    ctx: ActorContextRef,
    info: ExecutorInfo,
    input_l: Option<Executor>,
    input_r: Option<Executor>,
    actual_output_data_types: Vec<DataType>,
    side_l: JoinSide<S, E>,
    side_r: JoinSide<S, E>,
    null_safe: Vec<bool>,
    metrics: Arc<StreamingMetrics>,
    chunk_size: usize,
    watermark_buffers: BTreeMap<usize, BufferedWatermarks<u8>>,
    asof_desc: AsOfDesc,
    cnt_rows_received: u32,
    join_cache_evict_interval_rows: u32,
    high_join_amplification_threshold: usize,
}

AsOfJoinExecutor for streaming as-of joins.

This executor uses AsOfJoinHashMap to support both execution modes: it can either maintain an LRU cache or query the state table directly, depending on the use_cache configuration.

Fields

ctx: ActorContextRef

info: ExecutorInfo

input_l: Option<Executor>

Left input executor

input_r: Option<Executor>

Right input executor

actual_output_data_types: Vec<DataType>

The data types of the formed new columns

side_l: JoinSide<S, E>

The parameters of the left join executor

side_r: JoinSide<S, E>

The parameters of the right join executor

null_safe: Vec<bool>

Whether nulls are considered equal for each join key column

metrics: Arc<StreamingMetrics>

chunk_size: usize

The maximum size of the chunk produced by executor at a time

watermark_buffers: BTreeMap<usize, BufferedWatermarks<u8>>

watermark column index -> BufferedWatermarks

asof_desc: AsOfDesc

AsOf join description

cnt_rows_received: u32

Row counter for periodic cache eviction

join_cache_evict_interval_rows: u32

Number of processed rows between periodic manual evictions of the join cache.

high_join_amplification_threshold: usize

Threshold for logging high join amplification warnings.

Implementations

impl<S: StateStore, const T: u8, E: AsOfRowEncoding> AsOfJoinExecutor<S, T, E>

pub fn new( ctx: ActorContextRef, info: ExecutorInfo, input_l: Executor, input_r: Executor, params_l: JoinParams, params_r: JoinParams, null_safe: Vec<bool>, output_indices: Vec<usize>, state_table_l: StateTable<S>, state_table_r: StateTable<S>, watermark_epoch: AtomicU64Ref, metrics: Arc<StreamingMetrics>, chunk_size: usize, asof_desc: AsOfDesc, use_cache: bool, high_join_amplification_threshold: usize, ) -> Self

fn evict_cache( side_l: &mut JoinSide<S, E>, side_r: &mut JoinSide<S, E>, cnt_rows_received: &mut u32, join_cache_evict_interval_rows: u32, )

Periodically evict the LRU cache to prevent memory buildup between barriers.

fn into_stream(self) -> impl Stream<Item = Result<Message, StreamExecutorError>>

async fn flush_data( &mut self, epoch: EpochPair, ) -> StreamExecutorResult<(AsOfJoinHashMapPostCommit<'_, S, E>, AsOfJoinHashMapPostCommit<'_, S, E>)>

async fn try_flush_data(&mut self) -> StreamExecutorResult<()>

fn handle_watermark( &mut self, side: u8, watermark: Watermark, ) -> StreamExecutorResult<Vec<Watermark>>

fn eq_join_left<'_async0>( args: EqJoinArgs<'_async0, S, E>, ) -> impl Stream<Item = Result<StreamChunk, StreamExecutorError>> + '_async0

fn cmp_pk_rows(pk1: &impl Row, pk2: &impl Row) -> Ordering

fn eq_join_right<'_async0>( args: EqJoinArgs<'_async0, S, E>, ) -> impl Stream<Item = Result<StreamChunk, StreamExecutorError>> + '_async0

Trait Implementations

impl<S: StateStore, const T: u8, E: AsOfRowEncoding> Debug for AsOfJoinExecutor<S, T, E>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<S: StateStore, const T: u8, E: AsOfRowEncoding> Execute for AsOfJoinExecutor<S, T, E>

fn execute(self: Box<Self>) -> BoxedMessageStream

fn execute_with_epoch(self: Box<Self>, _epoch: u64) -> BoxedMessageStream

fn boxed(self) -> Box<dyn Execute>

where Self: Sized + Send + 'static,