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

Code before:
                .await?;

                tracing::trace!(
                    barrier = ?barrier,
                    "barrier persisted"
                );

                // handle mutations
                if let Some(mutation) = barrier.mutation.as_deref() {
                    use crate::executor::Mutation;
                    match mutation {
                        Mutation::Pause => {
                            global_pause = true;
                        }
                        Mutation::Resume => {
                            global_pause = false;
                        }
                        Mutation::StartFragmentBackfill { fragment_ids } if backfill_paused => {
                            if fragment_ids.contains(&self.fragment_id) {
                                backfill_paused = false;
                            }
                        }
                        Mutation::Throttle(fragment_to_apply) => {
                            let entry = fragment_to_apply.get(&self.fragment_id);
                            if let Some(entry) = entry
                                && entry.throttle_type() == ThrottleType::Backfill
                            {
                                let new_rate_limit = entry.rate_limit.into();
                                let old_rate_limit = self.rate_limiter.update(new_rate_limit);
                                if old_rate_limit != new_rate_limit {
                                    tracing::info!(
                                        old_rate_limit = ?old_rate_limit,
                                        new_rate_limit = ?new_rate_limit,
                                        %upstream_table_id,
                                        actor_id = %self.actor_id,
                                        "backfill rate limit changed",
                                    );
                                    builders = upstream_table
                                        .vnodes()
                                        .iter_vnodes()
                                        .map(|vnode| {
                                            let builder = create_builder(
                                                new_rate_limit,
                                                self.chunk_size,
                                                snapshot_data_types.clone(),
                                            );
                                            (vnode, builder)
                                        })
                                        .collect();
                                }
                            }

Review comment (suggestion):
The variable `entry` is retrieved on line 562 and then immediately checked on line 563. This can be simplified by combining them into a single `if let Some(entry) = fragment_to_apply.get(&self.fragment_id)` statement on line 563, which would be more concise and consistent with similar patterns used in other executor files.

Code after:
                .await?;

                tracing::trace!(
                    barrier = ?barrier,
                    "barrier persisted"
                );

                // handle mutations
                if let Some(mutation) = barrier.mutation.as_deref() {
                    use crate::executor::Mutation;
                    match mutation {
                        Mutation::Pause => {
                            global_pause = true;
                        }
                        Mutation::Resume => {
                            global_pause = false;
                        }
                        Mutation::StartFragmentBackfill { fragment_ids } if backfill_paused => {
                            if fragment_ids.contains(&self.fragment_id) {
                                backfill_paused = false;
                            }
                        }
                        Mutation::Throttle(fragment_to_apply) => {
                            if let Some(entry) = fragment_to_apply.get(&self.fragment_id)
                                && entry.throttle_type() == ThrottleType::Backfill
                            {
                                let new_rate_limit = entry.rate_limit.into();
                                let old_rate_limit = self.rate_limiter.update(new_rate_limit);
                                if old_rate_limit != new_rate_limit {
                                    tracing::info!(
                                        old_rate_limit = ?old_rate_limit,
                                        new_rate_limit = ?new_rate_limit,
                                        %upstream_table_id,
                                        actor_id = %self.actor_id,
                                        "backfill rate limit changed",
                                    );
                                    builders = upstream_table
                                        .vnodes()
                                        .iter_vnodes()
                                        .map(|vnode| {
                                            let builder = create_builder(
                                                new_rate_limit,
                                                self.chunk_size,
                                                snapshot_data_types.clone(),
                                            );
                                            (vnode, builder)
                                        })
                                        .collect();
                                }
                            }
                        }