Repository: risingwavelabs/risingwave
File: src/frontend/src/stream_fragmenter/mod.rs (Rust)

Code before:
                state.tables.insert(table.id, table);
            }

            NodeBody::StreamScan(node) => {
                current_fragment
                    .fragment_type_mask
                    .add(FragmentTypeFlag::StreamScan);
                match node.stream_scan_type() {
                    StreamScanType::SnapshotBackfill => {
                        current_fragment
                            .fragment_type_mask
                            .add(FragmentTypeFlag::SnapshotBackfillStreamScan);
                        state.has_snapshot_backfill = true;
                        state.has_any_backfill = true;
                    }
                    StreamScanType::Backfill | StreamScanType::ArrangementBackfill => {
                        state.has_any_backfill = true;
                    }
                    StreamScanType::CrossDbSnapshotBackfill => {
                        current_fragment
                            .fragment_type_mask
                            .add(FragmentTypeFlag::CrossDbSnapshotBackfillStreamScan);
                        state.has_cross_db_snapshot_backfill = true;
                        state.has_any_backfill = true;
                    }
                    StreamScanType::Unspecified
                    | StreamScanType::Chain
                    | StreamScanType::Rearrange
                    | StreamScanType::UpstreamOnly => {}
                }
                // memorize table id for later use
                // The table id could be a upstream CDC source
                state.dependent_table_ids.insert(node.table_id);

                // Add state table if present
                if let Some(state_table) = &node.state_table {
                    let table = state_table.clone();
                    state.tables.insert(table.id, table);
                }
            }

            NodeBody::StreamCdcScan(node) => {
                if let Some(o) = node.options
                    && CdcScanOptions::from_proto(&o).is_parallelized_backfill()
                {
                    // Use parallel CDC backfill.
                    current_fragment
                        .fragment_type_mask
                        .add(FragmentTypeFlag::StreamCdcScan);
                } else {
                    current_fragment

Review comment (suggestion):
The `has_any_backfill` flag is set for `StreamScanType::Backfill` and `StreamScanType::ArrangementBackfill` cases, but these types were previously falling through to the empty match arm. While this change is intentional and correct for the feature, it represents a behavior change that could affect existing logic. Verify that setting `has_any_backfill = true` for these scan types doesn't have unintended side effects beyond parallelism control.

Code after:
                let table = node.buffer_table.as_ref().unwrap().clone();
                state.tables.insert(table.id, table);
                let table = node.prev_row_table.as_ref().unwrap().clone();
                state.tables.insert(table.id, table);
            }

            NodeBody::GapFill(node) => {
                let table = node.state_table.as_ref().unwrap().clone();
                state.tables.insert(table.id, table);
            }

            NodeBody::StreamScan(node) => {
                current_fragment
                    .fragment_type_mask
                    .add(FragmentTypeFlag::StreamScan);
                match node.stream_scan_type() {
                    StreamScanType::SnapshotBackfill => {
                        current_fragment
                            .fragment_type_mask
                            .add(FragmentTypeFlag::SnapshotBackfillStreamScan);
                        state.has_snapshot_backfill = true;
                        state.has_any_backfill = true;
                    }
                    StreamScanType::Backfill | StreamScanType::ArrangementBackfill => {
                        state.has_any_backfill = true;
                    }
                    StreamScanType::CrossDbSnapshotBackfill => {
                        current_fragment
                            .fragment_type_mask
                            .add(FragmentTypeFlag::CrossDbSnapshotBackfillStreamScan);
                        state.has_cross_db_snapshot_backfill = true;
                        state.has_any_backfill = true;
                    }
                    StreamScanType::Unspecified
                    | StreamScanType::Chain
                    | StreamScanType::Rearrange
                    | StreamScanType::UpstreamOnly => {}
                }
                // memorize table id for later use
                // The table id could be a upstream CDC source
                state.dependent_table_ids.insert(node.table_id);

                // Add state table if present
                if let Some(state_table) = &node.state_table {
                    let table = state_table.clone();
                    state.tables.insert(table.id, table);
                }
            }

            NodeBody::StreamCdcScan(node) => {
                if let Some(o) = node.options