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

Code before:
            );
        }

        Ok(result)
    }

    async fn exists(
        &self,
        key_bytes: TableKey<Bytes>,
        prefix_hint: Option<Bytes>,
    ) -> StreamExecutorResult<bool> {
        if let Some(m) = &self.metrics {
            m.get_count.inc();
        }
        if let Some(rows) = &self.all_rows {
            let (vnode, key) = key_bytes.split_vnode_bytes();
            return Ok(rows.get(&vnode).expect("covered vnode").contains_key(&key));
        }

        // Try to prune using vnode statistics
        let should_prune = if let Some(stats) = &self.vnode_stats
            && let (vnode, key) = key_bytes.split_vnode_bytes()
            && let Some(vnode_stat) = stats.get(&vnode)
            && vnode_stat.can_prune(&key)
        {
            if let Some(m) = &self.metrics {
                m.get_vnode_pruned_count.inc();
            }
            true
        } else {
            false
        };

        // In dry-run mode, we verify pruning correctness but don't apply it
        if should_prune && self.enable_state_table_vnode_stats_pruning {
            return Ok(false);
        }

        let read_options = ReadOptions {
            prefix_hint,
            cache_policy: CachePolicy::Fill(Hint::Normal),
            ..Default::default()
        };
        let result = self
            .state_store
            .on_key_value(key_bytes, read_options, move |_, _| Ok(()))
            .await?;
        let exists = result.is_some();

        // In dry-run mode, verify that pruning would have been correct
        if should_prune && exists {

Review comment (refactor):
Regardless of whether it is a dry-run or not, **get_vnode_pruned_count** will be counted. It is impossible to tell at a glance from the metrics whether pruning is enabled. Should we add an additional actual_pruned_count metric to make a semantic distinction?

Code after:
        }

        Ok(result)
    }

    async fn exists(
        &self,
        key_bytes: TableKey<Bytes>,
        prefix_hint: Option<Bytes>,
    ) -> StreamExecutorResult<bool> {
        if let Some(m) = &self.metrics {
            m.get_count.inc();
        }
        if let Some(rows) = &self.all_rows {
            let (vnode, key) = key_bytes.split_vnode_bytes();
            return Ok(rows.get(&vnode).expect("covered vnode").contains_key(&key));
        }

        // Try to prune using vnode statistics
        let should_prune = if let Some(stats) = &self.vnode_stats
            && let (vnode, key) = key_bytes.split_vnode_bytes()
            && let Some(vnode_stat) = stats.get(&vnode)
            && vnode_stat.can_prune(&key)
        {
            if let Some(m) = &self.metrics {
                m.get_vnode_pruned_count.inc();
            }
            true
        } else {
            false
        };

        if should_prune && self.enable_state_table_vnode_stats_pruning {
            return Ok(false);
        }

        let read_options = ReadOptions {
            prefix_hint,
            cache_policy: CachePolicy::Fill(Hint::Normal),
            ..Default::default()
        };
        let result = self
            .state_store
            .on_key_value(key_bytes, read_options, move |_, _| Ok(()))
            .await?;
        let exists = result.is_some();

        // In dry-run mode, verify that pruning would have been correct
        if should_prune && exists {
            tracing::warn!(
                table_id = %self.table_id,