feat(query): support SHOW DROP DATABASES (databendlabs#16811)

add system.databases_with_history

src/meta/api/src/schema_api_impl.rs

@@ -669,7 +669,6 @@ impl<KV: kvapi::KVApi<Error = MetaError> + ?Sized> SchemaApi for KV {
                     name_ident: DatabaseNameIdent::new_from(db_id_list_key.clone()),
                     meta: db_meta,
                 };
-
                 dbs.insert(db_id.db_id, Arc::new(db));
             }
         }

src/query/ast/src/ast/statements/database.rs

@@ -49,6 +49,26 @@ impl Display for ShowDatabasesStmt {
     }
 }
 
+#[derive(Debug, Clone, PartialEq, Drive, DriveMut)]
+pub struct ShowDropDatabasesStmt {
+    pub catalog: Option<Identifier>,
+    pub limit: Option<ShowLimit>,
+}
+
+impl Display for ShowDropDatabasesStmt {
+    fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
+        write!(f, "SHOW DROP DATABASES")?;
+        if let Some(catalog) = &self.catalog {
+            write!(f, " FROM {catalog}")?;
+        }
+        if let Some(limit) = &self.limit {
+            write!(f, " {limit}")?;
+        }
+
+        Ok(())
+    }
+}
+
 #[derive(Debug, Clone, PartialEq, Eq, Drive, DriveMut)]
 pub struct ShowCreateDatabaseStmt {
     pub catalog: Option<Identifier>,

src/query/ast/src/ast/statements/statement.rs

@@ -127,6 +127,7 @@ pub enum Statement {
 
     // Databases
     ShowDatabases(ShowDatabasesStmt),
+    ShowDropDatabases(ShowDropDatabasesStmt),
     ShowCreateDatabase(ShowCreateDatabaseStmt),
     CreateDatabase(CreateDatabaseStmt),
     DropDatabase(DropDatabaseStmt),
@@ -548,6 +549,7 @@ impl Display for Statement {
             Statement::CreateCatalog(stmt) => write!(f, "{stmt}")?,
             Statement::DropCatalog(stmt) => write!(f, "{stmt}")?,
             Statement::ShowDatabases(stmt) => write!(f, "{stmt}")?,
+            Statement::ShowDropDatabases(stmt) => write!(f, "{stmt}")?,
             Statement::ShowCreateDatabase(stmt) => write!(f, "{stmt}")?,
             Statement::CreateDatabase(stmt) => write!(f, "{stmt}")?,
             Statement::DropDatabase(stmt) => write!(f, "{stmt}")?,

src/query/ast/src/ast/statements/table.rs

@@ -116,7 +116,7 @@ pub struct ShowDropTablesStmt {
 
 impl Display for ShowDropTablesStmt {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
-        write!(f, "SHOW DROP TABLE")?;
+        write!(f, "SHOW DROP TABLES")?;
         if let Some(database) = &self.database {
             write!(f, " FROM {database}")?;
         }

src/query/ast/src/parser/statement.rs

@@ -529,6 +529,19 @@ pub fn statement_body(i: Input) -> IResult<Statement> {
             })
         },
     );
+
+    let show_drop_databases = map(
+        rule! {
+            SHOW ~ DROP ~ ( DATABASES | DATABASES ) ~ ( FROM ~ ^#ident )? ~ #show_limit?
+        },
+        |(_, _, _, opt_catalog, limit)| {
+            Statement::ShowDropDatabases(ShowDropDatabasesStmt {
+                catalog: opt_catalog.map(|(_, catalog)| catalog),
+                limit,
+            })
+        },
+    );
+
     let show_create_database = map(
         rule! {
             SHOW ~ CREATE ~ ( DATABASE | SCHEMA ) ~ #dot_separated_idents_1_to_2
@@ -2284,6 +2297,7 @@ pub fn statement_body(i: Input) -> IResult<Statement> {
             #show_databases : "`SHOW [FULL] DATABASES [(FROM | IN) <catalog>] [<show_limit>]`"
             | #undrop_database : "`UNDROP DATABASE <database>`"
             | #show_create_database : "`SHOW CREATE DATABASE <database>`"
+            | #show_drop_databases : "`SHOW DROP DATABASES [FROM <database>] [<show_limit>]`"
             | #create_database : "`CREATE [OR REPLACE] DATABASE [IF NOT EXISTS] <database> [ENGINE = <engine>]`"
             | #drop_database : "`DROP DATABASE [IF EXISTS] <database>`"
             | #alter_database : "`ALTER DATABASE [IF EXISTS] <action>`"

src/query/ast/tests/it/parser.rs

@@ -87,6 +87,8 @@ fn test_statement() {
     let file = &mut mint.new_goldenfile("stmt.txt").unwrap();
     let cases = &[
         r#"show databases"#,
+        r#"show drop databases"#,
+        r#"show drop databases like 'db%'"#,
         r#"show databases format TabSeparatedWithNamesAndTypes;"#,
         r#"show tables"#,
         r#"show drop tables"#,

src/query/ast/tests/it/testdata/stmt.txt

@@ -12,6 +12,36 @@ ShowDatabases(
 )
 
 
+---------- Input ----------
+show drop databases
+---------- Output ---------
+SHOW DROP DATABASES
+---------- AST ------------
+ShowDropDatabases(
+    ShowDropDatabasesStmt {
+        catalog: None,
+        limit: None,
+    },
+)
+
+
+---------- Input ----------
+show drop databases like 'db%'
+---------- Output ---------
+SHOW DROP DATABASES LIKE 'db%'
+---------- AST ------------
+ShowDropDatabases(
+    ShowDropDatabasesStmt {
+        catalog: None,
+        limit: Some(
+            Like {
+                pattern: "db%",
+            },
+        ),
+    },
+)
+
+
 ---------- Input ----------
 show databases format TabSeparatedWithNamesAndTypes;
 ---------- Output ---------
@@ -49,7 +79,7 @@ ShowTables(
 ---------- Input ----------
 show drop tables
 ---------- Output ---------
-SHOW DROP TABLE
+SHOW DROP TABLES
 ---------- AST ------------
 ShowDropTables(
     ShowDropTablesStmt {
@@ -62,7 +92,7 @@ ShowDropTables(
 ---------- Input ----------
 show drop tables like 't%'
 ---------- Output ---------
-SHOW DROP TABLE LIKE 't%'
+SHOW DROP TABLES LIKE 't%'
 ---------- AST ------------
 ShowDropTables(
     ShowDropTablesStmt {
@@ -79,7 +109,7 @@ ShowDropTables(
 ---------- Input ----------
 show drop tables where name='t'
 ---------- Output ---------
-SHOW DROP TABLE WHERE name = 't'
+SHOW DROP TABLES WHERE name = 't'
 ---------- AST ------------
 ShowDropTables(
     ShowDropTablesStmt {

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

@@ -152,6 +152,9 @@ pub trait Catalog: DynClone + Send + Sync + Debug {
     // Get the database by name.
     async fn get_database(&self, tenant: &Tenant, db_name: &str) -> Result<Arc<dyn Database>>;
 
+    // List all databases history
+    async fn list_databases_history(&self, tenant: &Tenant) -> Result<Vec<Arc<dyn Database>>>;
+
     // Get all the databases.
     async fn list_databases(&self, tenant: &Tenant) -> Result<Vec<Arc<dyn Database>>>;
 

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

@@ -187,6 +187,17 @@ impl Catalog for DatabaseCatalog {
         }
     }
 
+    #[async_backtrace::framed]
+    async fn list_databases_history(&self, tenant: &Tenant) -> Result<Vec<Arc<dyn Database>>> {
+        let mut dbs = self
+            .immutable_catalog
+            .list_databases_history(tenant)
+            .await?;
+        let mut other = self.mutable_catalog.list_databases_history(tenant).await?;
+        dbs.append(&mut other);
+        Ok(dbs)
+    }
+
     #[async_backtrace::framed]
     async fn list_databases(&self, tenant: &Tenant) -> Result<Vec<Arc<dyn Database>>> {
         let mut dbs = self.immutable_catalog.list_databases(tenant).await?;

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

@@ -170,6 +170,10 @@ impl Catalog for ImmutableCatalog {
         }
     }
 
+    async fn list_databases_history(&self, _tenant: &Tenant) -> Result<Vec<Arc<dyn Database>>> {
+        Ok(vec![self.sys_db.clone(), self.info_schema_db.clone()])
+    }
+
     #[async_backtrace::framed]
     async fn list_databases(&self, _tenant: &Tenant) -> Result<Vec<Arc<dyn Database>>> {
         Ok(vec![self.sys_db.clone(), self.info_schema_db.clone()])

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

@@ -243,6 +243,33 @@ impl Catalog for MutableCatalog {
         self.build_db_instance(&db_info)
     }
 
+    #[async_backtrace::framed]
+    async fn list_databases_history(&self, tenant: &Tenant) -> Result<Vec<Arc<dyn Database>>> {
+        let dbs = self
+            .ctx
+            .meta
+            .get_tenant_history_databases(
+                ListDatabaseReq {
+                    tenant: tenant.clone(),
+                },
+                false,
+            )
+            .await?;
+
+        dbs.iter()
+            .try_fold(vec![], |mut acc, item: &Arc<DatabaseInfo>| {
+                let db_result = self.build_db_instance(item);
+                match db_result {
+                    Ok(db) => acc.push(db),
+                    Err(err) => {
+                        // Ignore the error and continue, allow partial failure.
+                        warn!("Failed to build database '{:?}': {:?}", item, err);
+                    }
+                }
+                Ok(acc)
+            })
+    }
+
     #[async_backtrace::framed]
     async fn list_databases(&self, tenant: &Tenant) -> Result<Vec<Arc<dyn Database>>> {
         let dbs = self

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

@@ -153,6 +153,11 @@ impl Catalog for SessionCatalog {
         self.inner.get_database(tenant, db_name).await
     }
 
+    // List all the databases history.
+    async fn list_databases_history(&self, tenant: &Tenant) -> Result<Vec<Arc<dyn Database>>> {
+        self.inner.list_databases_history(tenant).await
+    }
+
     // Get all the databases.
     async fn list_databases(&self, tenant: &Tenant) -> Result<Vec<Arc<dyn Database>>> {
         self.inner.list_databases(tenant).await

src/query/service/src/databases/system/system_database.rs

@@ -34,7 +34,8 @@ use databend_common_storages_system::ColumnsTable;
 use databend_common_storages_system::ConfigsTable;
 use databend_common_storages_system::ContributorsTable;
 use databend_common_storages_system::CreditsTable;
-use databend_common_storages_system::DatabasesTable;
+use databend_common_storages_system::DatabasesTableWithHistory;
+use databend_common_storages_system::DatabasesTableWithoutHistory;
 use databend_common_storages_system::DictionariesTable;
 use databend_common_storages_system::EnginesTable;
 use databend_common_storages_system::FullStreamsTable;
@@ -103,7 +104,8 @@ impl SystemDatabase {
             TablesTableWithoutHistory::create(sys_db_meta.next_table_id()),
             TablesTableWithHistory::create(sys_db_meta.next_table_id()),
             ClustersTable::create(sys_db_meta.next_table_id()),
-            DatabasesTable::create(sys_db_meta.next_table_id()),
+            DatabasesTableWithHistory::create(sys_db_meta.next_table_id()),
+            DatabasesTableWithoutHistory::create(sys_db_meta.next_table_id()),
             FullStreamsTable::create(sys_db_meta.next_table_id()),
             TerseStreamsTable::create(sys_db_meta.next_table_id()),
             ProcessesTable::create(sys_db_meta.next_table_id()),

src/query/service/src/interpreters/access/management_mode_access.rs

@@ -45,6 +45,7 @@ impl AccessChecker for ManagementModeAccess {
                         match rewrite_kind  {
                             Some(ref v) => matches!(v,
                             RewriteKind::ShowDatabases
+                            | RewriteKind::ShowDropDatabases
                             | RewriteKind::ShowTables(_, _)
                             | RewriteKind::ShowColumns(_, _, _)
                             | RewriteKind::ShowEngines

src/query/service/src/interpreters/access/privilege_access.rs

@@ -61,10 +61,11 @@ enum ObjectId {
 // some statements like `SELECT 1`, `SHOW USERS`, `SHOW ROLES`, `SHOW TABLES` will be
 // rewritten to the queries on the system tables, we need to skip the privilege check on
 // these tables.
-const SYSTEM_TABLES_ALLOW_LIST: [&str; 20] = [
+const SYSTEM_TABLES_ALLOW_LIST: [&str; 21] = [
     "catalogs",
     "columns",
     "databases",
+    "databases_with_history",
     "dictionaries",
     "tables",
     "views",
@@ -708,6 +709,7 @@ impl AccessChecker for PrivilegeAccess {
             } => {
                 match rewrite_kind {
                     Some(RewriteKind::ShowDatabases)
+                    | Some(RewriteKind::ShowDropDatabases)
                     | Some(RewriteKind::ShowEngines)
                     | Some(RewriteKind::ShowFunctions)
                     | Some(RewriteKind::ShowUserFunctions)

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

@@ -174,6 +174,10 @@ impl Catalog for FakedCatalog {
         todo!()
     }
 
+    async fn list_databases_history(&self, _tenant: &Tenant) -> Result<Vec<Arc<dyn Database>>> {
+        todo!()
+    }
+
     async fn list_databases(&self, _tenant: &Tenant) -> Result<Vec<Arc<dyn Database>>> {
         todo!()
     }

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

@@ -918,6 +918,10 @@ impl Catalog for FakedCatalog {
         todo!()
     }
 
+    async fn list_databases_history(&self, _tenant: &Tenant) -> Result<Vec<Arc<dyn Database>>> {
+        todo!()
+    }
+
     async fn create_database(&self, _req: CreateDatabaseReq) -> Result<CreateDatabaseReply> {
         todo!()
     }

src/query/service/tests/it/storages/system.rs

@@ -37,7 +37,8 @@ use databend_common_storages_system::ColumnsTable;
 use databend_common_storages_system::ConfigsTable;
 use databend_common_storages_system::ContributorsTable;
 use databend_common_storages_system::CreditsTable;
-use databend_common_storages_system::DatabasesTable;
+use databend_common_storages_system::DatabasesTableWithHistory;
+use databend_common_storages_system::DatabasesTableWithoutHistory;
 use databend_common_storages_system::EnginesTable;
 use databend_common_storages_system::FunctionsTable;
 use databend_common_storages_system::MetricsTable;
@@ -261,7 +262,7 @@ async fn test_databases_table() -> Result<()> {
     let fixture = TestFixture::setup_with_config(&config).await?;
     let ctx = fixture.new_query_ctx().await?;
 
-    let table = DatabasesTable::create(1);
+    let table = DatabasesTableWithoutHistory::create(1);
 
     let mut mint = Mint::new("tests/it/storages/testdata");
     let file = &mut mint.new_goldenfile("databases_table.txt").unwrap();
@@ -270,6 +271,24 @@ async fn test_databases_table() -> Result<()> {
     Ok(())
 }
 
+#[tokio::test(flavor = "multi_thread")]
+async fn test_databases_history_table() -> Result<()> {
+    let mut config = ConfigBuilder::create().build();
+    config.storage.params = StorageParams::Fs(StorageFsConfig::default());
+    let fixture = TestFixture::setup_with_config(&config).await?;
+    let ctx = fixture.new_query_ctx().await?;
+
+    let table = DatabasesTableWithHistory::create(1);
+
+    let mut mint = Mint::new("tests/it/storages/testdata");
+    let file = &mut mint
+        .new_goldenfile("databases_with_history_table.txt")
+        .unwrap();
+    run_table_tests(file, ctx, table).await?;
+
+    Ok(())
+}
+
 #[tokio::test(flavor = "multi_thread")]
 async fn test_engines_table() -> Result<()> {
     let fixture = TestFixture::setup().await?;