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

Code before:
                );
            }
            return Ok(specified);
        }

        // Use default parallelism when no specific parallelism is provided by the user.
        let default_parallelism = match default_parallelism {
            DefaultParallelism::Full => available,
            DefaultParallelism::Default(num) => {
                if *num > available {
                    tracing::warn!(
                        resource_group,
                        configured_parallelism = num.get(),
                        available_parallelism = available.get(),
                        "default parallelism exceeds available slots, scheduling with configured value",
                    );
                }
                *num
            }
        };

        if default_parallelism > max {
            tracing::warn!(
                max_parallelism = max.get(),
                resource_group,
                "too many parallelism available, use max parallelism instead",
            );
        }
        Ok(default_parallelism.min(max))
    }

    /// Builds the actor graph:
    /// - Add the upstream fragments to the fragment graph
    /// - Schedule the fragments based on their distribution
    /// - Expand each fragment into one or several actors
    /// - Construct the fragment level backfill order control.
    #[await_tree::instrument]
    pub(crate) async fn build_stream_job(
        &self,
        stream_ctx: StreamContext,
        mut stream_job: StreamingJob,
        fragment_graph: StreamFragmentGraph,
        specific_resource_group: Option<String>,
    ) -> MetaResult<(CreateStreamingJobContext, StreamJobFragmentsToCreate)> {
        let id = stream_job.id();
        let specified_parallelism = fragment_graph.specified_parallelism();
        let max_parallelism = NonZeroUsize::new(fragment_graph.max_parallelism()).unwrap();

        // 1. Fragment Level ordering graph
        let fragment_backfill_ordering = fragment_graph.create_fragment_backfill_ordering();


Review comment (bug):
The warning message "too many parallelism available" is grammatically incorrect and unclear. The issue is not that there is "too much parallelism available", but rather that the configured default parallelism exceeds the max limit. Consider changing to "default parallelism exceeds max parallelism, capping to max" or "capping default parallelism to max parallelism limit".
```suggestion
                "default parallelism exceeds max parallelism, capping to max",
```

Code after:
                );
            }
            return Ok(specified);
        }

        // Use default parallelism when no specific parallelism is provided by the user.
        let default_parallelism = match default_parallelism {
            DefaultParallelism::Full => available,
            DefaultParallelism::Default(num) => {
                if *num > available {
                    tracing::warn!(
                        resource_group,
                        configured_parallelism = num.get(),
                        available_parallelism = available.get(),
                        "default parallelism exceeds available slots, scheduling with configured value",
                    );
                }
                *num
            }
        };

        if default_parallelism > max {
            tracing::warn!(
                max_parallelism = max.get(),
                resource_group,
                "default parallelism exceeds max parallelism, capping to max",
            );
        }
        Ok(default_parallelism.min(max))
    }

    /// Builds the actor graph:
    /// - Add the upstream fragments to the fragment graph
    /// - Schedule the fragments based on their distribution
    /// - Expand each fragment into one or several actors
    /// - Construct the fragment level backfill order control.
    #[await_tree::instrument]
    pub(crate) async fn build_stream_job(
        &self,
        stream_ctx: StreamContext,
        mut stream_job: StreamingJob,
        fragment_graph: StreamFragmentGraph,
        specific_resource_group: Option<String>,
    ) -> MetaResult<(CreateStreamingJobContext, StreamJobFragmentsToCreate)> {
        let id = stream_job.id();
        let specified_parallelism = fragment_graph.specified_parallelism();
        let max_parallelism = NonZeroUsize::new(fragment_graph.max_parallelism()).unwrap();

        // 1. Fragment Level ordering graph
        let fragment_backfill_ordering = fragment_graph.create_fragment_backfill_ordering();