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

Code before:
                        tracing::warn!(
                            "source paused, wait barrier for {:?}",
                            last_barrier_time.elapsed()
                        );
                        stream.pause_stream();

                        // Only update `max_wait_barrier_time_ms` to capture
                        // `barrier_interval_ms`
                        // changes here to avoid frequently accessing the shared
                        // `system_params`.
                        max_wait_barrier_time_ms = self.system_params.load().barrier_interval_ms()
                            as u128
                            * WAIT_BARRIER_MULTIPLE_TIMES;
                    }

                    latest_state.iter().for_each(|(split_id, new_split_impl)| {
                        if let Some(split_impl) =
                            self.stream_source_core.latest_split_info.get_mut(split_id)
                        {
                            *split_impl = new_split_impl.clone();
                        }
                    });

                    self.stream_source_core
                        .updated_splits_in_epoch
                        .extend(latest_state.clone());

                    let card = chunk.cardinality();
                    if card == 0 {
                        continue;
                    }
                    source_output_row_count.inc_by(card as u64);
                    let to_remove_col_indices =
                        if let Some(pulsar_message_id_idx) = pulsar_message_id_idx {
                            vec![split_idx, offset_idx, pulsar_message_id_idx]
                        } else {
                            vec![split_idx, offset_idx]
                        };
                    let chunk =
                        prune_additional_cols(&chunk, &to_remove_col_indices, &source_desc.columns);
                    yield Message::Chunk(chunk);
                    self.try_flush_data().await?;
                }
            }
        }

        // The source executor should only be stopped by the actor when finding a `Stop` mutation.
        tracing::error!(
            actor_id = %self.actor_ctx.id,
            "source executor exited unexpectedly"
        )

Review comment (performance):
The addition of `.clone()` here appears unnecessary. The `latest_state` HashMap is only borrowed (not moved) in earlier lines (895, 902, 928), and those borrows end before this line. Since `latest_state` is not used after this `extend` call, it can be moved directly without cloning, which would avoid an unnecessary allocation and copy operation. Unless there's a specific reason related to the CDC offset reporting changes that requires this clone, it should be removed to maintain the original performance characteristics.
```suggestion
                        .extend(latest_state);
```

Code after:
                    unreachable!();
                }

                Either::Right((chunk, latest_state)) => {
                    if let Some(task_builder) = &mut wait_checkpoint_task_builder {
                        if let Some(pulsar_message_id_idx) = pulsar_message_id_idx {
                            let pulsar_message_id_col = chunk.column_at(pulsar_message_id_idx);
                            task_builder.update_task_on_chunk(
                                source_id,
                                &latest_state,
                                pulsar_message_id_col.clone(),
                            );
                        } else {
                            let offset_col = chunk.column_at(offset_idx);
                            task_builder.update_task_on_chunk(
                                source_id,
                                &latest_state,
                                offset_col.clone(),
                            );
                        }
                    }
                    if last_barrier_time.elapsed().as_millis() > max_wait_barrier_time_ms {
                        // Exceeds the max wait barrier time, the source will be paused.
                        // Currently we can guarantee the
                        // source is not paused since it received stream
                        // chunks.
                        self_paused = true;
                        tracing::warn!(
                            "source paused, wait barrier for {:?}",
                            last_barrier_time.elapsed()
                        );
                        stream.pause_stream();

                        // Only update `max_wait_barrier_time_ms` to capture
                        // `barrier_interval_ms`
                        // changes here to avoid frequently accessing the shared
                        // `system_params`.
                        max_wait_barrier_time_ms = self.system_params.load().barrier_interval_ms()
                            as u128
                            * WAIT_BARRIER_MULTIPLE_TIMES;
                    }

                    latest_state.iter().for_each(|(split_id, new_split_impl)| {
                        if let Some(split_impl) =
                            self.stream_source_core.latest_split_info.get_mut(split_id)
                        {
                            *split_impl = new_split_impl.clone();
                        }
                    });

                    self.stream_source_core