refactor: improve stream parallelism resolution and error handling (#23921) (#23926)

Signed-off-by: Peng Chen <[email protected]>
Co-authored-by: Shanicky Chen <[email protected]>
Co-authored-by: Copilot <[email protected]>

Author: 3 people

src/meta/src/rpc/ddl_controller.rs

@@ -1589,8 +1589,24 @@ impl DdlController {
         cluster_info: &StreamingClusterInfo,
         resource_group: String,
     ) -> MetaResult<NonZeroUsize> {
-        let available = cluster_info.parallelism(&resource_group);
-        let Some(available) = NonZeroUsize::new(available) else {
+        let available = NonZeroUsize::new(cluster_info.parallelism(&resource_group));
+        DdlController::resolve_stream_parallelism_inner(
+            specified,
+            max,
+            available,
+            &self.env.opts.default_parallelism,
+            &resource_group,
+        )
+    }
+
+    fn resolve_stream_parallelism_inner(
+        specified: Option<NonZeroUsize>,
+        max: NonZeroUsize,
+        available: Option<NonZeroUsize>,
+        default_parallelism: &DefaultParallelism,
+        resource_group: &str,
+    ) -> MetaResult<NonZeroUsize> {
+        let Some(available) = available else {
             bail_unavailable!(
                 "no available slots to schedule in resource group \"{}\", \
                  have you allocated any compute nodes within this resource group?",
@@ -1607,42 +1623,40 @@ impl DdlController {
                 );
             }
             if specified > available {
-                bail_unavailable!(
-                    "insufficient parallelism to schedule in resource group \"{}\", \
-                     required: {}, available: {}",
+                tracing::warn!(
                     resource_group,
-                    specified,
-                    available,
+                    specified_parallelism = specified.get(),
+                    available_parallelism = available.get(),
+                    "specified parallelism exceeds available slots, scheduling with specified value",
                 );
             }
-            Ok(specified)
-        } else {
-            // Use configured parallelism if no default parallelism is specified.
-            let default_parallelism = match self.env.opts.default_parallelism {
-                DefaultParallelism::Full => available,
-                DefaultParallelism::Default(num) => {
-                    if num > available {
-                        bail_unavailable!(
-                            "insufficient parallelism to schedule in resource group \"{}\", \
-                            required: {}, available: {}",
-                            resource_group,
-                            num,
-                            available,
-                        );
-                    }
-                    num
-                }
-            };
+            return Ok(specified);
+        }
 
-            if default_parallelism > max {
-                tracing::warn!(
-                    max_parallelism = max.get(),
-                    resource_group,
-                    "too many parallelism available, use max parallelism instead",
-                );
+        // 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
             }
-            Ok(default_parallelism.min(max))
+        };
+
+        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:
@@ -2356,3 +2370,79 @@ pub fn refill_upstream_sink_union_in_table(
         }
     });
 }
+
+#[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]
+    fn test_full_parallelism_capped_by_max() {
+        let result = DdlController::resolve_stream_parallelism_inner(
+            None,
+            NonZeroUsize::new(6).unwrap(),
+            Some(NonZeroUsize::new(10).unwrap()),
+            &DefaultParallelism::Full,
+            "default",
+        )
+        .unwrap();
+        assert_eq!(result.get(), 6);
+    }
+
+    #[test]
+    fn test_no_available_slots_returns_error() {
+        let result = DdlController::resolve_stream_parallelism_inner(
+            None,
+            NonZeroUsize::new(4).unwrap(),
+            None,
+            &DefaultParallelism::Full,
+            "default",
+        );
+        assert!(matches!(
+            result,
+            Err(ref e) if matches!(e.inner(), MetaErrorInner::Unavailable(_))
+        ));
+    }
+
+    #[test]
+    fn test_specified_over_max_returns_error() {
+        let result = DdlController::resolve_stream_parallelism_inner(
+            Some(NonZeroUsize::new(8).unwrap()),
+            NonZeroUsize::new(4).unwrap(),
+            Some(NonZeroUsize::new(10).unwrap()),
+            &DefaultParallelism::Full,
+            "default",
+        );
+        assert!(matches!(
+            result,
+            Err(ref e) if matches!(e.inner(), MetaErrorInner::InvalidParameter(_))
+        ));
+    }
+}

src/tests/simulation/tests/integration_tests/scale/mod.rs

@@ -23,6 +23,7 @@ mod nexmark_chaos;
 mod nexmark_q4;
 mod nexmark_source;
 mod no_shuffle;
+mod parallelism_exceeds_cores;
 mod resource_group;
 mod schedulability;
 mod shared_source;

src/tests/simulation/tests/integration_tests/scale/parallelism_exceeds_cores.rs

@@ -0,0 +1,83 @@
+// Copyright 2025 RisingWave Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use anyhow::Result;
+use risingwave_common::hash::VirtualNode;
+use risingwave_simulation::cluster::{Cluster, Configuration};
+use risingwave_simulation::ctl_ext::predicate::identity_contains;
+
+#[tokio::test]
+async fn test_fragment_parallelism_can_exceed_physical_cores() -> Result<()> {
+    let configuration = Configuration::for_scale();
+    let total_cores = configuration.total_streaming_cores() as usize;
+    let mut cluster = Cluster::start(configuration).await?;
+    let max_parallelism = VirtualNode::COUNT_FOR_COMPAT;
+
+    // Expect creating a job with parallelism larger than max_parallelism to be rejected.
+    let mut session = cluster.start_session();
+
+    session
+        .run(format!(
+            "set streaming_parallelism = {}",
+            max_parallelism + 1
+        ))
+        .await?;
+
+    assert!(session.run("create table t (v int);").await.is_err());
+
+    let new_parallelism = total_cores + 1;
+
+    session
+        .run(format!("set streaming_parallelism = {new_parallelism}"))
+        .await?;
+
+    // With a value within max_parallelism, creation succeeds; we will bump further afterward.
+    session.run("create table t (v int);").await?;
+
+    let fragment = cluster
+        .locate_one_fragment([identity_contains("materialize")])
+        .await?;
+    let fragment_id = fragment.id();
+    assert_eq!(fragment.inner.actors.len(), new_parallelism);
+
+    let new_parallelism = total_cores + 2;
+
+    // Raise table parallelism beyond physical cores; fragment actor count should follow.
+    session
+        .run(&format!(
+            "alter table t set parallelism = {new_parallelism};"
+        ))
+        .await?;
+
+    let updated_fragment = cluster
+        .locate_one_fragment([identity_contains("materialize")])
+        .await?;
+    assert_eq!(updated_fragment.inner.actors.len(), new_parallelism);
+
+    let new_parallelism = total_cores + 3;
+
+    // Alter fragment directly to an even higher parallelism and verify actor count.
+    session
+        .run(&format!(
+            "alter fragment {fragment_id} set parallelism = {new_parallelism};"
+        ))
+        .await?;
+
+    let updated_fragment = cluster
+        .locate_one_fragment([identity_contains("materialize")])
+        .await?;
+    assert_eq!(updated_fragment.inner.actors.len(), new_parallelism);
+
+    Ok(())
+}