diff --git a/Cargo.lock b/Cargo.lock index f4536c134de..812190d8130 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -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", diff --git a/Cargo.toml b/Cargo.toml index 7d61010864a..0c8fcb60fec 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -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"] } diff --git a/src/api/greptime/v1/meta/route.proto b/src/api/greptime/v1/meta/route.proto index d5bd2e043ba..1b1cc675077 100644 --- a/src/api/greptime/v1/meta/route.proto +++ b/src/api/greptime/v1/meta/route.proto @@ -36,6 +36,7 @@ message CreateRequest { TableName table_name = 2; repeated Partition partitions = 3; + bytes table_info = 4; } message RouteRequest { diff --git a/src/catalog/Cargo.toml b/src/catalog/Cargo.toml index 5aecb1f2bf8..9168dae7a2b 100644 --- a/src/catalog/Cargo.toml +++ b/src/catalog/Cargo.toml @@ -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" } diff --git a/src/catalog/src/error.rs b/src/catalog/src/error.rs index d706b59871a..6a053fc4691 100644 --- a/src/catalog/src/error.rs +++ b/src/catalog/src/error.rs @@ -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 = std::result::Result; @@ -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, } diff --git a/src/catalog/src/remote/manager.rs b/src/catalog/src/remote/manager.rs index 68686388530..ce7d24b621b 100644 --- a/src/catalog/src/remote/manager.rs +++ b/src/catalog/src/remote/manager.rs @@ -430,11 +430,18 @@ impl CatalogManager for RemoteCatalogManager { Ok(true) } - async fn deregister_table(&self, _request: DeregisterTableRequest) -> Result { - UnimplementedSnafu { - operation: "deregister table", - } - .fail() + async fn deregister_table(&self, request: DeregisterTableRequest) -> Result { + 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 { diff --git a/src/common/catalog/Cargo.toml b/src/common/catalog/Cargo.toml index 37c7b5ceff7..2ed50f13a04 100644 --- a/src/common/catalog/Cargo.toml +++ b/src/common/catalog/Cargo.toml @@ -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 diff --git a/src/common/time/Cargo.toml b/src/common/time/Cargo.toml index a55ab53eb14..419bf652531 100644 --- a/src/common/time/Cargo.toml +++ b/src/common/time/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" diff --git a/src/frontend/Cargo.toml b/src/frontend/Cargo.toml index 904a18dd4fb..27380222528 100644 --- a/src/frontend/Cargo.toml +++ b/src/frontend/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" } diff --git a/src/frontend/src/catalog.rs b/src/frontend/src/catalog.rs index b7a7926df93..ca7e44f41a5 100644 --- a/src/frontend/src/catalog.rs +++ b/src/frontend/src/catalog.rs @@ -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 { self.datanode_clients.clone() } @@ -79,15 +77,13 @@ impl CatalogManager for FrontendCatalogManager { Ok(()) } - async fn register_table(&self, _request: RegisterTableRequest) -> catalog::error::Result { - unimplemented!() + // TODO(LFC): Handle the table caching in (de)register_table. + async fn register_table(&self, _request: RegisterTableRequest) -> CatalogResult { + Ok(true) } - async fn deregister_table( - &self, - _request: DeregisterTableRequest, - ) -> catalog::error::Result { - unimplemented!() + async fn deregister_table(&self, _request: DeregisterTableRequest) -> CatalogResult { + 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, catalog::error::Error> = std::thread::spawn(|| { + let result: CatalogResult> = std::thread::spawn(|| { common_runtime::block_on_read(async move { let res = match backend.get(table_global_key.to_string().as_bytes()).await? { None => { diff --git a/src/frontend/src/error.rs b/src/frontend/src/error.rs index 488c0cb0387..75422511575 100644 --- a/src/frontend/src/error.rs +++ b/src/frontend/src/error.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, diff --git a/src/frontend/src/expr_factory.rs b/src/frontend/src/expr_factory.rs index f6bf9184785..4f2f34a5cab 100644 --- a/src/frontend/src/expr_factory.rs +++ b/src/frontend/src/expr_factory.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; #[async_trait::async_trait] pub trait CreateExprFactory { - async fn create_expr_by_stmt(&self, stmt: &CreateTable) -> Result; - 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 { - 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, - region_ids: Vec, +pub(crate) fn create_to_expr( create: &CreateTable, + query_ctx: QueryContextRef, ) -> Result { 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) } diff --git a/src/frontend/src/instance.rs b/src/frontend/src/instance.rs index 642eba1f3e1..b92c2ddfdc6 100644 --- a/src/frontend/src/instance.rs +++ b/src/frontend/src/instance.rs @@ -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> { impl Instance { async fn query_statement(&self, stmt: Statement, query_ctx: QueryContextRef) -> Result { - // 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, + ) { + 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::().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::>(); + 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")] diff --git a/src/frontend/src/instance/distributed.rs b/src/frontend/src/instance/distributed.rs index edd511898b5..53987620066 100644 --- a/src/frontend/src/instance/distributed.rs +++ b/src/frontend/src/instance/distributed.rs @@ -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, ) -> Result { - 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 { + 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, + table_info: &RawTableInfo, ) -> Result { 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 { - 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 { 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); diff --git a/src/frontend/src/instance/distributed/grpc.rs b/src/frontend/src/instance/distributed/grpc.rs index 77a8e98f0a6..20ddc4975c1 100644 --- a/src/frontend/src/instance/distributed/grpc.rs +++ b/src/frontend/src/instance/distributed/grpc.rs @@ -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 } } } diff --git a/src/frontend/src/instance/grpc.rs b/src/frontend/src/instance/grpc.rs index 462c2b5ce3b..6b61894687d 100644 --- a/src/frontend/src/instance/grpc.rs +++ b/src/frontend/src/instance/grpc.rs @@ -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) { - 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, 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, 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) { + 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 = "\ diff --git a/src/frontend/src/table.rs b/src/frontend/src/table.rs index 1327bf1cb31..65a18ef6159 100644 --- a/src/frontend/src/table.rs +++ b/src/frontend/src/table.rs @@ -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 diff --git a/src/meta-client/Cargo.toml b/src/meta-client/Cargo.toml index a0e268fca8c..678691c1c17 100644 --- a/src/meta-client/Cargo.toml +++ b/src/meta-client/Cargo.toml @@ -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" diff --git a/src/meta-client/examples/meta_client.rs b/src/meta-client/examples/meta_client.rs index 7558a4c8f22..4604662892e 100644 --- a/src/meta-client/examples/meta_client.rs +++ b/src/meta-client/examples/meta_client.rs @@ -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, + } +} diff --git a/src/meta-client/src/client.rs b/src/meta-client/src/client.rs index ffaf40c8099..8ae5682e9e2 100644 --- a/src/meta-client/src/client.rs +++ b/src/meta-client/src/client.rs @@ -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 { - self.router_client()?.create(req.into()).await?.try_into() + pub async fn create_route(&self, req: CreateRequest<'_>) -> Result { + 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] diff --git a/src/meta-client/src/error.rs b/src/meta-client/src/error.rs index 51ad1732787..8d246ceb876 100644 --- a/src/meta-client/src/error.rs +++ b/src/meta-client/src/error.rs @@ -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, } } diff --git a/src/meta-client/src/rpc/router.rs b/src/meta-client/src/rpc/router.rs index a5b3d47fd27..b3dd523039f 100644 --- a/src/meta-client/src/rpc/router.rs +++ b/src/meta-client/src/rpc/router.rs @@ -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, + pub table_info: &'a RawTableInfo, } -impl From for PbCreateRequest { - fn from(mut req: CreateRequest) -> Self { - Self { +impl TryFrom> for PbCreateRequest { + type Error = error::Error; + + fn try_from(mut req: CreateRequest) -> Result { + 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; diff --git a/src/meta-srv/Cargo.toml b/src/meta-srv/Cargo.toml index 0ccdeba585c..3b63781b7c1 100644 --- a/src/meta-srv/Cargo.toml +++ b/src/meta-srv/Cargo.toml @@ -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 diff --git a/src/meta-srv/src/error.rs b/src/meta-srv/src/error.rs index 81f8d4a4c6d..ad57f8d1bb3 100644 --- a/src/meta-srv/src/error.rs +++ b/src/meta-srv/src/error.rs @@ -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(), } } } diff --git a/src/meta-srv/src/keys.rs b/src/meta-srv/src/keys.rs index df4cde082b4..4b9e031af09 100644 --- a/src/meta-srv/src/keys.rs +++ b/src/meta-srv/src/keys.rs @@ -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, diff --git a/src/meta-srv/src/service/router.rs b/src/meta-srv/src/service/router.rs index 07fa2a7e96e..f32e7f1b3aa 100644 --- a/src/meta-srv/src/service/router.rs +++ b/src/meta-srv/src/service/router.rs @@ -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 { + 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 { let RouteRequest { header, @@ -187,6 +273,9 @@ async fn handle_delete(req: DeleteRequest, ctx: Context) -> Result Result<(Vec, 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>, - value: impl Into>, -) -> 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(()) -} diff --git a/src/mito/Cargo.toml b/src/mito/Cargo.toml index 26921b9e0f6..994bfc7b985 100644 --- a/src/mito/Cargo.toml +++ b/src/mito/Cargo.toml @@ -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" } diff --git a/src/table/Cargo.toml b/src/table/Cargo.toml index e7a0efc6c97..d46719df67a 100644 --- a/src/table/Cargo.toml +++ b/src/table/Cargo.toml @@ -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" }