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,
    asof_desc: Option<AsOfDesc>,
    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

asof_desc: Option<AsOfDesc>

Whether the join is an as-of join

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(super) fn probe_and_project<'a>( params: EquiJoinParams<K>, join_type: JoinType, cond: Option<&'a BoxedExpression>, output_indices: &'a [usize], output_data_types: Vec<DataType>, ) -> Pin<Box<dyn Stream<Item = Result<DataChunk, BatchError>> + Send + 'a>>

Probes the hash table, projects the output to output_indices, and yields chunks of params.chunk_size rows.

AS OF joins apply their condition through asof_desc, so cond is ignored for them.

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

Probes the hash table with each probe row and produces the join output.

The candidates of a probe row are the build rows with the same key, or for an AS OF join, the closest of them. A candidate pair matches if it satisfies the non-equi condition cond, or always if there is none. A probe or build row is matched if it is part of a matched pair. JoinOutput gives what each join type outputs:

join type	matched pairs	probe rows	build rows
inner	yes
left outer	yes	unmatched, with NULLs
left semi / anti		matched / unmatched
right outer	yes		unmatched, with NULLs
right semi / anti			matched / unmatched
full outer	yes	unmatched, with NULLs	unmatched, with NULLs

With a condition, candidate pairs are buffered and the condition is evaluated on a whole chunk of them at a time. Probe rows are output per probe chunk: if the join outputs them, all candidates of a probe chunk are evaluated before moving on to the next one, so that they have their final match state. Build rows are output after the last probe chunk.

fn null_row(data_types: &[DataType]) -> DataChunk

fn append_probe_row_output( chunk_builder: &mut DataChunkBuilder, output: RowOutput, probe_chunk: &DataChunk, probe_row_id: usize, null_build_row: &DataChunk, ) -> Option<DataChunk>

fn append_build_row_output( chunk_builder: &mut DataChunkBuilder, output: RowOutput, build_chunk: &DataChunk, build_row_id: usize, null_probe_row: &DataChunk, ) -> Option<DataChunk>

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 find_asof_matched_rows( probe_row_ref: RowRef<'_>, build_side: &[DataChunk], build_side_row_iter: RowIdIter<'_>, asof_join_condition: &AsOfDesc, ) -> Option<RowId>

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, asof_desc: Option<AsOfDesc>, 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, asof_desc: Option<AsOfDesc>, 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<()>

async fn new_boxed_executor( context: &ExecutorBuilder<'_>, inputs: Vec<BoxedExecutor>, ) -> Result<BoxedExecutor>

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.