refactor: catalog crate (#331)

* chore: refactor dir for local catalog manager

* refactor: CatalogProvider returns Result

* refactor: SchemaProvider returns Result

* feat: add kv operations to remote catalog

* chore: refactor some code

* feat: impl catalog initialization

* feat: add register table and register system table function

* refactor: add table_info method for Table trait

* chore: add some tests

* chore: add register schema test

* chore: fix build issue after rebase onto develop

* refactor: mock to separate file

* build: failed to compile

* fix: use a container struct to bridge KvBackend and Accessor trait

* feat: upgrade opendal to 0.17

* test: add more tests

* chore: add catalog name and schema name to table info

* chore: add catalog name and schema name to table info

* chore: rebase onto develop

* refactor: common-catalog crate

* refactor: remove remote catalog related files

* fix: compilation

* feat: add table version to TableKey

* feat: add node id to TableValue

* fix: some CR comments

* chore: change async fn create_expr_to_request to sync

* fix: add backtrace to errors

* fix: code style

* fix: CatalogManager::table also requires both catalog_name and schema_name

* chore: merge develop

Author: v0y4g3r

Cargo.lock

@@ -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",

Cargo.toml

@@ -7,9 +7,11 @@ 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 = [
@@ -18,12 +20,18 @@ datafusion = { git = "https://github.com/apache/arrow-datafusion.git", branch =
 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"

src/catalog/Cargo.toml

@@ -115,6 +115,17 @@ pub enum Error {
         #[snafu(backtrace)]
         source: common_query::error::Error,
     },
+    #[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>;
@@ -129,12 +140,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/error.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;
 pub 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>;
@@ -79,8 +82,8 @@ pub trait CatalogManager: CatalogList {
     /// Returns the table by catalog, schema and table name.
     fn table(
         &self,
-        catalog: Option<&str>,
-        schema: Option<&str>,
+        catalog: &str,
+        schema: &str,
         table_name: &str,
     ) -> error::Result<Option<TableRef>>;
 }
@@ -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,53 @@ 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(catalog_name, 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/lib.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,
+};