Struct StateTableInner 

pub struct StateTableInner<S, SD = BasicSerde, const IS_REPLICATED: bool = false, const USE_WATERMARK_CACHE: bool = false>
where
    S: StateStore,
    SD: ValueRowSerde,
{

    table_id: TableId,
    row_store: StateTableRowStore<S::Local, SD>,
    store: S,
    epoch: Option<EpochPair>,
    pk_serde: OrderedRowSerde,
    pk_indices: Vec<usize>,
    distribution: TableDistribution,
    prefix_hint_len: usize,
    value_indices: Option<Vec<usize>>,
    pending_watermark: Option<ScalarImpl>,
    committed_watermark: Option<ScalarImpl>,
    watermark_cache: StateTableWatermarkCache,
    data_types: Vec<DataType>,
    i2o_mapping: ColIndexMapping,
    pub output_indices: Vec<usize>,
    op_consistency_level: StateTableOpConsistencyLevel,
    clean_watermark_index_in_pk: Option<i32>,
    on_post_commit: bool,
}

StateTableInner is the interface accessing relational data in KV(StateStore) with row-based encoding.

Fields

table_id: TableId

Id for this table.

row_store: StateTableRowStore<S::Local, SD>

State store backend.

store: S

State store for accessing snapshot data

epoch: Option<EpochPair>

Current epoch

pk_serde: OrderedRowSerde

Used for serializing and deserializing the primary key.

pk_indices: Vec<usize>

Indices of primary key. Note that the index is based on the all columns of the table, instead of the output ones.

distribution: TableDistribution

Distribution of the state table.

It holds vnode bitmap. Only the rows whose vnode of the primary key is in this set will be visible to the executor. The table will also check whether the written rows conform to this partition.

prefix_hint_len: usize

value_indices: Option<Vec<usize>>

pending_watermark: Option<ScalarImpl>

Pending watermark for state cleaning. Old states below this watermark will be cleaned when committing.

committed_watermark: Option<ScalarImpl>

Last committed watermark for state cleaning. Will be restored on state table recovery.

watermark_cache: StateTableWatermarkCache

Cache for the top-N primary keys for reducing unnecessary range deletion.

data_types: Vec<DataType>

Data Types We will need to use to build data chunks from state table rows.

i2o_mapping: ColIndexMapping

“i” here refers to the base state_table’s actual schema. “o” here refers to the replicated state table’s output schema. This mapping is used to reconstruct a row being written from replicated state table. Such that the schema of this row will match the full schema of the base state table. It is only applicable for replication.

output_indices: Vec<usize>

Output indices Used for:

1. Computing output_value_indices to ser/de replicated rows.
2. Computing output pk indices to used them for backfill state.

op_consistency_level: StateTableOpConsistencyLevel

clean_watermark_index_in_pk: Option<i32>

on_post_commit: bool

Flag to indicate whether the state table has called commit, but has not called post_yield_barrier on the StateTablePostCommit callback yet.

Implementations

impl<S, SD, const IS_REPLICATED: bool, const USE_WATERMARK_CACHE: bool> StateTableInner<S, SD, IS_REPLICATED, USE_WATERMARK_CACHE>

where S: StateStore, SD: ValueRowSerde,

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

In streaming executors, this methods must be called after receiving and yielding the first barrier, and otherwise, deadlock can be likely to happen.

pub async fn try_wait_committed_epoch( &self, prev_epoch: u64, ) -> StorageResult<()>

pub fn state_store(&self) -> &S

impl<S, SD, const IS_REPLICATED: bool, const USE_WATERMARK_CACHE: bool> StateTableInner<S, SD, IS_REPLICATED, USE_WATERMARK_CACHE>

where S: StateStore, SD: ValueRowSerde,

pub async fn from_table_catalog( table_catalog: &Table, store: S, vnodes: Option<Arc<Bitmap>>, ) -> Self

Create state table from table catalog and store.

If vnodes is None, [TableDistribution::singleton()] will be used.

pub async fn from_table_catalog_inconsistent_op( table_catalog: &Table, store: S, vnodes: Option<Arc<Bitmap>>, ) -> Self

Create state table from table catalog and store with sanity check disabled.

async fn from_table_catalog_inner( table_catalog: &Table, store: S, vnodes: Option<Arc<Bitmap>>, op_consistency_level: StateTableOpConsistencyLevel, output_column_ids: Vec<ColumnId>, preload_all_rows: bool, ) -> Self

Create state table from table catalog and store.

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

pub fn table_id(&self) -> TableId

fn compute_prefix_vnode(&self, pk_prefix: &impl Row) -> VirtualNode

Get the vnode value with given (prefix of) primary key

pub fn compute_vnode_by_pk(&self, pk: impl Row) -> VirtualNode

Get the vnode value of the given primary key

pub fn pk_indices(&self) -> &[usize]

NOTE(kwannoel): This is used by backfill. We want to check pk indices of upstream table.

pub fn pk_in_output_indices(&self) -> Option<Vec<usize>>

Get the indices of the primary key columns in the output columns.

Returns None if any of the primary key columns is not in the output columns.

pub fn pk_serde(&self) -> &OrderedRowSerde

pub fn vnodes(&self) -> &Arc<Bitmap>

pub fn value_indices(&self) -> &Option<Vec<usize>>

pub fn is_consistent_op(&self) -> bool

impl<S, SD, const USE_WATERMARK_CACHE: bool> StateTableInner<S, SD, true, USE_WATERMARK_CACHE>

where S: StateStore, SD: ValueRowSerde,

pub async fn new_replicated( table_catalog: &Table, store: S, vnodes: Option<Arc<Bitmap>>, output_column_ids: Vec<ColumnId>, ) -> Self

Create replicated state table from table catalog with output indices

impl<S, SD, const IS_REPLICATED: bool, const USE_WATERMARK_CACHE: bool> StateTableInner<S, SD, IS_REPLICATED, USE_WATERMARK_CACHE>

where S: StateStore, SD: ValueRowSerde,

pub async fn get_row( &self, pk: impl Row, ) -> StreamExecutorResult<Option<OwnedRow>>

Get a single row from state table.

pub async fn exists(&self, pk: impl Row) -> StreamExecutorResult<bool>

Get a raw encoded row from state table.

fn serialize_pk(&self, pk: &impl Row) -> TableKey<Bytes>

fn serialize_pk_and_get_prefix_hint( &self, pk: &impl Row, ) -> (TableKey<Bytes>, Option<Bytes>)

impl<S, SD, const IS_REPLICATED: bool, const USE_WATERMARK_CACHE: bool> StateTableInner<S, SD, IS_REPLICATED, USE_WATERMARK_CACHE>

where S: StateStore, SD: ValueRowSerde,

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

Insert a row into state table. Must provide a full row corresponding to the column desc of the table.

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

Delete a row from state table. Must provide a full row of old value corresponding to the column desc of the table.

pub fn update(&mut self, old_value: impl Row, new_value: impl Row)

Update a row. The old and new value should have the same pk.

pub fn write_record(&mut self, record: Record<impl Row>)

Write a record into state table. Must have the same schema with the table.

fn fill_non_output_indices(&self, chunk: StreamChunk) -> StreamChunk

pub fn write_chunk(&mut self, chunk: StreamChunk)

Write batch with a StreamChunk which should have the same schema with the table.

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

Update watermark for state cleaning.

Arguments

• watermark - Latest watermark received.

pub fn get_committed_watermark(&self) -> Option<&ScalarImpl>

Get the committed watermark of the state table. Watermarks should be fed into the state table through update_watermark method.

pub async fn commit( &mut self, new_epoch: EpochPair, ) -> StreamExecutorResult<StateTablePostCommit<'_, S, SD, IS_REPLICATED, USE_WATERMARK_CACHE>>

pub async fn commit_assert_no_update_vnode_bitmap( &mut self, new_epoch: EpochPair, ) -> StreamExecutorResult<()>

pub async fn commit_may_switch_consistent_op( &mut self, new_epoch: EpochPair, op_consistency_level: StateTableOpConsistencyLevel, ) -> StreamExecutorResult<StateTablePostCommit<'_, S, SD, IS_REPLICATED, USE_WATERMARK_CACHE>>

async fn commit_inner( &mut self, new_epoch: EpochPair, switch_consistent_op: Option<StateTableOpConsistencyLevel>, ) -> StreamExecutorResult<StateTablePostCommit<'_, S, SD, IS_REPLICATED, USE_WATERMARK_CACHE>>

fn commit_pending_watermark( &mut self, ) -> Option<(WatermarkDirection, Vec<VnodeWatermark>, WatermarkSerdeType)>

Commit pending watermark and return vnode bitmap-watermark pairs to seal.

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

impl<S, SD, const IS_REPLICATED: bool, const USE_WATERMARK_CACHE: bool> StateTableInner<S, SD, IS_REPLICATED, USE_WATERMARK_CACHE>

where S: StateStore, SD: ValueRowSerde,

pub async fn iter_with_vnode( &self, vnode: VirtualNode, pk_range: &(Bound<impl Row>, Bound<impl Row>), prefetch_options: PrefetchOptions, ) -> StreamExecutorResult<impl RowStream<'_>>

This function scans rows from the relational table with specific pk_range under the same vnode.

pub async fn iter_keyed_row_with_vnode( &self, vnode: VirtualNode, pk_range: &(Bound<impl Row>, Bound<impl Row>), prefetch_options: PrefetchOptions, ) -> StreamExecutorResult<impl KeyedRowStream<'_>>

pub async fn iter_with_vnode_and_output_indices( &self, vnode: VirtualNode, pk_range: &(Bound<impl Row>, Bound<impl Row>), prefetch_options: PrefetchOptions, ) -> StreamExecutorResult<impl RowStream<'_>>

impl<S, SD, const IS_REPLICATED: bool, const USE_WATERMARK_CACHE: bool> StateTableInner<S, SD, IS_REPLICATED, USE_WATERMARK_CACHE>

where S: StateStore, SD: ValueRowSerde,

pub async fn iter_with_prefix( &self, pk_prefix: impl Row, sub_range: &(Bound<impl Row>, Bound<impl Row>), prefetch_options: PrefetchOptions, ) -> StreamExecutorResult<impl RowStream<'_>>

This function scans rows from the relational table with specific prefix and sub_range under the same vnode. If sub_range is (Unbounded, Unbounded), it scans rows from the relational table with specific pk_prefix. pk_prefix is used to identify the exact vnode the scan should perform on.

pub async fn get_from_one_row_table( &self, ) -> StreamExecutorResult<Option<OwnedRow>>

Get the row from a state table with only 1 row.

pub async fn get_from_one_value_table( &self, ) -> StreamExecutorResult<Option<ScalarImpl>>

Get the row from a state table with only 1 row, and the row has only 1 col.

None can mean either the row is never persisted, or is a persisted NULL, which does not matter in the use case.

pub async fn iter_keyed_row_with_prefix( &self, pk_prefix: impl Row, sub_range: &(Bound<impl Row>, Bound<impl Row>), prefetch_options: PrefetchOptions, ) -> StreamExecutorResult<impl KeyedRowStream<'_>>

pub async fn rev_iter_with_prefix( &self, pk_prefix: impl Row, sub_range: &(Bound<impl Row>, Bound<impl Row>), prefetch_options: PrefetchOptions, ) -> StreamExecutorResult<impl RowStream<'_>>

This function scans the table just like iter_with_prefix, but in reverse order.

async fn iter_with_prefix_inner<const REVERSE: bool, K: CopyFromSlice + FromVnodeBytes>( &self, pk_prefix: impl Row, sub_range: &(Bound<impl Row>, Bound<impl Row>), prefetch_options: PrefetchOptions, ) -> StreamExecutorResult<impl PkRowStream<'_, K>>

async fn iter_kv_with_pk_range<'a, K: CopyFromSlice + FromVnodeBytes>( &'a self, pk_range: &(Bound<impl Row>, Bound<impl Row>), vnode: VirtualNode, prefetch_options: PrefetchOptions, ) -> StreamExecutorResult<impl PkRowStream<'a, K>>

This function scans raw key-values from the relational table with specific pk_range under the same vnode.