fix(cluster): skip the recovery for warehouses that are not running state. (#17193)

* fix(cluster): fix recovery suspended warehouse

* fix(cluster): fix recovery suspended warehouse

Author: zhang2014

src/query/management/src/warehouse/warehouse_mgr.rs

@@ -373,6 +373,10 @@ impl WarehouseMgr {
                 continue;
             };
 
+            if wh.status.to_uppercase() != "RUNNING" {
+                continue;
+            }
+
             for (cluster_id, cluster) in wh.clusters {
                 let mut lost_nodes = cluster.nodes;
 

src/query/management/tests/it/warehouse.rs

@@ -534,6 +534,48 @@ async fn test_create_warehouse_with_no_resources() -> Result<()> {
     Ok(())
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+async fn test_recovery_with_suspended_warehouse() -> Result<()> {
+    let (_, warehouse_manager, nodes) = nodes(Duration::from_mins(30), 2).await?;
+
+    let create_warehouse = warehouse_manager.create_warehouse(
+        String::from("test_warehouse"),
+        vec![SelectedNode::Random(None); 2],
+    );
+    create_warehouse.await?;
+
+    let list_warehouse_nodes =
+        warehouse_manager.list_warehouse_nodes(String::from("test_warehouse"));
+
+    assert_eq!(list_warehouse_nodes.await?.len(), 2);
+
+    warehouse_manager
+        .suspend_warehouse(String::from("test_warehouse"))
+        .await?;
+
+    let shutdown_node = warehouse_manager.shutdown_node(nodes[0].clone());
+    shutdown_node.await?;
+
+    let shutdown_node = warehouse_manager.shutdown_node(nodes[1].clone());
+    shutdown_node.await?;
+
+    let list_warehouse_nodes =
+        warehouse_manager.list_warehouse_nodes(String::from("test_warehouse"));
+
+    assert_eq!(list_warehouse_nodes.await?.len(), 0);
+
+    let node_1 = GlobalUniqName::unique();
+    let start_node_1 = warehouse_manager.start_node(system_managed_node(&node_1));
+    assert!(start_node_1.await.is_ok());
+
+    let list_warehouse_nodes =
+        warehouse_manager.list_warehouse_nodes(String::from("test_warehouse"));
+
+    let nodes = list_warehouse_nodes.await?;
+    assert_eq!(nodes.len(), 0);
+    Ok(())
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
 async fn test_recovery_create_warehouse() -> Result<()> {
     let (_, warehouse_manager, nodes) = nodes(Duration::from_mins(30), 2).await?;

src/query/service/src/interpreters/interpreter_factory.rs

@@ -161,6 +161,11 @@ impl InterpreterFactory {
             Plan::UnassignWarehouseNodes(v) => Ok(Arc::new(
                 UnassignWarehouseNodesInterpreter::try_create(ctx.clone(), *v.clone())?,
             )),
+            // We allow the execution of SET statements because this could be SET GLOBAL enterprise_license.
+            Plan::Set(set_variable) => Ok(Arc::new(SetInterpreter::try_create(
+                ctx,
+                *set_variable.clone(),
+            )?)),
             Plan::Query { metadata, .. } => {
                 let read_guard = metadata.read();
                 for table in read_guard.tables() {