diff --git a/src/catalog/src/lib.rs b/src/catalog/src/lib.rs index 926674003f..81436d9657 100644 --- a/src/catalog/src/lib.rs +++ b/src/catalog/src/lib.rs @@ -59,6 +59,9 @@ pub trait CatalogManager: Send + Sync { /// This method will/should fail if catalog not exist async fn register_schema(&self, request: RegisterSchemaRequest) -> Result; + /// Deregisters a database within given catalog/schema to catalog manager + async fn deregister_schema(&self, request: DeregisterSchemaRequest) -> Result; + /// Registers a table within given catalog/schema to catalog manager, /// returns whether the table registered. /// @@ -149,6 +152,12 @@ pub struct DeregisterTableRequest { pub table_name: String, } +#[derive(Debug, Clone)] +pub struct DeregisterSchemaRequest { + pub catalog: String, + pub schema: String, +} + #[derive(Debug, Clone)] pub struct RegisterSchemaRequest { pub catalog: String, diff --git a/src/catalog/src/local/manager.rs b/src/catalog/src/local/manager.rs index 5708db2a5b..39f0323274 100644 --- a/src/catalog/src/local/manager.rs +++ b/src/catalog/src/local/manager.rs @@ -41,7 +41,7 @@ use crate::error::{ self, CatalogNotFoundSnafu, IllegalManagerStateSnafu, OpenTableSnafu, ReadSystemCatalogSnafu, Result, SchemaExistsSnafu, SchemaNotFoundSnafu, SystemCatalogSnafu, SystemCatalogTypeMismatchSnafu, TableEngineNotFoundSnafu, TableExistsSnafu, TableNotExistSnafu, - TableNotFoundSnafu, + TableNotFoundSnafu, UnimplementedSnafu, }; use crate::information_schema::InformationSchemaProvider; use crate::local::memory::MemoryCatalogManager; @@ -51,8 +51,9 @@ use crate::system::{ }; use crate::tables::SystemCatalog; use crate::{ - handle_system_table_request, CatalogManager, CatalogManagerRef, DeregisterTableRequest, - RegisterSchemaRequest, RegisterSystemTableRequest, RegisterTableRequest, RenameTableRequest, + handle_system_table_request, CatalogManager, CatalogManagerRef, DeregisterSchemaRequest, + DeregisterTableRequest, RegisterSchemaRequest, RegisterSystemTableRequest, + RegisterTableRequest, RenameTableRequest, }; /// A `CatalogManager` consists of a system catalog and a bunch of user catalogs. @@ -516,6 +517,13 @@ impl CatalogManager for LocalCatalogManager { } } + async fn deregister_schema(&self, _request: DeregisterSchemaRequest) -> Result { + UnimplementedSnafu { + operation: "deregister schema", + } + .fail() + } + async fn register_system_table(&self, request: RegisterSystemTableRequest) -> Result<()> { self.check_state().await?; diff --git a/src/catalog/src/local/memory.rs b/src/catalog/src/local/memory.rs index 84b73091a3..0d9bf7e686 100644 --- a/src/catalog/src/local/memory.rs +++ b/src/catalog/src/local/memory.rs @@ -29,8 +29,8 @@ use crate::error::{ CatalogNotFoundSnafu, Result, SchemaNotFoundSnafu, TableExistsSnafu, TableNotFoundSnafu, }; use crate::{ - CatalogManager, DeregisterTableRequest, RegisterSchemaRequest, RegisterSystemTableRequest, - RegisterTableRequest, RenameTableRequest, + CatalogManager, DeregisterSchemaRequest, DeregisterTableRequest, RegisterSchemaRequest, + RegisterSystemTableRequest, RegisterTableRequest, RenameTableRequest, }; type SchemaEntries = HashMap>; @@ -150,6 +150,34 @@ impl CatalogManager for MemoryCatalogManager { Ok(registered) } + async fn deregister_schema(&self, request: DeregisterSchemaRequest) -> Result { + let mut catalogs = self.catalogs.write().unwrap(); + let schemas = catalogs + .get_mut(&request.catalog) + .with_context(|| CatalogNotFoundSnafu { + catalog_name: &request.catalog, + })?; + let table_count = schemas + .remove(&request.schema) + .with_context(|| SchemaNotFoundSnafu { + catalog: &request.catalog, + schema: &request.schema, + })? + .len(); + decrement_gauge!( + crate::metrics::METRIC_CATALOG_MANAGER_TABLE_COUNT, + table_count as f64, + &[crate::metrics::db_label(&request.catalog, &request.schema)], + ); + + decrement_gauge!( + crate::metrics::METRIC_CATALOG_MANAGER_SCHEMA_COUNT, + 1.0, + &[crate::metrics::db_label(&request.catalog, &request.schema)], + ); + Ok(true) + } + async fn register_system_table(&self, _request: RegisterSystemTableRequest) -> Result<()> { // TODO(ruihang): support register system table request Ok(()) @@ -517,4 +545,42 @@ mod tests { .unwrap() .is_none()); } + + #[tokio::test] + async fn test_catalog_deregister_schema() { + let catalog = MemoryCatalogManager::default(); + + // Registers a catalog, a schema, and a table. + let catalog_name = "foo_catalog".to_string(); + let schema_name = "foo_schema".to_string(); + let table_name = "foo_table".to_string(); + let schema = RegisterSchemaRequest { + catalog: catalog_name.clone(), + schema: schema_name.clone(), + }; + let table = RegisterTableRequest { + catalog: catalog_name.clone(), + schema: schema_name.clone(), + table_name, + table_id: 0, + table: Arc::new(NumbersTable::default()), + }; + catalog + .register_catalog(catalog_name.clone()) + .await + .unwrap(); + catalog.register_schema(schema).await.unwrap(); + catalog.register_table(table).await.unwrap(); + + let request = DeregisterSchemaRequest { + catalog: catalog_name.clone(), + schema: schema_name.clone(), + }; + + assert!(catalog.deregister_schema(request).await.unwrap()); + assert!(!catalog + .schema_exist(&catalog_name, &schema_name) + .await + .unwrap()); + } } diff --git a/src/catalog/src/remote/manager.rs b/src/catalog/src/remote/manager.rs index e826341af4..7b36cb7413 100644 --- a/src/catalog/src/remote/manager.rs +++ b/src/catalog/src/remote/manager.rs @@ -34,7 +34,7 @@ use tokio::sync::Mutex; use crate::error::{ CatalogNotFoundSnafu, CreateTableSnafu, InvalidCatalogValueSnafu, OpenTableSnafu, ParallelOpenTableSnafu, Result, SchemaNotFoundSnafu, TableEngineNotFoundSnafu, - TableMetadataManagerSnafu, + TableMetadataManagerSnafu, UnimplementedSnafu, }; use crate::helper::{ build_catalog_prefix, build_schema_prefix, build_table_global_prefix, @@ -43,8 +43,8 @@ use crate::helper::{ }; use crate::remote::region_alive_keeper::RegionAliveKeepers; use crate::{ - handle_system_table_request, CatalogManager, DeregisterTableRequest, RegisterSchemaRequest, - RegisterSystemTableRequest, RegisterTableRequest, RenameTableRequest, + handle_system_table_request, CatalogManager, DeregisterSchemaRequest, DeregisterTableRequest, + RegisterSchemaRequest, RegisterSystemTableRequest, RegisterTableRequest, RenameTableRequest, }; /// Catalog manager based on metasrv. @@ -726,6 +726,13 @@ impl CatalogManager for RemoteCatalogManager { Ok(true) } + async fn deregister_schema(&self, _request: DeregisterSchemaRequest) -> Result { + UnimplementedSnafu { + operation: "deregister schema", + } + .fail() + } + async fn rename_table(&self, request: RenameTableRequest) -> Result { let catalog_name = request.catalog.clone(); let schema_name = request.schema.clone(); diff --git a/src/frontend/src/catalog.rs b/src/frontend/src/catalog.rs index 5fc45f87c3..328346a3db 100644 --- a/src/frontend/src/catalog.rs +++ b/src/frontend/src/catalog.rs @@ -28,8 +28,8 @@ use catalog::helper::{ use catalog::information_schema::InformationSchemaProvider; use catalog::remote::KvCacheInvalidatorRef; use catalog::{ - CatalogManager, DeregisterTableRequest, RegisterSchemaRequest, RegisterSystemTableRequest, - RegisterTableRequest, RenameTableRequest, + CatalogManager, DeregisterSchemaRequest, DeregisterTableRequest, RegisterSchemaRequest, + RegisterSystemTableRequest, RegisterTableRequest, RenameTableRequest, }; use client::client_manager::DatanodeClients; use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, INFORMATION_SCHEMA_NAME}; @@ -130,7 +130,7 @@ impl CatalogManager for FrontendCatalogManager { } async fn register_catalog(&self, _name: String) -> CatalogResult { - unimplemented!("FrontendCatalogManager does not support register catalog") + unimplemented!("FrontendCatalogManager does not support registering catalog") } // TODO(LFC): Handle the table caching in (de)register_table. @@ -151,7 +151,14 @@ impl CatalogManager for FrontendCatalogManager { &self, _request: RegisterSchemaRequest, ) -> catalog::error::Result { - unimplemented!("FrontendCatalogManager does not support register schema") + unimplemented!("FrontendCatalogManager does not support registering schema") + } + + async fn deregister_schema( + &self, + _request: DeregisterSchemaRequest, + ) -> catalog_err::Result { + unimplemented!("FrontendCatalogManager does not support deregistering schema") } async fn rename_table(&self, _request: RenameTableRequest) -> catalog_err::Result {