feat: introduce cluster limit

e2e_test/batch/catalog/pg_settings.slt.part

@@ -22,6 +22,7 @@ user backfill_rate_limit
 user background_ddl
 user batch_enable_distributed_dml
 user batch_parallelism
+user bypass_cluster_limits
 user bytea_output
 user cdc_source_wait_streaming_start_timeout
 user client_encoding

proto/meta.proto

@@ -789,3 +789,30 @@ message RelationIdInfos {
   // relation_id -> FragmentIdToActorIdMap
   map<uint32, FragmentIdToActorIdMap> map = 1;
 }
+
+message ActorCountPerParallelism {
+  message WorkerActorCount {
+    uint64 actor_count = 1;
+    uint64 parallelism = 2;
+  }
+  map<uint32, WorkerActorCount> worker_id_to_actor_count = 1;
+  uint64 hard_limit = 2;
+  uint64 soft_limit = 3;
+}
+
+message ClusterLimit {
+  oneof limit {
+    ActorCountPerParallelism actor_count = 1;
+    // TODO: limit DDL using compaction pending bytes
+  }
+}
+
+message GetClusterLimitsRequest {}
+
+message GetClusterLimitsResponse {
+  repeated ClusterLimit active_limits = 1;
+}
+
+service ClusterLimitService {
+  rpc GetClusterLimits(GetClusterLimitsRequest) returns (GetClusterLimitsResponse);
+}

src/common/src/config.rs

@@ -466,6 +466,16 @@ pub struct MetaDeveloperConfig {
 
     #[serde(default = "default::developer::max_get_task_probe_times")]
     pub max_get_task_probe_times: usize,
+
+    /// Max number of actor allowed per parallelism (default = 100).
+    /// CREATE MV/Table will be noticed when the number of actors exceeds this limit.
+    #[serde(default = "default::developer::actor_cnt_per_worker_parallelism_soft_limit")]
+    pub actor_cnt_per_worker_parallelism_soft_limit: usize,
+
+    /// Max number of actor allowed per parallelism (default = 400).
+    /// CREATE MV/Table will be rejected when the number of actors exceeds this limit.
+    #[serde(default = "default::developer::actor_cnt_per_worker_parallelism_hard_limit")]
+    pub actor_cnt_per_worker_parallelism_hard_limit: usize,
 }
 
 /// The section `[server]` in `risingwave.toml`.
@@ -1859,6 +1869,14 @@ pub mod default {
             5
         }
 
+        pub fn actor_cnt_per_worker_parallelism_soft_limit() -> usize {
+            100
+        }
+
+        pub fn actor_cnt_per_worker_parallelism_hard_limit() -> usize {
+            400
+        }
+
         pub fn memory_controller_threshold_aggressive() -> f64 {
             0.9
         }

src/common/src/session_config/mod.rs

@@ -292,6 +292,12 @@ pub struct SessionConfig {
 
     #[parameter(default = "hex", check_hook = check_bytea_output)]
     bytea_output: String,
+
+    /// Bypass checks on cluster limits
+    ///
+    /// When enabled, `CREATE MATERIALIZED VIEW` will not fail if the cluster limit is hit.
+    #[parameter(default = false)]
+    bypass_cluster_limits: bool,
 }
 
 fn check_timezone(val: &str) -> Result<(), String> {

src/common/src/util/cluster_limit.rs

@@ -0,0 +1,134 @@
+// Copyright 2024 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 std::collections::HashMap;
+use std::fmt::{self, Display, Formatter};
+
+use risingwave_pb::meta::actor_count_per_parallelism::PbWorkerActorCount;
+use risingwave_pb::meta::cluster_limit::PbLimit;
+use risingwave_pb::meta::{PbActorCountPerParallelism, PbClusterLimit};
+pub enum ClusterLimit {
+    ActorCount(ActorCountPerParallelism),
+}
+
+impl From<ClusterLimit> for PbClusterLimit {
+    fn from(limit: ClusterLimit) -> Self {
+        match limit {
+            ClusterLimit::ActorCount(actor_count_per_parallelism) => PbClusterLimit {
+                limit: Some(PbLimit::ActorCount(actor_count_per_parallelism.into())),
+            },
+        }
+    }
+}
+
+impl From<PbClusterLimit> for ClusterLimit {
+    fn from(pb_limit: PbClusterLimit) -> Self {
+        match pb_limit.limit.unwrap() {
+            PbLimit::ActorCount(actor_count_per_parallelism) => {
+                ClusterLimit::ActorCount(actor_count_per_parallelism.into())
+            }
+        }
+    }
+}
+
+#[derive(Debug)]
+pub struct WorkerActorCount {
+    pub actor_count: usize,
+    pub parallelism: usize,
+}
+
+impl From<WorkerActorCount> for PbWorkerActorCount {
+    fn from(worker_actor_count: WorkerActorCount) -> Self {
+        PbWorkerActorCount {
+            actor_count: worker_actor_count.actor_count as u64,
+            parallelism: worker_actor_count.parallelism as u64,
+        }
+    }
+}
+
+impl From<PbWorkerActorCount> for WorkerActorCount {
+    fn from(pb_worker_actor_count: PbWorkerActorCount) -> Self {
+        WorkerActorCount {
+            actor_count: pb_worker_actor_count.actor_count as usize,
+            parallelism: pb_worker_actor_count.parallelism as usize,
+        }
+    }
+}
+
+pub struct ActorCountPerParallelism {
+    pub worker_id_to_actor_count: HashMap<u32, WorkerActorCount>,
+    pub hard_limit: usize,
+    pub soft_limit: usize,
+}
+
+impl From<ActorCountPerParallelism> for PbActorCountPerParallelism {
+    fn from(actor_count_per_parallelism: ActorCountPerParallelism) -> Self {
+        PbActorCountPerParallelism {
+            worker_id_to_actor_count: actor_count_per_parallelism
+                .worker_id_to_actor_count
+                .into_iter()
+                .map(|(k, v)| (k, v.into()))
+                .collect(),
+            hard_limit: actor_count_per_parallelism.hard_limit as u64,
+            soft_limit: actor_count_per_parallelism.soft_limit as u64,
+        }
+    }
+}
+
+impl From<PbActorCountPerParallelism> for ActorCountPerParallelism {
+    fn from(pb_actor_count_per_parallelism: PbActorCountPerParallelism) -> Self {
+        ActorCountPerParallelism {
+            worker_id_to_actor_count: pb_actor_count_per_parallelism
+                .worker_id_to_actor_count
+                .into_iter()
+                .map(|(k, v)| (k, v.into()))
+                .collect(),
+            hard_limit: pb_actor_count_per_parallelism.hard_limit as usize,
+            soft_limit: pb_actor_count_per_parallelism.soft_limit as usize,
+        }
+    }
+}
+
+impl ActorCountPerParallelism {
+    pub fn exceed_hard_limit(&self) -> bool {
+        self.worker_id_to_actor_count
+            .values()
+            .any(|v| v.actor_count > self.hard_limit.saturating_mul(v.parallelism))
+    }
+
+    pub fn exceed_soft_limit(&self) -> bool {
+        self.worker_id_to_actor_count
+            .values()
+            .any(|v| v.actor_count > self.soft_limit.saturating_mul(v.parallelism))
+    }
+
+    pub fn exceed_limit(&self) -> bool {
+        self.exceed_soft_limit() || self.exceed_hard_limit()
+    }
+}
+
+impl Display for ActorCountPerParallelism {
+    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
+        let worker_id_to_actor_count_str: Vec<_> = self
+            .worker_id_to_actor_count
+            .iter()
+            .map(|(k, v)| format!("{} -> {:?}", k, v))
+            .collect();
+        write!(
+            f,
+            "ActorCountPerParallelism {{ critical limit: {:?}, recommended limit: {:?}. worker_id_to_actor_count: {:?} }}",
+            self.hard_limit, self.soft_limit, worker_id_to_actor_count_str
+        )
+    }
+}

src/common/src/util/mod.rs

@@ -42,3 +42,4 @@ pub mod tracing;
 pub mod value_encoding;
 pub mod worker_util;
 pub use tokio_util;
+pub mod cluster_limit;

src/config/example.toml

@@ -81,6 +81,8 @@ meta_enable_trivial_move = true
 meta_enable_check_task_level_overlap = false
 meta_max_trivial_move_task_count_per_loop = 256
 meta_max_get_task_probe_times = 5
+meta_actor_cnt_per_worker_parallelism_soft_limit = 100
+meta_actor_cnt_per_worker_parallelism_hard_limit = 400
 
 [batch]
 enable_barrier_read = false

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

@@ -59,3 +59,4 @@ mod rw_worker_nodes;
 
 mod rw_actor_id_to_ddl;
 mod rw_fragment_id_to_ddl;
+mod rw_worker_actor_count;

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

@@ -0,0 +1,31 @@
+// Copyright 2024 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 risingwave_common::types::Fields;
+use risingwave_frontend_macro::system_catalog;
+
+#[system_catalog(
+    view,
+    "rw_catalog.rw_worker_actor_count",
+    "SELECT t2.id as worker_id, parallelism, count(*) as actor_count
+     FROM rw_actors t1, rw_worker_nodes t2
+     where t1.worker_id = t2.id
+     GROUP  BY t2.id, t2.parallelism;"
+)]
+#[derive(Fields)]
+struct RwWorkerActorCount {
+    worker_id: i32,
+    parallelism: i32,
+    actor_count: i64,
+}

src/frontend/src/handler/create_mv.rs

@@ -205,6 +205,9 @@ pub async fn handle_create_mv_bound(
 ) -> Result<RwPgResponse> {
     let session = handler_args.session.clone();
 
+    // Check cluster limits
+    session.check_cluster_limits().await?;
+
     if let Either::Right(resp) = session.check_relation_name_duplicated(
         name.clone(),
         StatementType::CREATE_MATERIALIZED_VIEW,

src/frontend/src/handler/create_sink.rs

@@ -415,6 +415,8 @@ pub async fn handle_create_sink(
 ) -> Result<RwPgResponse> {
     let session = handle_args.session.clone();
 
+    session.check_cluster_limits().await?;
+
     if let Either::Right(resp) = session.check_relation_name_duplicated(
         stmt.sink_name.clone(),
         StatementType::CREATE_SINK,

src/frontend/src/handler/create_table.rs

@@ -1197,6 +1197,8 @@ pub async fn handle_create_table(
         session.notice_to_user("APPEND ONLY TABLE is currently an experimental feature.");
     }
 
+    session.check_cluster_limits().await?;
+
     if let Either::Right(resp) = session.check_relation_name_duplicated(
         table_name.clone(),
         StatementType::CREATE_TABLE,

src/frontend/src/meta_client.rs

@@ -17,6 +17,7 @@ use std::collections::HashMap;
 use anyhow::Context;
 use risingwave_common::session_config::SessionConfig;
 use risingwave_common::system_param::reader::SystemParamsReader;
+use risingwave_common::util::cluster_limit::ClusterLimit;
 use risingwave_hummock_sdk::version::{HummockVersion, HummockVersionDelta};
 use risingwave_hummock_sdk::HummockVersionId;
 use risingwave_pb::backup_service::MetaSnapshotMetadata;
@@ -136,6 +137,8 @@ pub trait FrontendMetaClient: Send + Sync {
     ) -> Result<Vec<u64>>;
 
     async fn get_cluster_recovery_status(&self) -> Result<RecoveryStatus>;
+
+    async fn get_cluster_limits(&self) -> Result<Vec<ClusterLimit>>;
 }
 
 pub struct FrontendMetaClientImpl(pub MetaClient);
@@ -345,4 +348,8 @@ impl FrontendMetaClient for FrontendMetaClientImpl {
     async fn get_cluster_recovery_status(&self) -> Result<RecoveryStatus> {
         self.0.get_cluster_recovery_status().await
     }
+
+    async fn get_cluster_limits(&self) -> Result<Vec<ClusterLimit>> {
+        self.0.get_cluster_limits().await
+    }
 }

src/frontend/src/session.rs

@@ -59,9 +59,10 @@ use risingwave_common::telemetry::manager::TelemetryManager;
 use risingwave_common::telemetry::telemetry_env_enabled;
 use risingwave_common::types::DataType;
 use risingwave_common::util::addr::HostAddr;
+use risingwave_common::util::cluster_limit::ActorCountPerParallelism;
 use risingwave_common::util::iter_util::ZipEqFast;
-use risingwave_common::util::resource_util;
 use risingwave_common::util::runtime::BackgroundShutdownRuntime;
+use risingwave_common::util::{cluster_limit, resource_util};
 use risingwave_common::{GIT_SHA, RW_VERSION};
 use risingwave_common_heap_profiling::HeapProfiler;
 use risingwave_common_service::{MetricsManager, ObserverManager};
@@ -1194,6 +1195,47 @@ impl SessionImpl {
     pub fn temporary_source_manager(&self) -> TemporarySourceManager {
         self.temporary_source_manager.lock().clone()
     }
+
+    pub async fn check_cluster_limits(&self) -> Result<()> {
+        if self.config().bypass_cluster_limits() {
+            return Ok(());
+        }
+
+        let gen_message = |violated_limit: &ActorCountPerParallelism,
+                           exceed_hard_limit: bool|
+         -> String {
+            let (limit_type, action) = if exceed_hard_limit {
+                ("critical", "Please scale the cluster before proceeding!")
+            } else {
+                ("recommended", "Scaling the cluster is recommended.")
+            };
+            format!(
+                "\n- {}\n- {}\n- {}\n- {}\n- {}\n{}",
+                format_args!("Actor count per parallelism exceeds the {} limit.", limit_type),
+                format_args!("Depending on your workload, this may overload the cluster and cause performance/stability issues. {}", action),
+                "Contact us via slack or https://risingwave.com/contact-us/ for further enquiry.",
+                "You can bypass this check via SQL `SET bypass_cluster_limits TO true`.",
+                "You can check actor count distribution via SQL `SELECT * FROM rw_worker_actor_count`.",
+                violated_limit,
+            )
+        };
+
+        let limits = self.env().meta_client().get_cluster_limits().await?;
+        for limit in limits {
+            match limit {
+                cluster_limit::ClusterLimit::ActorCount(l) => {
+                    if l.exceed_hard_limit() {
+                        return Err(RwError::from(ErrorCode::ProtocolError(gen_message(
+                            &l, true,
+                        ))));
+                    } else if l.exceed_soft_limit() {
+                        self.notice_to_user(gen_message(&l, false));
+                    }
+                }
+            }
+        }
+        Ok(())
+    }
 }
 
 pub static SESSION_MANAGER: std::sync::OnceLock<Arc<SessionManagerImpl>> =