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

Code before:
                        .await?,
                    &*self.row_serde,
                )
                .inspect(inspect_fn),
            ),
        ))
    }

    async fn rev_iter_kv<K: CopyFromSlice + FromVnodeBytes>(
        &self,
        vnode: VirtualNode,
        (start, end): (Bound<Bytes>, Bound<Bytes>),
        prefix_hint: Option<Bytes>,
        prefetch_options: PrefetchOptions,
    ) -> StreamExecutorResult<impl PkRowStream<'_, K>> {
        if let Some(m) = &self.metrics {
            m.iter_count.inc();
        }
        // Check if we can prune the entire range using vnode statistics
        let (pruned_start, pruned_end, should_prune_entirely) = if let Some(stats) =
            &self.vnode_stats
            && let Some(vnode_stat) = stats.get(&vnode)
        {
            match vnode_stat.pruned_key_range(&start, &end) {
                Some((new_start, new_end)) => {
                    if self.enable_state_table_vnode_stats_prunning {
                        (new_start, new_end, false)
                    } else {
                        // In dry-run mode, we don't apply pruning but verify correctness
                        (start, end, false)
                    }
                }
                None => {
                    if let Some(m) = &self.metrics {
                        m.iter_vnode_pruned_count.inc();
                    }
                    // Mark that we should prune entirely, but handle dry-run below
                    (start, end, true)
                }
            }
        } else {
            (start, end, false)
        };

        // In dry-run mode, we don't apply entire range pruning but verify correctness
        if should_prune_entirely && self.enable_state_table_vnode_stats_prunning {
            return Ok(futures::future::Either::Left(futures::stream::empty()));
        }

        let table_id = self.table_id;
        let inspect_fn = move |result: &StreamExecutorResult<(K, OwnedRow)>| {

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

Code after:
        ))
    }

    async fn rev_iter_kv<K: CopyFromSlice + FromVnodeBytes>(
        &self,
        vnode: VirtualNode,
        (start, end): (Bound<Bytes>, Bound<Bytes>),
        prefix_hint: Option<Bytes>,
        prefetch_options: PrefetchOptions,
    ) -> StreamExecutorResult<impl PkRowStream<'_, K>> {
        if let Some(m) = &self.metrics {
            m.iter_count.inc();
        }
        // Check if we can prune the entire range using vnode statistics
        let (pruned_start, pruned_end, should_prune_entirely) = if let Some(stats) =
            &self.vnode_stats
            && let Some(vnode_stat) = stats.get(&vnode)
        {
            match vnode_stat.pruned_key_range(&start, &end) {
                Some((new_start, new_end)) => {
                    if self.enable_state_table_vnode_stats_pruning {
                        (new_start, new_end, false)
                    } else {
                        // In dry-run mode, we don't apply pruning but verify correctness
                        (start, end, false)
                    }
                }
                None => {
                    if let Some(m) = &self.metrics {
                        m.iter_vnode_pruned_count.inc();
                    }
                    // Mark that we should prune entirely, but handle dry-run below
                    (start, end, true)
                }
            }
        } else {
            (start, end, false)
        };

        if should_prune_entirely && self.enable_state_table_vnode_stats_pruning {
            return Ok(futures::future::Either::Left(futures::stream::empty()));
        }

        let table_id = self.table_id;
        let inspect_fn = move |result: &StreamExecutorResult<(K, OwnedRow)>| {
            // Only log when in dry-run mode and we would have pruned but got results
            if should_prune_entirely && result.is_ok() {
                tracing::warn!(
                    table_id = %table_id,
                    "vnode stats pruning dry run fails for rev_iter. This will not affect correctness."
                );