refactor: SHOW GRANTS should not display drop on table/database (#14931)

feat: show grants not display droped table/database

src/meta/api/src/schema_api.rs

@@ -216,13 +216,13 @@ pub trait SchemaApi: Send + Sync {
     async fn mget_table_names_by_ids(
         &self,
         table_ids: &[MetaId],
-    ) -> Result<Vec<String>, KVAppError>;
+    ) -> Result<Vec<Option<String>>, KVAppError>;
 
     async fn get_table_name_by_id(&self, table_id: MetaId) -> Result<String, KVAppError>;
     async fn mget_database_names_by_ids(
         &self,
         db_ids: &[MetaId],
-    ) -> Result<Vec<String>, KVAppError>;
+    ) -> Result<Vec<Option<String>>, KVAppError>;
 
     async fn get_db_name_by_id(&self, db_id: MetaId) -> Result<String, KVAppError>;
 

src/meta/api/src/schema_api_impl.rs

@@ -2441,24 +2441,41 @@ impl<KV: kvapi::KVApi<Error = MetaError> + ?Sized> SchemaApi for KV {
     async fn mget_table_names_by_ids(
         &self,
         table_ids: &[MetaId],
-    ) -> Result<Vec<String>, KVAppError> {
+    ) -> Result<Vec<Option<String>>, KVAppError> {
         debug!(req :? =(&table_ids); "SchemaApi: {}", func_name!());
 
-        let mut kv_keys = Vec::with_capacity(table_ids.len());
+        let mut id_name_kv_keys = Vec::with_capacity(table_ids.len());
         for id in table_ids {
             let k = TableIdToName { table_id: *id }.to_string_key();
-            kv_keys.push(k);
+            id_name_kv_keys.push(k);
         }
 
         // Batch get all table-name by id
-        let seq_names = self.mget_kv(&kv_keys).await?;
-        let mut table_names = Vec::with_capacity(kv_keys.len());
+        let seq_names = self.mget_kv(&id_name_kv_keys).await?;
+        let mut table_names = Vec::with_capacity(id_name_kv_keys.len());
 
-        // None means table_name not found, maybe immuteable table id. Ignore it
+        // None means table_name not found, maybe immutable table id. Ignore it
         for seq_name in seq_names.into_iter().flatten() {
             let name_ident: DBIdTableName = deserialize_struct(&seq_name.data)?;
-            table_names.push(name_ident.table_name);
+            table_names.push(Some(name_ident.table_name));
+        }
+
+        let mut meta_kv_keys = Vec::with_capacity(table_ids.len());
+        for id in table_ids {
+            let k = TableId { table_id: *id }.to_string_key();
+            meta_kv_keys.push(k);
+        }
+
+        let seq_metas = self.mget_kv(&meta_kv_keys).await?;
+        for (i, seq_meta_opt) in seq_metas.iter().enumerate() {
+            if let Some(seq_meta) = seq_meta_opt {
+                let table_meta: TableMeta = deserialize_struct(&seq_meta.data)?;
+                if table_meta.drop_on.is_some() {
+                    table_names[i] = None;
+                }
+            }
         }
+
         Ok(table_names)
     }
 
@@ -2488,7 +2505,7 @@ impl<KV: kvapi::KVApi<Error = MetaError> + ?Sized> SchemaApi for KV {
     async fn mget_database_names_by_ids(
         &self,
         db_ids: &[MetaId],
-    ) -> Result<Vec<String>, KVAppError> {
+    ) -> Result<Vec<Option<String>>, KVAppError> {
         debug!(req :? =(&db_ids); "SchemaApi: {}", func_name!());
 
         let mut kv_keys = Vec::with_capacity(db_ids.len());
@@ -2501,10 +2518,26 @@ impl<KV: kvapi::KVApi<Error = MetaError> + ?Sized> SchemaApi for KV {
         let seq_names = self.mget_kv(&kv_keys).await?;
         let mut db_names = Vec::with_capacity(kv_keys.len());
 
-        // None means db_name not found, maybe immuteable database id. Ignore it
+        // None means db_name not found, maybe immutable database id. Ignore it
         for seq_name in seq_names.into_iter().flatten() {
             let name_ident: DatabaseNameIdent = deserialize_struct(&seq_name.data)?;
-            db_names.push(name_ident.db_name);
+            db_names.push(Some(name_ident.db_name));
+        }
+
+        let mut meta_kv_keys = Vec::with_capacity(db_ids.len());
+        for id in db_ids {
+            let k = DatabaseId { db_id: *id }.to_string_key();
+            meta_kv_keys.push(k);
+        }
+
+        let seq_metas = self.mget_kv(&meta_kv_keys).await?;
+        for (i, seq_meta_opt) in seq_metas.iter().enumerate() {
+            if let Some(seq_meta) = seq_meta_opt {
+                let db_meta: DatabaseMeta = deserialize_struct(&seq_meta.data)?;
+                if db_meta.drop_on.is_some() {
+                    db_names[i] = None;
+                }
+            }
         }
         Ok(db_names)
     }

src/query/catalog/src/catalog/interface.rs

@@ -197,7 +197,7 @@ pub trait Catalog: DynClone + Send + Sync + Debug {
     async fn mget_table_names_by_ids(
         &self,
         table_ids: &[MetaId],
-    ) -> databend_common_exception::Result<Vec<String>>;
+    ) -> databend_common_exception::Result<Vec<Option<String>>>;
 
     // Mget the db name by meta id.
     async fn get_db_name_by_id(&self, db_ids: MetaId) -> databend_common_exception::Result<String>;
@@ -206,7 +206,7 @@ pub trait Catalog: DynClone + Send + Sync + Debug {
     async fn mget_database_names_by_ids(
         &self,
         db_ids: &[MetaId],
-    ) -> databend_common_exception::Result<Vec<String>>;
+    ) -> databend_common_exception::Result<Vec<Option<String>>>;
 
     // Get one table by db and table name.
     async fn get_table(

src/query/catalog/src/catalog/session_catalog.rs

@@ -252,7 +252,7 @@ impl Catalog for SessionCatalog {
     async fn mget_table_names_by_ids(
         &self,
         table_ids: &[MetaId],
-    ) -> databend_common_exception::Result<Vec<String>> {
+    ) -> databend_common_exception::Result<Vec<Option<String>>> {
         self.inner.mget_table_names_by_ids(table_ids).await
     }
 
@@ -265,7 +265,7 @@ impl Catalog for SessionCatalog {
     async fn mget_database_names_by_ids(
         &self,
         db_ids: &[MetaId],
-    ) -> databend_common_exception::Result<Vec<String>> {
+    ) -> databend_common_exception::Result<Vec<Option<String>>> {
         self.inner.mget_database_names_by_ids(db_ids).await
     }
 

src/query/service/src/catalogs/default/database_catalog.rs

@@ -301,7 +301,7 @@ impl Catalog for DatabaseCatalog {
     }
 
     #[async_backtrace::framed]
-    async fn mget_table_names_by_ids(&self, table_ids: &[MetaId]) -> Result<Vec<String>> {
+    async fn mget_table_names_by_ids(&self, table_ids: &[MetaId]) -> Result<Vec<Option<String>>> {
         let mut tables = self
             .immutable_catalog
             .mget_table_names_by_ids(table_ids)
@@ -326,7 +326,7 @@ impl Catalog for DatabaseCatalog {
     }
 
     #[async_backtrace::framed]
-    async fn mget_database_names_by_ids(&self, db_ids: &[MetaId]) -> Result<Vec<String>> {
+    async fn mget_database_names_by_ids(&self, db_ids: &[MetaId]) -> Result<Vec<Option<String>>> {
         let mut dbs = self
             .immutable_catalog
             .mget_database_names_by_ids(db_ids)

src/query/service/src/catalogs/default/immutable_catalog.rs

@@ -205,11 +205,11 @@ impl Catalog for ImmutableCatalog {
     async fn mget_table_names_by_ids(
         &self,
         table_ids: &[MetaId],
-    ) -> databend_common_exception::Result<Vec<String>> {
+    ) -> databend_common_exception::Result<Vec<Option<String>>> {
         let mut table_name = Vec::with_capacity(table_ids.len());
         for id in table_ids {
             if let Some(table) = self.sys_db_meta.get_by_id(id) {
-                table_name.push(table.name().to_string());
+                table_name.push(Some(table.name().to_string()));
             }
         }
         Ok(table_name)
@@ -228,13 +228,13 @@ impl Catalog for ImmutableCatalog {
         }
     }
 
-    async fn mget_database_names_by_ids(&self, db_ids: &[MetaId]) -> Result<Vec<String>> {
+    async fn mget_database_names_by_ids(&self, db_ids: &[MetaId]) -> Result<Vec<Option<String>>> {
         let mut res = Vec::new();
         for id in db_ids {
             if self.sys_db.get_db_info().ident.db_id == *id {
-                res.push("system".to_string());
+                res.push(Some("system".to_string()));
             } else if self.info_schema_db.get_db_info().ident.db_id == *id {
-                res.push("information_schema".to_string());
+                res.push(Some("information_schema".to_string()));
             }
         }
         Ok(res)

src/query/service/src/catalogs/default/mutable_catalog.rs

@@ -374,7 +374,7 @@ impl Catalog for MutableCatalog {
     async fn mget_table_names_by_ids(
         &self,
         table_ids: &[MetaId],
-    ) -> databend_common_exception::Result<Vec<String>> {
+    ) -> databend_common_exception::Result<Vec<Option<String>>> {
         let res = self.ctx.meta.mget_table_names_by_ids(table_ids).await?;
         Ok(res)
     }
@@ -385,7 +385,7 @@ impl Catalog for MutableCatalog {
         Ok(res)
     }
 
-    async fn mget_database_names_by_ids(&self, db_ids: &[MetaId]) -> Result<Vec<String>> {
+    async fn mget_database_names_by_ids(&self, db_ids: &[MetaId]) -> Result<Vec<Option<String>>> {
         let res = self.ctx.meta.mget_database_names_by_ids(db_ids).await?;
         Ok(res)
     }

src/query/service/src/interpreters/interpreter_show_grants.rs

@@ -147,10 +147,12 @@ impl Interpreter for ShowGrantsInterpreter {
             let dbs_name = catalog.mget_database_names_by_ids(&db_ids).await?;
 
             for (i, db_name) in dbs_name.iter().enumerate() {
-                grant_list.push(format!(
-                    "GRANT {} ON '{}'.'{}'.* TO {}",
-                    &privileges_strs[i], catalog_name, db_name, identity
-                ));
+                if let Some(db_name) = db_name {
+                    grant_list.push(format!(
+                        "GRANT {} ON '{}'.'{}'.* TO {}",
+                        &privileges_strs[i], catalog_name, db_name, identity
+                    ));
+                }
             }
         }
 
@@ -166,10 +168,14 @@ impl Interpreter for ShowGrantsInterpreter {
             let tables_name = catalog.mget_table_names_by_ids(&table_ids).await?;
 
             for (i, table_name) in tables_name.iter().enumerate() {
-                grant_list.push(format!(
-                    "GRANT {} ON '{}'.'{}'.'{}' TO {}",
-                    &privileges_strs[i], catalog_name, dbs_name[i], table_name, identity
-                ));
+                if let Some(table_name) = table_name {
+                    if let Some(db_name) = &dbs_name[i] {
+                        grant_list.push(format!(
+                            "GRANT {} ON '{}'.'{}'.'{}' TO {}",
+                            &privileges_strs[i], catalog_name, db_name, table_name, identity
+                        ));
+                    }
+                }
             }
         }
 

src/query/service/tests/it/sql/exec/get_table_bind_test.rs

@@ -188,15 +188,15 @@ impl Catalog for FakedCatalog {
         self.cat.get_table_name_by_id(table_id).await
     }
 
-    async fn mget_table_names_by_ids(&self, table_ids: &[MetaId]) -> Result<Vec<String>> {
+    async fn mget_table_names_by_ids(&self, table_ids: &[MetaId]) -> Result<Vec<Option<String>>> {
         self.cat.mget_table_names_by_ids(table_ids).await
     }
 
     async fn get_db_name_by_id(&self, db_id: MetaId) -> Result<String> {
         self.cat.get_db_name_by_id(db_id).await
     }
 
-    async fn mget_database_names_by_ids(&self, db_ids: &[MetaId]) -> Result<Vec<String>> {
+    async fn mget_database_names_by_ids(&self, db_ids: &[MetaId]) -> Result<Vec<Option<String>>> {
         self.cat.mget_database_names_by_ids(db_ids).await
     }
 

src/query/service/tests/it/storages/fuse/operations/commit.rs

@@ -814,7 +814,7 @@ impl Catalog for FakedCatalog {
     }
 
     #[async_backtrace::framed]
-    async fn mget_table_names_by_ids(&self, table_id: &[MetaId]) -> Result<Vec<String>> {
+    async fn mget_table_names_by_ids(&self, table_id: &[MetaId]) -> Result<Vec<Option<String>>> {
         self.cat.mget_table_names_by_ids(table_id).await
     }
 
@@ -823,7 +823,7 @@ impl Catalog for FakedCatalog {
     }
 
     #[async_backtrace::framed]
-    async fn mget_database_names_by_ids(&self, db_ids: &[MetaId]) -> Result<Vec<String>> {
+    async fn mget_database_names_by_ids(&self, db_ids: &[MetaId]) -> Result<Vec<Option<String>>> {
         self.cat.mget_database_names_by_ids(db_ids).await
     }
 

src/query/storages/hive/hive/src/hive_catalog.rs

@@ -360,7 +360,7 @@ impl Catalog for HiveCatalog {
         ))
     }
 
-    async fn mget_table_names_by_ids(&self, _table_ids: &[MetaId]) -> Result<Vec<String>> {
+    async fn mget_table_names_by_ids(&self, _table_ids: &[MetaId]) -> Result<Vec<Option<String>>> {
         Err(ErrorCode::Unimplemented(
             "Cannot get tables name by ids in HIVE catalog",
         ))
@@ -372,7 +372,7 @@ impl Catalog for HiveCatalog {
         ))
     }
 
-    async fn mget_database_names_by_ids(&self, _db_ids: &[MetaId]) -> Result<Vec<String>> {
+    async fn mget_database_names_by_ids(&self, _db_ids: &[MetaId]) -> Result<Vec<Option<String>>> {
         Err(ErrorCode::Unimplemented(
             "Cannot get dbs name by ids in HIVE catalog",
         ))

src/query/storages/iceberg/src/catalog.rs

@@ -266,7 +266,7 @@ impl Catalog for IcebergCatalog {
         ))
     }
 
-    async fn mget_table_names_by_ids(&self, _table_ids: &[MetaId]) -> Result<Vec<String>> {
+    async fn mget_table_names_by_ids(&self, _table_ids: &[MetaId]) -> Result<Vec<Option<String>>> {
         Err(ErrorCode::Unimplemented(
             "Cannot get tables name by ids in HIVE catalog",
         ))
@@ -279,7 +279,7 @@ impl Catalog for IcebergCatalog {
         ))
     }
 
-    async fn mget_database_names_by_ids(&self, _db_ids: &[MetaId]) -> Result<Vec<String>> {
+    async fn mget_database_names_by_ids(&self, _db_ids: &[MetaId]) -> Result<Vec<Option<String>>> {
         Err(ErrorCode::Unimplemented(
             "Cannot get dbs name by ids in ICEBERG catalog",
         ))

tests/suites/0_stateless/18_rbac/18_0004_show_grants_with_dropped.result

@@ -0,0 +1,23 @@
+GRANT OWNERSHIP ON 'default'.'c_r'.* TO ROLE `role3`
+GRANT INSERT ON 'default'.'c_r1'.* TO ROLE `role3`
+GRANT SELECT,INSERT ON 'default'.'c_r2'.* TO ROLE `role3`
+GRANT SELECT,OWNERSHIP ON 'default'.'c_r'.'t' TO ROLE `role3`
+GRANT UPDATE,DELETE ON 'default'.'c_r1'.'t1' TO ROLE `role3`
+GRANT UPDATE,DELETE ON 'default'.'c_r2'.'t2' TO ROLE `role3`
+=== drop database c_r , c_r2 ===
+GRANT INSERT ON 'default'.'c_r1'.* TO ROLE `role3`
+GRANT UPDATE,DELETE ON 'default'.'c_r1'.'t1' TO ROLE `role3`
+=== undrop database c_r2 ===
+GRANT INSERT ON 'default'.'c_r1'.* TO ROLE `role3`
+GRANT SELECT,INSERT ON 'default'.'c_r2'.* TO ROLE `role3`
+GRANT UPDATE,DELETE ON 'default'.'c_r1'.'t1' TO ROLE `role3`
+GRANT UPDATE,DELETE ON 'default'.'c_r2'.'t2' TO ROLE `role3`
+=== undrop database c_r, contain table c_r.t's ownership ===
+GRANT INSERT ON 'default'.'c_r1'.* TO ROLE `role3`
+GRANT SELECT,INSERT ON 'default'.'c_r2'.* TO ROLE `role3`
+GRANT SELECT,OWNERSHIP ON 'default'.'c_r'.'t' TO ROLE `role3`
+GRANT UPDATE,DELETE ON 'default'.'c_r1'.'t1' TO ROLE `role3`
+GRANT UPDATE,DELETE ON 'default'.'c_r2'.'t2' TO ROLE `role3`
+=== drop database c_r, c_r2, re-create c_r, c_r2 ===
+GRANT INSERT ON 'default'.'c_r1'.* TO ROLE `role3`
+GRANT UPDATE,DELETE ON 'default'.'c_r1'.'t1' TO ROLE `role3`

tests/suites/0_stateless/18_rbac/18_0004_show_grants_with_dropped.sh

@@ -0,0 +1,53 @@
+#!/usr/bin/env bash
+
+CURDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+. "$CURDIR"/../../../shell_env.sh
+
+
+echo "drop role if exists role3;" | $BENDSQL_CLIENT_CONNECT
+echo "drop database if exists c_r;" | $BENDSQL_CLIENT_CONNECT
+echo "drop database if exists c_r1;" | $BENDSQL_CLIENT_CONNECT
+echo "drop database if exists c_r2;" | $BENDSQL_CLIENT_CONNECT
+
+echo "create role role3;" | $BENDSQL_CLIENT_CONNECT
+echo "create database c_r1" | $BENDSQL_CLIENT_CONNECT
+echo "create table c_r1.t1(id int)" | $BENDSQL_CLIENT_CONNECT
+echo "create database c_r2" | $BENDSQL_CLIENT_CONNECT
+echo "create table c_r2.t2(id int)" | $BENDSQL_CLIENT_CONNECT
+echo "create database c_r;" | $BENDSQL_CLIENT_CONNECT
+echo "create table c_r.t(id int);" | $BENDSQL_CLIENT_CONNECT
+echo "grant ownership on c_r.* to role role3;" | $BENDSQL_CLIENT_CONNECT
+echo "grant ownership on c_r.t to role role3;" | $BENDSQL_CLIENT_CONNECT
+
+echo "grant select on c_r.t to role role3;" | $BENDSQL_CLIENT_CONNECT
+echo "grant insert on c_r1.* to role role3;" | $BENDSQL_CLIENT_CONNECT
+echo "grant update, delete on c_r1.t1 to role role3;" | $BENDSQL_CLIENT_CONNECT
+echo "grant select, insert on c_r2.* to role role3;" | $BENDSQL_CLIENT_CONNECT
+echo "grant update, delete on c_r2.t2 to role role3;" | $BENDSQL_CLIENT_CONNECT
+
+echo "show grants for role role3;" | $BENDSQL_CLIENT_CONNECT
+echo "drop database c_r;" | $BENDSQL_CLIENT_CONNECT
+echo "drop database c_r2;" | $BENDSQL_CLIENT_CONNECT
+
+echo "=== drop database c_r , c_r2 ==="
+echo "show grants for role role3;" | $BENDSQL_CLIENT_CONNECT
+
+echo "=== undrop database c_r2 ==="
+echo "undrop database c_r2" | $BENDSQL_CLIENT_CONNECT
+echo "show grants for role role3;" | $BENDSQL_CLIENT_CONNECT
+
+echo "=== undrop database c_r, contain table c_r.t's ownership ==="
+echo "undrop database c_r" | $BENDSQL_CLIENT_CONNECT
+echo "show grants for role role3;" | $BENDSQL_CLIENT_CONNECT
+
+echo "=== drop database c_r, c_r2, re-create c_r, c_r2 ==="
+echo "drop database c_r" | $BENDSQL_CLIENT_CONNECT
+echo "drop database c_r2" | $BENDSQL_CLIENT_CONNECT
+echo "create database c_r" | $BENDSQL_CLIENT_CONNECT
+echo "create database c_r2" | $BENDSQL_CLIENT_CONNECT
+echo "show grants for role role3;" | $BENDSQL_CLIENT_CONNECT
+
+echo "drop role if exists role3;" | $BENDSQL_CLIENT_CONNECT
+echo "drop database if exists c_r;" | $BENDSQL_CLIENT_CONNECT
+echo "drop database if exists c_r1;" | $BENDSQL_CLIENT_CONNECT
+echo "drop database if exists c_r2;" | $BENDSQL_CLIENT_CONNECT