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,
    entry_state_max_rows: 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

When to alert high join amplification

entry_state_max_rows: usize

Max number of rows that will be cached in the entry state.

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

pub fn new_with_cache_size( 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, entry_state_max_rows: Option<usize>, ) -> Self

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

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

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>>

fn row_concat( row_update: impl Row, update_start_pos: usize, row_matched: impl Row, 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

fn handle_match_rows<'a, const SIDE: u8, const JOIN_OP: bool>( cached_lookup_result: CacheResult, row: RowRef<'a>, key: &'a K, hashjoin_chunk_builder: &'a mut JoinChunkBuilder<T, SIDE>, side_match: &'a mut JoinSide<K, S>, side_update: &'a mut JoinSide<K, S>, useful_state_clean_columns: &'a [(usize, &'a Watermark)], cond: &'a mut Option<NonStrictExpression>, append_only_optimize: bool, entry_state_max_rows: usize, ) -> impl Stream<Item = Result<StreamChunk, StreamExecutorError>> + 'a

For the probe-side row, we need to check if it has values in cache, if not, we need to fetch the matched rows from the state table.

Every matched build-side row being processed needs to go through the following phases:

1. Handle join condition evaluation.
2. Always do cache refill, if the state count is good.
3. Handle state cleaning.
4. Handle degree table update.

async fn handle_match_row<'a, const SIDE: u8, const JOIN_OP: bool, const MATCHED_ROWS_FROM_CACHE: bool>( update_row: RowRef<'a>, matched_row: JoinRow<OwnedRow>, matched_row_cache_ref: Option<&mut EncodedJoinRow>, hashjoin_chunk_builder: &'a mut JoinChunkBuilder<T, SIDE>, match_order_key_indices: &[usize], match_degree_table: &mut Option<TableInner<S>>, side_update_start_pos: usize, side_match_start_pos: usize, cond: &Option<NonStrictExpression>, update_row_degree: &mut u64, useful_state_clean_columns: &[(usize, &'a Watermark)], append_only_optimize: bool, append_only_matched_row: &mut Option<JoinRow<OwnedRow>>, matched_rows_to_clean: &mut Vec<JoinRow<OwnedRow>>, ) -> Option<StreamChunk>

async fn check_join_condition( row: impl Row, side_update_start_pos: usize, matched_row: impl Row, side_match_start_pos: usize, join_condition: &Option<NonStrictExpression>, ) -> bool

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,