Struct HashJoinExecutor

pub struct HashJoinExecutor<K> {

    join_type: JoinType,
    original_schema: Schema,
    schema: Schema,
    output_indices: Vec<usize>,
    probe_side_source: BoxedExecutor,
    build_side_source: BoxedExecutor,
    probe_key_idxs: Vec<usize>,
    build_key_idxs: Vec<usize>,
    cond: Option<Arc<BoxedExpression>>,
    null_matched: Vec<bool>,
    identity: String,
    chunk_size: usize,
    spill_backend: Option<SpillBackend>,
    spill_metrics: Arc<BatchSpillMetrics>,
    memory_upper_bound: Option<u64>,
    shutdown_rx: ShutdownToken,
    mem_ctx: MemoryContext,
    _phantom: PhantomData<K>,
}

Hash Join Executor

High-level idea:

1. Iterate over the build side (i.e. right table) and build a hash map.
2. Iterate over the probe side (i.e. left table) and compute the hash value of each row. Then find the matched build side row for each probe side row in the hash map.
3. Concatenate the matched pair of probe side row and build side row into a single row and push it into the data chunk builder.
4. Yield chunks from the builder.

Fields

join_type: JoinType

Join type e.g. inner, left outer, …

original_schema: Schema

Output schema without applying output_indices

schema: Schema

Output schema after applying output_indices

output_indices: Vec<usize>

output_indices are the indices of the columns that we needed.

probe_side_source: BoxedExecutor

Left child executor

build_side_source: BoxedExecutor

Right child executor

probe_key_idxs: Vec<usize>

Column indices of left keys in equi join

build_key_idxs: Vec<usize>

Column indices of right keys in equi join

cond: Option<Arc<BoxedExpression>>

Non-equi join condition (optional)

null_matched: Vec<bool>

Whether or not to enable ‘IS NOT DISTINCT FROM’ semantics for a specific probe/build key column

identity: String

chunk_size: usize

spill_backend: Option<SpillBackend>

spill_metrics: Arc<BatchSpillMetrics>

memory_upper_bound: Option<u64>

The upper bound of memory usage for this executor.

shutdown_rx: ShutdownToken

mem_ctx: MemoryContext

_phantom: PhantomData<K>

Implementations

impl<K: HashKey> HashJoinExecutor<K>

fn do_execute( self: Box<Self>, ) -> Pin<Box<dyn Stream<Item = Result<DataChunk, BatchError>> + Send>>

pub fn do_inner_join( _: EquiJoinParams<K>, ) -> Pin<Box<dyn Stream<Item = Result<DataChunk, BatchError>> + Send>>

pub fn do_inner_join_with_non_equi_condition<'_async0>( params: EquiJoinParams<K>, cond: &'_async0 BoxedExpression, ) -> Pin<Box<dyn Stream<Item = Result<DataChunk, BatchError>> + Send + '_async0>>

pub fn do_left_outer_join( _: EquiJoinParams<K>, ) -> Pin<Box<dyn Stream<Item = Result<DataChunk, BatchError>> + Send>>

pub fn do_left_outer_join_with_non_equi_condition<'_async0>( _: EquiJoinParams<K>, cond: &'_async0 BoxedExpression, ) -> Pin<Box<dyn Stream<Item = Result<DataChunk, BatchError>> + Send + '_async0>>

pub fn do_left_semi_anti_join<const ANTI_JOIN: bool>( _: EquiJoinParams<K>, ) -> Pin<Box<dyn Stream<Item = Result<DataChunk, BatchError>> + Send>>

pub fn do_left_semi_join_with_non_equi_condition<'a>( _: EquiJoinParams<K>, cond: &'a BoxedExpression, ) -> Pin<Box<dyn Stream<Item = Result<DataChunk, BatchError>> + Send + 'a>>

High-level idea:

1. For each probe_row, append candidate rows to buffer. Candidate rows: Those satisfying equi_predicate (==).
2. If buffer becomes full, process it. Apply non_equi_join predicates e.g. >=, <= to filter rows. Track if probe_row is matched to avoid duplicates.
3. If we matched probe_row in spilled chunk, stop appending its candidate rows, to avoid matching it again in next spilled chunk.

pub fn do_left_anti_join_with_non_equi_condition<'_async0>( _: EquiJoinParams<K>, cond: &'_async0 BoxedExpression, ) -> Pin<Box<dyn Stream<Item = Result<DataChunk, BatchError>> + Send + '_async0>>

pub fn do_right_outer_join( _: EquiJoinParams<K>, ) -> Pin<Box<dyn Stream<Item = Result<DataChunk, BatchError>> + Send>>

pub fn do_right_outer_join_with_non_equi_condition<'_async0>( _: EquiJoinParams<K>, cond: &'_async0 BoxedExpression, ) -> Pin<Box<dyn Stream<Item = Result<DataChunk, BatchError>> + Send + '_async0>>

pub fn do_right_semi_anti_join<const ANTI_JOIN: bool>( _: EquiJoinParams<K>, ) -> Pin<Box<dyn Stream<Item = Result<DataChunk, BatchError>> + Send>>

pub fn do_right_semi_anti_join_with_non_equi_condition<'_async0, const ANTI_JOIN: bool>( _: EquiJoinParams<K>, cond: &'_async0 BoxedExpression, ) -> Pin<Box<dyn Stream<Item = Result<DataChunk, BatchError>> + Send + '_async0>>

pub fn do_full_outer_join( _: EquiJoinParams<K>, ) -> Pin<Box<dyn Stream<Item = Result<DataChunk, BatchError>> + Send>>

pub fn do_full_outer_join_with_non_equi_condition<'_async0>( _: EquiJoinParams<K>, cond: &'_async0 BoxedExpression, ) -> Pin<Box<dyn Stream<Item = Result<DataChunk, BatchError>> + Send + '_async0>>

async fn process_left_outer_join_non_equi_condition( chunk: DataChunk, cond: &dyn Expression, __arg2: &mut LeftNonEquiJoinState, ) -> Result<DataChunk>

Process output chunk for left outer join when non-equi condition is presented.

Arguments

• chunk - Output chunk from do_left_outer_join_with_non_equi_condition, containing:
  • Concatenation of probe row and its corresponding build row according to the hash map.
  • Concatenation of probe row and NULL build row, if there is no matched build row found for the probe row.
• cond - Non-equi join condition.
• probe_column_count - The number of columns in the probe side.
• first_output_row_id - The offset of the first output row in chunk for each probe side row that has been processed.
• has_more_output_rows - Whether the probe row being processed currently has output rows in next output chunk.
• found_matched - Whether the probe row being processed currently has matched non-NULL build rows in last output chunk.

Examples

Assume we have two tables t1 and t2 as probe side and build side, respectively.

CREATE TABLE t1 (v1 int, v2 int);
CREATE TABLE t2 (v3 int);

Now we de left outer join on t1 and t2, as the following query shows:

SELECT * FROM t1 LEFT JOIN t2 ON t1.v1 = t2.v3 AND t1.v2 <> t2.v3;

Assume the chunk builder in do_left_outer_join_with_non_equi_condition has buffer size 5, and we have the following chunk as the first output (‘-’ represents NULL).

offset	v1	v2	v3
0	1	2	1
1	1	1	1
2	2	3	-
3	3	3	3
4	3	3	3

We have the following precondition:

assert_eq!(probe_column_count, 2);
assert_eq!(first_out_row_id, vec![0, 1, 2, 3]);
assert_eq!(has_more_output_rows);
assert_eq!(!found_matched);

In process_left_outer_join_non_equi_condition, we transform the chunk in following steps.

1. Evaluate the non-equi condition on the chunk. Here the condition is t1.v2 <> t2.v3.

We get the result array:

offset	value
0	true
1	false
2	false
3	false
4	false

2. Set the build side columns to NULL if the corresponding result value is false.

The chunk is changed to:

offset	v1	v2	v3
0	1	2	1
1	1	1	-
2	2	3	-
3	3	3	-
4	3	3	-

3. Remove duplicate rows with NULL build side. This is done by setting the visibility bitmap of the chunk.

offset	v1	v2	v3
0	1	2	1
1	1	1	-
2	2	3	-
3	3	3	-
4	3	3	-

For the probe row being processed currently ((3, 3) here), we don’t have output rows with non-NULL build side, so we set found_matched to false.

In do_left_outer_join_with_non_equi_condition, we have next output chunk as follows:

offset	v1	v2	v3
0	3	3	3
1	3	3	3
2	5	5	-
3	5	3	-
4	5	3	-

This time We have the following precondition:

assert_eq!(probe_column_count, 2);
assert_eq!(first_out_row_id, vec![2, 3]);
assert_eq!(!has_more_output_rows);
assert_eq!(!found_matched);

The transformed chunk is as follows after the same steps.

offset	v1	v2	v3
0	3	3	3
1	3	3	-
2	5	5	-
3	5	3	-
4	5	3	-

After we add these chunks to output chunk builder in do_execute, we get the final output:

Chunk 1

offset	v1	v2	v3
0	1	2	1
1	1	1	-
2	2	3	-
3	3	3	-
4	5	5	-

Chunk 2

offset	v1	v2	v3
0	5	3	-

For more information about how process_*_join_non_equi_condition work, see their unit tests.

async fn process_left_semi_anti_join_non_equi_condition<const ANTI_JOIN: bool>( chunk: DataChunk, cond: &dyn Expression, __arg2: &mut LeftNonEquiJoinState, ) -> Result<DataChunk>

Filters for candidate rows which satisfy non_equi predicate. Removes duplicate rows.

async fn process_right_outer_join_non_equi_condition( chunk: DataChunk, cond: &dyn Expression, __arg2: &mut RightNonEquiJoinState, ) -> Result<DataChunk>

async fn process_right_semi_anti_join_non_equi_condition( chunk: DataChunk, cond: &dyn Expression, __arg2: &mut RightNonEquiJoinState, ) -> Result<()>

async fn process_full_outer_join_non_equi_condition( chunk: DataChunk, cond: &dyn Expression, left_non_equi_state: &mut LeftNonEquiJoinState, right_non_equi_state: &mut RightNonEquiJoinState, ) -> Result<DataChunk>

fn handle_remaining_build_rows_for_right_outer_join<'a>( chunk_builder: &'a mut DataChunkBuilder, build_side: &'a [DataChunk], build_row_matched: &'a ChunkedData<bool>, probe_column_count: usize, ) -> impl Stream<Item = Result<DataChunk, BatchError>> + 'a

fn handle_remaining_build_rows_for_right_semi_anti_join<'a, const ANTI_JOIN: bool>( chunk_builder: &'a mut DataChunkBuilder, build_side: &'a [DataChunk], build_row_matched: &'a ChunkedData<bool>, ) -> impl Stream<Item = Result<DataChunk, BatchError>> + 'a

fn append_one_row( chunk_builder: &mut DataChunkBuilder, probe_chunk: &DataChunk, probe_row_id: usize, build_chunk: &DataChunk, build_row_id: usize, ) -> Option<DataChunk>

fn append_one_probe_row( chunk_builder: &mut DataChunkBuilder, probe_chunk: &DataChunk, probe_row_id: usize, ) -> Option<DataChunk>

fn append_one_build_row( chunk_builder: &mut DataChunkBuilder, build_chunk: &DataChunk, build_row_id: usize, ) -> Option<DataChunk>

fn append_one_row_with_null_build_side( chunk_builder: &mut DataChunkBuilder, probe_row_ref: RowRef<'_>, build_column_count: usize, ) -> Option<DataChunk>

fn append_one_row_with_null_probe_side( chunk_builder: &mut DataChunkBuilder, build_row_ref: RowRef<'_>, probe_column_count: usize, ) -> Option<DataChunk>

impl<K> HashJoinExecutor<K>

pub fn new( join_type: JoinType, output_indices: Vec<usize>, probe_side_source: BoxedExecutor, build_side_source: BoxedExecutor, probe_key_idxs: Vec<usize>, build_key_idxs: Vec<usize>, null_matched: Vec<bool>, cond: Option<Arc<BoxedExpression>>, identity: String, chunk_size: usize, spill_backend: Option<SpillBackend>, spill_metrics: Arc<BatchSpillMetrics>, shutdown_rx: ShutdownToken, mem_ctx: MemoryContext, ) -> Self

fn new_inner( join_type: JoinType, output_indices: Vec<usize>, probe_side_source: BoxedExecutor, build_side_source: BoxedExecutor, probe_key_idxs: Vec<usize>, build_key_idxs: Vec<usize>, null_matched: Vec<bool>, cond: Option<Arc<BoxedExpression>>, identity: String, chunk_size: usize, spill_backend: Option<SpillBackend>, spill_metrics: Arc<BatchSpillMetrics>, memory_upper_bound: Option<u64>, shutdown_rx: ShutdownToken, mem_ctx: MemoryContext, ) -> Self

Trait Implementations

impl BoxedExecutorBuilder for HashJoinExecutor<()>

fn new_boxed_executor<'life0, 'life1, 'async_trait, C>( context: &'life0 ExecutorBuilder<'life1, C>, inputs: Vec<BoxedExecutor>, ) -> Pin<Box<dyn Future<Output = Result<BoxedExecutor>> + Send + 'async_trait>>

where C: 'async_trait + BatchTaskContext, 'life0: 'async_trait, 'life1: 'async_trait,

impl<K: HashKey> Executor for HashJoinExecutor<K>

fn schema(&self) -> &Schema

Returns the schema of the executor’s return data.

fn identity(&self) -> &str

Identity string of the executor

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

Executes to return the data chunk stream.