Repository: risingwavelabs/risingwave
File: src/stream/src/executor/locality_provider.rs (Rust)

Code before:
                            processed_rows
                        }
                        LocalityBackfillProgress::Completed { total_rows, .. } => total_rows,
                        LocalityBackfillProgress::NotStarted => 0,
                    })
                    .sum();

                self.progress.update_with_buffered_rows(
                    barrier.epoch,
                    barrier.epoch.curr, // Use barrier epoch as snapshot read epoch
                    total_snapshot_processed_rows,
                    buffered_rows,
                );

                // Persist backfill progress
                Self::persist_backfill_state(&mut progress_table, &backfill_state).await?;
                let barrier_epoch = barrier.epoch;
                let post_commit2 = progress_table.commit(barrier_epoch).await?;

                metrics
                    .backfill_snapshot_read_row_count
                    .inc_by(cur_barrier_snapshot_processed_rows);
                metrics
                    .backfill_upstream_output_row_count
                    .inc_by(cur_barrier_upstream_processed_rows);

                yield Message::Barrier(barrier);
                post_commit1.post_yield_barrier(None).await?;
                post_commit2.post_yield_barrier(None).await?;
            }
        }

        tracing::debug!("Locality provider backfill finished, forwarding upstream directly");

        // Wait for first barrier after backfill completion to mark progress as finished
        if need_backfill && !backfill_state.is_completed() {
            while let Some(Ok(msg)) = upstream.next().await {
                match msg {
                    Message::Barrier(barrier) => {
                        // no-op commit state table
                        state_table
                            .commit_assert_no_update_vnode_bitmap(barrier.epoch)
                            .await?;

                        // Mark all vnodes as completed
                        for vnode in state_table.vnodes().iter_vnodes() {
                            backfill_state.finish_vnode(vnode, pk_indices.len());
                        }

                        // Calculate final total processed rows
                        let total_snapshot_processed_rows: u64 = backfill_state

Review comment (suggestion):
`commit_assert_no_update_vnode_bitmap` doesn't mean no-op, but means that there must no vnode update bitmap happen. 

Since here we always pass `None` to `post_yield_barrier`, we can assert that there is no update vnode bitmap, we therefore we can call `commit_assert_no_update_vnode_bitmap` directly.

Code after:
                        LocalityBackfillProgress::Completed { total_rows, .. } => total_rows,
                        LocalityBackfillProgress::NotStarted => 0,
                    })
                    .sum();

                self.progress.update_with_buffered_rows(
                    barrier.epoch,
                    barrier.epoch.curr, // Use barrier epoch as snapshot read epoch
                    total_snapshot_processed_rows,
                    buffered_rows,
                );

                // Persist backfill progress
                Self::persist_backfill_state(&mut progress_table, &backfill_state).await?;
                let barrier_epoch = barrier.epoch;
                let post_commit2 = progress_table.commit(barrier_epoch).await?;

                metrics
                    .backfill_snapshot_read_row_count
                    .inc_by(cur_barrier_snapshot_processed_rows);
                metrics
                    .backfill_upstream_output_row_count
                    .inc_by(cur_barrier_upstream_processed_rows);

                yield Message::Barrier(barrier);
                post_commit1.post_yield_barrier(None).await?;
                post_commit2.post_yield_barrier(None).await?;
            }
        }

        tracing::debug!("Locality provider backfill finished, forwarding upstream directly");

        // Wait for first barrier after backfill completion to mark progress as finished
        if need_backfill && !backfill_state.is_completed() {
            while let Some(Ok(msg)) = upstream.next().await {
                match msg {
                    Message::Barrier(barrier) => {
                        // no-op commit state table
                        state_table
                            .commit_assert_no_update_vnode_bitmap(barrier.epoch)
                            .await?;

                        // Mark all vnodes as completed
                        for vnode in state_table.vnodes().iter_vnodes() {
                            backfill_state.finish_vnode(vnode, pk_indices.len());
                        }

                        // Calculate final total processed rows
                        let total_snapshot_processed_rows: u64 = backfill_state
                            .vnodes()
                            .map(|(_, progress)| match *progress {