filter out singletons in table fragments max parallelism

Signed-off-by: Bugen Zhao <[email protected]>

proto/meta.proto

@@ -243,7 +243,7 @@ message ListTableFragmentStatesResponse {
     uint32 table_id = 1;
     TableFragments.State state = 2;
     TableParallelism parallelism = 3;
-    uint32 max_vnode_count = 4;
+    uint32 max_parallelism = 4;
   }
   repeated TableFragmentState states = 1;
 }

src/frontend/src/catalog/system_catalog/rw_catalog/rw_table_fragments.rs

@@ -41,7 +41,7 @@ async fn read_rw_table_fragments_info(
                 table_id: state.table_id as i32,
                 status: state.state().as_str_name().into(),
                 parallelism: parallelism.to_uppercase(),
-                max_parallelism: state.max_vnode_count as i32,
+                max_parallelism: state.max_parallelism as i32,
             }
         })
         .collect())

src/meta/service/src/stream_service.rs

@@ -271,7 +271,7 @@ impl StreamManagerService for StreamServiceImpl {
                             table_id: tf.table_id().table_id,
                             state: tf.state() as i32,
                             parallelism: Some(tf.assigned_parallelism.into()),
-                            max_vnode_count: tf.max_vnode_count() as _,
+                            max_parallelism: tf.max_parallelism() as _,
                         },
                     )
                     .collect_vec()
@@ -293,7 +293,7 @@ impl StreamManagerService for StreamServiceImpl {
                             table_id: table_id as _,
                             state: PbState::from(state) as _,
                             parallelism: Some(parallelism.into()),
-                            max_vnode_count: 0, // TODO(var-vnode): write query to obtain this from `fragment`
+                            max_parallelism: 0, // TODO(var-vnode): write query to obtain this from `fragment`
                         }
                     })
                     .collect_vec()

src/meta/src/model/stream.rs

@@ -21,6 +21,7 @@ use risingwave_common::hash::{VirtualNode, VnodeCountCompat, WorkerSlotId};
 use risingwave_connector::source::SplitImpl;
 use risingwave_pb::common::PbActorLocation;
 use risingwave_pb::meta::table_fragments::actor_status::ActorState;
+use risingwave_pb::meta::table_fragments::fragment::FragmentDistributionType;
 use risingwave_pb::meta::table_fragments::{ActorStatus, Fragment, State};
 use risingwave_pb::meta::table_parallelism::{
     FixedParallelism, Parallelism, PbAdaptiveParallelism, PbCustomParallelism, PbFixedParallelism,
@@ -282,16 +283,15 @@ impl TableFragments {
         self.ctx.timezone.clone()
     }
 
-    /// Returns the max vnode count of all fragments in the table.
-    ///
-    /// Setting the parallelism of a streaming job to a value greater than this will result in
-    /// an error, as there won't be any fragment that fits that value.
-    pub fn max_vnode_count(&self) -> usize {
+    /// Returns the maximum value of the `vnode_count` of all hash-distributed fragments.
+    /// Returns 1 if all fragments are singleton.
+    pub fn max_parallelism(&self) -> usize {
         self.fragments
             .values()
+            .filter(|f| matches!(f.distribution_type(), FragmentDistributionType::Hash))
             .map(|f| f.vnode_count())
             .max()
-            .expect("should be at least one fragment")
+            .unwrap_or(1) // if all fragments are singleton, return 1
     }
 
     /// Returns whether the table fragments is in `Created` state.