Struct NestedLoopJoinStream 

pub(crate) struct NestedLoopJoinStream {

    pub(crate) output_schema: Arc<Schema>,
    pub(crate) join_filter: Option<JoinFilter>,
    pub(crate) join_type: JoinType,
    pub(crate) right_data: SendableRecordBatchStream,
    pub(crate) left_data: OnceFut<JoinLeftData>,
    pub(crate) column_indices: Vec<ColumnIndex>,
    pub(crate) metrics: NestedLoopJoinMetrics,
    batch_size: usize,
    should_track_unmatched_right: bool,
    state: NLJState,
    output_buffer: Box<BatchCoalescer>,
    handled_empty_output: bool,
    buffered_left_data: Option<Arc<JoinLeftData>>,
    left_probe_idx: usize,
    left_emit_idx: usize,
    left_exhausted: bool,
    left_buffered_in_one_pass: bool,
    current_right_batch: Option<RecordBatch>,
    current_right_batch_matched: Option<BooleanArray>,
}

Fields

output_schema: Arc<Schema>

Output schema

join_filter: Option<JoinFilter>

join filter

join_type: JoinType

type of the join

right_data: SendableRecordBatchStream

the probe-side(right) table data of the nested loop join

left_data: OnceFut<JoinLeftData>

the build-side table data of the nested loop join

column_indices: Vec<ColumnIndex>

Projection to construct the output schema from the left and right tables. Example:

• output_schema: [‘a’, ‘c’]
• left_schema: [‘a’, ‘b’]
• right_schema: [‘c’]

The column indices would be [(left, 0), (right, 0)] – taking the left 0th column and right 0th column can construct the output schema.

Note there are other columns (‘b’ in the example) still kept after projection pushdown; this is because they might be used to evaluate the join filter (e.g., JOIN ON (b+c)>0).

metrics: NestedLoopJoinMetrics

Join execution metrics

batch_size: usize

batch_size from configuration

should_track_unmatched_right: bool

See comments in need_produce_right_in_final for more detail

state: NLJState

State Tracking

output_buffer: Box<BatchCoalescer>

Output buffer holds the join result to output. It will emit eagerly when the threshold is reached.

handled_empty_output: bool

See comments in NLJState::Done for its purpose

buffered_left_data: Option<Arc<JoinLeftData>>

The current buffered left data to join

left_probe_idx: usize

Index into the left buffered batch. Used in ProbeRight state

left_emit_idx: usize

Index into the left buffered batch. Used in EmitLeftUnmatched state

left_exhausted: bool

Should we go back to BufferingLeft state again after EmitLeftUnmatched state is over.

left_buffered_in_one_pass: bool

If we can buffer all left data in one pass TODO(now): this is for the (unimplemented) memory-limited execution

current_right_batch: Option<RecordBatch>

The current probe batch to process

current_right_batch_matched: Option<BooleanArray>

Implementations

impl NestedLoopJoinStream

pub(crate) fn new( schema: Arc<Schema>, filter: Option<JoinFilter>, join_type: JoinType, right_data: SendableRecordBatchStream, left_data: OnceFut<JoinLeftData>, column_indices: Vec<ColumnIndex>, metrics: NestedLoopJoinMetrics, batch_size: usize, ) -> Self

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

Handle BufferingLeft state - prepare left side batches

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

Handle FetchingRight state - fetch next right batch and prepare for processing

fn handle_probe_right( &mut self, ) -> ControlFlow<Poll<Option<Result<RecordBatch>>>>

Handle ProbeRight state - process current probe batch

fn handle_emit_right_unmatched( &mut self, ) -> ControlFlow<Poll<Option<Result<RecordBatch>>>>

Handle EmitRightUnmatched state - emit unmatched right rows

fn handle_emit_left_unmatched( &mut self, ) -> ControlFlow<Poll<Option<Result<RecordBatch>>>>

Handle EmitLeftUnmatched state - emit unmatched left rows

fn handle_done(&mut self) -> Poll<Option<Result<RecordBatch>>>

Handle Done state - final state processing

fn process_probe_batch(&mut self) -> Result<bool>

Returns bool to indicate should it continue probing true -> continue in the same ProbeRight state false -> It has done with the (buffered_left x cur_right_batch), go to next state (ProbeRight)

fn process_single_left_row_join( &mut self, left_data: &JoinLeftData, right_batch: &RecordBatch, l_index: usize, ) -> Result<Option<RecordBatch>>

Process a single left row join with the current right batch. Returns a RecordBatch containing the join results (None if empty)

fn process_left_unmatched(&mut self) -> Result<bool>

Returns bool to indicate should it continue processing unmatched rows true -> continue in the same EmitLeftUnmatched state false -> next state (Done)

fn process_left_unmatched_range( &self, left_data: &JoinLeftData, start_idx: usize, end_idx: usize, ) -> Result<Option<RecordBatch>>

Process unmatched rows from the left data within the specified range. Returns a RecordBatch containing the unmatched rows (None if empty).

Arguments

• left_data - The left side data containing the batch and bitmap
• start_idx - Start index (inclusive) of the range to process
• end_idx - End index (exclusive) of the range to process

Safety

The caller is responsible for ensuring that start_idx and end_idx are within valid bounds of the left batch. This function does not perform bounds checking.

fn process_right_unmatched(&mut self) -> Result<Option<RecordBatch>>

Process unmatched rows from the current right batch and reset the bitmap. Returns a RecordBatch containing the unmatched right rows (None if empty).

fn get_left_data(&self) -> Result<&Arc<JoinLeftData>>

Get the build-side data of the left input, errors if it’s None

fn maybe_flush_ready_batch( &mut self, ) -> Option<Poll<Option<Result<RecordBatch>>>>

Flush the output_buffer if there are batches ready to output None if no result batch ready.

fn update_matched_bitmap( &mut self, l_index: usize, r_matched_bitmap: &BooleanArray, ) -> Result<()>

After joining (l_index@left_buffer x current_right_batch), it will result in a bitmap (the same length as current_right_batch) as the join match result. Use this bitmap to update the global bitmap, for special join types like full joins.

Example: After joining l_index=1 (1-indexed row in the left buffer), and the current right batch with 3 elements, this function will be called with arguments: l_index = 1, r_matched = [false, false, true]

• If the join type is FullJoin, the 1-index in the left bitmap will be set to true, and also the right bitmap will be bitwise-ORed with the input r_matched bitmap.
• For join types that don’t require output unmatched rows, this function can be a no-op. For inner joins, this function is a no-op; for left joins, only the left bitmap may be updated.

Trait Implementations

impl RecordBatchStream for NestedLoopJoinStream

fn schema(&self) -> SchemaRef

Returns the schema of this RecordBatchStream.

impl Stream for NestedLoopJoinStream

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

See the comments NestedLoopJoinExec for high-level design ideas.

Implementation

This function is the entry point of NLJ operator’s state machine transitions. The rough state transition graph is as follow, for more details see the comment in each state’s matching arm.

============================ State transition graph:

(start) –> BufferingLeft

BufferingLeft → FetchingRight

FetchingRight → ProbeRight (if right batch available) FetchingRight → EmitLeftUnmatched (if right exhausted)

ProbeRight → ProbeRight (next left row or after yielding output) ProbeRight → EmitRightUnmatched (for special join types like right join) ProbeRight → FetchingRight (done with the current right batch)

EmitRightUnmatched → FetchingRight

EmitLeftUnmatched → EmitLeftUnmatched (only process 1 chunk for each iteration) EmitLeftUnmatched → Done (if finished)

Done → (end)

type Item = Result<RecordBatch, DataFusionError>

Values yielded by the stream.

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

Returns the bounds on the remaining length of the stream.