Struct datafusion_physical_plan::joins::SymmetricHashJoinExec

pub struct SymmetricHashJoinExec { /* private fields */ }

A symmetric hash join with range conditions is when both streams are hashed on the join key and the resulting hash tables are used to join the streams. The join is considered symmetric because the hash table is built on the join keys from both streams, and the matching of rows is based on the values of the join keys in both streams. This type of join is efficient in streaming context as it allows for fast lookups in the hash table, rather than having to scan through one or both of the streams to find matching rows, also it only considers the elements from the stream that fall within a certain sliding window (w/ range conditions), making it more efficient and less likely to store stale data. This enables operating on unbounded streaming data without any memory issues.

For each input stream, create a hash table.

• For each new RecordBatch in build side, hash and insert into inputs hash table. Update offsets.
• Test if input is equal to a predefined set of other inputs.
• If so record the visited rows. If the matched row results must be produced (INNER, LEFT), output the RecordBatch.
• Try to prune other side (probe) with new RecordBatch.
• If the join type indicates that the unmatched rows results must be produced (LEFT, FULL etc.), output the RecordBatch when a pruning happens or at the end of the data.

                       +-------------------------+
                       |                         |
  left stream ---------|  Left OneSideHashJoiner |---+
                       |                         |   |
                       +-------------------------+   |
                                                     |
                                                     |--------- Joined output
                                                     |
                       +-------------------------+   |
                       |                         |   |
 right stream ---------| Right OneSideHashJoiner |---+
                       |                         |
                       +-------------------------+

Prune build side when the new RecordBatch comes to the probe side. We utilize interval arithmetic
on JoinFilter's sorted PhysicalExprs to calculate the joinable range.


              PROBE SIDE          BUILD SIDE
                BUFFER              BUFFER
            +-------------+     +------------+
            |             |     |            |    Unjoinable
            |             |     |            |    Range
            |             |     |            |
            |             |  |---------------------------------
            |             |  |  |            |
            |             |  |  |            |
            |             | /   |            |
            |             | |   |            |
            |             | |   |            |
            |             | |   |            |
            |             | |   |            |
            |             | |   |            |    Joinable
            |             |/    |            |    Range
            |             ||    |            |
            |+-----------+||    |            |
            || Record    ||     |            |
            || Batch     ||     |            |
            |+-----------+||    |            |
            +-------------+\    +------------+
                            |
                            \
                             |---------------------------------

 This happens when range conditions are provided on sorted columns. E.g.

       SELECT * FROM left_table, right_table
       ON
         left_key = right_key AND
         left_time > right_time - INTERVAL 12 MINUTES AND left_time < right_time + INTERVAL 2 HOUR

or
      SELECT * FROM left_table, right_table
       ON
         left_key = right_key AND
         left_sorted > right_sorted - 3 AND left_sorted < right_sorted + 10

For general purpose, in the second scenario, when the new data comes to probe side, the conditions can be used to
determine a specific threshold for discarding rows from the inner buffer. For example, if the sort order the
two columns ("left_sorted" and "right_sorted") are ascending (it can be different in another scenarios)
and the join condition is "left_sorted > right_sorted - 3" and the latest value on the right input is 1234, meaning
that the left side buffer must only keep rows where "leftTime > rightTime - 3 > 1234 - 3 > 1231" ,
making the smallest value in 'left_sorted' 1231 and any rows below (since ascending)
than that can be dropped from the inner buffer.

Implementations

impl SymmetricHashJoinExec

pub fn try_new( left: Arc<dyn ExecutionPlan>, right: Arc<dyn ExecutionPlan>, on: JoinOn, filter: Option<JoinFilter>, join_type: &JoinType, null_equals_null: bool, left_sort_exprs: Option<Vec<PhysicalSortExpr>>, right_sort_exprs: Option<Vec<PhysicalSortExpr>>, mode: StreamJoinPartitionMode, ) -> Result<Self>

Tries to create a new SymmetricHashJoinExec.

Error

This function errors when:

• It is not possible to join the left and right sides on keys on, or
• It fails to construct SortedFilterExprs, or
• It fails to create the ExprIntervalGraph.

pub fn left(&self) -> &Arc<dyn ExecutionPlan>

left stream

pub fn right(&self) -> &Arc<dyn ExecutionPlan>

right stream

pub fn on(&self) -> &[(PhysicalExprRef, PhysicalExprRef)]

Set of common columns used to join on

pub fn filter(&self) -> Option<&JoinFilter>

Filters applied before join output

pub fn join_type(&self) -> &JoinType

How the join is performed

pub fn null_equals_null(&self) -> bool

Get null_equals_null

pub fn partition_mode(&self) -> StreamJoinPartitionMode

Get partition mode

pub fn left_sort_exprs(&self) -> Option<&[PhysicalSortExpr]>

Get left_sort_exprs

pub fn right_sort_exprs(&self) -> Option<&[PhysicalSortExpr]>

Get right_sort_exprs

pub fn check_if_order_information_available(&self) -> Result<bool>

Check if order information covers every column in the filter expression.

Trait Implementations

impl Debug for SymmetricHashJoinExec

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl DisplayAs for SymmetricHashJoinExec

fn fmt_as(&self, t: DisplayFormatType, f: &mut Formatter<'_>) -> Result

Format according to DisplayFormatType, used when verbose representation looks different from the default one

impl ExecutionPlan for SymmetricHashJoinExec

fn name(&self) -> &'static str

Short name for the ExecutionPlan, such as ‘ParquetExec’.

fn as_any(&self) -> &dyn Any

Returns the execution plan as Any so that it can be downcast to a specific implementation.

fn properties(&self) -> &PlanProperties

Return properties of the output of the ExecutionPlan, such as output ordering(s), partitioning information etc.

fn required_input_distribution(&self) -> Vec<Distribution>

Specifies the data distribution requirements for all the children for this ExecutionPlan, By default it’s [Distribution::UnspecifiedDistribution] for each child,

fn required_input_ordering(&self) -> Vec<Option<Vec<PhysicalSortRequirement>>>

Specifies the ordering required for all of the children of this ExecutionPlan.

fn children(&self) -> Vec<&Arc<dyn ExecutionPlan>>

Get a list of children ExecutionPlans that act as inputs to this plan. The returned list will be empty for leaf nodes such as scans, will contain a single value for unary nodes, or two values for binary nodes (such as joins).

fn with_new_children( self: Arc<Self>, children: Vec<Arc<dyn ExecutionPlan>>, ) -> Result<Arc<dyn ExecutionPlan>>

Returns a new ExecutionPlan where all existing children were replaced by the children, in order

fn metrics(&self) -> Option<MetricsSet>

Return a snapshot of the set of Metrics for this ExecutionPlan. If no Metrics are available, return None.

fn statistics(&self) -> Result<Statistics>

Returns statistics for this ExecutionPlan node. If statistics are not available, should return Statistics::new_unknown (the default), not an error.

fn execute( &self, partition: usize, context: Arc<TaskContext>, ) -> Result<SendableRecordBatchStream>

Begin execution of partition, returning a Stream of RecordBatches.

fn static_name() -> &'static str

where Self: Sized,

Short name for the ExecutionPlan, such as ‘ParquetExec’. Like name but can be called without an instance.

fn schema(&self) -> SchemaRef

Get the schema for this execution plan

fn maintains_input_order(&self) -> Vec<bool>

Returns false if this ExecutionPlan’s implementation may reorder rows within or between partitions.

fn benefits_from_input_partitioning(&self) -> Vec<bool>

Specifies whether the ExecutionPlan benefits from increased parallelization at its input for each child.

fn repartitioned( &self, _target_partitions: usize, _config: &ConfigOptions, ) -> Result<Option<Arc<dyn ExecutionPlan>>>

If supported, attempt to increase the partitioning of this ExecutionPlan to produce target_partitions partitions.

fn supports_limit_pushdown(&self) -> bool

Returns true if a limit can be safely pushed down through this ExecutionPlan node.

fn with_fetch(&self, _limit: Option<usize>) -> Option<Arc<dyn ExecutionPlan>>

Returns a fetching variant of this ExecutionPlan node, if it supports fetch limits. Returns None otherwise.