Struct RefreshProgressTable

pub struct RefreshProgressTable<S: StateStore> {
    pub state_table: StateTable<S>,
    cache: HashMap<VirtualNode, RefreshProgressEntry>,
    pk_len: usize,
}

Schema for the refresh progress table (simplified, following backfill pattern):

• vnode (i32): VirtualNode identifier
• current_pos (variable): Current processing position (primary key of last processed row)
• is_completed (bool): Whether this vnode has completed processing
• processed_rows (i64): Number of rows processed so far in this vnode

Fields

state_table: StateTable<S>

The underlying state table for persistence

cache: HashMap<VirtualNode, RefreshProgressEntry>

In-memory cache of progress information for quick access

pk_len: usize

Primary key length for upstream table (needed for schema)

Implementations

impl<S: StateStore> RefreshProgressTable<S>

pub fn new(state_table: StateTable<S>, pk_len: usize) -> Self

Create a new RefreshProgressTable

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

Initialize the progress table by loading existing entries from storage

pub fn set_progress( &mut self, vnode: VirtualNode, current_pos: Option<OwnedRow>, is_completed: bool, processed_rows: u64, ) -> StreamExecutorResult<()>

Set progress for a specific VirtualNode

pub fn get_progress(&self, vnode: VirtualNode) -> Option<&RefreshProgressEntry>

Get progress for a specific VirtualNode

pub fn get_all_progress(&self) -> &HashMap<VirtualNode, RefreshProgressEntry>

Get progress for all VirtualNodes

pub fn get_completed_vnodes(&self) -> HashSet<VirtualNode>

Get all VirtualNodes that have completed processing

pub fn get_vnodes_in_stage( &self, _stage: ProgressRefreshStage, ) -> Vec<VirtualNode>

Get all VirtualNodes in a specific stage (now tracked by executor memory) This method is kept for compatibility but should not be used with simplified schema

pub fn clear_progress(&mut self, vnode: VirtualNode) -> StreamExecutorResult<()>

Clear progress for a specific VirtualNode

pub fn clear_all_progress(&mut self) -> StreamExecutorResult<()>

Clear all progress entries

pub fn get_total_processed_rows(&self) -> u64

Get the total number of processed rows across all VirtualNodes

pub fn get_progress_stats(&self) -> RefreshProgressStats

Get progress statistics

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

Commit changes to storage

fn entry_to_row(&self, entry: &RefreshProgressEntry, pk_len: usize) -> OwnedRow

Convert RefreshProgressEntry to OwnedRow for storage New schema: | vnode | current_pos… | is_completed | processed_rows |

fn parse_row_to_entry( &self, row: &impl Row, pk_len: usize, ) -> Option<RefreshProgressEntry>

Parse OwnedRow from storage to RefreshProgressEntry New schema: | vnode | current_pos… | is_completed | processed_rows | Note: The length depends on the primary key length of upstream table

pub fn expected_schema(pk_data_types: &[DataType]) -> Vec<DataType>

Get the expected schema for the progress table Schema: | vnode | current_pos… | is_completed | processed_rows |

pub fn column_names(pk_column_names: &[&str]) -> Vec<String>

Get column names for the progress table schema Schema: | vnode | current_pos… | is_completed | processed_rows |

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