Struct OverPartition

pub(super) struct OverPartition<'a, S: StateStore> {
    deduped_part_key: &'a OwnedRow,
    range_cache: &'a mut EstimatedBTreeMap<Sentinelled<StateKey>, OwnedRow>,
    cache_policy: OverWindowCachePolicy,
    calls: &'a Calls,
    row_conv: RowConverter<'a>,
    stats: OverPartitionStats,
    _phantom: PhantomData<S>,
}

A wrapper of PartitionCache that provides helper methods to manipulate the cache. By putting this type inside private module, we can avoid misuse of the internal fields and methods.

Fields

deduped_part_key: &'a OwnedRow

range_cache: &'a mut EstimatedBTreeMap<Sentinelled<StateKey>, OwnedRow>

cache_policy: OverWindowCachePolicy

calls: &'a Calls

row_conv: RowConverter<'a>

stats: OverPartitionStats

_phantom: PhantomData<S>

Implementations

impl<'a, S: StateStore> OverPartition<'a, S>

pub fn new( deduped_part_key: &'a OwnedRow, cache: &'a mut EstimatedBTreeMap<Sentinelled<StateKey>, OwnedRow>, cache_policy: CachePolicy, calls: &'a Calls, row_conv: RowConverter<'a>, ) -> Self

pub fn summarize(self) -> OverPartitionStats

Get a summary for the execution happened in the OverPartition in current round. This will consume the OverPartition value itself.

pub fn cache_real_len(&self) -> usize

Get the number of cached entries ignoring sentinels.

fn cache_real_first_key(&self) -> Option<&StateKey>

fn cache_real_last_key(&self) -> Option<&StateKey>

fn cache_left_is_sentinel(&self) -> bool

fn cache_right_is_sentinel(&self) -> bool

pub async fn build_changes( &mut self, table: &StateTable<S>, delta: BTreeMap<Sentinelled<StateKey>, Change<OwnedRow>>, ) -> StreamExecutorResult<(BTreeMap<StateKey, Record<OwnedRow>>, Option<RangeInclusive<StateKey>>)>

Build changes for the partition, with the given delta. Necessary maintenance of the range cache will be done during this process, like loading rows from the table into the cache.

pub fn write_record( &mut self, table: &mut StateTable<S>, key: StateKey, record: Record<OwnedRow>, )

Write a change record to state table and cache. This function must be called after finding affected ranges, which means the change records should never exceed the cached range.

async fn find_affected_ranges<'s, 'delta>( &'s mut self, table: &StateTable<S>, delta: &'delta mut BTreeMap<Sentinelled<StateKey>, Change<OwnedRow>>, ) -> StreamExecutorResult<(DeltaBTreeMap<'delta, Sentinelled<StateKey>, OwnedRow>, Vec<AffectedRange<'delta>>)>

where 'a: 'delta, 's: 'delta,

Find all ranges in the partition that are affected by the given delta. The returned ranges are guaranteed to be sorted and non-overlapping. All keys in the ranges are guaranteed to be cached, which means they should be Sentinelled::Normals.

async fn ensure_delta_in_cache( &mut self, table: &StateTable<S>, delta: &mut BTreeMap<Sentinelled<StateKey>, Change<OwnedRow>>, ) -> StreamExecutorResult<()>

fn find_affected_ranges_readonly<'delta>( &self, part_with_delta: DeltaBTreeMap<'delta, Sentinelled<StateKey>, OwnedRow>, range_frame_logical_curr: Option<&(Sentinelled<Datum>, Sentinelled<Datum>)>, ) -> Result<Vec<AffectedRange<'delta>>, (bool, bool)>

Try to find affected ranges on immutable range cache + delta. If the algorithm reaches any sentinel node in the cache, which means some entries in the affected range may be in the state table, it returns an Err((bool, bool)) to notify the caller that the left side or the right side or both sides of the cache should be extended.

TODO(rc): Currently at most one range will be in the result vector. Ideally we should recognize uncontinuous changes in the delta and find multiple ranges, but that will be too complex for now.

async fn extend_cache_to_boundary( &mut self, table: &StateTable<S>, ) -> StreamExecutorResult<()>

async fn extend_cache_by_range( &mut self, table: &StateTable<S>, range: RangeInclusive<&StateKey>, ) -> StreamExecutorResult<()>

Try to load the given range of entries from table into cache. When the function returns, it’s guaranteed that there’s no entry in the table that is within the given range but not in the cache.

async fn extend_cache_leftward_by_n( &mut self, table: &StateTable<S>, hint_key: &StateKey, ) -> StreamExecutorResult<()>

async fn extend_cache_rightward_by_n( &mut self, table: &StateTable<S>, hint_key: &StateKey, ) -> StreamExecutorResult<()>

async fn extend_cache_by_range_inner( &mut self, table: &StateTable<S>, table_sub_range: (Bound<impl Row>, Bound<impl Row>), ) -> StreamExecutorResult<()>

async fn extend_cache_leftward_by_n_inner( &mut self, table: &StateTable<S>, range_to_exclusive: &StateKey, ) -> StreamExecutorResult<()>

async fn extend_cache_rightward_by_n_inner( &mut self, table: &StateTable<S>, range_from_exclusive: &StateKey, ) -> StreamExecutorResult<()>