fix: drop tenant not drop members bug(#1452) (#1626)

* fix: drop tenant not drop members bug(#1452)

* ci: add test case for #1452

* refactor: move drop members and roles to meta

* refactor: be in common use drop database

coordinator/src/resource_manager.rs

@@ -3,7 +3,6 @@ use std::sync::Arc;
 
 use meta::model::meta_admin::AdminMeta;
 use models::meta_data::ReplicationSet;
-use models::oid::Identifier;
 use models::schema::{ResourceInfo, ResourceOperator, ResourceStatus, TableSchema};
 use models::utils::now_timestamp_nanos;
 use protos::kv_service::admin_command_request::Command::{self, DropDb, DropTab, UpdateTags};
@@ -165,27 +164,12 @@ impl ResourceManager {
                     name: tenant_name.to_string(),
                 })?;
 
-        // drop role in the tenant
-        let all_roles = tenant
-            .custom_roles()
-            .await
-            .map_err(|err| CoordinatorError::Meta { source: err })?;
-        for role in all_roles {
-            tenant
-                .drop_custom_role(role.name())
-                .await
-                .map_err(|err| CoordinatorError::Meta { source: err })?;
-        }
-
         // drop database in the tenant
         let all_dbs = tenant
             .list_databases()
             .map_err(|err| CoordinatorError::Meta { source: err })?;
         for db_name in all_dbs {
-            tenant
-                .drop_db(&db_name)
-                .await
-                .map_err(|err| CoordinatorError::Meta { source: err })?;
+            ResourceManager::drop_database(coord.clone(), tenant_name, &db_name).await?;
         }
 
         // drop tenant metadata

meta/src/store/storage.rs

@@ -1241,6 +1241,21 @@ impl StateMachine {
     }
 
     fn process_drop_tenant(&self, cluster: &str, name: &str) -> MetaResult<()> {
+        // remove members in the tenant
+        let members = self.process_read_members(cluster, name)?;
+        let mut users = self.process_read_users(cluster)?;
+        users.retain(|user| members.iter().any(|member| user.name() == member.0));
+        for user in users {
+            self.process_remove_member_to_tenant(cluster, user.id(), name)?;
+        }
+
+        // drop role in the tenant
+        let roles = self.process_read_roles(cluster, name)?;
+        for role in roles {
+            self.process_drop_role(cluster, role.name(), name)?;
+        }
+
+        // drop tenant meta
         let key = KeyPath::tenant(cluster, name);
         let limiter_key = KeyPath::limiter(cluster, name);
 

query_server/sqllogicaltests/cases/ddl/drop_tenant.slt

@@ -0,0 +1,20 @@
+statement ok
+drop tenant if exists tenant1;
+
+statement ok
+create tenant if not exists tenant1;
+
+statement ok
+create user if not exists user1;
+
+statement ok
+alter tenant tenant1 add user user1 as owner;
+
+statement ok
+drop tenant if exists tenant1;
+
+statement ok
+create tenant if not exists tenant1;
+
+statement ok
+alter tenant tenant1 add user user1 as owner;