Struct HashJoinStream 

pub(super) struct HashJoinStream {

    partition: usize,
    schema: Arc<Schema>,
    on_right: Vec<PhysicalExprRef>,
    filter: Option<JoinFilter>,
    join_type: JoinType,
    right: SendableRecordBatchStream,
    random_state: RandomState,
    join_metrics: BuildProbeJoinMetrics,
    column_indices: Vec<ColumnIndex>,
    null_equality: NullEquality,
    state: HashJoinStreamState,
    build_side: BuildSide,
    batch_size: usize,
    hashes_buffer: Vec<u64>,
    right_side_ordered: bool,
    bounds_accumulator: Option<Arc<SharedBoundsAccumulator>>,
    bounds_waiter: Option<OnceFut<()>>,
    mode: PartitionMode,
}

[Stream] for super::HashJoinExec that does the actual join.

This stream:

• Collecting the build side (left input) into a hash map
• Iterating over the probe side (right input) in streaming fashion
• Looking up matches against the hash table and applying join filters
• Producing joined [RecordBatch]es incrementally
• Emitting unmatched rows for outer/semi/anti joins in the final stage

Fields

partition: usize

Partition identifier for debugging and determinism

schema: Arc<Schema>

Input schema

on_right: Vec<PhysicalExprRef>

equijoin columns from the right (probe side)

filter: Option<JoinFilter>

optional join filter

join_type: JoinType

type of the join (left, right, semi, etc)

right: SendableRecordBatchStream

right (probe) input

random_state: RandomState

Random state used for hashing initialization

join_metrics: BuildProbeJoinMetrics

Metrics

column_indices: Vec<ColumnIndex>

Information of index and left / right placement of columns

null_equality: NullEquality

Defines the null equality for the join.

state: HashJoinStreamState

State of the stream

build_side: BuildSide

Build side

batch_size: usize

Maximum output batch size

hashes_buffer: Vec<u64>

Scratch space for computing hashes

right_side_ordered: bool

Specifies whether the right side has an ordering to potentially preserve

bounds_accumulator: Option<Arc<SharedBoundsAccumulator>>

Shared bounds accumulator for coordinating dynamic filter updates (optional)

bounds_waiter: Option<OnceFut<()>>

Optional future to signal when bounds have been reported by all partitions and the dynamic filter has been updated

mode: PartitionMode

Partitioning mode to use

Implementations

impl HashJoinStream

pub(super) fn new( partition: usize, schema: Arc<Schema>, on_right: Vec<PhysicalExprRef>, filter: Option<JoinFilter>, join_type: JoinType, right: SendableRecordBatchStream, random_state: RandomState, join_metrics: BuildProbeJoinMetrics, column_indices: Vec<ColumnIndex>, null_equality: NullEquality, state: HashJoinStreamState, build_side: BuildSide, batch_size: usize, hashes_buffer: Vec<u64>, right_side_ordered: bool, bounds_accumulator: Option<Arc<SharedBoundsAccumulator>>, mode: PartitionMode, ) -> Self

fn poll_next_impl( &mut self, cx: &mut Context<'_>, ) -> Poll<Option<Result<RecordBatch>>>

Separate implementation function that unpins the HashJoinStream so that partial borrows work correctly

fn wait_for_partition_bounds_report( &mut self, cx: &mut Context<'_>, ) -> Poll<Result<StatefulStreamResult<Option<RecordBatch>>>>

Optional step to wait until bounds have been reported by all partitions. This state is only entered if a bounds accumulator is present.

Why wait?

The dynamic filter is only built once all partitions have reported their bounds. If we do not wait here, the probe-side scan may start before the filter is ready. This can lead to the probe-side scan missing the opportunity to apply the filter and skip reading unnecessary data.

fn collect_build_side( &mut self, cx: &mut Context<'_>, ) -> Poll<Result<StatefulStreamResult<Option<RecordBatch>>>>

Collects build-side data by polling OnceFut future from initialized build-side

Updates build-side to Ready, and state to FetchProbeSide

fn fetch_probe_batch( &mut self, cx: &mut Context<'_>, ) -> Poll<Result<StatefulStreamResult<Option<RecordBatch>>>>

Fetches next batch from probe-side

If non-empty batch has been fetched, updates state to ProcessProbeBatchState, otherwise updates state to ExhaustedProbeSide

fn process_probe_batch( &mut self, ) -> Result<StatefulStreamResult<Option<RecordBatch>>>

Joins current probe batch with build-side data and produces batch with matched output

Updates state to FetchProbeBatch

fn process_unmatched_build_batch( &mut self, ) -> Result<StatefulStreamResult<Option<RecordBatch>>>

Processes unmatched build-side rows for certain join types and produces output batch

Updates state to Completed

Trait Implementations

impl RecordBatchStream for HashJoinStream

fn schema(&self) -> SchemaRef

Returns the schema of this RecordBatchStream.

impl Stream for HashJoinStream

type Item = Result<RecordBatch, DataFusionError>

Values yielded by the stream.

fn poll_next( self: Pin<&mut Self>, cx: &mut Context<'_>, ) -> Poll<Option<Self::Item>>

Attempt to pull out the next value of this stream, registering the current task for wakeup if the value is not yet available, and returning None if the stream is exhausted.

fn size_hint(&self) -> (usize, Option<usize>)

Returns the bounds on the remaining length of the stream.