Struct SymmetricHashJoinStream 

struct SymmetricHashJoinStream<T> {

    left_stream: SendableRecordBatchStream,
    right_stream: SendableRecordBatchStream,
    schema: Arc<Schema>,
    filter: Option<JoinFilter>,
    join_type: JoinType,
    left: OneSideHashJoiner,
    right: OneSideHashJoiner,
    column_indices: Vec<ColumnIndex>,
    graph: Option<ExprIntervalGraph>,
    left_sorted_filter_expr: Option<SortedFilterExpr>,
    right_sorted_filter_expr: Option<SortedFilterExpr>,
    random_state: RandomState,
    null_equality: NullEquality,
    metrics: StreamJoinMetrics,
    reservation: Arc<Mutex<MemoryReservation>>,
    state: SHJStreamState,
    batch_transformer: T,
}

A stream that issues [RecordBatch]es as they arrive from the right of the join.

Fields

left_stream: SendableRecordBatchStream

Input streams

right_stream: SendableRecordBatchStream

schema: Arc<Schema>

Input schema

filter: Option<JoinFilter>

join filter

join_type: JoinType

type of the join

left: OneSideHashJoiner

right: OneSideHashJoiner

right hash joiner

column_indices: Vec<ColumnIndex>

Information of index and left / right placement of columns

graph: Option<ExprIntervalGraph>

left_sorted_filter_expr: Option<SortedFilterExpr>

right_sorted_filter_expr: Option<SortedFilterExpr>

random_state: RandomState

Random state used for hashing initialization

null_equality: NullEquality

Defines the null equality for the join.

metrics: StreamJoinMetrics

Metrics

reservation: Arc<Mutex<MemoryReservation>>

Memory reservation

state: SHJStreamState

State machine for input execution

batch_transformer: T

Transforms the output batch before returning.

Implementations

impl<T: BatchTransformer> SymmetricHashJoinStream<T>

SymmetricHashJoinStream manages incremental join operations between two streams. Unlike traditional join approaches that need to scan one side of the join fully before proceeding, SymmetricHashJoinStream facilitates more dynamic join operations by working with streams as they emit data. This approach allows for more efficient processing, particularly in scenarios where waiting for complete data materialization is not feasible or optimal. The trait provides a framework for handling various states of such a join process, ensuring that join logic is efficiently executed as data becomes available from either stream.

This implementation performs eager joins of data from two different asynchronous streams, typically referred to as left and right streams. The implementation provides a comprehensive set of methods to control and execute the join process, leveraging the states defined in SHJStreamState. Methods are primarily focused on asynchronously fetching data batches from each stream, processing them, and managing transitions between various states of the join.

This implementations use a state machine approach to navigate different stages of the join operation, handling data from both streams and determining when the join completes.

State Transitions:

• From PullLeft to PullRight or LeftExhausted:
  • In fetch_next_from_left_stream, when fetching a batch from the left stream:
    • On success (Some(Ok(batch))), state transitions to PullRight for processing the batch.
    • On error (Some(Err(e))), the error is returned, and the state remains unchanged.
    • On no data (None), state changes to LeftExhausted, returning Continue to proceed with the join process.
• From PullRight to PullLeft or RightExhausted:
  • In fetch_next_from_right_stream, when fetching from the right stream:
    • If a batch is available, state changes to PullLeft for processing.
    • On error, the error is returned without changing the state.
    • If right stream is exhausted (None), state transitions to RightExhausted, with a Continue result.
• Handling RightExhausted and LeftExhausted:
  • Methods handle_right_stream_end and handle_left_stream_end manage scenarios when streams are exhausted:
    • They attempt to continue processing with the other stream.
    • If both streams are exhausted, state changes to BothExhausted { final_result: false }.
• Transition to BothExhausted { final_result: true }:
  • Occurs in prepare_for_final_results_after_exhaustion when both streams are exhausted, indicating completion of processing and availability of final results.

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

Implements the main polling logic for the join stream.

This method continuously checks the state of the join stream and acts accordingly by delegating the handling to appropriate sub-methods depending on the current state.

Arguments

• cx - A context that facilitates cooperative non-blocking execution within a task.

Returns

• Poll<Option<Result<RecordBatch>>> - A polled result, either a RecordBatch or None.

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

Asynchronously pulls the next batch from the right stream.

This default implementation checks for the next value in the right stream. If a batch is found, the state is switched to PullLeft, and the batch handling is delegated to process_batch_from_right. If the stream ends, the state is set to RightExhausted.

Returns

• Result<StatefulStreamResult<Option<RecordBatch>>> - The state result after pulling the batch.

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

Asynchronously pulls the next batch from the left stream.

This default implementation checks for the next value in the left stream. If a batch is found, the state is switched to PullRight, and the batch handling is delegated to process_batch_from_left. If the stream ends, the state is set to LeftExhausted.

Returns

• Result<StatefulStreamResult<Option<RecordBatch>>> - The state result after pulling the batch.

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

Asynchronously handles the scenario when the right stream is exhausted.

In this default implementation, when the right stream is exhausted, it attempts to pull from the left stream. If a batch is found in the left stream, it delegates the handling to process_batch_from_left. If both streams are exhausted, the state is set to indicate both streams are exhausted without final results yet.

Returns

• Result<StatefulStreamResult<Option<RecordBatch>>> - The state result after checking the exhaustion state.

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

Asynchronously handles the scenario when the left stream is exhausted.

When the left stream is exhausted, this default implementation tries to pull from the right stream and delegates the batch handling to process_batch_after_left_end. If both streams are exhausted, the state is updated to indicate so.

Returns

• Result<StatefulStreamResult<Option<RecordBatch>>> - The state result after checking the exhaustion state.

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

Handles the state when both streams are exhausted and final results are yet to be produced.

This default implementation switches the state to indicate both streams are exhausted with final results and then invokes the handling for this specific scenario via process_batches_before_finalization.

Returns

• Result<StatefulStreamResult<Option<RecordBatch>>> - The state result after both streams are exhausted.

fn process_batch_from_right( &mut self, batch: RecordBatch, ) -> Result<StatefulStreamResult<Option<RecordBatch>>>

fn process_batch_from_left( &mut self, batch: RecordBatch, ) -> Result<StatefulStreamResult<Option<RecordBatch>>>

fn process_batch_after_left_end( &mut self, right_batch: RecordBatch, ) -> Result<StatefulStreamResult<Option<RecordBatch>>>

fn process_batch_after_right_end( &mut self, left_batch: RecordBatch, ) -> Result<StatefulStreamResult<Option<RecordBatch>>>

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

fn right_stream(&mut self) -> &mut SendableRecordBatchStream

fn left_stream(&mut self) -> &mut SendableRecordBatchStream

fn set_state(&mut self, state: SHJStreamState)

fn state(&mut self) -> SHJStreamState

fn size(&self) -> usize

fn perform_join_for_given_side( &mut self, probe_batch: RecordBatch, probe_side: JoinSide, ) -> Result<Option<RecordBatch>>

Performs a join operation for the specified probe_side (either left or right). This function:

1. Determines which side is the probe and which is the build side.
2. Updates metrics based on the batch that was polled.
3. Executes the join with the given probe_batch.
4. Optionally computes anti-join results if all conditions are met.
5. Combines the results and returns a combined batch or None if no batch was produced.

Trait Implementations

impl<T: BatchTransformer + Unpin + Send> RecordBatchStream for SymmetricHashJoinStream<T>

fn schema(&self) -> SchemaRef

Returns the schema of this RecordBatchStream.

impl<T: BatchTransformer + Unpin + Send> Stream for SymmetricHashJoinStream<T>

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.