diff --git a/Cargo.lock b/Cargo.lock index 6125713dddd..59291525d99 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -546,6 +546,7 @@ dependencies = [ "common-query", "common-recordbatch", "common-telemetry", + "common-time", "datafusion", "datatypes", "futures", @@ -856,6 +857,9 @@ dependencies = [ [[package]] name = "common-time" version = "0.1.0" +dependencies = [ + "chrono", +] [[package]] name = "concurrent-queue" diff --git a/src/catalog/Cargo.toml b/src/catalog/Cargo.toml index 09311839ff5..eca9c984117 100644 --- a/src/catalog/Cargo.toml +++ b/src/catalog/Cargo.toml @@ -12,6 +12,7 @@ common-error = { path = "../common/error" } common-query = { path = "../common/query" } common-recordbatch = { path = "../common/recordbatch" } common-telemetry = { path = "../common/telemetry" } +common-time = { path = "../common/time" } datafusion = { git = "https://github.com/apache/arrow-datafusion.git", branch = "arrow2", features = ["simd"] } datatypes = { path = "../datatypes" } futures = "0.3" diff --git a/src/catalog/src/error.rs b/src/catalog/src/error.rs index f527b3d845b..b98d7a31c0e 100644 --- a/src/catalog/src/error.rs +++ b/src/catalog/src/error.rs @@ -86,6 +86,15 @@ pub enum Error { #[snafu(backtrace)] source: datatypes::error::Error, }, + + #[snafu(display( + "Failed to insert table creation record to system catalog, source: {}", + source + ))] + InsertTableRecord { + #[snafu(backtrace)] + source: table::error::Error, + }, } pub type Result = std::result::Result; @@ -105,7 +114,8 @@ impl ErrorExt for Error { | Error::ValueDeserialize { .. } | Error::CatalogNotFound { .. } | Error::OpenTable { .. } - | Error::ReadSystemCatalog { .. } => StatusCode::StorageUnavailable, + | Error::ReadSystemCatalog { .. } + | Error::InsertTableRecord { .. } => StatusCode::StorageUnavailable, Error::RegisterTable { .. } | Error::SystemCatalogSchema { .. } => StatusCode::Internal, Error::TableExists { .. } => StatusCode::TableAlreadyExists, } diff --git a/src/catalog/src/lib.rs b/src/catalog/src/lib.rs index 39581c4f5e4..cf74028790e 100644 --- a/src/catalog/src/lib.rs +++ b/src/catalog/src/lib.rs @@ -3,9 +3,13 @@ use std::any::Any; use std::sync::Arc; +use table::metadata::TableId; +use table::TableRef; + pub use crate::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME}; pub use crate::manager::LocalCatalogManager; pub use crate::schema::{SchemaProvider, SchemaProviderRef}; + mod consts; pub mod error; mod manager; @@ -57,7 +61,28 @@ pub type CatalogProviderRef = Arc; #[async_trait::async_trait] pub trait CatalogManager: CatalogList { + /// Starts a catalog manager. async fn start(&self) -> error::Result<()>; + + /// Returns next available table id. + fn next_table_id(&self) -> TableId; + + /// Registers a table given given catalog/schema to catalog manager, + /// returns table registered. + async fn register_table(&self, request: RegisterTableRequest) -> error::Result; } pub type CatalogManagerRef = Arc; + +pub struct RegisterTableRequest { + pub catalog: Option, + pub schema: Option, + pub table_name: String, + pub table_id: TableId, + pub table: TableRef, +} + +/// Formats table fully-qualified name +pub fn format_full_table_name(catalog: &str, schema: &str, table: &str) -> String { + format!("{}.{}.{}", catalog, schema, table) +} diff --git a/src/catalog/src/manager.rs b/src/catalog/src/manager.rs index 6d4718dcf06..affec868153 100644 --- a/src/catalog/src/manager.rs +++ b/src/catalog/src/manager.rs @@ -1,13 +1,16 @@ use std::any::Any; +use std::sync::atomic::{AtomicU32, Ordering}; use std::sync::Arc; use common_recordbatch::RecordBatch; -use common_telemetry::info; +use common_telemetry::{debug, info}; use datatypes::prelude::ScalarVector; use datatypes::vectors::{BinaryVector, UInt8Vector}; +use futures_util::lock::Mutex; use futures_util::StreamExt; use snafu::{ensure, OptionExt, ResultExt}; use table::engine::{EngineContext, TableEngineRef}; +use table::metadata::TableId; use table::requests::OpenTableRequest; use table::table::numbers::NumbersTable; @@ -15,14 +18,17 @@ use super::error::Result; use crate::consts::{INFORMATION_SCHEMA_NAME, SYSTEM_CATALOG_NAME, SYSTEM_CATALOG_TABLE_NAME}; use crate::error::{ CatalogNotFoundSnafu, OpenTableSnafu, ReadSystemCatalogSnafu, SchemaNotFoundSnafu, - SystemCatalogSnafu, SystemCatalogTypeMismatchSnafu, TableNotFoundSnafu, + SystemCatalogSnafu, SystemCatalogTypeMismatchSnafu, TableExistsSnafu, TableNotFoundSnafu, }; use crate::memory::{MemoryCatalogList, MemoryCatalogProvider, MemorySchemaProvider}; -use crate::system::{decode_system_catalog, Entry, SystemCatalogTable, TableEntry}; +use crate::system::{ + decode_system_catalog, Entry, SystemCatalogTable, TableEntry, ENTRY_TYPE_INDEX, KEY_INDEX, + VALUE_INDEX, +}; use crate::tables::SystemCatalog; use crate::{ - CatalogList, CatalogManager, CatalogProvider, CatalogProviderRef, SchemaProvider, - DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, + format_full_table_name, CatalogList, CatalogManager, CatalogProvider, CatalogProviderRef, + RegisterTableRequest, SchemaProvider, DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, }; /// A `CatalogManager` consists of a system catalog and a bunch of user catalogs. @@ -30,6 +36,8 @@ pub struct LocalCatalogManager { system: Arc, catalogs: Arc, engine: TableEngineRef, + next_table_id: AtomicU32, + lock: Mutex<()>, } impl LocalCatalogManager { @@ -46,6 +54,8 @@ impl LocalCatalogManager { system: system_catalog, catalogs: memory_catalog_list, engine, + next_table_id: AtomicU32::new(0), + lock: Mutex::new(()), }) } @@ -53,14 +63,22 @@ impl LocalCatalogManager { pub async fn init(&self) -> Result<()> { self.init_system_catalog()?; let mut system_records = self.system.information_schema.system.records().await?; + let mut max_table_id = 0; while let Some(records) = system_records .next() .await .transpose() .context(ReadSystemCatalogSnafu)? { - self.handle_system_catalog_entries(records).await?; + let table_id = self.handle_system_catalog_entries(records).await?; + max_table_id = max_table_id.max(table_id); } + info!( + "All system catalog entries processed, max table id: {}", + max_table_id + ); + self.next_table_id + .store(max_table_id + 1, Ordering::Relaxed); Ok(()) } @@ -89,9 +107,11 @@ impl LocalCatalogManager { Ok(()) } - async fn handle_system_catalog_entries(&self, records: RecordBatch) -> Result<()> { + /// Processes records from system catalog table and returns the max table id persisted + /// in system catalog table. + async fn handle_system_catalog_entries(&self, records: RecordBatch) -> Result { ensure!( - records.df_recordbatch.columns().len() >= 4, + records.df_recordbatch.columns().len() >= 6, SystemCatalogSnafu { msg: format!( "Length mismatch: {}", @@ -102,19 +122,26 @@ impl LocalCatalogManager { let entry_type = UInt8Vector::try_from_arrow_array(&records.df_recordbatch.columns()[0]) .with_context(|_| SystemCatalogTypeMismatchSnafu { - data_type: records.df_recordbatch.columns()[0].data_type().clone(), + data_type: records.df_recordbatch.columns()[ENTRY_TYPE_INDEX] + .data_type() + .clone(), })?; let key = BinaryVector::try_from_arrow_array(&records.df_recordbatch.columns()[1]) .with_context(|_| SystemCatalogTypeMismatchSnafu { - data_type: records.df_recordbatch.columns()[1].data_type().clone(), + data_type: records.df_recordbatch.columns()[KEY_INDEX] + .data_type() + .clone(), })?; - let value = BinaryVector::try_from_arrow_array(&records.df_recordbatch.columns()[1]) + let value = BinaryVector::try_from_arrow_array(&records.df_recordbatch.columns()[3]) .with_context(|_| SystemCatalogTypeMismatchSnafu { - data_type: records.df_recordbatch.columns()[3].data_type().clone(), + data_type: records.df_recordbatch.columns()[VALUE_INDEX] + .data_type() + .clone(), })?; + let mut max_table_id = 0; for ((t, k), v) in entry_type .iter_data() .zip(key.iter_data()) @@ -143,13 +170,14 @@ impl LocalCatalogManager { info!("Registered schema: {:?}", s); } Entry::Table(t) => { + debug!("t: {:?}", t); self.open_and_register_table(&t).await?; - info!("Registered table: {:?}", t) + info!("Registered table: {:?}", t); + max_table_id = max_table_id.max(t.table_id); } } } - - Ok(()) + Ok(max_table_id) } async fn open_and_register_table(&self, t: &TableEntry) -> Result<()> { @@ -230,4 +258,50 @@ impl CatalogManager for LocalCatalogManager { async fn start(&self) -> Result<()> { self.init().await } + + #[inline] + fn next_table_id(&self) -> TableId { + self.next_table_id.fetch_add(1, Ordering::Relaxed) + } + + async fn register_table(&self, request: RegisterTableRequest) -> Result { + let _lock = self.lock.lock().await; + let catalog_name = request + .catalog + .unwrap_or_else(|| DEFAULT_CATALOG_NAME.to_string()); + let schema_name = request + .schema + .unwrap_or_else(|| DEFAULT_SCHEMA_NAME.to_string()); + + let catalog = self + .catalogs + .catalog(&catalog_name) + .context(CatalogNotFoundSnafu { + catalog_name: &catalog_name, + })?; + let schema = catalog + .schema(&schema_name) + .with_context(|| SchemaNotFoundSnafu { + schema_info: format!("{}.{}", catalog_name, schema_name), + })?; + + if schema.table_exist(&request.table_name) { + return TableExistsSnafu { + table: format_full_table_name(&catalog_name, &schema_name, &request.table_name), + } + .fail(); + } + + self.system + .register_table( + catalog_name, + schema_name, + request.table_name.clone(), + request.table_id, + ) + .await?; + + schema.register_table(request.table_name, request.table)?; + Ok(1) + } } diff --git a/src/catalog/src/system.rs b/src/catalog/src/system.rs index 406fcba870b..b9fc5d25def 100644 --- a/src/catalog/src/system.rs +++ b/src/catalog/src/system.rs @@ -1,15 +1,18 @@ use std::any::Any; +use std::collections::HashMap; use std::sync::Arc; use common_query::logical_plan::Expr; use common_recordbatch::SendableRecordBatchStream; -use datatypes::prelude::ConcreteDataType; +use common_telemetry::debug; +use datatypes::prelude::{ConcreteDataType, ScalarVector}; use datatypes::schema::{ColumnSchema, Schema, SchemaRef}; +use datatypes::vectors::{BinaryVector, Int64Vector, UInt8Vector}; use serde::{Deserialize, Serialize}; use snafu::{ensure, OptionExt, ResultExt}; use table::engine::{EngineContext, TableEngineRef}; use table::metadata::TableId; -use table::requests::{CreateTableRequest, OpenTableRequest}; +use table::requests::{CreateTableRequest, InsertRequest, OpenTableRequest}; use table::{Table, TableRef}; use crate::consts::{ @@ -21,6 +24,11 @@ use crate::error::{ OpenSystemCatalogSnafu, Result, SystemCatalogSchemaSnafu, ValueDeserializeSnafu, }; +pub const ENTRY_TYPE_INDEX: usize = 0; +pub const KEY_INDEX: usize = 1; +pub const TIMESTAMP_INDEX: usize = 2; +pub const VALUE_INDEX: usize = 3; + pub struct SystemCatalogTable { schema: SchemaRef, pub table: TableRef, @@ -44,6 +52,11 @@ impl Table for SystemCatalogTable { ) -> table::Result { panic!("System catalog table does not support scan!") } + + /// Insert values into table. + async fn insert(&self, request: InsertRequest) -> table::error::Result { + self.table.insert(request).await + } } impl SystemCatalogTable { @@ -67,10 +80,12 @@ impl SystemCatalogTable { // system catalog table is not yet created, try to create let request = CreateTableRequest { id: SYSTEM_CATALOG_TABLE_ID, - name: SYSTEM_CATALOG_TABLE_NAME.to_string(), + catalog_name: Some(SYSTEM_CATALOG_NAME.to_string()), + schema_name: Some(INFORMATION_SCHEMA_NAME.to_string()), + table_name: SYSTEM_CATALOG_TABLE_NAME.to_string(), desc: Some("System catalog table".to_string()), schema: schema.clone(), - primary_key_indices: vec![0, 1, 2], + primary_key_indices: vec![ENTRY_TYPE_INDEX, KEY_INDEX, TIMESTAMP_INDEX], create_if_not_exists: true, }; @@ -97,28 +112,83 @@ impl SystemCatalogTable { /// - timestamp: currently not used. /// - value: JSON-encoded value of entry's metadata. fn build_system_catalog_schema() -> Result { - let mut cols = Vec::with_capacity(6); - cols.push(ColumnSchema::new( + let cols = vec![ + ColumnSchema::new( + "entry_type".to_string(), + ConcreteDataType::uint8_datatype(), + false, + ), + ColumnSchema::new( + "key".to_string(), + ConcreteDataType::binary_datatype(), + false, + ), + ColumnSchema::new( + "timestamp".to_string(), + ConcreteDataType::int64_datatype(), + false, + ), + ColumnSchema::new( + "value".to_string(), + ConcreteDataType::binary_datatype(), + false, + ), + ColumnSchema::new( + "gmt_created".to_string(), + ConcreteDataType::int64_datatype(), + false, + ), + ColumnSchema::new( + "gmt_modified".to_string(), + ConcreteDataType::int64_datatype(), + false, + ), + ]; + + Schema::with_timestamp_index(cols, TIMESTAMP_INDEX).context(SystemCatalogSchemaSnafu) +} + +pub fn build_table_insert_request(full_table_name: String, table_id: TableId) -> InsertRequest { + let mut columns_values = HashMap::with_capacity(6); + columns_values.insert( "entry_type".to_string(), - ConcreteDataType::uint8_datatype(), - false, - )); - cols.push(ColumnSchema::new( + Arc::new(UInt8Vector::from_vec(vec![EntryType::Table as u8])) as _, + ); + + columns_values.insert( "key".to_string(), - ConcreteDataType::binary_datatype(), - false, - )); - cols.push(ColumnSchema::new( + Arc::new(BinaryVector::from_slice(&[full_table_name.as_bytes()])) as _, + ); + + // Timestamp in key part is intentionally left to 0 + columns_values.insert( "timestamp".to_string(), - ConcreteDataType::int64_datatype(), - false, - )); - cols.push(ColumnSchema::new( + Arc::new(Int64Vector::from_vec(vec![0])) as _, + ); + + columns_values.insert( "value".to_string(), - ConcreteDataType::binary_datatype(), - false, - )); - Schema::with_timestamp_index(cols, 2).context(SystemCatalogSchemaSnafu) + Arc::new(BinaryVector::from_slice(&[serde_json::to_string( + &TableEntryValue { table_id }, + ) + .unwrap() + .as_bytes()])) as _, + ); + + columns_values.insert( + "gmt_created".to_string(), + Arc::new(Int64Vector::from_vec(vec![0])) as _, + ); + + columns_values.insert( + "gmt_modified".to_string(), + Arc::new(Int64Vector::from_vec(vec![0])) as _, + ); + + InsertRequest { + table_name: SYSTEM_CATALOG_TABLE_NAME.to_string(), + columns_values, + } } pub fn decode_system_catalog( @@ -126,6 +196,10 @@ pub fn decode_system_catalog( key: Option<&[u8]>, value: Option<&[u8]>, ) -> Result { + debug!( + "Decode system catalog entry: {:?}, {:?}, {:?}", + entry_type, key, value + ); let entry_type = entry_type.context(InvalidKeySnafu { key: None })?; let key = String::from_utf8_lossy(key.context(InvalidKeySnafu { key: None })?); @@ -163,6 +237,7 @@ pub fn decode_system_catalog( } ); let value = value.context(EmptyValueSnafu)?; + debug!("Table meta value: {}", String::from_utf8_lossy(value)); let table_meta: TableEntryValue = serde_json::from_slice(value).context(ValueDeserializeSnafu)?; Ok(Entry::Table(TableEntry { diff --git a/src/catalog/src/tables.rs b/src/catalog/src/tables.rs index c83f6e0b357..40d04329ce3 100644 --- a/src/catalog/src/tables.rs +++ b/src/catalog/src/tables.rs @@ -14,26 +14,31 @@ use datatypes::schema::{ColumnSchema, Schema, SchemaRef}; use datatypes::value::Value; use datatypes::vectors::VectorRef; use futures::Stream; +use snafu::ResultExt; use table::engine::TableEngineRef; +use table::metadata::TableId; use table::{Table, TableRef}; use crate::consts::{INFORMATION_SCHEMA_NAME, SYSTEM_CATALOG_TABLE_NAME}; -use crate::system::SystemCatalogTable; -use crate::{CatalogListRef, CatalogProvider, SchemaProvider, SchemaProviderRef}; +use crate::error::InsertTableRecordSnafu; +use crate::system::{build_table_insert_request, SystemCatalogTable}; +use crate::{ + format_full_table_name, CatalogListRef, CatalogProvider, SchemaProvider, SchemaProviderRef, +}; /// Tables holds all tables created by user. pub struct Tables { schema: SchemaRef, catalogs: CatalogListRef, - engine: TableEngineRef, + engine_name: String, } impl Tables { - pub fn new(catalogs: CatalogListRef, engine: TableEngineRef) -> Self { + pub fn new(catalogs: CatalogListRef, engine_name: String) -> Self { Self { schema: Arc::new(build_schema_for_tables()), catalogs, - engine, + engine_name, } } } @@ -56,7 +61,7 @@ impl Table for Tables { ) -> table::error::Result { let catalogs = self.catalogs.clone(); let schema_ref = self.schema.clone(); - let engine_name = self.engine.name().to_string(); + let engine_name = self.engine_name.clone(); let stream = stream!({ for catalog_name in catalogs.catalog_names() { @@ -189,13 +194,29 @@ impl SystemCatalog { engine: TableEngineRef, ) -> Self { let schema = InformationSchema { - tables: Arc::new(Tables::new(catalogs, engine)), + tables: Arc::new(Tables::new(catalogs, engine.name().to_string())), system: Arc::new(system), }; Self { information_schema: Arc::new(schema), } } + + pub async fn register_table( + &self, + catalog: String, + schema: String, + table_name: String, + table_id: TableId, + ) -> crate::error::Result { + let full_table_name = format_full_table_name(&catalog, &schema, &table_name); + let request = build_table_insert_request(full_table_name, table_id); + self.information_schema + .system + .insert(request) + .await + .context(InsertTableRecordSnafu) + } } impl CatalogProvider for SystemCatalog { @@ -249,3 +270,80 @@ fn build_schema_for_tables() -> Schema { ]; Schema::new(cols) } + +#[cfg(test)] +mod tests { + use datatypes::arrow::array::Utf8Array; + use datatypes::arrow::datatypes::DataType; + use futures_util::StreamExt; + use table::table::numbers::NumbersTable; + + use super::*; + use crate::memory::{new_memory_catalog_list, MemoryCatalogProvider, MemorySchemaProvider}; + use crate::CatalogList; + + #[tokio::test] + async fn test_tables() { + let catalog_list = new_memory_catalog_list().unwrap(); + let catalog_provider = Arc::new(MemoryCatalogProvider::default()); + let schema = Arc::new(MemorySchemaProvider::new()); + schema + .register_table("test_table".to_string(), Arc::new(NumbersTable::default())) + .unwrap(); + catalog_provider.register_schema("test_schema".to_string(), schema); + catalog_list.register_catalog("test_catalog".to_string(), catalog_provider); + let tables = Tables::new(catalog_list, "test_engine".to_string()); + + let mut tables_stream = tables.scan(&None, &[], None).await.unwrap(); + if let Some(t) = tables_stream.next().await { + let batch = t.unwrap().df_recordbatch; + assert_eq!(1, batch.num_rows()); + assert_eq!(4, batch.num_columns()); + assert_eq!(&DataType::LargeUtf8, batch.column(0).data_type()); + assert_eq!(&DataType::LargeUtf8, batch.column(1).data_type()); + assert_eq!(&DataType::LargeUtf8, batch.column(2).data_type()); + assert_eq!(&DataType::LargeUtf8, batch.column(3).data_type()); + assert_eq!( + "test_catalog", + batch + .column(0) + .as_any() + .downcast_ref::>() + .unwrap() + .value(0) + ); + + assert_eq!( + "test_schema", + batch + .column(1) + .as_any() + .downcast_ref::>() + .unwrap() + .value(0) + ); + + assert_eq!( + "test_table", + batch + .column(2) + .as_any() + .downcast_ref::>() + .unwrap() + .value(0) + ); + + assert_eq!( + "test_engine", + batch + .column(3) + .as_any() + .downcast_ref::>() + .unwrap() + .value(0) + ); + } else { + panic!("Record batch should not be empty!") + } + } +} diff --git a/src/common/time/Cargo.toml b/src/common/time/Cargo.toml index 9652cb17c31..52a85c20f31 100644 --- a/src/common/time/Cargo.toml +++ b/src/common/time/Cargo.toml @@ -6,3 +6,4 @@ edition = "2021" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html [dependencies] +chrono = "0.4" \ No newline at end of file diff --git a/src/common/time/src/lib.rs b/src/common/time/src/lib.rs index 84b9bfaf125..23e98710814 100644 --- a/src/common/time/src/lib.rs +++ b/src/common/time/src/lib.rs @@ -1,5 +1,6 @@ pub mod range; pub mod timestamp; +pub mod util; pub use range::RangeMillis; pub use timestamp::TimestampMillis; diff --git a/src/common/time/src/util.rs b/src/common/time/src/util.rs new file mode 100644 index 00000000000..c42e169a3ce --- /dev/null +++ b/src/common/time/src/util.rs @@ -0,0 +1,4 @@ +/// Calculates the time duration since UNIX_EPOCH in milliseconds. +pub fn current_timestamp() -> i64 { + chrono::Utc::now().timestamp() +} diff --git a/src/datanode/Cargo.toml b/src/datanode/Cargo.toml index d9cde636c04..8bfdd028330 100644 --- a/src/datanode/Cargo.toml +++ b/src/datanode/Cargo.toml @@ -33,7 +33,7 @@ sql = { path = "../sql" } storage = { path = "../storage" } store-api = { path = "../store-api" } table = { path = "../table" } -table-engine = { path = "../table-engine" } +table-engine = { path = "../table-engine", features = ["test"] } tokio = { version = "1.18", features = ["full"] } tonic = "0.8" tokio-stream = { version = "0.1.8", features = ["net"] } diff --git a/src/datanode/src/error.rs b/src/datanode/src/error.rs index 043921861c1..6601208481e 100644 --- a/src/datanode/src/error.rs +++ b/src/datanode/src/error.rs @@ -121,6 +121,39 @@ pub enum Error { #[snafu(display("Unsupported expr type: {}", name))] UnsupportedExpr { name: String }, + + #[snafu(display("Invalid CREATE TABLE sql statement, cause: {}", msg))] + InvalidCreateTableSql { msg: String, backtrace: Backtrace }, + + #[snafu(display("Failed to create schema when creating table: {}", source))] + CreateSchema { + #[snafu(backtrace)] + source: datatypes::error::Error, + }, + + #[snafu(display("SQL data type not supported yet: {:?}", t))] + SqlTypeNotSupported { + t: sql::ast::DataType, + backtrace: Backtrace, + }, + + #[snafu(display("Specified timestamp key or primary key column not found: {}", name))] + KeyColumnNotFound { name: String, backtrace: Backtrace }, + + #[snafu(display( + "Constraint in CREATE TABLE statement is not supported yet: {}", + constraint + ))] + ConstraintNotSupported { + constraint: String, + backtrace: Backtrace, + }, + + #[snafu(display("Failed to insert into system catalog table: {}", source))] + InsertSystemCatalog { + #[snafu(backtrace)] + source: catalog::error::Error, + }, } pub type Result = std::result::Result; @@ -139,13 +172,19 @@ impl ErrorExt for Error { | Error::ParseSqlValue { .. } | Error::ColumnTypeMismatch { .. } | Error::IllegalInsertData { .. } - | Error::DecodeInsert { .. } => StatusCode::InvalidArguments, + | Error::DecodeInsert { .. } + | Error::InvalidCreateTableSql { .. } + | Error::SqlTypeNotSupported { .. } + | Error::CreateSchema { .. } + | Error::KeyColumnNotFound { .. } + | Error::ConstraintNotSupported { .. } => StatusCode::InvalidArguments, // TODO(yingwen): Further categorize http error. Error::StartHttp { .. } | Error::ParseAddr { .. } | Error::TcpBind { .. } | Error::StartGrpc { .. } | Error::CreateDir { .. } + | Error::InsertSystemCatalog { .. } | Error::Conversion { .. } | Error::UnsupportedExpr { .. } => StatusCode::Internal, Error::InitBackend { .. } => StatusCode::StorageUnavailable, diff --git a/src/datanode/src/instance.rs b/src/datanode/src/instance.rs index ac2fb37ad29..f4edcd3e510 100644 --- a/src/datanode/src/instance.rs +++ b/src/datanode/src/instance.rs @@ -3,27 +3,23 @@ use std::{fs, path, sync::Arc}; use api::v1::InsertExpr; use catalog::{CatalogManagerRef, DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME}; use common_telemetry::logging::info; -use datatypes::prelude::ConcreteDataType; -use datatypes::schema::{ColumnSchema, Schema}; use log_store::fs::{config::LogConfig, log::LocalFileLogStore}; use object_store::{backend::fs::Backend, util, ObjectStore}; use query::query_engine::{Output, QueryEngineFactory, QueryEngineRef}; use snafu::{OptionExt, ResultExt}; use sql::statements::statement::Statement; use storage::{config::EngineConfig as StorageEngineConfig, EngineImpl}; -use table::engine::EngineContext; -use table::engine::TableEngine; -use table::requests::CreateTableRequest; +#[cfg(test)] +use table::engine::TableEngineRef; use table_engine::config::EngineConfig as TableEngineConfig; use table_engine::engine::MitoEngine; use crate::datanode::{DatanodeOptions, ObjectStoreConfig}; use crate::error::{ - self, CreateTableSnafu, ExecuteSqlSnafu, InsertSnafu, NewCatalogSnafu, Result, - TableNotFoundSnafu, + self, ExecuteSqlSnafu, InsertSnafu, NewCatalogSnafu, Result, TableNotFoundSnafu, }; use crate::server::grpc::insert::insertion_expr_to_request; -use crate::sql::SqlHandler; +use crate::sql::{SqlHandler, SqlRequest}; type DefaultEngine = MitoEngine>; @@ -31,7 +27,6 @@ type DefaultEngine = MitoEngine>; pub struct Instance { // Query service query_engine: QueryEngineRef, - table_engine: DefaultEngine, sql_handler: SqlHandler, // Catalog list catalog_manager: CatalogManagerRef, @@ -63,8 +58,7 @@ impl Instance { Ok(Self { query_engine, - sql_handler: SqlHandler::new(table_engine.clone()), - table_engine, + sql_handler: SqlHandler::new(table_engine, catalog_manager.clone()), catalog_manager, }) } @@ -110,7 +104,7 @@ impl Instance { .await .context(ExecuteSqlSnafu) } - Statement::Insert(_) => { + Statement::Insert(i) => { let schema_provider = self .catalog_manager .catalog(DEFAULT_CATALOG_NAME) @@ -118,11 +112,28 @@ impl Instance { .schema(DEFAULT_SCHEMA_NAME) .unwrap(); - let request = self - .sql_handler - .statement_to_request(schema_provider, stmt)?; + let request = self.sql_handler.insert_to_request(schema_provider, *i)?; self.sql_handler.execute(request).await } + + Statement::Create(c) => { + let table_id = self.catalog_manager.next_table_id(); + let _engine_name = c.engine.clone(); + // TODO(hl): Select table engine by engine_name + + let request = self.sql_handler.create_to_request(table_id, c)?; + let catalog_name = request.catalog_name.clone(); + let schema_name = request.schema_name.clone(); + let table_name = request.table_name.clone(); + let table_id = request.id; + info!( + "Creating table, catalog: {:?}, schema: {:?}, table name: {:?}, table id: {}", + catalog_name, schema_name, table_name, table_id + ); + + self.sql_handler.execute(SqlRequest::Create(request)).await + } + _ => unimplemented!(), } } @@ -132,7 +143,23 @@ impl Instance { .start() .await .context(NewCatalogSnafu)?; - // FIXME(dennis): create a demo table for test + Ok(()) + } + + #[cfg(test)] + pub fn table_engine(&self) -> TableEngineRef { + self.sql_handler.table_engine() + } + + #[cfg(test)] + pub async fn create_test_table(&self) -> Result<()> { + use datatypes::data_type::ConcreteDataType; + use datatypes::schema::{ColumnSchema, Schema}; + use table::engine::EngineContext; + use table::requests::CreateTableRequest; + + use crate::error::CreateTableSnafu; + let column_schemas = vec![ ColumnSchema::new("host", ConcreteDataType::string_datatype(), false), ColumnSchema::new("cpu", ConcreteDataType::float64_datatype(), true), @@ -142,12 +169,14 @@ impl Instance { let table_name = "demo"; let table = self - .table_engine + .table_engine() .create_table( &EngineContext::default(), CreateTableRequest { id: 1, - name: table_name.to_string(), + catalog_name: None, + schema_name: None, + table_name: table_name.to_string(), desc: Some(" a test table".to_string()), schema: Arc::new( Schema::with_timestamp_index(column_schemas, 3) @@ -228,6 +257,7 @@ mod tests { let (opts, _guard) = test_util::create_tmp_dir_and_datanode_opts(); let instance = Instance::new(&opts).await.unwrap(); instance.start().await.unwrap(); + instance.create_test_table().await.unwrap(); let output = instance .execute_sql( @@ -268,4 +298,29 @@ mod tests { _ => unreachable!(), } } + + #[tokio::test] + pub async fn test_execute_create() { + common_telemetry::init_default_ut_logging(); + let (opts, _guard) = test_util::create_tmp_dir_and_datanode_opts(); + let instance = Instance::new(&opts).await.unwrap(); + instance.start().await.unwrap(); + instance.create_test_table().await.unwrap(); + + let output = instance + .execute_sql( + r#"create table test_table( + host string, + ts bigint, + cpu double default 0, + memory double, + TIME INDEX (ts), + PRIMARY KEY(ts, host) + ) engine=mito with(regions=1);"#, + ) + .await + .unwrap(); + + assert!(matches!(output, Output::AffectedRows(1))); + } } diff --git a/src/datanode/src/lib.rs b/src/datanode/src/lib.rs index 7ee33a41e5d..2c3478b1381 100644 --- a/src/datanode/src/lib.rs +++ b/src/datanode/src/lib.rs @@ -1,3 +1,5 @@ +#![feature(assert_matches)] + pub mod datanode; pub mod error; pub mod instance; diff --git a/src/datanode/src/sql.rs b/src/datanode/src/sql.rs index 13a5353b514..f4e2ddeb188 100644 --- a/src/datanode/src/sql.rs +++ b/src/datanode/src/sql.rs @@ -1,34 +1,44 @@ //! sql handler -mod insert; -use catalog::SchemaProviderRef; +use std::sync::Arc; + +use catalog::CatalogManagerRef; use common_error::ext::BoxedError; use query::query_engine::Output; use snafu::{OptionExt, ResultExt}; -use sql::statements::statement::Statement; use table::engine::{EngineContext, TableEngine}; use table::requests::*; use table::TableRef; use crate::error::{GetTableSnafu, Result, TableNotFoundSnafu}; +mod create; +mod insert; + +#[derive(Debug)] pub enum SqlRequest { Insert(InsertRequest), + Create(CreateTableRequest), } // Handler to execute SQL except query pub struct SqlHandler { - table_engine: Engine, + table_engine: Arc, + catalog_manager: CatalogManagerRef, } impl SqlHandler { - pub fn new(table_engine: Engine) -> Self { - Self { table_engine } + pub fn new(table_engine: Engine, catalog_manager: CatalogManagerRef) -> Self { + Self { + table_engine: Arc::new(table_engine), + catalog_manager, + } } pub async fn execute(&self, request: SqlRequest) -> Result { match request { SqlRequest::Insert(req) => self.insert(req).await, + SqlRequest::Create(req) => self.create(req).await, } } @@ -40,16 +50,9 @@ impl SqlHandler { .context(TableNotFoundSnafu { table_name }) } - // Cast sql statement into sql request - pub(crate) fn statement_to_request( - &self, - schema_provider: SchemaProviderRef, - statement: Statement, - ) -> Result { - match statement { - Statement::Insert(stmt) => self.insert_to_request(schema_provider, *stmt), - _ => unimplemented!(), - } + #[cfg(test)] + pub fn table_engine(&self) -> Arc { + self.table_engine.clone() } } @@ -67,6 +70,7 @@ mod tests { use log_store::fs::noop::NoopLogStore; use object_store::{backend::fs::Backend, ObjectStore}; use query::QueryEngineFactory; + use sql::statements::statement::Statement; use storage::config::EngineConfig as StorageEngineConfig; use storage::EngineImpl; use table::error::Result as TableResult; @@ -143,10 +147,6 @@ mod tests { let accessor = Backend::build().root(&store_dir).finish().await.unwrap(); let object_store = ObjectStore::new(accessor); - let catalog_list = catalog::memory::new_memory_catalog_list().unwrap(); - let factory = QueryEngineFactory::new(catalog_list); - let query_engine = factory.query_engine().clone(); - let sql = r#"insert into demo(host, cpu, memory, ts) values ('host1', 66.6, 1024, 1655276557000), ('host2', 88.8, 333.3, 1655276558000) @@ -161,12 +161,25 @@ mod tests { ), object_store, ); - let sql_handler = SqlHandler::new(table_engine); - let stmt = query_engine.sql_to_statement(sql).unwrap(); + let catalog_list = Arc::new( + catalog::LocalCatalogManager::try_new(Arc::new(table_engine.clone())) + .await + .unwrap(), + ); + let factory = QueryEngineFactory::new(catalog_list.clone()); + let query_engine = factory.query_engine().clone(); + let sql_handler = SqlHandler::new(table_engine, catalog_list); + + let stmt = match query_engine.sql_to_statement(sql).unwrap() { + Statement::Insert(i) => i, + _ => { + unreachable!() + } + }; let schema_provider = Arc::new(MockSchemaProvider {}); let request = sql_handler - .statement_to_request(schema_provider, stmt) + .insert_to_request(schema_provider, *stmt) .unwrap(); match request { @@ -195,6 +208,9 @@ mod tests { assert_eq!(Value::from(1655276557000i64), ts.get(0)); assert_eq!(Value::from(1655276558000i64), ts.get(1)); } + _ => { + panic!("Not supposed to reach here") + } } } } diff --git a/src/datanode/src/sql/create.rs b/src/datanode/src/sql/create.rs new file mode 100644 index 00000000000..f0f39db0cd4 --- /dev/null +++ b/src/datanode/src/sql/create.rs @@ -0,0 +1,353 @@ +use std::collections::HashMap; +use std::sync::Arc; + +use catalog::RegisterTableRequest; +use common_telemetry::tracing::info; +use datatypes::prelude::ConcreteDataType; +use datatypes::schema::{ColumnSchema, Schema}; +use query::query_engine::Output; +use snafu::{OptionExt, ResultExt}; +use sql::ast::{ColumnDef, ColumnOption, DataType as SqlDataType, ObjectName, TableConstraint}; +use sql::statements::create_table::CreateTable; +use store_api::storage::consts::TIME_INDEX_NAME; +use table::engine::{EngineContext, TableEngine}; +use table::metadata::TableId; +use table::requests::*; + +use crate::error::{ + ConstraintNotSupportedSnafu, CreateSchemaSnafu, CreateTableSnafu, InsertSystemCatalogSnafu, + InvalidCreateTableSqlSnafu, KeyColumnNotFoundSnafu, Result, SqlTypeNotSupportedSnafu, +}; +use crate::sql::SqlHandler; + +impl SqlHandler { + pub(crate) async fn create(&self, req: CreateTableRequest) -> Result { + let ctx = EngineContext {}; + let catalog_name = req.catalog_name.clone(); + let schema_name = req.schema_name.clone(); + let table_name = req.table_name.clone(); + let table_id = req.id; + + let table = self + .table_engine + .create_table(&ctx, req) + .await + .with_context(|_| CreateTableSnafu { + table_name: &table_name, + })?; + + let register_req = RegisterTableRequest { + catalog: catalog_name, + schema: schema_name, + table_name: table_name.clone(), + table_id, + table, + }; + + self.catalog_manager + .register_table(register_req) + .await + .context(InsertSystemCatalogSnafu)?; + info!("Successfully created table: {:?}", table_name); + // TODO(hl): maybe support create multiple tables + Ok(Output::AffectedRows(1)) + } + + /// Converts [CreateTable] to [SqlRequest::Create]. + pub(crate) fn create_to_request( + &self, + table_id: TableId, + stmt: CreateTable, + ) -> Result { + let mut ts_index = usize::MAX; + let mut primary_keys = vec![]; + + let (catalog_name, schema_name, table_name) = table_idents_to_full_name(stmt.name)?; + + let col_map = stmt + .columns + .iter() + .map(|e| e.name.value.clone()) + .enumerate() + .map(|(k, v)| (v, k)) + .collect::>(); + + for c in stmt.constraints { + match c { + TableConstraint::Unique { + name, + columns, + is_primary, + } => { + if let Some(name) = name { + if name.value == TIME_INDEX_NAME { + ts_index = *col_map.get(&columns[0].value).context( + KeyColumnNotFoundSnafu { + name: columns[0].value.to_string(), + }, + )?; + } else { + return InvalidCreateTableSqlSnafu { + msg: format!("Cannot recognize named UNIQUE constraint: {}", name), + } + .fail(); + } + } else if is_primary { + for col in columns { + primary_keys.push(*col_map.get(&col.value).context( + KeyColumnNotFoundSnafu { + name: col.value.to_string(), + }, + )?); + } + } else { + return InvalidCreateTableSqlSnafu { + msg: format!( + "Unrecognized non-primary unnamed UNIQUE constraint: {:?}", + name + ), + } + .fail(); + } + } + _ => { + return ConstraintNotSupportedSnafu { + constraint: format!("{:?}", c), + } + .fail() + } + } + } + + if primary_keys.is_empty() { + info!( + "Creating table: {:?}.{:?}.{} but primary key not set, use time index column: {}", + catalog_name, schema_name, table_name, ts_index + ); + primary_keys.push(ts_index); + } + + let columns_schemas: Vec<_> = stmt + .columns + .iter() + .map(column_def_to_schema) + .collect::>>()?; + + let schema = Arc::new( + Schema::with_timestamp_index(columns_schemas, ts_index).context(CreateSchemaSnafu)?, + ); + + let request = CreateTableRequest { + id: table_id, + catalog_name, + schema_name, + table_name, + desc: None, + schema, + primary_key_indices: primary_keys, + create_if_not_exists: stmt.if_not_exists, + }; + Ok(request) + } +} + +/// Converts maybe fully-qualified table name (`..` or `
` when catalog and schema are default) +/// to tuples +fn table_idents_to_full_name( + obj_name: ObjectName, +) -> Result<(Option, Option, String)> { + match &obj_name.0[..] { + [table] => Ok((None, None, table.value.clone())), + [catalog, schema, table] => Ok(( + Some(catalog.value.clone()), + Some(schema.value.clone()), + table.value.clone(), + )), + _ => InvalidCreateTableSqlSnafu { + msg: format!( + "table name can only be ..
or
, but found: {}", + obj_name + ), + } + .fail(), + } +} + +fn column_def_to_schema(def: &ColumnDef) -> Result { + let is_nullable = def + .options + .iter() + .any(|o| matches!(o.option, ColumnOption::Null)); + Ok(ColumnSchema { + name: def.name.value.clone(), + data_type: sql_data_type_to_concrete_data_type(&def.data_type)?, + is_nullable, + }) +} + +fn sql_data_type_to_concrete_data_type(t: &SqlDataType) -> Result { + match t { + SqlDataType::BigInt(_) => Ok(ConcreteDataType::int64_datatype()), + SqlDataType::Int(_) => Ok(ConcreteDataType::int32_datatype()), + SqlDataType::SmallInt(_) => Ok(ConcreteDataType::int16_datatype()), + SqlDataType::Char(_) + | SqlDataType::Varchar(_) + | SqlDataType::Text + | SqlDataType::String => Ok(ConcreteDataType::string_datatype()), + SqlDataType::Float(_) => Ok(ConcreteDataType::float32_datatype()), + SqlDataType::Double => Ok(ConcreteDataType::float64_datatype()), + SqlDataType::Boolean => Ok(ConcreteDataType::boolean_datatype()), + // TODO(hl): Date/DateTime/Timestamp not supported + _ => SqlTypeNotSupportedSnafu { t: t.clone() }.fail(), + } +} + +#[cfg(test)] +mod tests { + use std::assert_matches::assert_matches; + + use sql::dialect::GenericDialect; + use sql::parser::ParserContext; + use sql::statements::statement::Statement; + use table_engine::config::EngineConfig; + use table_engine::engine::MitoEngine; + use table_engine::table::test_util::{new_test_object_store, MockEngine, MockMitoEngine}; + + use super::*; + use crate::error::Error; + + async fn create_mock_sql_handler() -> SqlHandler> { + let (_dir, object_store) = new_test_object_store("setup_mock_engine_and_table").await; + let mock_engine = + MockMitoEngine::new(EngineConfig::default(), MockEngine::default(), object_store); + let catalog_manager = Arc::new( + catalog::LocalCatalogManager::try_new(Arc::new(mock_engine.clone())) + .await + .unwrap(), + ); + SqlHandler::new(mock_engine, catalog_manager) + } + + fn sql_to_statement(sql: &str) -> CreateTable { + let mut res = ParserContext::create_with_dialect(sql, &GenericDialect {}).unwrap(); + assert_eq!(1, res.len()); + match res.pop().unwrap() { + Statement::Create(c) => c, + _ => { + panic!("Unexpected statement!") + } + } + } + + #[tokio::test] + pub async fn test_create_to_request() { + let handler = create_mock_sql_handler().await; + let parsed_stmt = sql_to_statement("create table demo_table( host string, ts bigint, cpu double default 0, memory double, TIME INDEX (ts), PRIMARY KEY(ts, host)) engine=mito with(regions=1);"); + let c = handler.create_to_request(42, parsed_stmt).unwrap(); + assert_eq!("demo_table", c.table_name); + assert_eq!(42, c.id); + assert!(c.schema_name.is_none()); + assert!(c.catalog_name.is_none()); + assert!(!c.create_if_not_exists); + assert_eq!(vec![1, 0], c.primary_key_indices); + assert_eq!(1, c.schema.timestamp_index().unwrap()); + assert_eq!(4, c.schema.column_schemas().len()); + } + + /// Time index not specified in sql + #[tokio::test] + pub async fn test_time_index_not_specified() { + let handler = create_mock_sql_handler().await; + let parsed_stmt = sql_to_statement("create table demo_table( host string, ts bigint, cpu double default 0, memory double, PRIMARY KEY(ts, host)) engine=mito with(regions=1);"); + let error = handler.create_to_request(42, parsed_stmt).unwrap_err(); + assert_matches!(error, Error::CreateSchema { .. }); + } + + /// If primary key is not specified, time index should be used as primary key. + #[tokio::test] + pub async fn test_primary_key_not_specified() { + let handler = create_mock_sql_handler().await; + + let parsed_stmt = sql_to_statement("create table demo_table( host string, ts bigint, cpu double default 0, memory double, TIME INDEX (ts)) engine=mito with(regions=1);"); + + let c = handler.create_to_request(42, parsed_stmt).unwrap(); + assert_eq!(1, c.primary_key_indices.len()); + assert_eq!( + c.schema.timestamp_index().unwrap(), + c.primary_key_indices[0] + ); + } + + /// Constraints specified, not column cannot be found. + #[tokio::test] + pub async fn test_key_not_found() { + let handler = create_mock_sql_handler().await; + + let parsed_stmt = sql_to_statement( + "create table demo_table( host string, TIME INDEX (ts)) engine=mito with(regions=1);", + ); + + let error = handler.create_to_request(42, parsed_stmt).unwrap_err(); + assert_matches!(error, Error::KeyColumnNotFound { .. }); + } + + #[tokio::test] + pub async fn test_parse_create_sql() { + let create_table = sql_to_statement( + r"create table c.s.demo( + host string, + ts bigint, + cpu double default 0, + memory double, + TIME INDEX (ts), + PRIMARY KEY(ts, host)) engine=mito + with(regions=1); + ", + ); + + let handler = create_mock_sql_handler().await; + + let request = handler.create_to_request(42, create_table).unwrap(); + + assert_eq!(42, request.id); + assert_eq!(Some("c".to_string()), request.catalog_name); + assert_eq!(Some("s".to_string()), request.schema_name); + assert_eq!("demo".to_string(), request.table_name); + assert!(!request.create_if_not_exists); + assert_eq!(4, request.schema.column_schemas().len()); + + assert_eq!(vec![1, 0], request.primary_key_indices); + assert_eq!( + ConcreteDataType::string_datatype(), + request + .schema + .column_schema_by_name("host") + .unwrap() + .data_type + ); + assert_eq!( + ConcreteDataType::int64_datatype(), + request + .schema + .column_schema_by_name("ts") + .unwrap() + .data_type + ); + assert_eq!( + ConcreteDataType::float64_datatype(), + request + .schema + .column_schema_by_name("cpu") + .unwrap() + .data_type + ); + assert_eq!( + ConcreteDataType::float64_datatype(), + request + .schema + .column_schema_by_name("memory") + .unwrap() + .data_type + ); + } +} diff --git a/src/datatypes/src/vectors.rs b/src/datatypes/src/vectors.rs index 29f7410b2ac..f7b249fd441 100644 --- a/src/datatypes/src/vectors.rs +++ b/src/datatypes/src/vectors.rs @@ -10,6 +10,7 @@ pub mod primitive; mod string; use std::any::Any; +use std::fmt::Debug; use std::sync::Arc; use arrow::array::ArrayRef; @@ -51,7 +52,7 @@ impl<'a> Validity<'a> { } /// Vector of data values. -pub trait Vector: Send + Sync + Serializable { +pub trait Vector: Send + Sync + Serializable + Debug { /// Returns the data type of the vector. /// /// This may require heap allocation. diff --git a/src/log-store/src/fs/file.rs b/src/log-store/src/fs/file.rs index c1fa7642954..ff621346131 100644 --- a/src/log-store/src/fs/file.rs +++ b/src/log-store/src/fs/file.rs @@ -360,7 +360,8 @@ impl LogFile { let entry_stream = stream!({ let mut chunks = ChunkList::new(); while let Some(chunk) = chunk_stream.next().await { - chunks.push(chunk.unwrap()); + let chunk = chunk.unwrap(); + chunks.push(chunk); let mut batch = vec![]; loop { match EntryImpl::decode(&mut chunks) { diff --git a/src/sql/src/ast.rs b/src/sql/src/ast.rs index 335c84040ef..c97ebf657f3 100644 --- a/src/sql/src/ast.rs +++ b/src/sql/src/ast.rs @@ -1 +1,4 @@ -pub use sqlparser::ast::{ColumnDef, Expr, Ident, ObjectName, SqlOption, TableConstraint, Value}; +pub use sqlparser::ast::{ + ColumnDef, ColumnOption, ColumnOptionDef, DataType, Expr, Ident, ObjectName, SqlOption, + TableConstraint, Value, +}; diff --git a/src/sql/src/parsers/create_parser.rs b/src/sql/src/parsers/create_parser.rs index 3fa42a3ceb4..6bd893b96f9 100644 --- a/src/sql/src/parsers/create_parser.rs +++ b/src/sql/src/parsers/create_parser.rs @@ -14,7 +14,7 @@ use crate::statements::statement::Statement; const ENGINE: &str = "ENGINE"; -/// Pasre create [table] statement +/// Parses create [table] statement impl<'a> ParserContext<'a> { pub(crate) fn parse_create(&mut self) -> Result { self.parser @@ -42,6 +42,7 @@ impl<'a> ParserContext<'a> { engine, constraints, options, + table_id: 0, // table id is assigned by catalog manager })) } diff --git a/src/sql/src/statements/create_table.rs b/src/sql/src/statements/create_table.rs index 2d3dd610023..4dda3e1641c 100644 --- a/src/sql/src/statements/create_table.rs +++ b/src/sql/src/statements/create_table.rs @@ -7,6 +7,7 @@ pub const TIME_INDEX: &str = "__time_index"; pub struct CreateTable { /// Create if not exists pub if_not_exists: bool, + pub table_id: u32, /// Table name pub name: ObjectName, pub columns: Vec, diff --git a/src/store-api/src/storage/consts.rs b/src/store-api/src/storage/consts.rs index 54b2e862369..536e0608cd8 100644 --- a/src/store-api/src/storage/consts.rs +++ b/src/store-api/src/storage/consts.rs @@ -26,13 +26,17 @@ pub const VERSION_COLUMN_ID: ColumnId = 1; /// Name of version column. pub const VERSION_COLUMN_NAME: &str = "__version"; -// Names for default column family. + +/// Names for default column family. pub const DEFAULT_CF_NAME: &str = "default"; -// Name for reserved column: sequence +/// Name for reserved column: sequence pub const SEQUENCE_COLUMN_NAME: &str = "__sequence"; -// Name for reserved column: value_type +/// Name for time index constraint name. +pub const TIME_INDEX_NAME: &str = "__time_index"; + +/// Name for reserved column: value_type pub const VALUE_TYPE_COLUMN_NAME: &str = "__value_type"; // ----------------------------------------------------------------------------- diff --git a/src/table-engine/Cargo.toml b/src/table-engine/Cargo.toml index 5d73710a215..52c5bbf87b1 100644 --- a/src/table-engine/Cargo.toml +++ b/src/table-engine/Cargo.toml @@ -3,16 +3,20 @@ name = "table-engine" version = "0.1.0" edition = "2021" +[features] +default = [] +test = ["tempdir"] + [dependencies] arc-swap = "1.0" async-stream = "0.3" async-trait = "0.1" chrono = { version = "0.4", features = ["serde"] } -common-error = {path = "../common/error" } -common-query = {path = "../common/query" } -common-recordbatch = {path = "../common/recordbatch" } -common-telemetry = {path = "../common/telemetry" } -datafusion-common = { git = "https://github.com/apache/arrow-datafusion.git" , branch = "arrow2"} +common-error = { path = "../common/error" } +common-query = { path = "../common/query" } +common-recordbatch = { path = "../common/recordbatch" } +common-telemetry = { path = "../common/telemetry" } +datafusion-common = { git = "https://github.com/apache/arrow-datafusion.git", branch = "arrow2" } datatypes = { path = "../datatypes" } futures = "0.3" log-store = { path = "../log-store" } @@ -20,12 +24,12 @@ object-store = { path = "../object-store" } serde = { version = "1.0", features = ["derive"] } serde_json = "1.0" snafu = { version = "0.7", features = ["backtraces"] } -storage ={ path = "../storage" } -store-api ={ path = "../store-api" } +storage = { path = "../storage" } +store-api = { path = "../store-api" } table = { path = "../table" } +tempdir = { version = "0.3", optional = true } tokio = { version = "1.0", features = ["full"] } [dev-dependencies] datatypes = { path = "../datatypes" } -tempdir = "0.3" tokio = { version = "1.18", features = ["full"] } diff --git a/src/table-engine/src/engine.rs b/src/table-engine/src/engine.rs index de7e693a5cb..bf6f84b6510 100644 --- a/src/table-engine/src/engine.rs +++ b/src/table-engine/src/engine.rs @@ -134,7 +134,7 @@ fn build_row_key_desc_from_schema( .schema .timestamp_column() .context(MissingTimestampIndexSnafu { - table_name: &request.name, + table_name: &request.table_name, })?; let timestamp_index = request.schema.timestamp_index().unwrap(); @@ -147,7 +147,7 @@ fn build_row_key_desc_from_schema( .build() .context(BuildColumnDescriptorSnafu { column_name: &ts_column_schema.name, - table_name: &request.name, + table_name: &request.table_name, })?; column_id += 1; @@ -172,7 +172,7 @@ fn build_row_key_desc_from_schema( .build() .context(BuildColumnDescriptorSnafu { column_name: &column_schema.name, - table_name: &request.name, + table_name: &request.table_name, })?; builder = builder.push_column(column); @@ -182,7 +182,7 @@ fn build_row_key_desc_from_schema( Ok(( column_id, builder.build().context(BuildRowKeyDescriptorSnafu { - table_name: &request.name, + table_name: &request.table_name, })?, )) } @@ -198,7 +198,7 @@ fn build_column_family_from_request( .schema .timestamp_index() .context(MissingTimestampIndexSnafu { - table_name: &request.name, + table_name: &request.table_name, })?; let column_schemas = request .schema @@ -217,7 +217,7 @@ fn build_column_family_from_request( .build() .context(BuildColumnDescriptorSnafu { column_name: &column_schema.name, - table_name: &request.name, + table_name: &request.table_name, })?; builder = builder.push_column(column); @@ -227,7 +227,7 @@ fn build_column_family_from_request( Ok(( column_id, builder.build().context(BuildColumnFamilyDescriptorSnafu { - table_name: &request.name, + table_name: &request.table_name, })?, )) } @@ -238,7 +238,7 @@ impl MitoEngineInner { _ctx: &EngineContext, request: CreateTableRequest, ) -> Result { - let table_name = &request.name; + let table_name = &request.table_name; if let Some(table) = self.get_table(table_name) { if request.create_if_not_exists { @@ -486,7 +486,9 @@ mod tests { let request = CreateTableRequest { id: 1, - name: table_info.name.to_string(), + catalog_name: None, + schema_name: None, + table_name: table_info.name.to_string(), schema: table_info.meta.schema.clone(), create_if_not_exists: true, desc: None, @@ -506,7 +508,9 @@ mod tests { // test create_if_not_exists=false let request = CreateTableRequest { id: 1, - name: table_info.name.to_string(), + catalog_name: None, + schema_name: None, + table_name: table_info.name.to_string(), schema: table_info.meta.schema.clone(), create_if_not_exists: false, desc: None, diff --git a/src/table-engine/src/table.rs b/src/table-engine/src/table.rs index e1cfbc4f2b2..3a803c87b7e 100644 --- a/src/table-engine/src/table.rs +++ b/src/table-engine/src/table.rs @@ -1,4 +1,4 @@ -#[cfg(test)] +#[cfg(any(test, feature = "test"))] pub mod test_util; use std::any::Any; diff --git a/src/table-engine/src/table/test_util.rs b/src/table-engine/src/table/test_util.rs index 180ebefc397..293db4b89f0 100644 --- a/src/table-engine/src/table/test_util.rs +++ b/src/table-engine/src/table/test_util.rs @@ -86,7 +86,9 @@ pub async fn setup_test_engine_and_table() -> ( &EngineContext::default(), CreateTableRequest { id: 1, - name: TABLE_NAME.to_string(), + catalog_name: None, + schema_name: None, + table_name: TABLE_NAME.to_string(), desc: Some("a test table".to_string()), schema: schema.clone(), create_if_not_exists: true, @@ -115,7 +117,9 @@ pub async fn setup_mock_engine_and_table( &EngineContext::default(), CreateTableRequest { id: 1, - name: TABLE_NAME.to_string(), + catalog_name: None, + schema_name: None, + table_name: TABLE_NAME.to_string(), desc: None, schema: schema.clone(), create_if_not_exists: true, diff --git a/src/table-engine/src/table/test_util/mock_engine.rs b/src/table-engine/src/table/test_util/mock_engine.rs index aacf2109d72..be8fd4348b4 100644 --- a/src/table-engine/src/table/test_util/mock_engine.rs +++ b/src/table-engine/src/table/test_util/mock_engine.rs @@ -5,7 +5,6 @@ use std::sync::{Arc, Mutex}; use async_trait::async_trait; use common_error::mock::MockError; -use common_error::prelude::StatusCode; use common_telemetry::logging; use storage::metadata::{RegionMetaImpl, RegionMetadataRef}; use storage::write_batch::WriteBatch; @@ -142,7 +141,7 @@ impl StorageEngine for MockEngine { return Ok(Some(region)); } - Err(MockError::with_backtrace(StatusCode::Unexpected)) + return Ok(None); } async fn close_region(&self, _ctx: &EngineContext, _region: MockRegion) -> Result<()> { diff --git a/src/table/src/requests.rs b/src/table/src/requests.rs index 2c60021a54c..4e640ae3ba6 100644 --- a/src/table/src/requests.rs +++ b/src/table/src/requests.rs @@ -7,6 +7,7 @@ use datatypes::schema::SchemaRef; use crate::metadata::TableId; /// Insert request +#[derive(Debug)] pub struct InsertRequest { pub table_name: String, pub columns_values: HashMap, @@ -16,12 +17,13 @@ pub struct InsertRequest { #[derive(Debug)] pub struct CreateTableRequest { pub id: TableId, - pub name: String, + pub catalog_name: Option, + pub schema_name: Option, + pub table_name: String, pub desc: Option, pub schema: SchemaRef, pub primary_key_indices: Vec, pub create_if_not_exists: bool, - // TODO(yingwen): 1. Add catalog_name/schema_name and other infos } /// Open table request @@ -30,7 +32,6 @@ pub struct OpenTableRequest { pub catalog_name: String, pub schema_name: String, pub table_name: String, - // TODO(yingwen): TableId could be recovered from the table metadata. pub table_id: TableId, }