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 ) -> 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) -> &[(Column, Column)]

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 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 ExecutionPlan for SymmetricHashJoinExec

fn as_any(&self) -> &dyn Any

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

fn schema(&self) -> SchemaRef

Get the schema for this execution plan

fn unbounded_output(&self, children: &[bool]) -> Result<bool>

Specifies whether this plan generates an infinite stream of records. If the plan does not support pipelining, but its input(s) are infinite, returns an error to indicate this.

fn benefits_from_input_partitioning(&self) -> bool

Returns true if this operator would benefit from partitioning its input (and thus from more parallelism). For operators that do very little work the overhead of extra parallelism may outweigh any benefits

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

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

fn output_partitioning(&self) -> Partitioning

Specifies the output partitioning scheme of this plan

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

If the output of this operator within each partition is sorted, returns Some(keys) with the description of how it was sorted.

fn equivalence_properties(&self) -> EquivalenceProperties

Get the EquivalenceProperties within the plan

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

Get a list of child execution plans that provide the input for this plan. The returned list will be empty for leaf nodes, 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 plan where all children were replaced by new plans.

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

Format this ExecutionPlan to f in the specified type.

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

Return a snapshot of the set of Metrics for this ExecutionPlan.

fn statistics(&self) -> Statistics

Returns the global output statistics for this ExecutionPlan node.

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

creates an iterator

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

Specifies the ordering requirements for all of the children For each child, it’s the local ordering requirement within each partition rather than the global ordering

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

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

fn ordering_equivalence_properties(&self) -> OrderingEquivalenceProperties

Get the OrderingEquivalenceProperties within the plan