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.

The degree of each row will tell us if we need to emit NULL for the row. For instance, given lhs LEFT JOIN rhs, If the degree of a row in lhs is 0, it means the row does not have a match in rhs. If the degree of a row in lhs is 2, it means the row has two matches in rhs. Now, when emitting the result of the join, we need to emit NULL for the row in lhs if the degree is 0.

Why don’t just use a boolean value instead of a degree count? Consider the case where we delete a matched record from rhs. Since we can delete a record, there must have been a record in rhs that matched the record in lhs. So this value is true. But we don’t know how many records are matched after removing this record, since we only stored a boolean value rather than the count. Hence we need to store the count of matched records.

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(crate) fn get_degree_state_mut_ref( &mut self, ) -> (&[usize], &mut Option<TableInner<S>>)

pub(crate) async fn fetch_matched_rows_and_get_degree_table_ref<'a>( &'a mut self, key: &'a K, ) -> StreamExecutorResult<(impl Stream<Item = StreamExecutorResult<(Vec<u8>, JoinRow<OwnedRow>)>> + 'a, &'a [usize], &'a mut Option<TableInner<S>>)>

NOTE(kwannoel): This allows us to concurrently stream records from the state_table, and update the degree table, without using unsafe code.

This is because we obtain separate references to separate parts of the JoinHashMap, instead of reusing the same reference to JoinHashMap for concurrent read access to state_table, and write access to the degree table.

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<()>

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

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

pub fn take_state_opt(&mut self, key: &K) -> CacheResult

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.

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<JoinHashMapPostCommit<'_, K, S>>

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

pub fn insert_handle_degree( &mut self, key: &K, value: JoinRow<impl Row>, ) -> 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_handle_degree( &mut self, key: &K, value: JoinRow<impl Row>, ) -> StreamExecutorResult<()>

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.

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>