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
2026-10-30 07:26:56 +00:00 · 2023-02-08 07:36:38 +00:00
parent 37ca5ba380
commit 803b7f0633
28 changed files with 763 additions and 395 deletions
@@ -3844,15 +3844,19 @@ version = "0.1.0"
 dependencies = [
 "api",
 "async-trait",
+ "chrono",
 "common-error",
 "common-grpc",
 "common-telemetry",
+ "datatypes",
 "etcd-client",
 "futures",
 "meta-srv",
 "rand 0.8.5",
 "serde",
+ "serde_json",
 "snafu",
+ "table",
 "tokio",
 "tokio-stream",
 "tonic",
@@ -3888,6 +3892,7 @@ dependencies = [
 "serde",
 "serde_json",
 "snafu",
+ "table",
 "tokio",
 "tokio-stream",
 "tonic",
@@ -4099,7 +4104,7 @@ dependencies = [
 "futures-sink",
 "futures-util",
 "lazy_static",
- "lru",
+ "lru 0.8.1",
 "mio",
 "mysql_common",
 "once_cell",
@@ -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"] }
@@ -36,6 +36,7 @@ message CreateRequest {

  TableName table_name = 2;
  repeated Partition partitions = 3;
+  bytes table_info = 4;
 }

 message RouteRequest {
@@ -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" }
@@ -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,
        }
@@ -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> {
@@ -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
@@ -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"
@@ -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" }
@@ -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 => {
@@ -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,

@@ -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)
 }
@@ -25,16 +25,14 @@ use std::time::Duration;
 use api::v1::alter_expr::Kind;
 use api::v1::ddl_request::Expr as DdlExpr;
 use api::v1::greptime_request::Request;
-use api::v1::{AddColumns, AlterExpr, Column, DdlRequest, DropTableExpr, InsertRequest};
+use api::v1::{AddColumns, AlterExpr, Column, DdlRequest, InsertRequest};
 use async_trait::async_trait;
 use catalog::remote::MetaKvBackend;
 use catalog::CatalogManagerRef;
-use common_error::ext::BoxedError;
 use common_grpc::channel_manager::{ChannelConfig, ChannelManager};
 use common_query::Output;
 use common_recordbatch::RecordBatches;
 use common_telemetry::logging::{debug, info};
-use datanode::instance::sql::table_idents_to_full_name;
 use datanode::instance::InstanceRef as DnInstanceRef;
 use datatypes::schema::Schema;
 use distributed::DistInstance;
@@ -372,9 +370,7 @@ fn parse_stmt(sql: &str) -> Result<Vec<Statement>> {

 impl Instance {
    async fn query_statement(&self, stmt: Statement, query_ctx: QueryContextRef) -> Result<Output> {
-        // TODO(sunng87): provide a better form to log or track statement
-        let query = &format!("{:?}", &stmt);
-        match stmt.clone() {
+        match stmt {
            Statement::CreateDatabase(_)
            | Statement::ShowDatabases(_)
            | Statement::CreateTable(_)
@@ -383,31 +379,13 @@ impl Instance {
            | Statement::Explain(_)
            | Statement::Query(_)
            | Statement::Insert(_)
-            | Statement::Alter(_) => {
-                return self.sql_handler.do_statement_query(stmt, query_ctx).await;
-            }
-            Statement::DropTable(drop_stmt) => {
-                let (catalog_name, schema_name, table_name) =
-                    table_idents_to_full_name(drop_stmt.table_name(), query_ctx.clone())
-                        .map_err(BoxedError::new)
-                        .context(error::ExternalSnafu)?;
-                let expr = DropTableExpr {
-                    catalog_name,
-                    schema_name,
-                    table_name,
-                };
-                return self
-                    .grpc_query_handler
-                    .do_query(
-                        Request::Ddl(DdlRequest {
-                            expr: Some(DdlExpr::DropTable(expr)),
-                        }),
-                        query_ctx,
-                    )
-                    .await;
-            }
-            Statement::ShowCreateTable(_) => error::NotSupportedSnafu { feat: query }.fail(),
+            | Statement::Alter(_)
+            | Statement::DropTable(_) => self.sql_handler.do_statement_query(stmt, query_ctx).await,
            Statement::Use(db) => self.handle_use(db, query_ctx),
+            _ => NotSupportedSnafu {
+                feat: format!("{stmt:?}"),
+            }
+            .fail(),
        }
    }
 }
@@ -546,107 +524,209 @@ impl PromqlHandler for Instance {
 #[cfg(test)]
 mod tests {
    use std::borrow::Cow;
+    use std::collections::HashMap;
    use std::sync::atomic::AtomicU32;

+    use catalog::helper::{TableGlobalKey, TableGlobalValue};
    use session::context::QueryContext;

    use super::*;
+    use crate::table::DistTable;
    use crate::tests;
+    use crate::tests::MockDistributedInstance;

    #[tokio::test(flavor = "multi_thread")]
-    async fn test_execute_sql() {
-        let query_ctx = Arc::new(QueryContext::new());
+    async fn test_standalone_exec_sql() {
+        let standalone = tests::create_standalone_instance("test_standalone_exec_sql").await;
+        let instance = standalone.instance.as_ref();

-        let standalone = tests::create_standalone_instance("test_execute_sql").await;
-        let instance = standalone.instance;
+        let sql = r#"
+            CREATE TABLE demo(
+                host STRING,
+                ts TIMESTAMP,
+                cpu DOUBLE NULL,
+                memory DOUBLE NULL,
+                disk_util DOUBLE DEFAULT 9.9,
+                TIME INDEX (ts),
+                PRIMARY KEY(host)
+            ) engine=mito"#;
+        create_table(instance, sql).await;

-        let sql = r#"CREATE TABLE demo(
-                            host STRING,
-                            ts TIMESTAMP,
-                            cpu DOUBLE NULL,
-                            memory DOUBLE NULL,
-                            disk_util DOUBLE DEFAULT 9.9,
-                            TIME INDEX (ts),
-                            PRIMARY KEY(host)
-                        ) engine=mito with(regions=1);"#;
-        let output = SqlQueryHandler::do_query(&*instance, sql, query_ctx.clone())
+        insert_and_query(instance).await;
+
+        drop_table(instance).await;
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn test_distributed_exec_sql() {
+        let distributed = tests::create_distributed_instance("test_distributed_exec_sql").await;
+        let instance = distributed.frontend.as_ref();
+
+        let sql = r#"
+            CREATE TABLE demo(
+                host STRING,
+                ts TIMESTAMP,
+                cpu DOUBLE NULL,
+                memory DOUBLE NULL,
+                disk_util DOUBLE DEFAULT 9.9,
+                TIME INDEX (ts),
+                PRIMARY KEY(host)
+            )
+            PARTITION BY RANGE COLUMNS (host) (
+                PARTITION r0 VALUES LESS THAN ('550-A'),
+                PARTITION r1 VALUES LESS THAN ('550-W'),
+                PARTITION r2 VALUES LESS THAN ('MOSS'),
+                PARTITION r3 VALUES LESS THAN (MAXVALUE),
+            )
+            engine=mito"#;
+        create_table(instance, sql).await;
+
+        insert_and_query(instance).await;
+
+        verify_data_distribution(
+            &distributed,
+            HashMap::from([
+                (
+                    0u32,
+                    "\
+---------------------+------+
+| ts                  | host |
+---------------------+------+
+| 2013-12-31T16:00:00 | 490  |
+---------------------+------+",
+                ),
+                (
+                    1u32,
+                    "\
+---------------------+-------+
+| ts                  | host  |
+---------------------+-------+
+| 2022-12-31T16:00:00 | 550-A |
+---------------------+-------+",
+                ),
+                (
+                    2u32,
+                    "\
+---------------------+-------+
+| ts                  | host  |
+---------------------+-------+
+| 2023-12-31T16:00:00 | 550-W |
+---------------------+-------+",
+                ),
+                (
+                    3u32,
+                    "\
+---------------------+------+
+| ts                  | host |
+---------------------+------+
+| 2043-12-31T16:00:00 | MOSS |
+---------------------+------+",
+                ),
+            ]),
+        )
+        .await;
+
+        drop_table(instance).await;
+
+        verify_table_is_dropped(&distributed);
+    }
+
+    async fn query(instance: &Instance, sql: &str) -> Output {
+        SqlQueryHandler::do_query(instance, sql, QueryContext::arc())
            .await
            .remove(0)
-            .unwrap();
-        match output {
-            Output::AffectedRows(rows) => assert_eq!(rows, 0),
-            _ => unreachable!(),
-        }
+            .unwrap()
+    }

-        let sql = r#"insert into demo(host, cpu, memory, ts) values
-                                ('frontend.host1', 1.1, 100, 1000),
-                                ('frontend.host2', null, null, 2000),
-                                ('frontend.host3', 3.3, 300, 3000)
+    async fn create_table(instance: &Instance, sql: &str) {
+        let output = query(instance, sql).await;
+        let Output::AffectedRows(x) = output else { unreachable!() };
+        assert_eq!(x, 0);
+    }
+
+    async fn insert_and_query(instance: &Instance) {
+        let sql = r#"INSERT INTO demo(host, cpu, memory, ts) VALUES
+                                ('490', 0.1, 1, 1388505600000),
+                                ('550-A', 1, 100, 1672502400000),
+                                ('550-W', 10000, 1000000, 1704038400000),
+                                ('MOSS', 100000000, 10000000000, 2335190400000)
                                "#;
-        let output = SqlQueryHandler::do_query(&*instance, sql, query_ctx.clone())
-            .await
-            .remove(0)
+        let output = query(instance, sql).await;
+        let Output::AffectedRows(x) = output else { unreachable!() };
+        assert_eq!(x, 4);
+
+        let sql = "SELECT * FROM demo WHERE ts > cast(1000000000 as timestamp) ORDER BY host"; // use nanoseconds as where condition
+        let output = query(instance, sql).await;
+        let Output::Stream(s) = output else { unreachable!() };
+        let batches = common_recordbatch::util::collect_batches(s).await.unwrap();
+        let pretty_print = batches.pretty_print().unwrap();
+        let expected = "\
+-------+---------------------+-----------+-------------+-----------+
+| host  | ts                  | cpu       | memory      | disk_util |
+-------+---------------------+-----------+-------------+-----------+
+| 490   | 2013-12-31T16:00:00 | 0.1       | 1           | 9.9       |
+| 550-A | 2022-12-31T16:00:00 | 1         | 100         | 9.9       |
+| 550-W | 2023-12-31T16:00:00 | 10000     | 1000000     | 9.9       |
+| MOSS  | 2043-12-31T16:00:00 | 100000000 | 10000000000 | 9.9       |
+-------+---------------------+-----------+-------------+-----------+";
+        assert_eq!(pretty_print, expected);
+    }
+
+    async fn verify_data_distribution(
+        instance: &MockDistributedInstance,
+        expected_distribution: HashMap<u32, &str>,
+    ) {
+        let table = instance
+            .frontend
+            .catalog_manager()
+            .table("greptime", "public", "demo")
+            .unwrap()
            .unwrap();
-        match output {
-            Output::AffectedRows(rows) => assert_eq!(rows, 3),
-            _ => unreachable!(),
+        let table = table.as_any().downcast_ref::<DistTable>().unwrap();
+
+        let TableGlobalValue { regions_id_map, .. } = table
+            .table_global_value(&TableGlobalKey {
+                catalog_name: "greptime".to_string(),
+                schema_name: "public".to_string(),
+                table_name: "demo".to_string(),
+            })
+            .await
+            .unwrap()
+            .unwrap();
+        let region_to_dn_map = regions_id_map
+            .iter()
+            .map(|(k, v)| (v[0], *k))
+            .collect::<HashMap<u32, u64>>();
+        assert_eq!(region_to_dn_map.len(), expected_distribution.len());
+
+        for (region, dn) in region_to_dn_map.iter() {
+            let dn = instance.datanodes.get(dn).unwrap();
+            let output = dn
+                .execute_sql("SELECT ts, host FROM demo ORDER BY ts", QueryContext::arc())
+                .await
+                .unwrap();
+            let Output::Stream(stream) = output else { unreachable!() };
+            let recordbatches = RecordBatches::try_collect(stream).await.unwrap();
+            let actual = recordbatches.pretty_print().unwrap();
+
+            let expected = expected_distribution.get(region).unwrap();
+            assert_eq!(&actual, expected);
        }
+    }

-        let sql = "select * from demo";
-        let output = SqlQueryHandler::do_query(&*instance, sql, query_ctx.clone())
-            .await
-            .remove(0)
-            .unwrap();
-        match output {
-            Output::RecordBatches(_) => {
-                unreachable!("Output::RecordBatches");
-            }
-            Output::AffectedRows(_) => {
-                unreachable!("Output::AffectedRows");
-            }
-            Output::Stream(s) => {
-                let batches = common_recordbatch::util::collect_batches(s).await.unwrap();
-                let pretty_print = batches.pretty_print().unwrap();
-                let expected = "\
-+----------------+---------------------+-----+--------+-----------+
-| host           | ts                  | cpu | memory | disk_util |
-+----------------+---------------------+-----+--------+-----------+
-| frontend.host1 | 1970-01-01T00:00:01 | 1.1 | 100    | 9.9       |
-| frontend.host2 | 1970-01-01T00:00:02 |     |        | 9.9       |
-| frontend.host3 | 1970-01-01T00:00:03 | 3.3 | 300    | 9.9       |
-+----------------+---------------------+-----+--------+-----------+\
-                ";
-                assert_eq!(pretty_print, expected);
-            }
-        };
+    async fn drop_table(instance: &Instance) {
+        let sql = "DROP TABLE demo";
+        let output = query(instance, sql).await;
+        let Output::AffectedRows(x) = output else { unreachable!() };
+        assert_eq!(x, 1);
+    }

-        let sql = "select * from demo where ts>cast(1000000000 as timestamp)"; // use nanoseconds as where condition
-        let output = SqlQueryHandler::do_query(&*instance, sql, query_ctx.clone())
-            .await
-            .remove(0)
-            .unwrap();
-        match output {
-            Output::RecordBatches(_) => {
-                unreachable!("Output::RecordBatches")
-            }
-            Output::AffectedRows(_) => {
-                unreachable!("Output::AffectedRows")
-            }
-            Output::Stream(s) => {
-                let recordbatches = common_recordbatch::util::collect_batches(s).await.unwrap();
-                let pretty = recordbatches.pretty_print().unwrap();
-                let expected = "\
-+----------------+---------------------+-----+--------+-----------+
-| host           | ts                  | cpu | memory | disk_util |
-+----------------+---------------------+-----+--------+-----------+
-| frontend.host2 | 1970-01-01T00:00:02 |     |        | 9.9       |
-| frontend.host3 | 1970-01-01T00:00:03 | 3.3 | 300    | 9.9       |
-+----------------+---------------------+-----+--------+-----------+\
-                    "
-                .to_string();
-                assert_eq!(pretty, expected);
-            }
-        };
+    fn verify_table_is_dropped(instance: &MockDistributedInstance) {
+        assert!(instance.datanodes.iter().all(|(_, x)| x
+            .catalog_manager()
+            .table("greptime", "public", "demo")
+            .unwrap()
+            .is_none()))
    }

    #[tokio::test(flavor = "multi_thread")]
@@ -18,23 +18,26 @@ use std::collections::HashMap;
 use std::sync::Arc;

 use api::helper::ColumnDataTypeWrapper;
-use api::v1::{AlterExpr, CreateDatabaseExpr, CreateTableExpr, InsertRequest, TableId};
+use api::v1::{
+    AlterExpr, CreateDatabaseExpr, CreateTableExpr, DropTableExpr, InsertRequest, TableId,
+};
 use async_trait::async_trait;
-use catalog::helper::{SchemaKey, SchemaValue, TableGlobalKey, TableGlobalValue};
-use catalog::{CatalogList, CatalogManager};
+use catalog::helper::{SchemaKey, SchemaValue};
+use catalog::{CatalogList, CatalogManager, DeregisterTableRequest, RegisterTableRequest};
 use chrono::DateTime;
 use client::Database;
 use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME};
 use common_error::prelude::BoxedError;
 use common_query::Output;
-use common_telemetry::{debug, error, info};
+use common_telemetry::{debug, info};
 use datanode::instance::sql::table_idents_to_full_name;
 use datatypes::prelude::ConcreteDataType;
 use datatypes::schema::{RawSchema, Schema};
 use meta_client::client::MetaClient;
+use meta_client::rpc::router::DeleteRequest as MetaDeleteRequest;
 use meta_client::rpc::{
    CreateRequest as MetaCreateRequest, Partition as MetaPartition, PutRequest, RouteResponse,
-    TableName, TableRoute,
+    TableName,
 };
 use partition::partition::{PartitionBound, PartitionDef};
 use query::parser::QueryStatement;
@@ -58,9 +61,10 @@ use crate::error::{
    RequestDatanodeSnafu, RequestMetaSnafu, Result, SchemaNotFoundSnafu, StartMetaClientSnafu,
    TableNotFoundSnafu, TableSnafu, ToTableInsertRequestSnafu,
 };
-use crate::expr_factory::{CreateExprFactory, DefaultCreateExprFactory};
+use crate::expr_factory;
 use crate::instance::parse_stmt;
 use crate::sql::insert_to_request;
+use crate::table::DistTable;

 #[derive(Clone)]
 pub(crate) struct DistInstance {
@@ -90,7 +94,17 @@ impl DistInstance {
        create_table: &mut CreateTableExpr,
        partitions: Option<Partitions>,
    ) -> Result<Output> {
-        let response = self.create_table_in_meta(create_table, partitions).await?;
+        let table_name = TableName::new(
+            &create_table.catalog_name,
+            &create_table.schema_name,
+            &create_table.table_name,
+        );
+
+        let mut table_info = create_table_info(create_table)?;
+
+        let response = self
+            .create_table_in_meta(create_table, partitions, &table_info)
+            .await?;
        let table_routes = response.table_routes;
        ensure!(
            table_routes.len() == 1,
@@ -100,11 +114,9 @@ impl DistInstance {
        );
        let table_route = table_routes.first().unwrap();
        info!(
-            "Create table {:?}.{:?}.{}, table routes: {:?}",
-            create_table.catalog_name,
-            create_table.schema_name,
-            create_table.table_name,
-            table_route
+            "Creating distributed table {table_name} with table routes: {}",
+            serde_json::to_string_pretty(table_route)
+                .unwrap_or_else(|_| format!("{table_route:#?}"))
        );
        let region_routes = &table_route.region_routes;
        ensure!(
@@ -113,11 +125,37 @@ impl DistInstance {
                table_name: create_table.table_name.to_string()
            }
        );
-        create_table.table_id = Some(TableId {
-            id: table_route.table.id as u32,
-        });
-        self.put_table_global_meta(create_table, table_route)
-            .await?;
+
+        let table_id = table_route.table.id as u32;
+        table_info.ident.table_id = table_id;
+        let table_info = Arc::new(table_info.try_into().context(error::CreateTableInfoSnafu)?);
+
+        create_table.table_id = Some(TableId { id: table_id });
+
+        let table = DistTable::new(
+            table_name.clone(),
+            table_info,
+            self.catalog_manager.partition_manager(),
+            self.catalog_manager.datanode_clients(),
+            self.catalog_manager.backend(),
+        );
+
+        let request = RegisterTableRequest {
+            catalog: table_name.catalog_name.clone(),
+            schema: table_name.schema_name.clone(),
+            table_name: table_name.table_name.clone(),
+            table_id,
+            table: Arc::new(table),
+        };
+        ensure!(
+            self.catalog_manager
+                .register_table(request)
+                .await
+                .context(CatalogSnafu)?,
+            error::TableAlreadyExistSnafu {
+                table: table_name.to_string()
+            }
+        );

        for datanode in table_route.find_leaders() {
            let client = self.datanode_clients.get_client(&datanode).await;
@@ -142,6 +180,63 @@ impl DistInstance {
        Ok(Output::AffectedRows(0))
    }

+    async fn drop_table(&self, table_name: TableName) -> Result<Output> {
+        let _ = self
+            .catalog_manager
+            .table(
+                &table_name.catalog_name,
+                &table_name.schema_name,
+                &table_name.table_name,
+            )
+            .context(CatalogSnafu)?
+            .with_context(|| TableNotFoundSnafu {
+                table_name: table_name.to_string(),
+            })?;
+
+        let route_response = self
+            .meta_client
+            .delete_route(MetaDeleteRequest {
+                table_name: table_name.clone(),
+            })
+            .await
+            .context(RequestMetaSnafu)?;
+
+        let request = DeregisterTableRequest {
+            catalog: table_name.catalog_name.clone(),
+            schema: table_name.schema_name.clone(),
+            table_name: table_name.table_name.clone(),
+        };
+        ensure!(
+            self.catalog_manager
+                .deregister_table(request)
+                .await
+                .context(CatalogSnafu)?,
+            error::TableNotFoundSnafu {
+                table_name: table_name.to_string()
+            }
+        );
+
+        let expr = DropTableExpr {
+            catalog_name: table_name.catalog_name.clone(),
+            schema_name: table_name.schema_name.clone(),
+            table_name: table_name.table_name.clone(),
+        };
+        for table_route in route_response.table_routes.iter() {
+            for datanode in table_route.find_leaders() {
+                debug!("Dropping table {table_name} on Datanode {datanode:?}");
+
+                let client = self.datanode_clients.get_client(&datanode).await;
+                let client = Database::new(&expr.catalog_name, &expr.schema_name, client);
+                client
+                    .drop_table(expr.clone())
+                    .await
+                    .context(RequestDatanodeSnafu)?;
+            }
+        }
+
+        Ok(Output::AffectedRows(1))
+    }
+
    async fn handle_statement(
        &self,
        stmt: Statement,
@@ -163,9 +258,17 @@ impl DistInstance {
                Ok(self.handle_create_database(expr).await?)
            }
            Statement::CreateTable(stmt) => {
-                let create_expr = &mut DefaultCreateExprFactory.create_expr_by_stmt(&stmt).await?;
+                let create_expr = &mut expr_factory::create_to_expr(&stmt, query_ctx)?;
                Ok(self.create_table(create_expr, stmt.partitions).await?)
            }
+            Statement::DropTable(stmt) => {
+                let (catalog, schema, table) =
+                    table_idents_to_full_name(stmt.table_name(), query_ctx)
+                        .map_err(BoxedError::new)
+                        .context(error::ExternalSnafu)?;
+                let table_name = TableName::new(catalog, schema, table);
+                return self.drop_table(table_name).await;
+            }
            Statement::ShowDatabases(stmt) => show_databases(stmt, self.catalog_manager.clone()),
            Statement::ShowTables(stmt) => {
                show_tables(stmt, self.catalog_manager.clone(), query_ctx)
@@ -201,7 +304,12 @@ impl DistInstance {
                    table.insert(insert_request).await.context(TableSnafu)?,
                ));
            }
-            _ => unreachable!(),
+            _ => {
+                return error::NotSupportedSnafu {
+                    feat: format!("{stmt:?}"),
+                }
+                .fail()
+            }
        }
        .context(error::ExecuteStatementSnafu)
    }
@@ -293,6 +401,7 @@ impl DistInstance {
        &self,
        create_table: &CreateTableExpr,
        partitions: Option<Partitions>,
+        table_info: &RawTableInfo,
    ) -> Result<RouteResponse> {
        let mut catalog_name = create_table.catalog_name.clone();
        if catalog_name.is_empty() {
@@ -308,6 +417,7 @@ impl DistInstance {
        let request = MetaCreateRequest {
            table_name,
            partitions,
+            table_info,
        };
        self.meta_client
            .create_route(request)
@@ -315,47 +425,6 @@ impl DistInstance {
            .context(error::RequestMetaSnafu)
    }

-    // TODO(LFC): Maybe move this to FrontendCatalogManager's "register_table" method?
-    async fn put_table_global_meta(
-        &self,
-        create_table: &CreateTableExpr,
-        table_route: &TableRoute,
-    ) -> Result<()> {
-        let table_name = &table_route.table.table_name;
-        let key = TableGlobalKey {
-            catalog_name: table_name.catalog_name.clone(),
-            schema_name: table_name.schema_name.clone(),
-            table_name: table_name.table_name.clone(),
-        }
-        .to_string();
-
-        let value = create_table_global_value(create_table, table_route)?
-            .as_bytes()
-            .context(error::CatalogEntrySerdeSnafu)?;
-
-        if let Err(existing) = self
-            .catalog_manager
-            .backend()
-            .compare_and_set(key.as_bytes(), &[], &value)
-            .await
-            .context(CatalogSnafu)?
-        {
-            let existing_bytes = existing.unwrap(); // this unwrap is safe since we compare with empty bytes and failed
-            let existing_value =
-                TableGlobalValue::from_bytes(&existing_bytes).context(CatalogEntrySerdeSnafu)?;
-            if existing_value.table_info.ident.table_id
-                != create_table.table_id.as_ref().unwrap().id
-            {
-                error!(
-                    "Table with name {} already exists, value in catalog: {:?}",
-                    key, existing_bytes
-                );
-                return error::TableAlreadyExistSnafu { table: key }.fail();
-            }
-        }
-        Ok(())
-    }
-
    // TODO(LFC): Refactor insertion implementation for DistTable,
    // GRPC InsertRequest to Table InsertRequest, than split Table InsertRequest, than assemble each GRPC InsertRequest, is rather inefficient,
    // should operate on GRPC InsertRequest directly.
@@ -430,38 +499,7 @@ impl SqlQueryHandler for DistInstance {
    }
 }

-fn create_table_global_value(
-    create_table: &CreateTableExpr,
-    table_route: &TableRoute,
-) -> Result<TableGlobalValue> {
-    let table_name = &table_route.table.table_name;
-
-    let region_routes = &table_route.region_routes;
-    let node_id = region_routes[0]
-        .leader_peer
-        .as_ref()
-        .with_context(|| error::FindLeaderPeerSnafu {
-            region: region_routes[0].region.id,
-            table_name: table_name.to_string(),
-        })?
-        .id;
-
-    let mut regions_id_map = HashMap::new();
-    for route in region_routes.iter() {
-        let node_id = route
-            .leader_peer
-            .as_ref()
-            .with_context(|| error::FindLeaderPeerSnafu {
-                region: route.region.id,
-                table_name: table_name.to_string(),
-            })?
-            .id;
-        regions_id_map
-            .entry(node_id)
-            .or_insert_with(Vec::new)
-            .push(route.region.id as u32);
-    }
-
+fn create_table_info(create_table: &CreateTableExpr) -> Result<RawTableInfo> {
    let mut column_schemas = Vec::with_capacity(create_table.column_defs.len());
    let mut column_name_to_index_map = HashMap::new();

@@ -518,22 +556,18 @@ fn create_table_global_value(

    let table_info = RawTableInfo {
        ident: TableIdent {
-            table_id: table_route.table.id as u32,
+            // The table id of distributed table is assigned by Meta, set "0" here as a placeholder.
+            table_id: 0,
            version: 0,
        },
-        name: table_name.table_name.clone(),
+        name: create_table.table_name.clone(),
        desc,
-        catalog_name: table_name.catalog_name.clone(),
-        schema_name: table_name.schema_name.clone(),
+        catalog_name: create_table.catalog_name.clone(),
+        schema_name: create_table.schema_name.clone(),
        meta,
        table_type: TableType::Base,
    };
-
-    Ok(TableGlobalValue {
-        node_id,
-        regions_id_map,
-        table_info,
-    })
+    Ok(table_info)
 }

 fn parse_partitions(
@@ -628,7 +662,6 @@ mod test {
    use sql::statements::statement::Statement;

    use super::*;
-    use crate::expr_factory::{CreateExprFactory, DefaultCreateExprFactory};
    use crate::instance::standalone::StandaloneSqlQueryHandler;

    #[tokio::test]
@@ -662,10 +695,7 @@ ENGINE=mito",
            let result = ParserContext::create_with_dialect(sql, &GenericDialect {}).unwrap();
            match &result[0] {
                Statement::CreateTable(c) => {
-                    let expr = DefaultCreateExprFactory
-                        .create_expr_by_stmt(c)
-                        .await
-                        .unwrap();
+                    let expr = expr_factory::create_to_expr(c, QueryContext::arc()).unwrap();
                    let partitions = parse_partitions(&expr, c.partitions.clone()).unwrap();
                    let json = serde_json::to_string(&partitions).unwrap();
                    assert_eq!(json, expected);
@@ -16,6 +16,7 @@ use api::v1::ddl_request::Expr as DdlExpr;
 use api::v1::greptime_request::Request;
 use async_trait::async_trait;
 use common_query::Output;
+use meta_client::rpc::TableName;
 use servers::query_handler::grpc::GrpcQueryHandler;
 use session::context::QueryContextRef;
 use snafu::OptionExt;
@@ -45,10 +46,10 @@ impl GrpcQueryHandler for DistInstance {
                        self.create_table(&mut expr, None).await
                    }
                    DdlExpr::Alter(expr) => self.handle_alter_table(expr).await,
-                    DdlExpr::DropTable(_) => {
-                        // TODO(LFC): Implement distributed drop table.
-                        // Seems the whole "drop table through GRPC interface" feature is not implemented?
-                        unimplemented!()
+                    DdlExpr::DropTable(expr) => {
+                        let table_name =
+                            TableName::new(&expr.catalog_name, &expr.schema_name, &expr.table_name);
+                        self.drop_table(table_name).await
                    }
                }
            }
@@ -68,13 +68,13 @@ impl GrpcQueryHandler for Instance {
 #[cfg(test)]
 mod test {
    use std::collections::HashMap;
-    use std::sync::Arc;

    use api::v1::column::{SemanticType, Values};
    use api::v1::ddl_request::Expr as DdlExpr;
    use api::v1::{
        alter_expr, AddColumn, AddColumns, AlterExpr, Column, ColumnDataType, ColumnDef,
-        CreateDatabaseExpr, CreateTableExpr, DdlRequest, InsertRequest, QueryRequest,
+        CreateDatabaseExpr, CreateTableExpr, DdlRequest, DropTableExpr, InsertRequest,
+        QueryRequest,
    };
    use catalog::helper::{TableGlobalKey, TableGlobalValue};
    use common_query::Output;
@@ -92,7 +92,9 @@ mod test {
            tests::create_distributed_instance("test_distributed_handle_ddl_request").await;
        let frontend = &instance.frontend;

-        test_handle_ddl_request(frontend).await
+        test_handle_ddl_request(frontend.as_ref()).await;
+
+        verify_table_is_dropped(&instance);
    }

    #[tokio::test(flavor = "multi_thread")]
@@ -101,22 +103,26 @@ mod test {
            tests::create_standalone_instance("test_standalone_handle_ddl_request").await;
        let instance = &standalone.instance;

-        test_handle_ddl_request(instance).await
+        test_handle_ddl_request(instance.as_ref()).await;
    }

-    async fn test_handle_ddl_request(instance: &Arc<Instance>) {
-        let query = Request::Ddl(DdlRequest {
+    async fn query(instance: &Instance, request: Request) -> Output {
+        GrpcQueryHandler::do_query(instance, request, QueryContext::arc())
+            .await
+            .unwrap()
+    }
+
+    async fn test_handle_ddl_request(instance: &Instance) {
+        let request = Request::Ddl(DdlRequest {
            expr: Some(DdlExpr::CreateDatabase(CreateDatabaseExpr {
                database_name: "database_created_through_grpc".to_string(),
                create_if_not_exists: true,
            })),
        });
-        let output = GrpcQueryHandler::do_query(instance.as_ref(), query, QueryContext::arc())
-            .await
-            .unwrap();
+        let output = query(instance, request).await;
        assert!(matches!(output, Output::AffectedRows(1)));

-        let query = Request::Ddl(DdlRequest {
+        let request = Request::Ddl(DdlRequest {
            expr: Some(DdlExpr::CreateTable(CreateTableExpr {
                catalog_name: "greptime".to_string(),
                schema_name: "database_created_through_grpc".to_string(),
@@ -139,12 +145,10 @@ mod test {
                ..Default::default()
            })),
        });
-        let output = GrpcQueryHandler::do_query(instance.as_ref(), query, QueryContext::arc())
-            .await
-            .unwrap();
+        let output = query(instance, request).await;
        assert!(matches!(output, Output::AffectedRows(0)));

-        let query = Request::Ddl(DdlRequest {
+        let request = Request::Ddl(DdlRequest {
            expr: Some(DdlExpr::Alter(AlterExpr {
                catalog_name: "greptime".to_string(),
                schema_name: "database_created_through_grpc".to_string(),
@@ -162,28 +166,22 @@ mod test {
                })),
            })),
        });
-        let output = GrpcQueryHandler::do_query(instance.as_ref(), query, QueryContext::arc())
-            .await
-            .unwrap();
+        let output = query(instance, request).await;
        assert!(matches!(output, Output::AffectedRows(0)));

-        let query = Request::Query(QueryRequest {
+        let request = Request::Query(QueryRequest {
            query: Some(Query::Sql("INSERT INTO database_created_through_grpc.table_created_through_grpc (a, b, ts) VALUES ('s', 1, 1672816466000)".to_string()))
        });
-        let output = GrpcQueryHandler::do_query(instance.as_ref(), query, QueryContext::arc())
-            .await
-            .unwrap();
+        let output = query(instance, request).await;
        assert!(matches!(output, Output::AffectedRows(1)));

-        let query = Request::Query(QueryRequest {
+        let request = Request::Query(QueryRequest {
            query: Some(Query::Sql(
                "SELECT ts, a, b FROM database_created_through_grpc.table_created_through_grpc"
                    .to_string(),
            )),
        });
-        let output = GrpcQueryHandler::do_query(instance.as_ref(), query, QueryContext::arc())
-            .await
-            .unwrap();
+        let output = query(instance, request).await;
        let Output::Stream(stream) = output else { unreachable!() };
        let recordbatches = RecordBatches::try_collect(stream).await.unwrap();
        let expected = "\
@@ -193,6 +191,28 @@ mod test {
 | 2023-01-04T07:14:26 | s | 1 |
 +---------------------+---+---+";
        assert_eq!(recordbatches.pretty_print().unwrap(), expected);
+
+        let request = Request::Ddl(DdlRequest {
+            expr: Some(DdlExpr::DropTable(DropTableExpr {
+                catalog_name: "greptime".to_string(),
+                schema_name: "database_created_through_grpc".to_string(),
+                table_name: "table_created_through_grpc".to_string(),
+            })),
+        });
+        let output = query(instance, request).await;
+        assert!(matches!(output, Output::AffectedRows(1)));
+    }
+
+    fn verify_table_is_dropped(instance: &MockDistributedInstance) {
+        assert!(instance.datanodes.iter().all(|(_, x)| x
+            .catalog_manager()
+            .table(
+                "greptime",
+                "database_created_through_grpc",
+                "table_created_through_grpc"
+            )
+            .unwrap()
+            .is_none()))
    }

    #[tokio::test(flavor = "multi_thread")]
@@ -201,7 +221,7 @@ mod test {

        let instance =
            tests::create_distributed_instance("test_distributed_insert_and_query").await;
-        let frontend = &instance.frontend;
+        let frontend = instance.frontend.as_ref();

        let table_name = "my_dist_table";
        let sql = format!(
@@ -310,17 +330,15 @@ CREATE TABLE {table_name} (
        test_insert_and_query_on_auto_created_table(instance).await
    }

-    async fn create_table(frontend: &Arc<Instance>, sql: String) {
-        let query = Request::Query(QueryRequest {
+    async fn create_table(frontend: &Instance, sql: String) {
+        let request = Request::Query(QueryRequest {
            query: Some(Query::Sql(sql)),
        });
-        let output = GrpcQueryHandler::do_query(frontend.as_ref(), query, QueryContext::arc())
-            .await
-            .unwrap();
+        let output = query(frontend, request).await;
        assert!(matches!(output, Output::AffectedRows(0)));
    }

-    async fn test_insert_and_query_on_existing_table(instance: &Arc<Instance>, table_name: &str) {
+    async fn test_insert_and_query_on_existing_table(instance: &Instance, table_name: &str) {
        let insert = InsertRequest {
            table_name: table_name.to_string(),
            columns: vec![
@@ -358,20 +376,16 @@ CREATE TABLE {table_name} (
            ..Default::default()
        };

-        let query = Request::Insert(insert);
-        let output = GrpcQueryHandler::do_query(instance.as_ref(), query, QueryContext::arc())
-            .await
-            .unwrap();
+        let request = Request::Insert(insert);
+        let output = query(instance, request).await;
        assert!(matches!(output, Output::AffectedRows(8)));

-        let query = Request::Query(QueryRequest {
+        let request = Request::Query(QueryRequest {
            query: Some(Query::Sql(format!(
                "SELECT ts, a FROM {table_name} ORDER BY ts"
            ))),
        });
-        let output = GrpcQueryHandler::do_query(instance.as_ref(), query, QueryContext::arc())
-            .await
-            .unwrap();
+        let output = query(instance, request).await;
        let Output::Stream(stream) = output else { unreachable!() };
        let recordbatches = RecordBatches::try_collect(stream).await.unwrap();
        let expected = "\
@@ -436,7 +450,7 @@ CREATE TABLE {table_name} (
        }
    }

-    async fn test_insert_and_query_on_auto_created_table(instance: &Arc<Instance>) {
+    async fn test_insert_and_query_on_auto_created_table(instance: &Instance) {
        let insert = InsertRequest {
            table_name: "auto_created_table".to_string(),
            columns: vec![
@@ -466,10 +480,8 @@ CREATE TABLE {table_name} (
        };

        // Test auto create not existed table upon insertion.
-        let query = Request::Insert(insert);
-        let output = GrpcQueryHandler::do_query(instance.as_ref(), query, QueryContext::arc())
-            .await
-            .unwrap();
+        let request = Request::Insert(insert);
+        let output = query(instance, request).await;
        assert!(matches!(output, Output::AffectedRows(3)));

        let insert = InsertRequest {
@@ -501,20 +513,16 @@ CREATE TABLE {table_name} (
        };

        // Test auto add not existed column upon insertion.
-        let query = Request::Insert(insert);
-        let output = GrpcQueryHandler::do_query(instance.as_ref(), query, QueryContext::arc())
-            .await
-            .unwrap();
+        let request = Request::Insert(insert);
+        let output = query(instance, request).await;
        assert!(matches!(output, Output::AffectedRows(3)));

-        let query = Request::Query(QueryRequest {
+        let request = Request::Query(QueryRequest {
            query: Some(Query::Sql(
                "SELECT ts, a, b FROM auto_created_table".to_string(),
            )),
        });
-        let output = GrpcQueryHandler::do_query(instance.as_ref(), query, QueryContext::arc())
-            .await
-            .unwrap();
+        let output = query(instance, request).await;
        let Output::Stream(stream) = output else { unreachable!() };
        let recordbatches = RecordBatches::try_collect(stream).await.unwrap();
        let expected = "\
@@ -410,7 +410,7 @@ mod test {
    use table::TableRef;

    use super::*;
-    use crate::expr_factory::{CreateExprFactory, DefaultCreateExprFactory};
+    use crate::expr_factory;

    struct DummyKvBackend;

@@ -823,10 +823,7 @@ mod test {
                _ => unreachable!(),
            };

-        let mut expr = DefaultCreateExprFactory
-            .create_expr_by_stmt(&create_table)
-            .await
-            .unwrap();
+        let mut expr = expr_factory::create_to_expr(&create_table, QueryContext::arc()).unwrap();
        let _result = dist_instance
            .create_table(&mut expr, create_table.partitions)
            .await
@@ -7,18 +7,22 @@ license.workspace = true
 [dependencies]
 api = { path = "../api" }
 async-trait = "0.1"
+chrono.workspace = true
 common-error = { path = "../common/error" }
 common-grpc = { path = "../common/grpc" }
 common-telemetry = { path = "../common/telemetry" }
 etcd-client = "0.10"
 rand = "0.8"
-serde = "1.0"
+serde.workspace = true
+serde_json.workspace = true
 snafu.workspace = true
+table = { path = "../table" }
 tokio.workspace = true
 tokio-stream = { version = "0.1", features = ["net"] }
 tonic.workspace = true

 [dev-dependencies]
+datatypes = { path = "../datatypes" }
 futures = "0.3"
 meta-srv = { path = "../meta-srv", features = ["mock"] }
 tower = "0.4"
@@ -12,15 +12,20 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

+use std::collections::HashMap;
 use std::time::Duration;

 use api::v1::meta::{HeartbeatRequest, Peer};
+use chrono::DateTime;
 use common_grpc::channel_manager::{ChannelConfig, ChannelManager};
+use datatypes::prelude::ConcreteDataType;
+use datatypes::schema::{ColumnSchema, RawSchema};
 use meta_client::client::MetaClientBuilder;
 use meta_client::rpc::{
    BatchPutRequest, CompareAndPutRequest, CreateRequest, DeleteRangeRequest, Partition,
    PutRequest, RangeRequest, TableName,
 };
+use table::metadata::{RawTableInfo, RawTableMeta, TableIdent, TableType};
 use tracing::{event, subscriber, Level};
 use tracing_subscriber::FmtSubscriber;

@@ -81,8 +86,8 @@ async fn run() {
    };

    let table_name = TableName::new("test_catalog", "test_schema", "test_table");
-
-    let create_req = CreateRequest::new(table_name)
+    let table_info = new_table_info();
+    let create_req = CreateRequest::new(table_name, &table_info)
        .add_partition(p1)
        .add_partition(p2);

@@ -141,3 +146,40 @@ async fn run() {
    let res = meta_client.range(range).await.unwrap();
    event!(Level::INFO, "get range result: {:#?}", res);
 }
+
+fn new_table_info() -> RawTableInfo {
+    RawTableInfo {
+        ident: TableIdent {
+            table_id: 0,
+            version: 0,
+        },
+        name: "test_table".to_string(),
+        desc: None,
+        catalog_name: "test_catalog".to_string(),
+        schema_name: "test_schema".to_string(),
+        meta: RawTableMeta {
+            schema: RawSchema {
+                column_schemas: vec![
+                    ColumnSchema::new(
+                        "ts",
+                        ConcreteDataType::timestamp_millisecond_datatype(),
+                        false,
+                    ),
+                    ColumnSchema::new("col1", ConcreteDataType::string_datatype(), true),
+                    ColumnSchema::new("col2", ConcreteDataType::string_datatype(), true),
+                ],
+                timestamp_index: Some(0),
+                version: 0,
+            },
+            primary_key_indices: vec![],
+            value_indices: vec![],
+            engine: "mito".to_string(),
+            next_column_id: 0,
+            region_numbers: vec![],
+            engine_options: HashMap::new(),
+            options: HashMap::new(),
+            created_on: DateTime::default(),
+        },
+        table_type: TableType::Base,
+    }
+}
@@ -180,8 +180,11 @@ impl MetaClient {
    /// a list of `datanode` addresses that are generated based on the partition
    /// information contained in the request and using some intelligent policies,
    /// such as load-based.
-    pub async fn create_route(&self, req: CreateRequest) -> Result<RouteResponse> {
-        self.router_client()?.create(req.into()).await?.try_into()
+    pub async fn create_route(&self, req: CreateRequest<'_>) -> Result<RouteResponse> {
+        self.router_client()?
+            .create(req.try_into()?)
+            .await?
+            .try_into()
    }

    /// Fetch routing information for tables. The smallest unit is the complete
@@ -291,12 +294,17 @@ impl MetaClient {

 #[cfg(test)]
 mod tests {
+    use std::collections::HashMap;
    use std::sync::Arc;

    use api::v1::meta::{HeartbeatRequest, Peer};
+    use chrono::DateTime;
+    use datatypes::prelude::ConcreteDataType;
+    use datatypes::schema::{ColumnSchema, RawSchema};
    use meta_srv::metasrv::Context;
    use meta_srv::selector::{Namespace, Selector};
    use meta_srv::Result as MetaResult;
+    use table::metadata::{RawTableInfo, RawTableMeta, TableIdent, TableType};

    use super::*;
    use crate::mocks;
@@ -401,6 +409,39 @@ mod tests {
        assert!(matches!(res.err(), Some(error::Error::NotStarted { .. })));
    }

+    fn new_table_info() -> RawTableInfo {
+        RawTableInfo {
+            ident: TableIdent {
+                table_id: 0,
+                version: 0,
+            },
+            name: "t".to_string(),
+            desc: None,
+            catalog_name: "c".to_string(),
+            schema_name: "s".to_string(),
+            meta: RawTableMeta {
+                schema: RawSchema {
+                    column_schemas: vec![ColumnSchema::new(
+                        "ts",
+                        ConcreteDataType::timestamp_millisecond_datatype(),
+                        false,
+                    )],
+                    timestamp_index: Some(0),
+                    version: 0,
+                },
+                primary_key_indices: vec![],
+                value_indices: vec![],
+                engine: "mito".to_string(),
+                next_column_id: 0,
+                region_numbers: vec![],
+                engine_options: HashMap::new(),
+                options: HashMap::new(),
+                created_on: DateTime::default(),
+            },
+            table_type: TableType::Base,
+        }
+    }
+
    #[tokio::test]
    async fn test_not_start_router_client() {
        let urls = &["127.0.0.1:3001", "127.0.0.1:3002"];
@@ -409,7 +450,9 @@ mod tests {
            .enable_store()
            .build();
        meta_client.start(urls).await.unwrap();
-        let req = CreateRequest::new(TableName::new("c", "s", "t"));
+
+        let table_info = new_table_info();
+        let req = CreateRequest::new(TableName::new("c", "s", "t"), &table_info);
        let res = meta_client.create_route(req).await;
        assert!(matches!(res.err(), Some(error::Error::NotStarted { .. })));
    }
@@ -504,7 +547,8 @@ mod tests {
            value_list: vec![b"Max1".to_vec(), b"Max2".to_vec()],
        };
        let table_name = TableName::new("test_catalog", "test_schema", "test_table");
-        let req = CreateRequest::new(table_name.clone())
+        let table_info = new_table_info();
+        let req = CreateRequest::new(table_name.clone(), &table_info)
            .add_partition(p1)
            .add_partition(p2);

@@ -513,13 +557,11 @@ mod tests {

        let req = RouteRequest::new().add_table_name(table_name.clone());
        let res = client.route(req).await.unwrap();
-        // empty table_routes since no TableGlobalValue is stored by datanode
-        assert!(res.table_routes.is_empty());
+        assert!(!res.table_routes.is_empty());

        let req = DeleteRequest::new(table_name.clone());
        let res = client.delete_route(req).await;
-        // empty table_routes since no TableGlobalValue is stored by datanode
-        assert!(res.is_err());
+        assert!(res.is_ok());
    }

    #[tokio::test]
@@ -72,6 +72,12 @@ pub enum Error {
        err_msg: String,
        backtrace: Backtrace,
    },
+
+    #[snafu(display("Failed to serde json, source: {}", source))]
+    SerdeJson {
+        source: serde_json::error::Error,
+        backtrace: Backtrace,
+    },
 }

 #[allow(dead_code)]
@@ -97,7 +103,8 @@ impl ErrorExt for Error {
            | Error::SendHeartbeat { .. }
            | Error::CreateHeartbeatStream { .. }
            | Error::CreateChannel { .. }
-            | Error::IllegalServerState { .. } => StatusCode::Internal,
+            | Error::IllegalServerState { .. }
+            | Error::SerdeJson { .. } => StatusCode::Internal,
            Error::RouteInfoCorrupted { .. } => StatusCode::Unexpected,
        }
    }
@@ -20,34 +20,40 @@ use api::v1::meta::{
    Table as PbTable,
 };
 use serde::{Deserialize, Serialize, Serializer};
-use snafu::OptionExt;
+use snafu::{OptionExt, ResultExt};
+use table::metadata::RawTableInfo;

 use crate::error;
 use crate::error::Result;
 use crate::rpc::{util, Peer, TableName};

 #[derive(Debug, Clone)]
-pub struct CreateRequest {
+pub struct CreateRequest<'a> {
    pub table_name: TableName,
    pub partitions: Vec<Partition>,
+    pub table_info: &'a RawTableInfo,
 }

-impl From<CreateRequest> for PbCreateRequest {
-    fn from(mut req: CreateRequest) -> Self {
-        Self {
+impl TryFrom<CreateRequest<'_>> for PbCreateRequest {
+    type Error = error::Error;
+
+    fn try_from(mut req: CreateRequest) -> Result<Self> {
+        Ok(Self {
            header: None,
            table_name: Some(req.table_name.into()),
            partitions: req.partitions.drain(..).map(Into::into).collect(),
-        }
+            table_info: serde_json::to_vec(&req.table_info).context(error::SerdeJsonSnafu)?,
+        })
    }
 }

-impl CreateRequest {
+impl<'a> CreateRequest<'a> {
    #[inline]
-    pub fn new(table_name: TableName) -> Self {
+    pub fn new(table_name: TableName, table_info: &'a RawTableInfo) -> Self {
        Self {
            table_name,
            partitions: vec![],
+            table_info,
        }
    }

@@ -302,9 +308,50 @@ mod tests {
        RouteResponse as PbRouteResponse, Table as PbTable, TableName as PbTableName,
        TableRoute as PbTableRoute,
    };
+    use chrono::DateTime;
+    use datatypes::prelude::ConcreteDataType;
+    use datatypes::schema::{ColumnSchema, RawSchema};
+    use table::metadata::{RawTableMeta, TableIdent, TableType};

    use super::*;

+    fn new_table_info() -> RawTableInfo {
+        RawTableInfo {
+            ident: TableIdent {
+                table_id: 0,
+                version: 0,
+            },
+            name: "t1".to_string(),
+            desc: None,
+            catalog_name: "c1".to_string(),
+            schema_name: "s1".to_string(),
+            meta: RawTableMeta {
+                schema: RawSchema {
+                    column_schemas: vec![
+                        ColumnSchema::new(
+                            "ts",
+                            ConcreteDataType::timestamp_millisecond_datatype(),
+                            false,
+                        ),
+                        ColumnSchema::new("c1", ConcreteDataType::string_datatype(), true),
+                        ColumnSchema::new("c2", ConcreteDataType::string_datatype(), true),
+                    ],
+                    timestamp_index: Some(0),
+                    version: 0,
+                },
+                primary_key_indices: vec![],
+                value_indices: vec![],
+                engine: "mito".to_string(),
+                next_column_id: 0,
+                region_numbers: vec![],
+                engine_options: HashMap::new(),
+                options: HashMap::new(),
+                created_on: DateTime::default(),
+            },
+            table_type: TableType::Base,
+        }
+    }
+
    #[test]
    fn test_create_request_trans() {
        let req = CreateRequest {
@@ -319,9 +366,9 @@ mod tests {
                    value_list: vec![b"v11".to_vec(), b"v22".to_vec()],
                },
            ],
+            table_info: &new_table_info(),
        };
-
-        let into_req: PbCreateRequest = req.into();
+        let into_req: PbCreateRequest = req.try_into().unwrap();

        assert!(into_req.header.is_none());
        let table_name = into_req.table_name;
@@ -31,6 +31,7 @@ regex = "1.6"
 serde = "1.0"
 serde_json = "1.0"
 snafu.workspace = true
+table = { path = "../table" }
 tokio.workspace = true
 tokio-stream = { version = "0.1", features = ["net"] }
 tonic.workspace = true
@@ -161,10 +161,16 @@ pub enum Error {
        backtrace: Backtrace,
    },

-    #[snafu(display("An error occurred in Meta, source: {}", source))]
-    MetaBoxedError {
-        #[snafu(backtrace)]
-        source: BoxedError,
+    #[snafu(display("Failed to decode table global value, source: {}", source))]
+    DecodeTableGlobalValue {
+        source: prost::DecodeError,
+        backtrace: Backtrace,
+    },
+
+    #[snafu(display("Unexpected, violated: {}", violated))]
+    Unexpected {
+        violated: String,
+        backtrace: Backtrace,
    },

    #[snafu(display("Invalid KVs length, expected: {}, actual: {}", expected, actual))]
@@ -234,6 +240,7 @@ impl ErrorExt for Error {
            | Error::SerializeToJson { .. }
            | Error::DeserializeFromJson { .. }
            | Error::DecodeTableRoute { .. }
+            | Error::DecodeTableGlobalValue { .. }
            | Error::NoLeader { .. }
            | Error::CreateChannel { .. }
            | Error::BatchGet { .. }
@@ -258,10 +265,10 @@ impl ErrorExt for Error {
            | Error::NextSequence { .. }
            | Error::MoveValue { .. }
            | Error::InvalidKvsLength { .. }
-            | Error::InvalidTxnResult { .. } => StatusCode::Unexpected,
+            | Error::InvalidTxnResult { .. }
+            | Error::Unexpected { .. } => StatusCode::Unexpected,
            Error::TableNotFound { .. } => StatusCode::TableNotFound,
            Error::InvalidCatalogValue { source, .. } => source.status_code(),
-            Error::MetaBoxedError { source } => source.status_code(),
        }
    }
 }
@@ -170,10 +170,14 @@ impl<'a> TableRouteKey<'a> {

    #[inline]
    pub fn removed_key(&self) -> String {
-        format!("{}-{}", REMOVED_PREFIX, self.key())
+        to_removed_key(&self.key())
    }
 }

+pub(crate) fn to_removed_key(key: &str) -> String {
+    format!("{REMOVED_PREFIX}-{key}")
+}
+
 #[derive(Eq, PartialEq, Debug, Clone, Hash)]
 pub struct StatKey {
    pub cluster_id: u64,
@@ -12,14 +12,17 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

+use std::collections::HashMap;
+
 use api::v1::meta::{
-    router_server, CreateRequest, DeleteRequest, Error, MoveValueRequest, Peer, PeerDict,
-    PutRequest, Region, RegionRoute, ResponseHeader, RouteRequest, RouteResponse, Table, TableName,
-    TableRoute, TableRouteValue,
+    router_server, BatchPutRequest, CreateRequest, DeleteRequest, Error, KeyValue,
+    MoveValueRequest, Peer, PeerDict, Region, RegionRoute, ResponseHeader, RouteRequest,
+    RouteResponse, Table, TableName, TableRoute, TableRouteValue,
 };
 use catalog::helper::{TableGlobalKey, TableGlobalValue};
 use common_telemetry::warn;
 use snafu::{OptionExt, ResultExt};
+use table::metadata::RawTableInfo;
 use tonic::{Request, Response};

 use crate::error;
@@ -90,10 +93,19 @@ async fn handle_create(
        header,
        table_name,
        partitions,
+        table_info,
    } = req;
    let table_name = table_name.context(error::EmptyTableNameSnafu)?;
-    let cluster_id = header.as_ref().map_or(0, |h| h.cluster_id);

+    let mut table_info: RawTableInfo =
+        serde_json::from_slice(&table_info).with_context(|_| error::DeserializeFromJsonSnafu {
+            input: format!(
+                "Corrupted table info: {}",
+                String::from_utf8_lossy(&table_info)
+            ),
+        })?;
+
+    let cluster_id = header.as_ref().map_or(0, |h| h.cluster_id);
    let peers = selector.select(cluster_id, &ctx).await?;
    if peers.is_empty() {
        let header = Some(ResponseHeader::failed(
@@ -106,13 +118,15 @@ async fn handle_create(
        });
    }
    let id = table_id_sequence.next().await?;
+    table_info.ident.table_id = id as u32;
+
    let table_route_key = TableRouteKey::with_table_name(id, &table_name)
        .key()
        .into_bytes();

    let table = Table {
        id,
-        table_name: Some(table_name),
+        table_name: Some(table_name.clone()),
        ..Default::default()
    };
    let mut region_routes = Vec::with_capacity(partitions.len());
@@ -139,7 +153,41 @@ async fn handle_create(
        peers: peers.clone(),
        table_route: Some(table_route.clone()),
    };
-    put_into_store(&ctx.kv_store, table_route_key, table_route_value).await?;
+
+    let table_global_key = TableGlobalKey {
+        catalog_name: table_name.catalog_name.clone(),
+        schema_name: table_name.schema_name.clone(),
+        table_name: table_name.table_name.clone(),
+    }
+    .to_string()
+    .into_bytes();
+
+    let table_global_value = create_table_global_value(&table_route_value, table_info)?
+        .as_bytes()
+        .context(error::InvalidCatalogValueSnafu)?;
+
+    let req = BatchPutRequest {
+        kvs: vec![
+            KeyValue {
+                key: table_global_key,
+                value: table_global_value,
+            },
+            KeyValue {
+                key: table_route_key,
+                value: table_route_value.into(),
+            },
+        ],
+        prev_kv: true,
+        ..Default::default()
+    };
+
+    let resp = ctx.kv_store.batch_put(req).await?;
+    if !resp.prev_kvs.is_empty() {
+        warn!(
+            "Caution: table meta values are replaced! \
+            Maybe last creation procedure for table {table_name:?} was aborted?"
+        );
+    }

    let header = Some(ResponseHeader::success(cluster_id));
    Ok(RouteResponse {
@@ -149,6 +197,44 @@ async fn handle_create(
    })
 }

+fn create_table_global_value(
+    table_route_value: &TableRouteValue,
+    table_info: RawTableInfo,
+) -> Result<TableGlobalValue> {
+    let peers = &table_route_value.peers;
+    let region_routes = &table_route_value
+        .table_route
+        .as_ref()
+        .context(error::UnexpectedSnafu {
+            violated: "table route should have been set",
+        })?
+        .region_routes;
+
+    let node_id = peers[region_routes[0].leader_peer_index as usize].id;
+
+    let mut regions_id_map = HashMap::with_capacity(region_routes.len());
+    for route in region_routes.iter() {
+        let node_id = peers[route.leader_peer_index as usize].id;
+        let region_id = route
+            .region
+            .as_ref()
+            .context(error::UnexpectedSnafu {
+                violated: "region should have been set",
+            })?
+            .id as u32;
+        regions_id_map
+            .entry(node_id)
+            .or_insert_with(Vec::new)
+            .push(region_id);
+    }
+
+    Ok(TableGlobalValue {
+        node_id,
+        regions_id_map,
+        table_info,
+    })
+}
+
 async fn handle_route(req: RouteRequest, ctx: Context) -> Result<RouteResponse> {
    let RouteRequest {
        header,
@@ -187,6 +273,9 @@ async fn handle_delete(req: DeleteRequest, ctx: Context) -> Result<RouteResponse
        .with_context(|| error::TableNotFoundSnafu {
            name: format!("{tgk}"),
        })?;
+
+    let _ = remove_table_global_value(&ctx.kv_store, &tgk).await?;
+
    let trk = TableRouteKey::with_table_global_key(tgv.table_id() as u64, &tgk);
    let (_, trv) = remove_table_route_value(&ctx.kv_store, &trk).await?;
    let (peers, table_routes) = fill_table_routes(vec![(tgv, trv)])?;
@@ -291,6 +380,20 @@ async fn remove_table_route_value(
    Ok((v.0, trv))
 }

+async fn remove_table_global_value(
+    kv_store: &KvStoreRef,
+    key: &TableGlobalKey,
+) -> Result<(Vec<u8>, TableGlobalValue)> {
+    let key = key.to_string();
+    let removed_key = crate::keys::to_removed_key(&key);
+    let kv = move_value(kv_store, key.as_bytes(), removed_key)
+        .await?
+        .context(error::TableNotFoundSnafu { name: key })?;
+    let value: TableGlobalValue =
+        TableGlobalValue::from_bytes(&kv.1).context(error::InvalidCatalogValueSnafu)?;
+    Ok((kv.0, value))
+}
+
 async fn get_table_global_value(
    kv_store: &KvStoreRef,
    key: &TableGlobalKey,
@@ -323,20 +426,3 @@ async fn move_value(

    Ok(res.kv.map(|kv| (kv.key, kv.value)))
 }
-
-async fn put_into_store(
-    kv_store: &KvStoreRef,
-    key: impl Into<Vec<u8>>,
-    value: impl Into<Vec<u8>>,
-) -> Result<()> {
-    let key = key.into();
-    let value = value.into();
-    let put_req = PutRequest {
-        key,
-        value,
-        ..Default::default()
-    };
-    let _ = kv_store.put(put_req).await?;
-
-    Ok(())
-}
@@ -13,7 +13,7 @@ anymap = "1.0.0-beta.2"
 arc-swap = "1.0"
 async-stream.workspace = true
 async-trait = "0.1"
-chrono = { version = "0.4", features = ["serde"] }
+chrono.workspace = true
 common-catalog = { path = "../common/catalog" }
 common-error = { path = "../common/error" }
 common-query = { path = "../common/query" }
@@ -7,7 +7,7 @@ license.workspace = true
 [dependencies]
 anymap = "1.0.0-beta.2"
 async-trait = "0.1"
-chrono = { version = "0.4", features = ["serde"] }
+chrono.workspace = true
 common-catalog = { path = "../common/catalog" }
 common-error = { path = "../common/error" }
 common-query = { path = "../common/query" }