refactor: catalog crate

Cargo.toml

@@ -5,6 +5,7 @@ members = [
     "src/client",
     "src/cmd",
     "src/common/base",
+    "src/common/catalog",
     "src/common/error",
     "src/common/function",
     "src/common/function-macro",

src/catalog/Cargo.toml

@@ -7,25 +7,34 @@ edition = "2021"
 [dependencies]
 async-stream = "0.3"
 async-trait = "0.1"
+common-catalog = { path = "../common/catalog" }
 common-error = { path = "../common/error" }
 common-query = { path = "../common/query" }
 common-recordbatch = { path = "../common/recordbatch" }
+common-runtime = { path = "../common/runtime" }
 common-telemetry = { path = "../common/telemetry" }
 common-time = { path = "../common/time" }
 datafusion = { git = "https://github.com/apache/arrow-datafusion.git", branch = "arrow2", features = ["simd"] }
 datatypes = { path = "../datatypes" }
 futures = "0.3"
 futures-util = "0.3"
+lazy_static = "1.4"
+opendal = "0.17"
+regex = "1.6"
 serde = "1.0"
 serde_json = "1.0"
 snafu = { version = "0.7", features = ["backtraces"] }
+storage = { path = "../storage" }
 table = { path = "../table" }
+tokio = { version = "1.18", features = ["full"] }
 
 [dev-dependencies]
+chrono = "0.4"
 log-store = { path = "../log-store" }
 object-store = { path = "../object-store" }
 opendal = "0.17"
 storage = { path = "../storage" }
 table-engine = { path = "../table-engine" }
 tempdir = "0.3"
+test-util = { path = "../../test-util" }
 tokio = { version = "1.0", features = ["full"] }

src/catalog/src/error.rs

@@ -103,6 +103,18 @@ pub enum Error {
 
     #[snafu(display("Illegal catalog manager state: {}", msg))]
     IllegalManagerState { backtrace: Backtrace, msg: String },
+
+    #[snafu(display("Cannot parse catalog value, source: {}", source))]
+    InvalidCatalogValue {
+        #[snafu(backtrace)]
+        source: common_catalog::error::Error,
+    },
+
+    #[snafu(display("IO error occurred while fetching catalog info, source: {}", source))]
+    Io {
+        backtrace: Backtrace,
+        source: std::io::Error,
+    },
 }
 
 pub type Result<T> = std::result::Result<T, Error>;
@@ -117,12 +129,14 @@ impl ErrorExt for Error {
             | Error::CatalogNotFound { .. }
             | Error::InvalidEntryType { .. } => StatusCode::Unexpected,
 
-            Error::SystemCatalog { .. } | Error::EmptyValue | Error::ValueDeserialize { .. } => {
-                StatusCode::StorageUnavailable
-            }
+            Error::SystemCatalog { .. }
+            | Error::EmptyValue
+            | Error::ValueDeserialize { .. }
+            | Error::Io { .. } => StatusCode::StorageUnavailable,
 
             Error::ReadSystemCatalog { source, .. } => source.status_code(),
             Error::SystemCatalogTypeMismatch { source, .. } => source.status_code(),
+            Error::InvalidCatalogValue { source, .. } => source.status_code(),
 
             Error::RegisterTable { .. } => StatusCode::Internal,
             Error::TableExists { .. } => StatusCode::TableAlreadyExists,

src/catalog/src/lib.rs

@@ -3,18 +3,18 @@
 use std::any::Any;
 use std::sync::Arc;
 
+use common_telemetry::info;
+use snafu::ResultExt;
+use table::engine::{EngineContext, TableEngineRef};
 use table::metadata::TableId;
 use table::requests::CreateTableRequest;
 use table::TableRef;
 
-pub use crate::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, MIN_USER_TABLE_ID};
-pub use crate::manager::LocalCatalogManager;
+use crate::error::{CreateTableSnafu, Result};
 pub use crate::schema::{SchemaProvider, SchemaProviderRef};
 
-pub mod consts;
 pub mod error;
-mod manager;
-pub mod memory;
+pub mod local;
 pub mod schema;
 mod system;
 pub mod tables;
@@ -31,13 +31,13 @@ pub trait CatalogList: Sync + Send {
         &self,
         name: String,
         catalog: CatalogProviderRef,
-    ) -> Option<CatalogProviderRef>;
+    ) -> Result<Option<CatalogProviderRef>>;
 
     /// Retrieves the list of available catalog names
-    fn catalog_names(&self) -> Vec<String>;
+    fn catalog_names(&self) -> Result<Vec<String>>;
 
     /// Retrieves a specific catalog by name, provided it exists.
-    fn catalog(&self, name: &str) -> Option<CatalogProviderRef>;
+    fn catalog(&self, name: &str) -> Result<Option<CatalogProviderRef>>;
 }
 
 /// Represents a catalog, comprising a number of named schemas.
@@ -47,14 +47,17 @@ pub trait CatalogProvider: Sync + Send {
     fn as_any(&self) -> &dyn Any;
 
     /// Retrieves the list of available schema names in this catalog.
-    fn schema_names(&self) -> Vec<String>;
+    fn schema_names(&self) -> Result<Vec<String>>;
 
     /// Registers schema to this catalog.
-    fn register_schema(&self, name: String, schema: SchemaProviderRef)
-        -> Option<SchemaProviderRef>;
+    fn register_schema(
+        &self,
+        name: String,
+        schema: SchemaProviderRef,
+    ) -> Result<Option<SchemaProviderRef>>;
 
     /// Retrieves a specific schema from the catalog by name, provided it exists.
-    fn schema(&self, name: &str) -> Option<SchemaProviderRef>;
+    fn schema(&self, name: &str) -> Result<Option<SchemaProviderRef>>;
 }
 
 pub type CatalogListRef = Arc<dyn CatalogList>;
@@ -99,9 +102,10 @@ pub struct RegisterSystemTableRequest {
     pub open_hook: Option<OpenSystemTableHook>,
 }
 
+#[derive(Clone)]
 pub struct RegisterTableRequest {
-    pub catalog: Option<String>,
-    pub schema: Option<String>,
+    pub catalog: String,
+    pub schema: String,
     pub table_name: String,
     pub table_id: TableId,
     pub table: TableRef,
@@ -111,3 +115,55 @@ pub struct RegisterTableRequest {
 pub fn format_full_table_name(catalog: &str, schema: &str, table: &str) -> String {
     format!("{}.{}.{}", catalog, schema, table)
 }
+
+pub trait CatalogProviderFactory {
+    fn create(&self, catalog_name: String) -> CatalogProviderRef;
+}
+
+pub trait SchemaProviderFactory {
+    fn create(&self, catalog_name: String, schema_name: String) -> SchemaProviderRef;
+}
+
+pub(crate) async fn handle_system_table_request<'a, M: CatalogManager>(
+    manager: &'a M,
+    engine: TableEngineRef,
+    sys_table_requests: &'a mut Vec<RegisterSystemTableRequest>,
+) -> Result<()> {
+    for req in sys_table_requests.drain(..) {
+        let catalog_name = &req.create_table_request.catalog_name;
+        let schema_name = &req.create_table_request.schema_name;
+        let table_name = &req.create_table_request.table_name;
+        let table_id = req.create_table_request.id;
+
+        let table = if let Some(table) =
+            manager.table(Some(catalog_name), Some(schema_name), table_name)?
+        {
+            table
+        } else {
+            let table = engine
+                .create_table(&EngineContext::default(), req.create_table_request.clone())
+                .await
+                .with_context(|_| CreateTableSnafu {
+                    table_info: format!(
+                        "{}.{}.{}, id: {}",
+                        catalog_name, schema_name, table_name, table_id,
+                    ),
+                })?;
+            manager
+                .register_table(RegisterTableRequest {
+                    catalog: catalog_name.clone(),
+                    schema: schema_name.clone(),
+                    table_name: table_name.clone(),
+                    table_id,
+                    table: table.clone(),
+                })
+                .await?;
+            info!("Created and registered system table: {}", table_name);
+            table
+        };
+        if let Some(hook) = req.open_hook {
+            (hook)(table)?;
+        }
+    }
+    Ok(())
+}

src/catalog/src/local.rs

@@ -0,0 +1,7 @@
+pub mod manager;
+pub mod memory;
+
+pub use manager::LocalCatalogManager;
+pub use memory::{
+    new_memory_catalog_list, MemoryCatalogList, MemoryCatalogProvider, MemorySchemaProvider,
+};