Struct AsOfJoinExecutor

pub struct AsOfJoinExecutor<K: HashKey, S: StateStore, const T: u8> {

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

AsOfJoinExecutor takes two input streams and runs equal hash join on them. The output columns are the concatenation of left and right columns.

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<K, S>

The parameters of the left join executor

side_r: JoinSide<K, S>

The parameters of the right join executor

metrics: Arc<StreamingMetrics>

chunk_size: usize

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

cnt_rows_received: u32

Count the messages received, clear to 0 when counted to EVICT_EVERY_N_MESSAGES

watermark_buffers: BTreeMap<usize, BufferedWatermarks<u8>>

watermark column index -> BufferedWatermarks

high_join_amplification_threshold: usize

asof_desc: AsOfDesc

AsOf join description

Implementations

impl<K: HashKey, S: StateStore, const T: u8> AsOfJoinExecutor<K, S, T>

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, high_join_amplification_threshold: usize, asof_desc: AsOfDesc, ) -> Self

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

async fn flush_data(&mut self, epoch: EpochPair) -> StreamExecutorResult<()>

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

fn evict_cache( side_update: &mut JoinSide<K, S>, side_match: &mut JoinSide<K, S>, cnt_rows_received: &mut u32, )

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

async fn hash_eq_match( key: &K, ht: &mut JoinHashMap<K, S>, ) -> StreamExecutorResult<Option<Box<JoinEntryState>>>

the data the hash table and match the coming data chunk with the executor state

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

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

Trait Implementations

impl<K: HashKey, S: StateStore, const T: u8> Debug for AsOfJoinExecutor<K, S, T>

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

Formats the value using the given formatter.

impl<K: HashKey, S: StateStore, const T: u8> Execute for AsOfJoinExecutor<K, S, T>

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,