Struct StateTableWatermarkCache

pub struct StateTableWatermarkCache {
    inner: TopNStateCache<DefaultOrdered<OwnedRow>, ()>,
}

Cache Invariants

• If cache is synced, it will ALWAYS contain TopN entries. NOTE: TopN here refers to the lowest N entries, where the first is the minimum and so on.
• Cache will not contain NULL values in the watermark column, they are ignored in watermark state cleaning. They will not be included in the cache’s table row count either, since they are treated as invisible.

Updates to Cache

INSERT A. Cache evicted. Update cached value. B. Cache uninitialized. Initialize cache, insert into TopNCache. C. Cache not empty. Insert into TopNCache.

DELETE A. Matches lowest value pk. Remove lowest value. Mark cache as Evicted. Later on Barrier we will refresh the cache with table scan. Since on barrier we will clean up all values before watermark, We have less rows to scan. B. Does not match. Do nothing.

UPDATE Nothing. Watermark is part of pk. Pk won’t change.

BARRIER State table commit. See below.

STATE TABLE COMMIT A. Decide whether to do state cleaning: if watermark_to_be_cleaned < smallest val OR no value in cache + cache is synced: No need issue delete range. if watermark_to_be_cleaned => smallest val OR cache not synced: Issue delete ranges.

B. Refreshing the cache: On barrier, do table scan from most_recently_cleaned_watermark (inclusive) to +inf. Take the Top N rows and insert into cache. This has to be implemented in state_table. We do not need to store any values, just the keys.

TODO(kwannoel): Optimization: If cache is not full, we can also do point delete for each cache entry. Not sure if this is more optimal, we have to measure this scenario to see if it is indeed better.

Fields

inner: TopNStateCache<DefaultOrdered<OwnedRow>, ()>

Implementations

impl StateTableWatermarkCache

pub fn new(size: usize) -> Self

fn new_with_row_count(size: usize, row_count: usize) -> Self

NOTE(kwannoel): Unused. Requires row count from State Table. On first initialization, we can use row_count 0. But if state table is reconstructed after recovery, we need to obtain row count meta-data.

fn first_key(&self) -> Option<&DefaultOrdered<OwnedRow>>

Get the lowest key.

pub fn lowest_key(&self) -> Option<ScalarRefImpl<'_>>

pub fn insert(&mut self, key: &impl Row)

Insert a new value.

pub fn delete(&mut self, key: &impl Row)

Delete a value If the watermark col is NULL, it will just be ignored.

pub fn capacity(&self) -> usize

pub fn len(&self) -> usize

pub fn set_table_row_count(&mut self, table_row_count: usize)

Trait Implementations

impl Clone for StateTableWatermarkCache

fn clone(&self) -> StateTableWatermarkCache

Returns a copy of the value.

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

Performs copy-assignment from source.

impl EstimateSize for StateTableWatermarkCache

fn estimated_heap_size(&self) -> usize

The estimated heap size of the current struct in bytes.

fn estimated_size(&self) -> usize

where Self: Sized,

The estimated total size of the current struct in bytes, including the estimated_heap_size and the size of Self.

impl StateCache for StateTableWatermarkCache

type Filler<'a> = &'a mut TopNStateCache<DefaultOrdered<OwnedRow>, ()>

Type of state cache filler, for syncing the cache with the state table.

type Key = DefaultOrdered<OwnedRow>

type Value = ()

fn is_synced(&self) -> bool

Check if the cache is synced with the state table.

fn begin_syncing(&mut self) -> Self::Filler<'_>

Begin syncing the cache with the state table.

fn insert(&mut self, key: Self::Key, value: Self::Value) -> Option<Self::Value>

Insert an entry into the cache. Should not break cache validity.

fn delete(&mut self, key: &Self::Key) -> Option<Self::Value>

Delete an entry from the cache. Should not break cache validity.

fn apply_batch( &mut self, batch: impl IntoIterator<Item = (Op, Self::Key, Self::Value)>, )

Apply a batch of operations to the cache. Should not break cache validity.

fn clear(&mut self)

Clear the cache.

fn values(&self) -> impl Iterator<Item = &Self::Value>

Iterate over the values in the cache.

fn first_key_value(&self) -> Option<(&Self::Key, &Self::Value)>

Get the reference of first key-value pair in the cache.