Repository: risingwavelabs/risingwave
File: src/stream/src/common/table/state_table.rs (Rust)

Code before:
            $body
        } else {
            $body
        }
    };
}

/// 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)`.
struct StateTableRowStore<LS: LocalStateStore, SD: ValueRowSerde> {
    state_store: LS,
    all_rows: Option<HashMap<VirtualNode, BTreeMap<Bytes, OwnedRow>>>,

    table_id: TableId,
    row_serde: Arc<SD>,
    // should be only used for debugging in panic message of handle_mem_table_error
    pk_serde: OrderedRowSerde,

    // Per-vnode min/max key statistics for pruning
    vnode_stats: Option<HashMap<VirtualNode, VnodeStatistics>>,
    /// 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.
    enable_state_table_vnode_stats_prunning: bool,
    // Optional metrics for state table operations
    pub metrics: Option<StateTableMetrics>,
}

impl<LS: LocalStateStore, SD: ValueRowSerde> StateTableRowStore<LS, SD> {
    async fn may_load_vnode_stats(&mut self, vnode_bitmap: &Bitmap) -> StreamExecutorResult<()> {
        if self.vnode_stats.is_none() {
            return Ok(());
        }

        // vnode stats must be disabled when all rows are preloaded
        assert!(self.all_rows.is_none());

        let start_time = Instant::now();
        let mut stats_map = HashMap::new();

        // Scan each vnode to get min/max keys
        for vnode in vnode_bitmap.iter_vnodes() {
            let mut stats = VnodeStatistics::new();

            // Get min key via forward iteration
            let memcomparable_range_with_vnode = prefixed_range_with_vnode::<Bytes>(.., vnode);
            let read_options = ReadOptions {
                cache_policy: CachePolicy::Fill(Hint::Low),
                ..Default::default()

Review comment (suggestion):
Spelling error: "prunning" should be "pruning" (single 'n'). This affects the field name which should be `enable_state_table_vnode_stats_pruning` instead of `enable_state_table_vnode_stats_prunning`.
```suggestion
    enable_state_table_vnode_stats_pruning: bool,
```

Code after:
            $body
        } else {
            $body
        }
    };
}

/// 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)`.
struct StateTableRowStore<LS: LocalStateStore, SD: ValueRowSerde> {
    state_store: LS,
    all_rows: Option<HashMap<VirtualNode, BTreeMap<Bytes, OwnedRow>>>,

    table_id: TableId,
    row_serde: Arc<SD>,
    // should be only used for debugging in panic message of handle_mem_table_error
    pk_serde: OrderedRowSerde,

    // Per-vnode min/max key statistics for pruning
    vnode_stats: Option<HashMap<VirtualNode, VnodeStatistics>>,
    /// 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.
    enable_state_table_vnode_stats_pruning: bool,
    // Optional metrics for state table operations
    pub metrics: Option<StateTableMetrics>,
}

impl<LS: LocalStateStore, SD: ValueRowSerde> StateTableRowStore<LS, SD> {
    async fn may_load_vnode_stats(&mut self, vnode_bitmap: &Bitmap) -> StreamExecutorResult<()> {
        if self.vnode_stats.is_none() {
            return Ok(());
        }

        // vnode stats must be disabled when all rows are preloaded
        assert!(self.all_rows.is_none());

        let start_time = Instant::now();
        let mut stats_map = HashMap::new();

        // Scan each vnode to get min/max keys
        for vnode in vnode_bitmap.iter_vnodes() {
            let mut stats = VnodeStatistics::new();

            // Get min key via forward iteration
            let memcomparable_range_with_vnode = prefixed_range_with_vnode::<Bytes>(.., vnode);
            let read_options = ReadOptions {
                cache_policy: CachePolicy::Fill(Hint::Low),
                ..Default::default()