rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-10-27 14:16:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: impl create table sql execution (#168)

Browse Source 
* catalog manager allocates table id

* rebase develop

* add some tests

* add some more test

* fix some cr comments

* insert into system catalog

* use slice pattern to simplify code

* add optional dependencies

* add sql-to-request test

* successfully recover

* fix unit tests

* rebase develop

* add some tests

* fix some cr comments

* fix some cr comments

* add a lock to CatalogManager

* feat: add gmt_created and gmt_modified columns to system catalog table
This commit is contained in:
Lei, Huang
2022-08-17 10:53:19 +08:00
committed by GitHub
parent 34133fae5a
commit a1c4921933
28 changed files with 902 additions and 121 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+4
View File
@@ -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"
src/catalog/Cargo.toml
+1
View File
@@ -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"
src/catalog/src/error.rs
+11 -1
View File
@@ -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<T> = std::result::Result<T, Error>;
@@ -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,
}
src/catalog/src/lib.rs
+25
View File
@@ -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<dyn CatalogProvider>;
#[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<usize>;
}
pub type CatalogManagerRef = Arc<dyn CatalogManager>;
pub struct RegisterTableRequest {
pub catalog: Option<String>,
pub schema: Option<String>,
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)
}
src/catalog/src/manager.rs
+89 -15
View File
@@ -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<SystemCatalog>,
catalogs: Arc<MemoryCatalogList>,
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<TableId> {
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<usize> {
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)
}
}
src/catalog/src/system.rs
+97 -22
View File
@@ -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<SendableRecordBatchStream> {
panic!("System catalog table does not support scan!")
}
/// Insert values into table.
async fn insert(&self, request: InsertRequest) -> table::error::Result<usize> {
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<Schema> {
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<Entry> {
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 {
src/catalog/src/tables.rs
+105 -7
View File
@@ -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<SendableRecordBatchStream> {
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<usize> {
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::<Utf8Array<i64>>()
.unwrap()
.value(0)
);
assert_eq!(
"test_schema",
batch
.column(1)
.as_any()
.downcast_ref::<Utf8Array<i64>>()
.unwrap()
.value(0)
);
assert_eq!(
"test_table",
batch
.column(2)
.as_any()
.downcast_ref::<Utf8Array<i64>>()
.unwrap()
.value(0)
);
assert_eq!(
"test_engine",
batch
.column(3)
.as_any()
.downcast_ref::<Utf8Array<i64>>()
.unwrap()
.value(0)
);
} else {
panic!("Record batch should not be empty!")
}
}
}
src/common/time/Cargo.toml
+1
View File
@@ -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"
src/common/time/src/lib.rs
+1
View File
@@ -1,5 +1,6 @@
pub mod range;
pub mod timestamp;
pub mod util;
pub use range::RangeMillis;
pub use timestamp::TimestampMillis;
src/common/time/src/util.rs
+4
View File
@@ -0,0 +1,4 @@
/// Calculates the time duration since UNIX_EPOCH in milliseconds.
pub fn current_timestamp() -> i64 {
chrono::Utc::now().timestamp()
}
src/datanode/Cargo.toml
+1 -1
View File
@@ -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"] }
src/datanode/src/error.rs
+40 -1
View File
@@ -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<T> = std::result::Result<T, Error>;
@@ -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,
src/datanode/src/instance.rs
+73 -18
View File
@@ -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<EngineImpl<LocalFileLogStore>>;
@@ -31,7 +27,6 @@ type DefaultEngine = MitoEngine<EngineImpl<LocalFileLogStore>>;
pub struct Instance {
// Query service
query_engine: QueryEngineRef,
table_engine: DefaultEngine,
sql_handler: SqlHandler<DefaultEngine>,
// 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)));
}
}
src/datanode/src/lib.rs
+2
View File
@@ -1,3 +1,5 @@
#![feature(assert_matches)]
pub mod datanode;
pub mod error;
pub mod instance;
src/datanode/src/sql.rs
+39 -23
View File
@@ -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<Engine: TableEngine> {
table_engine: Engine,
table_engine: Arc<Engine>,
catalog_manager: CatalogManagerRef,
}
impl<Engine: TableEngine> SqlHandler<Engine> {
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<Output> {
match request {
SqlRequest::Insert(req) => self.insert(req).await,
SqlRequest::Create(req) => self.create(req).await,
}
}
@@ -40,16 +50,9 @@ impl<Engine: TableEngine> SqlHandler<Engine> {
.context(TableNotFoundSnafu { table_name })
}
// Cast sql statement into sql request
pub(crate) fn statement_to_request(
&self,
schema_provider: SchemaProviderRef,
statement: Statement,
) -> Result<SqlRequest> {
match statement {
Statement::Insert(stmt) => self.insert_to_request(schema_provider, *stmt),
_ => unimplemented!(),
}
#[cfg(test)]
pub fn table_engine(&self) -> Arc<Engine> {
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")
}
}
}
}
src/datanode/src/sql/create.rs
+353
View File
@@ -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<Engine: TableEngine> SqlHandler<Engine> {
pub(crate) async fn create(&self, req: CreateTableRequest) -> Result<Output> {
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<CreateTableRequest> {
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::<HashMap<_, _>>();
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::<Result<Vec<_>>>()?;
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 (`<catalog>.<schema>.<table>` or `<table>` when catalog and schema are default)
/// to tuples
fn table_idents_to_full_name(
obj_name: ObjectName,
) -> Result<(Option<String>, Option<String>, 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 <catalog>.<schema>.<table> or <table>, but found: {}",
obj_name
),
}
.fail(),
}
}
fn column_def_to_schema(def: &ColumnDef) -> Result<ColumnSchema> {
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<ConcreteDataType> {
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<MitoEngine<MockEngine>> {
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
);
}
}
src/datatypes/src/vectors.rs
+2 -1
View File
@@ -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.
src/log-store/src/fs/file.rs
+2 -1
View File
@@ -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) {
src/sql/src/ast.rs
+4 -1
View File
@@ -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,
};
src/sql/src/parsers/create_parser.rs
+2 -1
View File
@@ -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<Statement> {
self.parser
@@ -42,6 +42,7 @@ impl<'a> ParserContext<'a> {
engine,
constraints,
options,
table_id: 0, // table id is assigned by catalog manager
}))
}
src/sql/src/statements/create_table.rs
+1
View File
@@ -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<ColumnDef>,
src/store-api/src/storage/consts.rs
+7 -3
View File
@@ -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";
// -----------------------------------------------------------------------------
src/table-engine/Cargo.toml
+12 -8
View File
@@ -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"] }
src/table-engine/src/engine.rs
+14 -10
View File
@@ -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<S: StorageEngine> MitoEngineInner<S> {
_ctx: &EngineContext,
request: CreateTableRequest,
) -> Result<TableRef> {
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,
src/table-engine/src/table.rs
+1 -1
View File
@@ -1,4 +1,4 @@
#[cfg(test)]
#[cfg(any(test, feature = "test"))]
pub mod test_util;
use std::any::Any;
src/table-engine/src/table/test_util.rs
+6 -2
View File
@@ -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,
src/table-engine/src/table/test_util/mock_engine.rs
+1 -2
View File
@@ -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<()> {
src/table/src/requests.rs
+4 -3
View File
@@ -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<String, VectorRef>,
@@ -16,12 +17,13 @@ pub struct InsertRequest {
#[derive(Debug)]
pub struct CreateTableRequest {
pub id: TableId,
pub name: String,
pub catalog_name: Option<String>,
pub schema_name: Option<String>,
pub table_name: String,
pub desc: Option<String>,
pub schema: SchemaRef,
pub primary_key_indices: Vec<usize>,
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,
}