Repository: risingwavelabs/risingwave
File: src/meta/src/rpc/ddl_controller.rs (Rust)

Code before:
    visit_stream_node_cont_mut(union_fragment_root, |node| {
        if let Some(NodeBody::UpstreamSinkUnion(upstream_sink_union)) = &mut node.node_body {
            let init_upstreams = upstream_sink_infos
                .iter()
                .map(|info| PbUpstreamSinkInfo {
                    upstream_fragment_id: info.sink_fragment_id,
                    sink_output_schema: info.sink_output_fields.clone(),
                    project_exprs: info.project_exprs.clone(),
                })
                .collect();
            upstream_sink_union.init_upstreams = init_upstreams;
            false
        } else {
            true
        }
    });
}

#[cfg(test)]
mod tests {
    use std::num::NonZeroUsize;

    use super::*;

    #[test]
    fn specified_parallelism_can_exceed_available() {
        let result = DdlController::resolve_stream_parallelism_inner(
            Some(NonZeroUsize::new(100).unwrap()),
            NonZeroUsize::new(256).unwrap(),
            Some(NonZeroUsize::new(4).unwrap()),
            &DefaultParallelism::Full,
            "default",
        )
        .unwrap();
        assert_eq!(result.get(), 100);
    }

    #[test]
    fn default_parallelism_can_exceed_available() {
        let result = DdlController::resolve_stream_parallelism_inner(
            None,
            NonZeroUsize::new(256).unwrap(),
            Some(NonZeroUsize::new(4).unwrap()),
            &DefaultParallelism::Default(NonZeroUsize::new(50).unwrap()),
            "default",
        )
        .unwrap();
        assert_eq!(result.get(), 50);
    }

    #[test]

Review comment (suggestion):
[nitpick] Test name `specified_parallelism_can_exceed_available` uses passive voice and doesn't clearly communicate what behavior is being tested. Consider renaming to `test_specified_parallelism_exceeds_available_logs_warning` or `test_allows_specified_parallelism_over_available` to better describe the expected behavior and follow common test naming conventions (starting with `test_`).
```suggestion
    fn test_specified_parallelism_exceeds_available() {
```

Code after:
    visit_stream_node_cont_mut(union_fragment_root, |node| {
        if let Some(NodeBody::UpstreamSinkUnion(upstream_sink_union)) = &mut node.node_body {
            let init_upstreams = upstream_sink_infos
                .iter()
                .map(|info| PbUpstreamSinkInfo {
                    upstream_fragment_id: info.sink_fragment_id,
                    sink_output_schema: info.sink_output_fields.clone(),
                    project_exprs: info.project_exprs.clone(),
                })
                .collect();
            upstream_sink_union.init_upstreams = init_upstreams;
            false
        } else {
            true
        }
    });
}

#[cfg(test)]
mod tests {
    use std::num::NonZeroUsize;

    use super::*;

    #[test]
    fn test_specified_parallelism_exceeds_available() {
        let result = DdlController::resolve_stream_parallelism_inner(
            Some(NonZeroUsize::new(100).unwrap()),
            NonZeroUsize::new(256).unwrap(),
            Some(NonZeroUsize::new(4).unwrap()),
            &DefaultParallelism::Full,
            "default",
        )
        .unwrap();
        assert_eq!(result.get(), 100);
    }

    #[test]
    fn test_allows_default_parallelism_over_available() {
        let result = DdlController::resolve_stream_parallelism_inner(
            None,
            NonZeroUsize::new(256).unwrap(),
            Some(NonZeroUsize::new(4).unwrap()),
            &DefaultParallelism::Default(NonZeroUsize::new(50).unwrap()),
            "default",
        )
        .unwrap();
        assert_eq!(result.get(), 50);
    }

    #[test]