Struct HashJoinExecutor

pub struct HashJoinExecutor<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>,
    cond: Option<NonStrictExpression>,
    inequality_pairs: Vec<(Vec<usize>, Option<NonStrictExpression>)>,
    inequality_watermarks: Vec<Option<Watermark>>,
    append_only_optimize: bool,
    metrics: Arc<StreamingMetrics>,
    chunk_size: usize,
    cnt_rows_received: u32,
    watermark_buffers: BTreeMap<usize, BufferedWatermarks<u8>>,
    high_join_amplification_threshold: usize,
}

HashJoinExecutor 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

cond: Option<NonStrictExpression>

Optional non-equi join conditions

inequality_pairs: Vec<(Vec<usize>, Option<NonStrictExpression>)>

Column indices of watermark output and offset expression of each inequality, respectively.

inequality_watermarks: Vec<Option<Watermark>>

The output watermark of each inequality condition and its value is the minimum of the calculation result of both side. It will be used to generate watermark into downstream and do state cleaning if clean_state field of that inequality is true.

append_only_optimize: bool

Whether the logic can be optimized for append-only stream

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

Implementations

impl<K: HashKey, S: StateStore, const T: u8> HashJoinExecutor<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>, cond: Option<NonStrictExpression>, inequality_pairs: Vec<(usize, usize, bool, Option<NonStrictExpression>)>, state_table_l: StateTable<S>, degree_state_table_l: StateTable<S>, state_table_r: StateTable<S>, degree_state_table_r: StateTable<S>, watermark_epoch: AtomicU64Ref, is_append_only: bool, metrics: Arc<StreamingMetrics>, chunk_size: usize, high_join_amplification_threshold: usize, ) -> 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, )

async 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 row_concat( row_update: &RowRef<'_>, update_start_pos: usize, row_matched: &OwnedRow, matched_start_pos: usize, ) -> OwnedRow

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

Used to forward eq_join_oneside to show join side in stack.

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

Used to forward eq_join_oneside to show join side in stack.

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

Trait Implementations

impl<K: HashKey, S: StateStore, const T: u8> Debug for HashJoinExecutor<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 HashJoinExecutor<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,