diff --git a/src/meta/service/src/stream_service.rs b/src/meta/service/src/stream_service.rs index f16a5a6ea1f5..90665a611cb9 100644 --- a/src/meta/service/src/stream_service.rs +++ b/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_parallelism: tf.max_parallelism() as _, + max_parallelism: tf.max_parallelism as _, }, ) .collect_vec() diff --git a/src/meta/src/model/stream.rs b/src/meta/src/model/stream.rs index da68ba469fa6..954ee820e6a7 100644 --- a/src/meta/src/model/stream.rs +++ b/src/meta/src/model/stream.rs @@ -17,11 +17,10 @@ use std::ops::AddAssign; use itertools::Itertools; use risingwave_common::catalog::TableId; -use risingwave_common::hash::{VirtualNode, VnodeCountCompat, WorkerSlotId}; +use risingwave_common::hash::{VirtualNode, 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, @@ -283,17 +282,6 @@ impl TableFragments { self.ctx.timezone.clone() } - /// 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() - .unwrap_or(1) // if all fragments are singleton, return 1 - } - /// Returns whether the table fragments is in `Created` state. pub fn is_created(&self) -> bool { self.state == State::Created