feat: implement "drop table" in distributed mode (both in SQL and gRPC) (#944)

* feat: implement "drop table" in distributed mode (both in SQL and gRPC)

refactor: create distributed table
some details:
- set table global value in Meta, as well as table routes value. Datanode only set table regional value
- complete instance SQL tests both in standalone and distributed mode

* fix: rebase develop

* fix: resolve PR comments

Author: MichaelScofield

Cargo.lock

@@ -52,6 +52,7 @@ arrow-flight = "29.0"
 arrow-schema = { version = "29.0", features = ["serde"] }
 async-stream = "0.3"
 async-trait = "0.1"
+chrono = { version = "0.4", features = ["serde"] }
 # TODO(LFC): Use released Datafusion when it officially dpendent on Arrow 29.0
 datafusion = { git = "https://github.com/apache/arrow-datafusion.git", rev = "4917235a398ae20145c87d20984e6367dc1a0c1e" }
 datafusion-common = { git = "https://github.com/apache/arrow-datafusion.git", rev = "4917235a398ae20145c87d20984e6367dc1a0c1e" }
@@ -65,6 +66,7 @@ parquet = "29.0"
 paste = "1.0"
 prost = "0.11"
 serde = { version = "1.0", features = ["derive"] }
+serde_json = "1.0"
 snafu = { version = "0.7", features = ["backtraces"] }
 sqlparser = "0.28"
 tokio = { version = "1.24.2", features = ["full"] }

Cargo.toml

@@ -36,6 +36,7 @@ message CreateRequest {
 
   TableName table_name = 2;
   repeated Partition partitions = 3;
+  bytes table_info = 4;
 }
 
 message RouteRequest {

src/api/greptime/v1/meta/route.proto

@@ -33,7 +33,7 @@ table = { path = "../table" }
 tokio.workspace = true
 
 [dev-dependencies]
-chrono = "0.4"
+chrono.workspace = true
 log-store = { path = "../log-store" }
 mito = { path = "../mito", features = ["test"] }
 object-store = { path = "../object-store" }

src/catalog/Cargo.toml

@@ -181,6 +181,12 @@ pub enum Error {
         source: BoxedError,
     },
 
+    #[snafu(display("{source}"))]
+    Internal {
+        #[snafu(backtrace)]
+        source: BoxedError,
+    },
+
     #[snafu(display("Failed to execute system catalog table scan, source: {}", source))]
     SystemCatalogTableScanExec {
         #[snafu(backtrace)]
@@ -203,6 +209,12 @@ pub enum Error {
         #[snafu(backtrace)]
         source: datatypes::error::Error,
     },
+
+    #[snafu(display("Failed to serialize or deserialize catalog entry: {}", source))]
+    CatalogEntrySerde {
+        #[snafu(backtrace)]
+        source: common_catalog::error::Error,
+    },
 }
 
 pub type Result<T> = std::result::Result<T, Error>;
@@ -224,7 +236,9 @@ impl ErrorExt for Error {
             Error::SystemCatalogTypeMismatch { .. } => StatusCode::Internal,
 
             Error::ReadSystemCatalog { source, .. } => source.status_code(),
-            Error::InvalidCatalogValue { source, .. } => source.status_code(),
+            Error::InvalidCatalogValue { source, .. } | Error::CatalogEntrySerde { source } => {
+                source.status_code()
+            }
 
             Error::TableExists { .. } => StatusCode::TableAlreadyExists,
             Error::TableNotExist { .. } => StatusCode::TableNotFound,
@@ -240,9 +254,11 @@ impl ErrorExt for Error {
             Error::MetaSrv { source, .. } => source.status_code(),
             Error::SystemCatalogTableScan { source } => source.status_code(),
             Error::SystemCatalogTableScanExec { source } => source.status_code(),
-            Error::InvalidTableSchema { source, .. } => source.status_code(),
-            Error::InvalidTableInfoInCatalog { .. } => StatusCode::Unexpected,
-            Error::SchemaProviderOperation { source } => source.status_code(),
+            Error::InvalidTableSchema { source, .. }
+            | Error::InvalidTableInfoInCatalog { source } => source.status_code(),
+            Error::SchemaProviderOperation { source } | Error::Internal { source } => {
+                source.status_code()
+            }
 
             Error::Unimplemented { .. } => StatusCode::Unsupported,
         }

src/catalog/src/error.rs

@@ -430,11 +430,18 @@ impl CatalogManager for RemoteCatalogManager {
         Ok(true)
     }
 
-    async fn deregister_table(&self, _request: DeregisterTableRequest) -> Result<bool> {
-        UnimplementedSnafu {
-            operation: "deregister table",
-        }
-        .fail()
+    async fn deregister_table(&self, request: DeregisterTableRequest) -> Result<bool> {
+        let catalog_name = &request.catalog;
+        let schema_name = &request.schema;
+        let schema = self
+            .schema(catalog_name, schema_name)?
+            .context(SchemaNotFoundSnafu {
+                catalog: catalog_name,
+                schema: schema_name,
+            })?;
+
+        let result = schema.deregister_table(&request.table_name)?;
+        Ok(result.is_none())
     }
 
     async fn register_schema(&self, request: RegisterSchemaRequest) -> Result<bool> {

src/catalog/src/remote/manager.rs

@@ -16,6 +16,6 @@ serde_json = "1.0"
 snafu = { version = "0.7", features = ["backtraces"] }
 
 [dev-dependencies]
-chrono = "0.4"
+chrono.workspace = true
 tempdir = "0.3"
 tokio.workspace = true

src/common/catalog/Cargo.toml

@@ -5,7 +5,7 @@ edition.workspace = true
 license.workspace = true
 
 [dependencies]
-chrono = "0.4"
+chrono.workspace = true
 common-error = { path = "../error" }
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"

src/common/time/Cargo.toml

@@ -10,7 +10,7 @@ api = { path = "../api" }
 async-stream.workspace = true
 async-trait = "0.1"
 catalog = { path = "../catalog" }
-chrono = "0.4"
+chrono.workspace = true
 client = { path = "../client" }
 common-base = { path = "../common/base" }
 common-catalog = { path = "../common/catalog" }

src/frontend/Cargo.toml

@@ -16,7 +16,7 @@ use std::any::Any;
 use std::collections::HashSet;
 use std::sync::Arc;
 
-use catalog::error::{self as catalog_err, InvalidCatalogValueSnafu};
+use catalog::error::{self as catalog_err, InvalidCatalogValueSnafu, Result as CatalogResult};
 use catalog::helper::{
     build_catalog_prefix, build_schema_prefix, build_table_global_prefix, CatalogKey, SchemaKey,
     TableGlobalKey, TableGlobalValue,
@@ -60,12 +60,10 @@ impl FrontendCatalogManager {
         self.backend.clone()
     }
 
-    #[cfg(test)]
     pub(crate) fn partition_manager(&self) -> PartitionRuleManagerRef {
         self.partition_manager.clone()
     }
 
-    #[cfg(test)]
     pub(crate) fn datanode_clients(&self) -> Arc<DatanodeClients> {
         self.datanode_clients.clone()
     }
@@ -79,15 +77,13 @@ impl CatalogManager for FrontendCatalogManager {
         Ok(())
     }
 
-    async fn register_table(&self, _request: RegisterTableRequest) -> catalog::error::Result<bool> {
-        unimplemented!()
+    // TODO(LFC): Handle the table caching in (de)register_table.
+    async fn register_table(&self, _request: RegisterTableRequest) -> CatalogResult<bool> {
+        Ok(true)
     }
 
-    async fn deregister_table(
-        &self,
-        _request: DeregisterTableRequest,
-    ) -> catalog::error::Result<bool> {
-        unimplemented!()
+    async fn deregister_table(&self, _request: DeregisterTableRequest) -> CatalogResult<bool> {
+        Ok(true)
     }
 
     async fn register_schema(
@@ -289,7 +285,7 @@ impl SchemaProvider for FrontendSchemaProvider {
         let partition_manager = self.partition_manager.clone();
         let datanode_clients = self.datanode_clients.clone();
         let table_name = TableName::new(&self.catalog_name, &self.schema_name, name);
-        let result: Result<Option<TableRef>, catalog::error::Error> = std::thread::spawn(|| {
+        let result: CatalogResult<Option<TableRef>> = std::thread::spawn(|| {
             common_runtime::block_on_read(async move {
                 let res = match backend.get(table_global_key.to_string().as_bytes()).await? {
                     None => {

src/frontend/src/catalog.rs

@@ -181,10 +181,10 @@ pub enum Error {
         source: partition::error::Error,
     },
 
-    #[snafu(display("Failed to create AlterExpr from Alter statement, source: {}", source))]
-    AlterExprFromStmt {
+    #[snafu(display("Failed to create table info, source: {}", source))]
+    CreateTableInfo {
         #[snafu(backtrace)]
-        source: sql::error::Error,
+        source: datatypes::error::Error,
     },
 
     #[snafu(display("Failed to build CreateExpr on insertion: {}", source))]
@@ -226,17 +226,6 @@ pub enum Error {
         backtrace: Backtrace,
     },
 
-    #[snafu(display(
-        "Failed to find leader peer for region {} in table {}",
-        region,
-        table_name
-    ))]
-    FindLeaderPeer {
-        region: u64,
-        table_name: String,
-        backtrace: Backtrace,
-    },
-
     #[snafu(display("Cannot find primary key column by name: {}", msg))]
     PrimaryKeyNotFound { msg: String, backtrace: Backtrace },
 
@@ -370,13 +359,12 @@ impl ErrorExt for Error {
             Error::SqlExecIntercepted { source, .. } => source.status_code(),
             Error::StartServer { source, .. } => source.status_code(),
 
-            Error::ParseSql { source } | Error::AlterExprFromStmt { source } => {
-                source.status_code()
-            }
+            Error::ParseSql { source } => source.status_code(),
 
             Error::Table { source } => source.status_code(),
 
-            Error::ConvertColumnDefaultConstraint { source, .. } => source.status_code(),
+            Error::ConvertColumnDefaultConstraint { source, .. }
+            | Error::CreateTableInfo { source } => source.status_code(),
 
             Error::RequestDatanode { source } => source.status_code(),
 
@@ -387,7 +375,6 @@ impl ErrorExt for Error {
             Error::FindDatanode { .. }
             | Error::CreateTableRoute { .. }
             | Error::FindRegionRoute { .. }
-            | Error::FindLeaderPeer { .. }
             | Error::BuildDfLogicalPlan { .. }
             | Error::BuildTableMeta { .. } => StatusCode::Internal,
 

src/frontend/src/error.rs

@@ -17,23 +17,24 @@ use std::sync::Arc;
 
 use api::helper::ColumnDataTypeWrapper;
 use api::v1::{Column, ColumnDataType, CreateTableExpr};
+use common_error::prelude::BoxedError;
+use datanode::instance::sql::table_idents_to_full_name;
 use datatypes::schema::ColumnSchema;
+use session::context::QueryContextRef;
 use snafu::{ensure, ResultExt};
 use sql::ast::{ColumnDef, TableConstraint};
+use sql::statements::column_def_to_schema;
 use sql::statements::create::{CreateTable, TIME_INDEX};
-use sql::statements::{column_def_to_schema, table_idents_to_full_name};
 
 use crate::error::{
-    BuildCreateExprOnInsertionSnafu, ColumnDataTypeSnafu, ConvertColumnDefaultConstraintSnafu,
-    InvalidSqlSnafu, ParseSqlSnafu, Result,
+    self, BuildCreateExprOnInsertionSnafu, ColumnDataTypeSnafu,
+    ConvertColumnDefaultConstraintSnafu, InvalidSqlSnafu, ParseSqlSnafu, Result,
 };
 
 pub type CreateExprFactoryRef = Arc<dyn CreateExprFactory + Send + Sync>;
 
 #[async_trait::async_trait]
 pub trait CreateExprFactory {
-    async fn create_expr_by_stmt(&self, stmt: &CreateTable) -> Result<CreateTableExpr>;
-
     async fn create_expr_by_columns(
         &self,
         catalog_name: &str,
@@ -48,10 +49,6 @@ pub struct DefaultCreateExprFactory;
 
 #[async_trait::async_trait]
 impl CreateExprFactory for DefaultCreateExprFactory {
-    async fn create_expr_by_stmt(&self, stmt: &CreateTable) -> Result<CreateTableExpr> {
-        create_to_expr(None, vec![0], stmt)
-    }
-
     async fn create_expr_by_columns(
         &self,
         catalog_name: &str,
@@ -74,13 +71,14 @@ impl CreateExprFactory for DefaultCreateExprFactory {
 }
 
 /// Convert `CreateTable` statement to `CreateExpr` gRPC request.
-fn create_to_expr(
-    table_id: Option<u32>,
-    region_ids: Vec<u32>,
+pub(crate) fn create_to_expr(
     create: &CreateTable,
+    query_ctx: QueryContextRef,
 ) -> Result<CreateTableExpr> {
     let (catalog_name, schema_name, table_name) =
-        table_idents_to_full_name(&create.name).context(ParseSqlSnafu)?;
+        table_idents_to_full_name(&create.name, query_ctx)
+            .map_err(BoxedError::new)
+            .context(error::ExternalSnafu)?;
 
     let time_index = find_time_index(&create.constraints)?;
     let expr = CreateTableExpr {
@@ -94,8 +92,8 @@ fn create_to_expr(
         create_if_not_exists: create.if_not_exists,
         // TODO(LFC): Fill in other table options.
         table_options: HashMap::from([("engine".to_string(), create.engine.clone())]),
-        table_id: table_id.map(|id| api::v1::TableId { id }),
-        region_ids,
+        table_id: None,
+        region_ids: vec![],
     };
     Ok(expr)
 }