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,
    local_store: S::Local,
    store: S,
    pk_serde: OrderedRowSerde,
    row_serde: SD,
    pk_indices: Vec<usize>,
    distribution: TableDistribution,
    prefix_hint_len: usize,
    table_option: TableOption,
    value_indices: Option<Vec<usize>>,
    pending_watermark: Option<ScalarImpl>,
    committed_watermark: Option<ScalarImpl>,
    watermark_cache: StateTableWatermarkCache,
    data_types: Vec<DataType>,
    i2o_mapping: ColIndexMapping,
    output_indices: Vec<usize>,
    op_consistency_level: StateTableOpConsistencyLevel,
}

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

Fields

table_id: TableId

Id for this table.

local_store: S::Local

State store backend.

store: S

State store for accessing snapshot data

pk_serde: OrderedRowSerde

Used for serializing and deserializing the primary key.

row_serde: SD

Row deserializer with value encoding

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

table_option: TableOption

Used for catalog table_properties

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

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 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.

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

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

Create state table from table catalog and store.

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

pub fn table_id(&self) -> u32

pub fn epoch(&self) -> u64

get the newest epoch of the state store and panic if the init_epoch() has never be called

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

fn is_dirty(&self) -> bool

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 from_table_catalog_with_output_column_ids( 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 get_encoded_row( &self, pk: impl Row, ) -> StreamExecutorResult<Option<Bytes>>

Get a raw encoded row from state table.

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

Get a row in value-encoding format from state table.

pub fn update_vnode_bitmap( &mut self, new_vnodes: Arc<Bitmap>, ) -> (Arc<Bitmap>, bool)

Update the vnode bitmap of the state table, returns the previous vnode bitmap.

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,

fn handle_mem_table_error(&self, e: StorageError)

fn serialize_value(&self, value: impl Row) -> Bytes

fn insert_inner(&mut self, key: TableKey<Bytes>, value_bytes: Bytes)

fn delete_inner(&mut self, key: TableKey<Bytes>, value_bytes: Bytes)

fn update_inner( &mut self, key_bytes: TableKey<Bytes>, old_value_bytes: Option<Bytes>, new_value_bytes: Bytes, )

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 update_without_old_value(&mut self, new_value: impl Row)

Update a row without giving old value.

op_consistency_level should be set to Inconsistent.

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

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

async fn commit_inner( &mut self, new_epoch: EpochPair, switch_consistent_op: Option<StateTableOpConsistencyLevel>, ) -> StreamExecutorResult<()>

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

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 KeyedRowStream<'_>>

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

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 Stream<Item = StreamExecutorResult<KeyedRow<Bytes>>> + '_>

async fn iter_kv( &self, table_key_range: TableKeyRange, prefix_hint: Option<Bytes>, prefetch_options: PrefetchOptions, ) -> StreamExecutorResult<<S::Local as LocalStateStore>::Iter<'_>>

async fn rev_iter_kv( &self, table_key_range: TableKeyRange, prefix_hint: Option<Bytes>, prefetch_options: PrefetchOptions, ) -> StreamExecutorResult<<S::Local as LocalStateStore>::RevIter<'_>>

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

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 rev_iter_with_prefix( &self, pk_prefix: impl Row, sub_range: &(Bound<impl Row>, Bound<impl Row>), prefetch_options: PrefetchOptions, ) -> StreamExecutorResult<impl KeyedRowStream<'_>>

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

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

async fn iter_kv_with_pk_range( &self, pk_range: &(Bound<impl Row>, Bound<impl Row>), vnode: VirtualNode, prefetch_options: PrefetchOptions, ) -> StreamExecutorResult<<S::Local as LocalStateStore>::Iter<'_>>

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

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_log_with_vnode( &self, vnode: VirtualNode, epoch_range: (u64, u64), pk_range: &(Bound<impl Row>, Bound<impl Row>), ) -> StreamExecutorResult<impl Stream<Item = StreamExecutorResult<ChangeLogRow>> + '_>

Trait Implementations

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

where S: StateStore + Clone, SD: ValueRowSerde + Clone, S::Local: Clone,

fn clone(&self) -> StateTableInner<S, SD, IS_REPLICATED, USE_WATERMARK_CACHE>

Returns a copy of the value.

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

Performs copy-assignment from source.