Struct StreamHashJoin

pub struct StreamHashJoin {
    pub base: PlanBase<Stream>,
    core: Join<PlanRef>,
    eq_join_predicate: EqJoinPredicate,
    inequality_pairs: Vec<(bool, InequalityInputPair)>,
    is_append_only: bool,
    clean_left_state_conjunction_idx: Option<usize>,
    clean_right_state_conjunction_idx: Option<usize>,
}

StreamHashJoin implements super::LogicalJoin with hash table. It builds a hash table from inner (right-side) relation and probes with data from outer (left-side) relation to get output rows.

Fields

base: PlanBase<Stream>

core: Join<PlanRef>

eq_join_predicate: EqJoinPredicate

The join condition must be equivalent to logical.on, but separated into equal and non-equal parts to facilitate execution later

inequality_pairs: Vec<(bool, InequalityInputPair)>

(do_state_cleaning, InequalityInputPair {key_required_larger, key_required_smaller, delta_expression}). View struct InequalityInputPair for details.

is_append_only: bool

Whether can optimize for append-only stream. It is true if input of both side is append-only

clean_left_state_conjunction_idx: Option<usize>

The conjunction index of the inequality which is used to clean left state table in HashJoinExecutor. If any equal condition is able to clean state table, this field will always be None.

clean_right_state_conjunction_idx: Option<usize>

The conjunction index of the inequality which is used to clean right state table in HashJoinExecutor. If any equal condition is able to clean state table, this field will always be None.

Implementations

impl StreamHashJoin

pub fn new(core: Join<PlanRef>, eq_join_predicate: EqJoinPredicate) -> Self

pub fn join_type(&self) -> JoinType

Get join type

pub fn eq_join_predicate(&self) -> &EqJoinPredicate

Get a reference to the hash join’s eq join predicate.

pub fn into_delta_join(self) -> StreamDeltaJoin

Convert this hash join to a delta join plan

pub fn derive_dist_key_in_join_key(&self) -> Vec<usize>

pub fn inequality_pairs(&self) -> &Vec<(bool, InequalityInputPair)>

Methods from Deref<Target = PlanBase<Stream>>

pub fn clone_with_new_plan_id(&self) -> Self

pub fn clone_with_new_distribution(&self, dist: Distribution) -> Self

Clone the plan node with a new distribution.

Panics if the plan node is not physical.

Trait Implementations

impl Clone for StreamHashJoin

fn clone(&self) -> StreamHashJoin

Returns a copy of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl ColPrunable for StreamHashJoin

fn prune_col( &self, _required_cols: &[usize], _ctx: &mut ColumnPruningContext, ) -> PlanRef

Transform the plan node to only output the required columns ordered by index number.

impl Debug for StreamHashJoin

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

Formats the value using the given formatter.

impl Deref for StreamHashJoin

type Target = PlanBase<Stream>

The resulting type after dereferencing.

fn deref(&self) -> &Self::Target

Dereferences the value.

impl Distill for StreamHashJoin

fn distill<'a>(&self) -> XmlNode<'a>

fn distill_to_string(&self) -> String

impl ExprRewritable for StreamHashJoin

fn has_rewritable_expr(&self) -> bool

fn rewrite_exprs(&self, r: &mut dyn ExprRewriter) -> PlanRef

impl ExprVisitable for StreamHashJoin

fn visit_exprs(&self, v: &mut dyn ExprVisitor)

impl Hash for StreamHashJoin

fn hash<__H: Hasher>(&self, state: &mut __H)

Feeds this value into the given Hasher.

fn hash_slice<H>(data: &[Self], state: &mut H)

where H: Hasher, Self: Sized,

Feeds a slice of this type into the given Hasher.

impl PartialEq for StreamHashJoin

fn eq(&self, other: &StreamHashJoin) -> bool

Tests for self and other values to be equal, and is used by ==.

fn ne(&self, other: &Rhs) -> bool

Tests for !=. The default implementation is almost always sufficient, and should not be overridden without very good reason.

impl PlanNodeMeta for StreamHashJoin

const NODE_TYPE: PlanNodeType = PlanNodeType::StreamHashJoin

type Convention = Stream

fn plan_base(&self) -> &PlanBase<Stream>

Get the reference to the PlanBase with corresponding convention.

fn plan_base_ref(&self) -> PlanBaseRef<'_>

Get the reference to the PlanBase with erased convention.

impl PlanTreeNode for StreamHashJoin

fn inputs(&self) -> SmallVec<[PlanRef; 2]>

Get input nodes of the plan.

fn clone_with_inputs(&self, inputs: &[PlanRef]) -> PlanRef

Clone the node with a list of new inputs.

impl PlanTreeNodeBinary for StreamHashJoin

fn left(&self) -> PlanRef

fn right(&self) -> PlanRef

fn clone_with_left_right(&self, left: PlanRef, right: PlanRef) -> Self

fn rewrite_with_left_right( &self, _left: PlanRef, _left_col_change: ColIndexMapping, _right: PlanRef, _right_col_change: ColIndexMapping, ) -> (Self, ColIndexMapping)

where Self: Sized,

Rewrites the plan node according to the schema change of its input nodes during rewriting.

impl PredicatePushdown for StreamHashJoin

fn predicate_pushdown( &self, _predicate: Condition, _ctx: &mut PredicatePushdownContext, ) -> PlanRef

Push predicate down for every logical plan node.

impl StreamNode for StreamHashJoin

fn to_stream_prost_body(&self, state: &mut BuildFragmentGraphState) -> NodeBody

impl ToBatch for StreamHashJoin

fn to_batch(&self) -> Result<PlanRef, RwError>

to_batch is equivalent to to_batch_with_order_required(&Order::any())

fn to_batch_with_order_required( &self, required_order: &Order, ) -> Result<PlanRef, RwError>

convert the plan to batch physical plan and satisfy the required Order

impl ToDistributedBatch for StreamHashJoin

fn to_distributed(&self) -> Result<PlanRef, RwError>

to_distributed is equivalent to to_distributed_with_required(&Order::any(), &RequiredDist::Any)

fn to_distributed_with_required( &self, required_order: &Order, required_dist: &RequiredDist, ) -> Result<PlanRef, RwError>

insert the exchange in batch physical plan to satisfy the required Distribution and Order.

impl ToLocalBatch for StreamHashJoin

fn to_local(&self) -> Result<PlanRef, RwError>

fn to_local_with_order_required( &self, required_order: &Order, ) -> Result<PlanRef, RwError>

Convert the plan to batch local physical plan and satisfy the required Order

impl ToStream for StreamHashJoin

fn to_stream(&self, _ctx: &mut ToStreamContext) -> Result<PlanRef, RwError>

to_stream is equivalent to to_stream_with_dist_required(RequiredDist::Any)

fn logical_rewrite_for_stream( &self, _ctx: &mut RewriteStreamContext, ) -> Result<(PlanRef, ColIndexMapping), RwError>

logical_rewrite_for_stream will rewrite the logical node, and return (new_plan_node, col_mapping), the col_mapping is for original columns have been changed into some other position.

fn to_stream_with_dist_required( &self, required_dist: &RequiredDist, ctx: &mut ToStreamContext, ) -> Result<PlanRef, RwError>

convert the plan to streaming physical plan and satisfy the required distribution

impl TryToBatchPb for StreamHashJoin

fn try_to_batch_prost_body(&self) -> Result<NodeBody, SchedulerError>

impl Eq for StreamHashJoin

impl PlanNode for StreamHashJoin

impl StructuralPartialEq for StreamHashJoin

impl ToPb for StreamHashJoin