Struct JoinHashMap

pub struct JoinHashMap<K: HashKey, S: StateStore> {
    inner: ManagedLruCache<K, HashValueWrapper, PrecomputedBuildHasher, SharedStatsAlloc<Global>>,
    join_key_data_types: Vec<DataType>,
    null_matched: K::Bitmap,
    pk_serializer: OrderedRowSerde,
    state: TableInner<S>,
    degree_state: Option<TableInner<S>>,
    need_degree_table: bool,
    pk_contained_in_jk: bool,
    inequality_key_desc: Option<InequalityKeyDesc>,
    metrics: JoinHashMapMetrics,
}

Fields

inner: ManagedLruCache<K, HashValueWrapper, PrecomputedBuildHasher, SharedStatsAlloc<Global>>

Store the join states.

join_key_data_types: Vec<DataType>

Data types of the join key columns

null_matched: K::Bitmap

Null safe bitmap for each join pair

pk_serializer: OrderedRowSerde

The memcomparable serializer of primary key.

state: TableInner<S>

State table. Contains the data from upstream.

degree_state: Option<TableInner<S>>

Degree table.

The degree is generated from the hash join executor. Each row in state has a corresponding degree in degree state. A degree value d in for a row means the row has d matched row in the other join side.

It will only be used when needed in a side.

• Full Outer: both side
• Left Outer/Semi/Anti: left side
• Right Outer/Semi/Anti: right side
• Inner: neither side.

Should be set to None if need_degree_table was set to false.

need_degree_table: bool

If degree table is need

pk_contained_in_jk: bool

Pk is part of the join key.

inequality_key_desc: Option<InequalityKeyDesc>

Inequality key description for AsOf join.

metrics: JoinHashMapMetrics

Metrics of the hash map

Implementations

impl<K: HashKey, S: StateStore> JoinHashMap<K, S>

pub fn new( watermark_sequence: AtomicU64Ref, join_key_data_types: Vec<DataType>, state_join_key_indices: Vec<usize>, state_all_data_types: Vec<DataType>, state_table: StateTable<S>, state_pk_indices: Vec<usize>, degree_state: Option<TableInner<S>>, null_matched: K::Bitmap, pk_contained_in_jk: bool, inequality_key_idx: Option<usize>, metrics: Arc<StreamingMetrics>, actor_id: ActorId, fragment_id: FragmentId, side: &'static str, ) -> Self

Create a JoinHashMap with the given LRU capacity.

pub async fn init(&mut self, epoch: EpochPair) -> StreamExecutorResult<()>

pub fn update_vnode_bitmap(&mut self, vnode_bitmap: Arc<Bitmap>) -> bool

Update the vnode bitmap and manipulate the cache if necessary.

pub fn update_watermark(&mut self, watermark: ScalarImpl)

pub async fn take_state<'a>( &mut self, key: &K, ) -> StreamExecutorResult<Box<JoinEntryState>>

Take the state for the given key out of the hash table and return it. One MUST call update_state after some operations to put the state back.

If the state does not exist in the cache, fetch the remote storage and return. If it still does not exist in the remote storage, a JoinEntryState with empty cache will be returned.

Note: This will NOT remove anything from remote storage.

async fn fetch_cached_state( &self, key: &K, ) -> StreamExecutorResult<JoinEntryState>

Fetch cache from the state store. Should only be called if the key does not exist in memory. Will return a empty JoinEntryState even when state does not exist in remote.

pub async fn flush(&mut self, epoch: EpochPair) -> StreamExecutorResult<()>

pub async fn try_flush(&mut self) -> StreamExecutorResult<()>

pub fn insert( &mut self, key: &K, value: JoinRow<impl Row>, ) -> StreamExecutorResult<()>

Insert a join row

pub fn insert_row( &mut self, key: &K, value: impl Row, ) -> StreamExecutorResult<()>

Insert a row. Used when the side does not need to update degree.

pub fn delete( &mut self, key: &K, value: JoinRow<impl Row>, ) -> StreamExecutorResult<()>

Delete a join row

pub fn delete_row( &mut self, key: &K, value: impl Row, ) -> StreamExecutorResult<()>

Delete a row Used when the side does not need to update degree.

pub fn update_state(&mut self, key: &K, state: Box<JoinEntryState>)

Update a JoinEntryState into the hash table.

fn manipulate_degree( &mut self, join_row_ref: &mut EncodedJoinRow, join_row: &mut JoinRow<OwnedRow>, action: impl Fn(&mut u64), )

Manipulate the degree of the given JoinRow and EncodedJoinRow with action, both in memory and in the degree table.

pub fn inc_degree( &mut self, join_row_ref: &mut EncodedJoinRow, join_row: &mut JoinRow<OwnedRow>, )

Increment the degree of the given JoinRow and EncodedJoinRow with action, both in memory and in the degree table.

pub fn dec_degree( &mut self, join_row_ref: &mut EncodedJoinRow, join_row: &mut JoinRow<OwnedRow>, )

Decrement the degree of the given JoinRow and EncodedJoinRow with action, both in memory and in the degree table.

pub fn evict(&mut self)

Evict the cache.

pub fn entry_count(&self) -> usize

Cached entry count for this hash table.

pub fn null_matched(&self) -> &K::Bitmap

pub fn table_id(&self) -> u32

pub fn join_key_data_types(&self) -> &[DataType]

pub fn check_inequal_key_null(&self, row: &impl Row) -> bool

Return true if the inequality key is null.

Panics

Panics if the inequality key is not set.

pub fn serialize_inequal_key_from_row(&self, row: impl Row) -> Vec<u8>

Serialize the inequality key from a row.

Panics

Panics if the inequality key is not set.

pub fn serialize_pk_from_row(&self, row: impl Row) -> Vec<u8>