Struct StateTableRowStore 

struct StateTableRowStore<LS: LocalStateStore, SD: ValueRowSerde> {
    state_store: LS,
    all_rows: Option<HashMap<VirtualNode, BTreeMap<Bytes, OwnedRow>>>,
    pending_watermark: Option<(WatermarkDirection, Bytes)>,
    table_id: TableId,
    row_serde: Arc<SD>,
    pk_serde: OrderedRowSerde,
    vnode_stats: Option<HashMap<VirtualNode, VnodeStatistics>>,
    enable_state_table_vnode_stats_pruning: bool,
    pub metrics: Option<StateTableMetrics>,
}

Extract the logic of StateTableRowStore from StateTable, which serves a similar functionality as a BTreeMap<TableKey<Bytes>, OwnedRow>, and provides method to read (get and iter) over serialized key (or key range) and returns OwnedRow, and write (insert, delete, update) on (TableKey<Bytes>, OwnedRow).

Fields

state_store: LS

all_rows: Option<HashMap<VirtualNode, BTreeMap<Bytes, OwnedRow>>>

pending_watermark: Option<(WatermarkDirection, Bytes)>

The pk-prefix watermark fed in the current epoch, with its direction. The state store only applies it once committed, so the reads from the state store hide the keys it cleans until then.

table_id: TableId

row_serde: Arc<SD>

pk_serde: OrderedRowSerde

vnode_stats: Option<HashMap<VirtualNode, VnodeStatistics>>

enable_state_table_vnode_stats_pruning: bool

When false, vnode stats pruning is in dry-run mode: we maintain stats and verify pruning correctness but don’t actually apply pruning. Reads still go to cache/storage even when pruning would indicate no results.

metrics: Option<StateTableMetrics>

Implementations

impl<LS: LocalStateStore, SD: ValueRowSerde> StateTableRowStore<LS, SD>

async fn may_load_vnode_stats( &mut self, vnode_bitmap: &Bitmap, ) -> StreamExecutorResult<()>

async fn may_reload_all_rows( &mut self, vnode_bitmap: &Bitmap, ) -> StreamExecutorResult<()>

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

async fn update_vnode_bitmap( &mut self, vnodes: Arc<Bitmap>, ) -> StreamExecutorResult<Arc<Bitmap>>

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

fn clean_preloaded_rows( &mut self, direction: WatermarkDirection, watermarks: &[VnodeWatermark], serde_type: WatermarkSerdeType, )

Remove the preloaded rows cleaned by watermarks.

fn update_pending_watermark( &mut self, direction: WatermarkDirection, vnode_watermark: VnodeWatermark, )

Feed the pk-prefix watermark of the current epoch: the preloaded rows below it are removed at once, and the reads from the state store hide them until the commit.

fn narrow_by_pending_watermark(&self, key_range: &mut TableKeyRange)

fn below_pending_watermark(&self, key: &TableKey<Bytes>) -> bool

async fn seal_current_epoch( &mut self, next_epoch: u64, table_watermarks: Option<(WatermarkDirection, Vec<VnodeWatermark>, WatermarkSerdeType)>, switch_consistent_op: Option<StateTableOpConsistencyLevel>, ) -> StreamExecutorResult<()>

impl<LS: LocalStateStore, SD: ValueRowSerde> StateTableRowStore<LS, SD>

async fn get( &self, key_bytes: TableKey<Bytes>, prefix_hint: Option<Bytes>, ) -> StreamExecutorResult<Option<OwnedRow>>

async fn exists( &self, key_bytes: TableKey<Bytes>, prefix_hint: Option<Bytes>, ) -> StreamExecutorResult<bool>

impl<LS: LocalStateStore, SD: ValueRowSerde> StateTableRowStore<LS, SD>

fn handle_mem_table_error(&self, e: StorageError)

fn insert(&mut self, key: TableKey<Bytes>, value: impl Row)

fn delete(&mut self, key: TableKey<Bytes>, value: impl Row)

fn update( &mut self, key_bytes: TableKey<Bytes>, old_value: impl Row, new_value: impl Row, )

impl<LS: LocalStateStore, SD: ValueRowSerde> StateTableRowStore<LS, SD>

async fn iter_kv<K: CopyFromSlice + FromVnodeBytes>( &self, vnode: VirtualNode, __arg2: (Bound<Bytes>, Bound<Bytes>), prefix_hint: Option<Bytes>, prefetch_options: PrefetchOptions, ) -> StreamExecutorResult<impl PkRowStream<'_, K>>

Middle-level APIs:

• StateTableInner::iter_with_prefix_inner
• StateTableInner::iter_kv_with_pk_range

async fn rev_iter_kv<K: CopyFromSlice + FromVnodeBytes>( &self, vnode: VirtualNode, __arg2: (Bound<Bytes>, Bound<Bytes>), prefix_hint: Option<Bytes>, prefetch_options: PrefetchOptions, ) -> StreamExecutorResult<impl PkRowStream<'_, K>>