feat: add risectl scale check command for sql in version 2.0.x (#19761)

Signed-off-by: Shanicky Chen <peng@risingwave-labs.com>

src/ctl/src/cmd_impl/meta.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 mod backup_meta;
+mod check;
 mod cluster_info;
 mod connection;
 mod migration;
@@ -21,6 +22,7 @@ mod reschedule;
 mod serving;
 
 pub use backup_meta::*;
+pub use check::*;
 pub use cluster_info::*;
 pub use connection::*;
 pub use migration::*;

src/ctl/src/cmd_impl/meta/check.rs

@@ -0,0 +1,33 @@
+// 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::process::exit;
+
+use risingwave_meta::controller::catalog::CatalogController;
+use sea_orm::TransactionTrait;
+
+pub async fn graph_check(endpoint: String) -> anyhow::Result<()> {
+    let conn = sea_orm::Database::connect(sea_orm::ConnectOptions::new(endpoint)).await?;
+    let txn = conn.begin().await?;
+    match CatalogController::graph_check(&txn).await {
+        Ok(_) => {
+            println!("integrity check passed!");
+            exit(0);
+        }
+        Err(_) => {
+            println!("integrity check failed!");
+            exit(1);
+        }
+    }
+}

src/ctl/src/lib.rs

@@ -514,6 +514,14 @@ enum MetaCommands {
         #[clap(short = 'f', long, default_value_t = false)]
         force_clean: bool,
     },
+
+    /// Performing graph check for scaling.
+    #[clap(verbatim_doc_comment)]
+    GraphCheck {
+        /// SQL endpoint
+        #[clap(long, required = true)]
+        endpoint: String,
+    },
 }
 
 #[derive(Subcommand, Clone, Debug)]
@@ -850,6 +858,9 @@ async fn start_impl(opts: CliOpts, context: &CtlContext) -> Result<()> {
         Commands::Meta(MetaCommands::ValidateSource { props }) => {
             cmd_impl::meta::validate_source(context, props).await?
         }
+        Commands::Meta(MetaCommands::GraphCheck { endpoint }) => {
+            cmd_impl::meta::graph_check(endpoint).await?
+        }
         Commands::Meta(MetaCommands::Migration {
             etcd_endpoints,
             etcd_user_password,
@@ -881,6 +892,7 @@ async fn start_impl(opts: CliOpts, context: &CtlContext) -> Result<()> {
         Commands::Profile(ProfileCommands::Heap { dir }) => {
             cmd_impl::profile::heap_profile(context, dir).await?
         }
+
         Commands::Scale(ScaleCommands::Cordon { workers }) => {
             cmd_impl::scale::update_schedulability(context, workers, Schedulability::Unschedulable)
                 .await?