rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-10-26 21:56:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: remove optional from the protos (#756)

Browse Source
This commit is contained in:
Jiachun Feng
2022-12-16 15:47:51 +08:00
committed by GitHub
parent 7c16a4a17b
commit 66bca11401
16 changed files with 235 additions and 189 deletions
Download Patch File Download Diff File Expand all files Collapse all files
benchmarks/src/bin/nyc-taxi.rs
+26 -26
View File
@@ -28,7 +28,7 @@ use arrow::record_batch::RecordBatch;
use clap::Parser;
use client::admin::Admin;
use client::api::v1::column::Values;
use client::api::v1::{Column, ColumnDataType, ColumnDef, CreateExpr, InsertExpr};
use client::api::v1::{Column, ColumnDataType, ColumnDef, CreateTableExpr, InsertExpr, TableId};
use client::{Client, Database, Select};
use indicatif::{MultiProgress, ProgressBar, ProgressStyle};
use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
@@ -219,126 +219,126 @@ fn build_values(column: &ArrayRef) -> Values {
}
}
fn create_table_expr() -> CreateExpr {
CreateExpr {
catalog_name: Some(CATALOG_NAME.to_string()),
schema_name: Some(SCHEMA_NAME.to_string()),
fn create_table_expr() -> CreateTableExpr {
CreateTableExpr {
catalog_name: CATALOG_NAME.to_string(),
schema_name: SCHEMA_NAME.to_string(),
table_name: TABLE_NAME.to_string(),
desc: None,
desc: "".to_string(),
column_defs: vec![
ColumnDef {
name: "VendorID".to_string(),
datatype: ColumnDataType::Int64 as i32,
is_nullable: true,
default_constraint: None,
default_constraint: vec![],
},
ColumnDef {
name: "tpep_pickup_datetime".to_string(),
datatype: ColumnDataType::Int64 as i32,
is_nullable: true,
default_constraint: None,
default_constraint: vec![],
},
ColumnDef {
name: "tpep_dropoff_datetime".to_string(),
datatype: ColumnDataType::Int64 as i32,
is_nullable: true,
default_constraint: None,
default_constraint: vec![],
},
ColumnDef {
name: "passenger_count".to_string(),
datatype: ColumnDataType::Float64 as i32,
is_nullable: true,
default_constraint: None,
default_constraint: vec![],
},
ColumnDef {
name: "trip_distance".to_string(),
datatype: ColumnDataType::Float64 as i32,
is_nullable: true,
default_constraint: None,
default_constraint: vec![],
},
ColumnDef {
name: "RatecodeID".to_string(),
datatype: ColumnDataType::Float64 as i32,
is_nullable: true,
default_constraint: None,
default_constraint: vec![],
},
ColumnDef {
name: "store_and_fwd_flag".to_string(),
datatype: ColumnDataType::String as i32,
is_nullable: true,
default_constraint: None,
default_constraint: vec![],
},
ColumnDef {
name: "PULocationID".to_string(),
datatype: ColumnDataType::Int64 as i32,
is_nullable: true,
default_constraint: None,
default_constraint: vec![],
},
ColumnDef {
name: "DOLocationID".to_string(),
datatype: ColumnDataType::Int64 as i32,
is_nullable: true,
default_constraint: None,
default_constraint: vec![],
},
ColumnDef {
name: "payment_type".to_string(),
datatype: ColumnDataType::Int64 as i32,
is_nullable: true,
default_constraint: None,
default_constraint: vec![],
},
ColumnDef {
name: "fare_amount".to_string(),
datatype: ColumnDataType::Float64 as i32,
is_nullable: true,
default_constraint: None,
default_constraint: vec![],
},
ColumnDef {
name: "extra".to_string(),
datatype: ColumnDataType::Float64 as i32,
is_nullable: true,
default_constraint: None,
default_constraint: vec![],
},
ColumnDef {
name: "mta_tax".to_string(),
datatype: ColumnDataType::Float64 as i32,
is_nullable: true,
default_constraint: None,
default_constraint: vec![],
},
ColumnDef {
name: "tip_amount".to_string(),
datatype: ColumnDataType::Float64 as i32,
is_nullable: true,
default_constraint: None,
default_constraint: vec![],
},
ColumnDef {
name: "tolls_amount".to_string(),
datatype: ColumnDataType::Float64 as i32,
is_nullable: true,
default_constraint: None,
default_constraint: vec![],
},
ColumnDef {
name: "improvement_surcharge".to_string(),
datatype: ColumnDataType::Float64 as i32,
is_nullable: true,
default_constraint: None,
default_constraint: vec![],
},
ColumnDef {
name: "total_amount".to_string(),
datatype: ColumnDataType::Float64 as i32,
is_nullable: true,
default_constraint: None,
default_constraint: vec![],
},
ColumnDef {
name: "congestion_surcharge".to_string(),
datatype: ColumnDataType::Float64 as i32,
is_nullable: true,
default_constraint: None,
default_constraint: vec![],
},
ColumnDef {
name: "airport_fee".to_string(),
datatype: ColumnDataType::Float64 as i32,
is_nullable: true,
default_constraint: None,
default_constraint: vec![],
},
],
time_index: "tpep_pickup_datetime".to_string(),
@@ -346,7 +346,7 @@ fn create_table_expr() -> CreateExpr {
create_if_not_exists: false,
table_options: Default::default(),
region_ids: vec![0],
table_id: Some(0),
table_id: Some(TableId { id: 0 }),
}
}
src/api/greptime/v1/admin.proto
+15 -12
View File
@@ -17,7 +17,7 @@ message AdminResponse {
message AdminExpr {
ExprHeader header = 1;
oneof expr {
CreateExpr create = 2;
CreateTableExpr create_table = 2;
AlterExpr alter = 3;
CreateDatabaseExpr create_database = 4;
DropTableExpr drop_table = 5;
@@ -31,24 +31,23 @@ message AdminResult {
}
}
// TODO(hl): rename to CreateTableExpr
message CreateExpr {
optional string catalog_name = 1;
optional string schema_name = 2;
message CreateTableExpr {
string catalog_name = 1;
string schema_name = 2;
string table_name = 3;
optional string desc = 4;
string desc = 4;
repeated ColumnDef column_defs = 5;
string time_index = 6;
repeated string primary_keys = 7;
bool create_if_not_exists = 8;
map<string, string> table_options = 9;
optional uint32 table_id = 10;
TableId table_id = 10;
repeated uint32 region_ids = 11;
}
message AlterExpr {
optional string catalog_name = 1;
optional string schema_name = 2;
string catalog_name = 1;
string schema_name = 2;
string table_name = 3;
oneof kind {
AddColumns add_columns = 4;
@@ -62,6 +61,11 @@ message DropTableExpr {
string table_name = 3;
}
message CreateDatabaseExpr {
//TODO(hl): maybe rename to schema_name?
string database_name = 1;
}
message AddColumns {
repeated AddColumn add_columns = 1;
}
@@ -79,7 +83,6 @@ message DropColumn {
string name = 1;
}
message CreateDatabaseExpr {
//TODO(hl): maybe rename to schema_name?
string database_name = 1;
message TableId {
uint32 id = 1;
}
src/api/greptime/v1/column.proto
+1 -1
View File
@@ -59,7 +59,7 @@ message ColumnDef {
string name = 1;
ColumnDataType datatype = 2;
bool is_nullable = 3;
optional bytes default_constraint = 4;
bytes default_constraint = 4;
}
enum ColumnDataType {
src/api/src/v1/column_def.rs
+6 -5
View File
@@ -23,12 +23,13 @@ impl ColumnDef {
pub fn try_as_column_schema(&self) -> Result<ColumnSchema> {
let data_type = ColumnDataTypeWrapper::try_new(self.datatype)?;
let constraint = match &self.default_constraint {
None => None,
Some(v) => Some(
ColumnDefaultConstraint::try_from(&v[..])
let constraint = if self.default_constraint.is_empty() {
None
} else {
Some(
ColumnDefaultConstraint::try_from(self.default_constraint.as_slice())
.context(error::ConvertColumnDefaultConstraintSnafu { column: &self.name })?,
),
)
};
ColumnSchema::new(&self.name, data_type.into(), self.is_nullable)
src/client/examples/logical.rs
+9 -9
View File
@@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use api::v1::{ColumnDataType, ColumnDef, CreateExpr};
use api::v1::{ColumnDataType, ColumnDef, CreateTableExpr, TableId};
use client::admin::Admin;
use client::{Client, Database};
use prost_09::Message;
@@ -33,36 +33,36 @@ fn main() {
async fn run() {
let client = Client::with_urls(vec!["127.0.0.1:3001"]);
let create_table_expr = CreateExpr {
catalog_name: Some("greptime".to_string()),
schema_name: Some("public".to_string()),
let create_table_expr = CreateTableExpr {
catalog_name: "greptime".to_string(),
schema_name: "public".to_string(),
table_name: "test_logical_dist_exec".to_string(),
desc: None,
desc: "".to_string(),
column_defs: vec![
ColumnDef {
name: "timestamp".to_string(),
datatype: ColumnDataType::TimestampMillisecond as i32,
is_nullable: false,
default_constraint: None,
default_constraint: vec![],
},
ColumnDef {
name: "key".to_string(),
datatype: ColumnDataType::Uint64 as i32,
is_nullable: false,
default_constraint: None,
default_constraint: vec![],
},
ColumnDef {
name: "value".to_string(),
datatype: ColumnDataType::Uint64 as i32,
is_nullable: false,
default_constraint: None,
default_constraint: vec![],
},
],
time_index: "timestamp".to_string(),
primary_keys: vec!["key".to_string()],
create_if_not_exists: false,
table_options: Default::default(),
table_id: Some(1024),
table_id: Some(TableId { id: 1024 }),
region_ids: vec![0],
};
src/client/src/admin.rs
+2 -2
View File
@@ -34,13 +34,13 @@ impl Admin {
}
}
pub async fn create(&self, expr: CreateExpr) -> Result<AdminResult> {
pub async fn create(&self, expr: CreateTableExpr) -> Result<AdminResult> {
let header = ExprHeader {
version: PROTOCOL_VERSION,
};
let expr = AdminExpr {
header: Some(header),
expr: Some(admin_expr::Expr::Create(expr)),
expr: Some(admin_expr::Expr::CreateTable(expr)),
};
self.do_request(expr).await
}
src/common/grpc-expr/src/alter.rs
+39 -19
View File
@@ -15,7 +15,7 @@
use std::sync::Arc;
use api::v1::alter_expr::Kind;
use api::v1::{AlterExpr, CreateExpr, DropColumns};
use api::v1::{AlterExpr, CreateTableExpr, DropColumns};
use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME};
use datatypes::schema::{ColumnSchema, SchemaBuilder, SchemaRef};
use snafu::{ensure, OptionExt, ResultExt};
@@ -29,6 +29,16 @@ use crate::error::{
/// Convert an [`AlterExpr`] to an optional [`AlterTableRequest`]
pub fn alter_expr_to_request(expr: AlterExpr) -> Result<Option<AlterTableRequest>> {
let catalog_name = if expr.catalog_name.is_empty() {
None
} else {
Some(expr.catalog_name)
};
let schema_name = if expr.schema_name.is_empty() {
None
} else {
Some(expr.schema_name)
};
match expr.kind {
Some(Kind::AddColumns(add_columns)) => {
let add_column_requests = add_columns
@@ -57,8 +67,8 @@ pub fn alter_expr_to_request(expr: AlterExpr) -> Result<Option<AlterTableRequest
};
let request = AlterTableRequest {
catalog_name: expr.catalog_name,
schema_name: expr.schema_name,
catalog_name,
schema_name,
table_name: expr.table_name,
alter_kind,
};
@@ -70,8 +80,8 @@ pub fn alter_expr_to_request(expr: AlterExpr) -> Result<Option<AlterTableRequest
};
let request = AlterTableRequest {
catalog_name: expr.catalog_name,
schema_name: expr.schema_name,
catalog_name,
schema_name,
table_name: expr.table_name,
alter_kind,
};
@@ -81,7 +91,7 @@ pub fn alter_expr_to_request(expr: AlterExpr) -> Result<Option<AlterTableRequest
}
}
pub fn create_table_schema(expr: &CreateExpr) -> Result<SchemaRef> {
pub fn create_table_schema(expr: &CreateTableExpr) -> Result<SchemaRef> {
let column_schemas = expr
.column_defs
.iter()
@@ -119,7 +129,10 @@ pub fn create_table_schema(expr: &CreateExpr) -> Result<SchemaRef> {
))
}
pub fn create_expr_to_request(table_id: TableId, expr: CreateExpr) -> Result<CreateTableRequest> {
pub fn create_expr_to_request(
table_id: TableId,
expr: CreateTableExpr,
) -> Result<CreateTableRequest> {
let schema = create_table_schema(&expr)?;
let primary_key_indices = expr
.primary_keys
@@ -134,12 +147,19 @@ pub fn create_expr_to_request(table_id: TableId, expr: CreateExpr) -> Result<Cre
})
.collect::<Result<Vec<usize>>>()?;
let catalog_name = expr
.catalog_name
.unwrap_or_else(|| DEFAULT_CATALOG_NAME.to_string());
let schema_name = expr
.schema_name
.unwrap_or_else(|| DEFAULT_SCHEMA_NAME.to_string());
let mut catalog_name = expr.catalog_name;
if catalog_name.is_empty() {
catalog_name = DEFAULT_CATALOG_NAME.to_string();
}
let mut schema_name = expr.schema_name;
if schema_name.is_empty() {
schema_name = DEFAULT_SCHEMA_NAME.to_string();
}
let desc = if expr.desc.is_empty() {
None
} else {
Some(expr.desc)
};
let region_ids = if expr.region_ids.is_empty() {
vec![0]
@@ -152,7 +172,7 @@ pub fn create_expr_to_request(table_id: TableId, expr: CreateExpr) -> Result<Cre
catalog_name,
schema_name,
table_name: expr.table_name,
desc: expr.desc,
desc,
schema,
region_numbers: region_ids,
primary_key_indices,
@@ -171,8 +191,8 @@ mod tests {
#[test]
fn test_alter_expr_to_request() {
let expr = AlterExpr {
catalog_name: None,
schema_name: None,
catalog_name: "".to_string(),
schema_name: "".to_string(),
table_name: "monitor".to_string(),
kind: Some(Kind::AddColumns(AddColumns {
@@ -181,7 +201,7 @@ mod tests {
name: "mem_usage".to_string(),
datatype: ColumnDataType::Float64 as i32,
is_nullable: false,
default_constraint: None,
default_constraint: vec![],
}),
is_key: false,
}],
@@ -208,8 +228,8 @@ mod tests {
#[test]
fn test_drop_column_expr() {
let expr = AlterExpr {
catalog_name: Some("test_catalog".to_string()),
schema_name: Some("test_schema".to_string()),
catalog_name: "test_catalog".to_string(),
schema_name: "test_schema".to_string(),
table_name: "monitor".to_string(),
kind: Some(Kind::DropColumns(DropColumns {
src/common/grpc-expr/src/insert.rs
+10 -10
View File
@@ -18,7 +18,7 @@ use std::sync::Arc;
use api::helper::ColumnDataTypeWrapper;
use api::v1::column::{SemanticType, Values};
use api::v1::{AddColumn, AddColumns, Column, ColumnDataType, ColumnDef, CreateExpr};
use api::v1::{AddColumn, AddColumns, Column, ColumnDataType, ColumnDef, CreateTableExpr};
use common_base::BitVec;
use common_time::timestamp::Timestamp;
use common_time::{Date, DateTime};
@@ -45,7 +45,7 @@ fn build_column_def(column_name: &str, datatype: i32, nullable: bool) -> ColumnD
name: column_name.to_string(),
datatype,
is_nullable: nullable,
default_constraint: None,
default_constraint: vec![],
}
}
@@ -214,7 +214,7 @@ pub fn build_create_expr_from_insertion(
table_id: Option<TableId>,
table_name: &str,
columns: &[Column],
) -> Result<CreateExpr> {
) -> Result<CreateTableExpr> {
let mut new_columns: HashSet<String> = HashSet::default();
let mut column_defs = Vec::default();
let mut primary_key_indices = Vec::default();
@@ -263,17 +263,17 @@ pub fn build_create_expr_from_insertion(
.map(|idx| columns[*idx].column_name.clone())
.collect::<Vec<_>>();
let expr = CreateExpr {
catalog_name: Some(catalog_name.to_string()),
schema_name: Some(schema_name.to_string()),
let expr = CreateTableExpr {
catalog_name: catalog_name.to_string(),
schema_name: schema_name.to_string(),
table_name: table_name.to_string(),
desc: Some("Created on insertion".to_string()),
desc: "Created on insertion".to_string(),
column_defs,
time_index: timestamp_field_name,
primary_keys,
create_if_not_exists: true,
table_options: Default::default(),
table_id,
table_id: table_id.map(|id| api::v1::TableId { id }),
region_ids: vec![0], // TODO:(hl): region id should be allocated by frontend
};
@@ -516,9 +516,9 @@ mod tests {
build_create_expr_from_insertion("", "", table_id, table_name, &insert_batch.0)
.unwrap();
assert_eq!(table_id, create_expr.table_id);
assert_eq!(table_id, create_expr.table_id.map(|x| x.id));
assert_eq!(table_name, create_expr.table_name);
assert_eq!(Some("Created on insertion".to_string()), create_expr.desc);
assert_eq!("Created on insertion".to_string(), create_expr.desc);
assert_eq!(
vec![create_expr.column_defs[0].name.clone()],
create_expr.primary_keys
src/datanode/src/instance/grpc.rs
+3 -1
View File
@@ -188,7 +188,9 @@ impl GrpcQueryHandler for Instance {
impl GrpcAdminHandler for Instance {
async fn exec_admin_request(&self, expr: AdminExpr) -> servers::error::Result<AdminResult> {
let admin_resp = match expr.expr {
Some(admin_expr::Expr::Create(create_expr)) => self.handle_create(create_expr).await,
Some(admin_expr::Expr::CreateTable(create_expr)) => {
self.handle_create(create_expr).await
}
Some(admin_expr::Expr::Alter(alter_expr)) => self.handle_alter(alter_expr).await,
Some(admin_expr::Expr::CreateDatabase(create_database_expr)) => {
self.execute_create_database(create_database_expr).await
src/datanode/src/server/grpc.rs
+22 -20
View File
@@ -15,7 +15,7 @@
use std::sync::Arc;
use api::result::AdminResultBuilder;
use api::v1::{AdminResult, AlterExpr, CreateExpr, DropTableExpr};
use api::v1::{AdminResult, AlterExpr, CreateTableExpr, DropTableExpr};
use common_error::prelude::{ErrorExt, StatusCode};
use common_grpc_expr::{alter_expr_to_request, create_expr_to_request};
use common_query::Output;
@@ -31,15 +31,15 @@ use crate::sql::SqlRequest;
impl Instance {
/// Handle gRPC create table requests.
pub(crate) async fn handle_create(&self, expr: CreateExpr) -> AdminResult {
pub(crate) async fn handle_create(&self, expr: CreateTableExpr) -> AdminResult {
// Respect CreateExpr's table id and region ids if present, or allocate table id
// from local table id provider and set region id to 0.
let table_id = if let Some(table_id) = expr.table_id {
let table_id = if let Some(table_id) = &expr.table_id {
info!(
"Creating table {:?}.{:?}.{:?} with table id from frontend: {}",
expr.catalog_name, expr.schema_name, expr.table_name, table_id
expr.catalog_name, expr.schema_name, expr.table_name, table_id.id
);
table_id
table_id.id
} else {
match self.table_id_provider.as_ref() {
None => {
@@ -157,7 +157,7 @@ impl Instance {
mod tests {
use std::sync::Arc;
use api::v1::{ColumnDataType, ColumnDef};
use api::v1::{ColumnDataType, ColumnDef, TableId};
use common_catalog::consts::MIN_USER_TABLE_ID;
use common_grpc_expr::create_table_schema;
use datatypes::prelude::ConcreteDataType;
@@ -175,7 +175,7 @@ mod tests {
assert_eq!(request.catalog_name, "greptime".to_string());
assert_eq!(request.schema_name, "public".to_string());
assert_eq!(request.table_name, "my-metrics");
assert_eq!(request.desc, Some("blabla".to_string()));
assert_eq!(request.desc, Some("blabla little magic fairy".to_string()));
assert_eq!(request.schema, expected_table_schema());
assert_eq!(request.primary_key_indices, vec![1, 0]);
assert!(request.create_if_not_exists);
@@ -214,7 +214,7 @@ mod tests {
name: "a".to_string(),
datatype: 1024,
is_nullable: true,
default_constraint: None,
default_constraint: vec![],
};
let result = column_def.try_as_column_schema();
assert!(matches!(
@@ -226,7 +226,7 @@ mod tests {
name: "a".to_string(),
datatype: ColumnDataType::String as i32,
is_nullable: true,
default_constraint: None,
default_constraint: vec![],
};
let column_schema = column_def.try_as_column_schema().unwrap();
assert_eq!(column_schema.name, "a");
@@ -238,7 +238,7 @@ mod tests {
name: "a".to_string(),
datatype: ColumnDataType::String as i32,
is_nullable: true,
default_constraint: Some(default_constraint.clone().try_into().unwrap()),
default_constraint: default_constraint.clone().try_into().unwrap(),
};
let column_schema = column_def.try_as_column_schema().unwrap();
assert_eq!(column_schema.name, "a");
@@ -250,44 +250,46 @@ mod tests {
);
}
fn testing_create_expr() -> CreateExpr {
fn testing_create_expr() -> CreateTableExpr {
let column_defs = vec![
ColumnDef {
name: "host".to_string(),
datatype: ColumnDataType::String as i32,
is_nullable: false,
default_constraint: None,
default_constraint: vec![],
},
ColumnDef {
name: "ts".to_string(),
datatype: ColumnDataType::TimestampMillisecond as i32,
is_nullable: false,
default_constraint: None,
default_constraint: vec![],
},
ColumnDef {
name: "cpu".to_string(),
datatype: ColumnDataType::Float32 as i32,
is_nullable: true,
default_constraint: None,
default_constraint: vec![],
},
ColumnDef {
name: "memory".to_string(),
datatype: ColumnDataType::Float64 as i32,
is_nullable: true,
default_constraint: None,
default_constraint: vec![],
},
];
CreateExpr {
catalog_name: None,
schema_name: None,
CreateTableExpr {
catalog_name: "".to_string(),
schema_name: "".to_string(),
table_name: "my-metrics".to_string(),
desc: Some("blabla".to_string()),
desc: "blabla little magic fairy".to_string(),
column_defs,
time_index: "ts".to_string(),
primary_keys: vec!["ts".to_string(), "host".to_string()],
create_if_not_exists: true,
table_options: Default::default(),
table_id: Some(MIN_USER_TABLE_ID),
table_id: Some(TableId {
id: MIN_USER_TABLE_ID,
}),
region_ids: vec![0],
}
}
src/frontend/src/expr_factory.rs
+19 -17
View File
@@ -16,7 +16,7 @@ use std::collections::HashMap;
use std::sync::Arc;
use api::helper::ColumnDataTypeWrapper;
use api::v1::{Column, ColumnDataType, CreateExpr};
use api::v1::{Column, ColumnDataType, CreateTableExpr};
use datatypes::schema::ColumnSchema;
use snafu::{ensure, ResultExt};
use sql::ast::{ColumnDef, TableConstraint};
@@ -32,7 +32,7 @@ pub type CreateExprFactoryRef = Arc<dyn CreateExprFactory + Send + Sync>;
#[async_trait::async_trait]
pub trait CreateExprFactory {
async fn create_expr_by_stmt(&self, stmt: &CreateTable) -> Result<CreateExpr>;
async fn create_expr_by_stmt(&self, stmt: &CreateTable) -> Result<CreateTableExpr>;
async fn create_expr_by_columns(
&self,
@@ -40,7 +40,7 @@ pub trait CreateExprFactory {
schema_name: &str,
table_name: &str,
columns: &[Column],
) -> crate::error::Result<CreateExpr>;
) -> crate::error::Result<CreateTableExpr>;
}
#[derive(Debug)]
@@ -48,7 +48,7 @@ pub struct DefaultCreateExprFactory;
#[async_trait::async_trait]
impl CreateExprFactory for DefaultCreateExprFactory {
async fn create_expr_by_stmt(&self, stmt: &CreateTable) -> Result<CreateExpr> {
async fn create_expr_by_stmt(&self, stmt: &CreateTable) -> Result<CreateTableExpr> {
create_to_expr(None, vec![0], stmt)
}
@@ -58,7 +58,7 @@ impl CreateExprFactory for DefaultCreateExprFactory {
schema_name: &str,
table_name: &str,
columns: &[Column],
) -> Result<CreateExpr> {
) -> Result<CreateTableExpr> {
let table_id = None;
let create_expr = common_grpc_expr::build_create_expr_from_insertion(
catalog_name,
@@ -78,23 +78,23 @@ fn create_to_expr(
table_id: Option<u32>,
region_ids: Vec<u32>,
create: &CreateTable,
) -> Result<CreateExpr> {
) -> Result<CreateTableExpr> {
let (catalog_name, schema_name, table_name) =
table_idents_to_full_name(&create.name).context(ParseSqlSnafu)?;
let time_index = find_time_index(&create.constraints)?;
let expr = CreateExpr {
catalog_name: Some(catalog_name),
schema_name: Some(schema_name),
let expr = CreateTableExpr {
catalog_name,
schema_name,
table_name,
desc: None,
desc: "".to_string(),
column_defs: columns_to_expr(&create.columns, &time_index)?,
time_index,
primary_keys: find_primary_keys(&create.constraints)?,
create_if_not_exists: create.if_not_exists,
// TODO(LFC): Fill in other table options.
table_options: HashMap::from([("engine".to_string(), create.engine.clone())]),
table_id,
table_id: table_id.map(|id| api::v1::TableId { id }),
region_ids,
};
Ok(expr)
@@ -171,12 +171,14 @@ fn columns_to_expr(
datatype: datatype as i32,
is_nullable: schema.is_nullable(),
default_constraint: match schema.default_constraint() {
None => None,
Some(v) => Some(v.clone().try_into().context(
ConvertColumnDefaultConstraintSnafu {
column_name: &schema.name,
},
)?),
None => vec![],
Some(v) => {
v.clone()
.try_into()
.context(ConvertColumnDefaultConstraintSnafu {
column_name: &schema.name,
})?
}
},
})
})
src/frontend/src/instance.rs
+19 -21
View File
@@ -25,7 +25,7 @@ use api::v1::alter_expr::Kind;
use api::v1::object_expr::Expr;
use api::v1::{
admin_expr, AddColumns, AdminExpr, AdminResult, AlterExpr, Column, CreateDatabaseExpr,
CreateExpr, DropTableExpr, ExprHeader, InsertExpr, ObjectExpr,
CreateTableExpr, DropTableExpr, ExprHeader, InsertExpr, ObjectExpr,
ObjectResult as GrpcObjectResult,
};
use async_trait::async_trait;
@@ -196,7 +196,7 @@ impl Instance {
/// Handle create expr.
pub async fn handle_create_table(
&self,
mut expr: CreateExpr,
mut expr: CreateTableExpr,
partitions: Option<Partitions>,
) -> Result<Output> {
if let Some(v) = &self.dist_instance {
@@ -206,7 +206,7 @@ impl Instance {
header: Some(ExprHeader {
version: PROTOCOL_VERSION,
}),
expr: Some(admin_expr::Expr::Create(expr)),
expr: Some(admin_expr::Expr::CreateTable(expr)),
};
let result = self
.grpc_admin_handler
@@ -359,8 +359,8 @@ impl Instance {
);
let expr = AlterExpr {
table_name: table_name.to_string(),
schema_name: Some(schema_name.to_string()),
catalog_name: Some(catalog_name.to_string()),
schema_name: schema_name.to_string(),
catalog_name: catalog_name.to_string(),
kind: Some(Kind::AddColumns(add_columns)),
};
@@ -630,7 +630,7 @@ impl GrpcAdminHandler for Instance {
async fn exec_admin_request(&self, mut expr: AdminExpr) -> server_error::Result<AdminResult> {
// Force the default to be `None` rather than `Some(0)` comes from gRPC decode.
// Related issue: #480
if let Some(api::v1::admin_expr::Expr::Create(create)) = &mut expr.expr {
if let Some(api::v1::admin_expr::Expr::CreateTable(create)) = &mut expr.expr {
create.table_id = None;
}
self.grpc_admin_handler.exec_admin_request(expr).await
@@ -808,7 +808,7 @@ mod tests {
let create_expr = create_expr();
let admin_expr = AdminExpr {
header: Some(ExprHeader::default()),
expr: Some(admin_expr::Expr::Create(create_expr)),
expr: Some(admin_expr::Expr::CreateTable(create_expr)),
};
let result = GrpcAdminHandler::exec_admin_request(&*instance, admin_expr)
.await
@@ -886,48 +886,46 @@ mod tests {
}
}
fn create_expr() -> CreateExpr {
fn create_expr() -> CreateTableExpr {
let column_defs = vec![
GrpcColumnDef {
name: "host".to_string(),
datatype: ColumnDataType::String as i32,
is_nullable: false,
default_constraint: None,
default_constraint: vec![],
},
GrpcColumnDef {
name: "cpu".to_string(),
datatype: ColumnDataType::Float64 as i32,
is_nullable: true,
default_constraint: None,
default_constraint: vec![],
},
GrpcColumnDef {
name: "memory".to_string(),
datatype: ColumnDataType::Float64 as i32,
is_nullable: true,
default_constraint: None,
default_constraint: vec![],
},
GrpcColumnDef {
name: "disk_util".to_string(),
datatype: ColumnDataType::Float64 as i32,
is_nullable: true,
default_constraint: Some(
ColumnDefaultConstraint::Value(Value::from(9.9f64))
.try_into()
.unwrap(),
),
default_constraint: ColumnDefaultConstraint::Value(Value::from(9.9f64))
.try_into()
.unwrap(),
},
GrpcColumnDef {
name: "ts".to_string(),
datatype: ColumnDataType::TimestampMillisecond as i32,
is_nullable: true,
default_constraint: None,
default_constraint: vec![],
},
];
CreateExpr {
catalog_name: None,
schema_name: None,
CreateTableExpr {
catalog_name: "".to_string(),
schema_name: "".to_string(),
table_name: "demo".to_string(),
desc: None,
desc: "".to_string(),
column_defs,
time_index: "ts".to_string(),
primary_keys: vec!["host".to_string()],
src/frontend/src/instance/distributed.rs
+42 -26
View File
@@ -18,8 +18,8 @@ use std::sync::Arc;
use api::helper::ColumnDataTypeWrapper;
use api::result::AdminResultBuilder;
use api::v1::{
admin_expr, AdminExpr, AdminResult, AlterExpr, CreateDatabaseExpr, CreateExpr, ObjectExpr,
ObjectResult,
admin_expr, AdminExpr, AdminResult, AlterExpr, CreateDatabaseExpr, CreateTableExpr, ObjectExpr,
ObjectResult, TableId,
};
use async_trait::async_trait;
use catalog::helper::{SchemaKey, SchemaValue, TableGlobalKey, TableGlobalValue};
@@ -86,7 +86,7 @@ impl DistInstance {
pub(crate) async fn create_table(
&self,
create_table: &mut CreateExpr,
create_table: &mut CreateTableExpr,
partitions: Option<Partitions>,
) -> Result<Output> {
let response = self.create_table_in_meta(create_table, partitions).await?;
@@ -112,7 +112,9 @@ impl DistInstance {
table_name: create_table.table_name.to_string()
}
);
create_table.table_id = Some(table_route.table.id as u32);
create_table.table_id = Some(TableId {
id: table_route.table.id as u32,
});
self.put_table_global_meta(create_table, table_route)
.await?;
@@ -194,8 +196,16 @@ impl DistInstance {
}
async fn handle_alter_table(&self, expr: AlterExpr) -> Result<AdminResult> {
let catalog_name = expr.catalog_name.as_deref().unwrap_or(DEFAULT_CATALOG_NAME);
let schema_name = expr.schema_name.as_deref().unwrap_or(DEFAULT_SCHEMA_NAME);
let catalog_name = if expr.catalog_name.is_empty() {
DEFAULT_CATALOG_NAME
} else {
expr.catalog_name.as_str()
};
let schema_name = if expr.schema_name.is_empty() {
DEFAULT_SCHEMA_NAME
} else {
expr.schema_name.as_str()
};
let table_name = expr.table_name.as_str();
let table = self
.catalog_manager
@@ -223,20 +233,18 @@ impl DistInstance {
async fn create_table_in_meta(
&self,
create_table: &CreateExpr,
create_table: &CreateTableExpr,
partitions: Option<Partitions>,
) -> Result<RouteResponse> {
let table_name = TableName::new(
create_table
.catalog_name
.clone()
.unwrap_or_else(|| DEFAULT_CATALOG_NAME.to_string()),
create_table
.schema_name
.clone()
.unwrap_or_else(|| DEFAULT_SCHEMA_NAME.to_string()),
create_table.table_name.clone(),
);
let mut catalog_name = create_table.catalog_name.clone();
if catalog_name.is_empty() {
catalog_name = DEFAULT_CATALOG_NAME.to_string();
}
let mut schema_name = create_table.schema_name.clone();
if schema_name.is_empty() {
schema_name = DEFAULT_SCHEMA_NAME.to_string();
}
let table_name = TableName::new(catalog_name, schema_name, create_table.table_name.clone());
let partitions = parse_partitions(create_table, partitions)?;
let request = MetaCreateRequest {
@@ -252,7 +260,7 @@ impl DistInstance {
// TODO(LFC): Maybe move this to FrontendCatalogManager's "register_table" method?
async fn put_table_global_meta(
&self,
create_table: &CreateExpr,
create_table: &CreateTableExpr,
table_route: &TableRoute,
) -> Result<()> {
let table_name = &table_route.table.table_name;
@@ -274,10 +282,12 @@ impl DistInstance {
.await
.context(CatalogSnafu)?
{
let existing_bytes = existing.unwrap(); //this unwrap is safe since we compare with empty bytes and failed
let existing_bytes = existing.unwrap(); // this unwrap is safe since we compare with empty bytes and failed
let existing_value =
TableGlobalValue::from_bytes(&existing_bytes).context(CatalogEntrySerdeSnafu)?;
if existing_value.table_info.ident.table_id != create_table.table_id.unwrap() {
if existing_value.table_info.ident.table_id
!= create_table.table_id.as_ref().unwrap().id
{
error!(
"Table with name {} already exists, value in catalog: {:?}",
key, existing_bytes
@@ -340,7 +350,7 @@ impl GrpcAdminHandler for DistInstance {
}
fn create_table_global_value(
create_table: &CreateExpr,
create_table: &CreateTableExpr,
table_route: &TableRoute,
) -> Result<TableGlobalValue> {
let table_name = &table_route.table.table_name;
@@ -419,13 +429,19 @@ fn create_table_global_value(
created_on: DateTime::default(),
};
let desc = if create_table.desc.is_empty() {
None
} else {
Some(create_table.desc.clone())
};
let table_info = RawTableInfo {
ident: TableIdent {
table_id: table_route.table.id as u32,
version: 0,
},
name: table_name.table_name.clone(),
desc: create_table.desc.clone(),
desc,
catalog_name: table_name.catalog_name.clone(),
schema_name: table_name.schema_name.clone(),
meta,
@@ -440,7 +456,7 @@ fn create_table_global_value(
}
fn parse_partitions(
create_table: &CreateExpr,
create_table: &CreateTableExpr,
partitions: Option<Partitions>,
) -> Result<Vec<MetaPartition>> {
// If partitions are not defined by user, use the timestamp column (which has to be existed) as
@@ -455,7 +471,7 @@ fn parse_partitions(
}
fn find_partition_entries(
create_table: &CreateExpr,
create_table: &CreateTableExpr,
partitions: &Option<Partitions>,
partition_columns: &[String],
) -> Result<Vec<Vec<PartitionBound>>> {
@@ -505,7 +521,7 @@ fn find_partition_entries(
}
fn find_partition_columns(
create_table: &CreateExpr,
create_table: &CreateTableExpr,
partitions: &Option<Partitions>,
) -> Result<Vec<String>> {
let columns = if let Some(partitions) = partitions {
src/sql/src/statements.rs
+2 -2
View File
@@ -283,7 +283,7 @@ pub fn sql_column_def_to_grpc_column_def(col: ColumnDef) -> Result<api::v1::Colu
name,
datatype: data_type,
is_nullable,
default_constraint,
default_constraint: default_constraint.unwrap_or_default(),
})
}
@@ -588,7 +588,7 @@ mod tests {
assert_eq!("col", grpc_column_def.name);
assert!(grpc_column_def.is_nullable); // nullable when options are empty
assert_eq!(ColumnDataType::Float64 as i32, grpc_column_def.datatype);
assert_eq!(None, grpc_column_def.default_constraint);
assert!(grpc_column_def.default_constraint.is_empty());
// test not null
let column_def = ColumnDef {
src/sql/src/statements/alter.rs
+4 -4
View File
@@ -56,7 +56,7 @@ impl TryFrom<AlterTable> for AlterExpr {
type Error = crate::error::Error;
fn try_from(value: AlterTable) -> Result<Self, Self::Error> {
let (catalog, schema, table) = table_idents_to_full_name(&value.table_name)?;
let (catalog_name, schema_name, table_name) = table_idents_to_full_name(&value.table_name)?;
let kind = match value.alter_operation {
AlterTableOperation::AddConstraint(_) => {
@@ -80,9 +80,9 @@ impl TryFrom<AlterTable> for AlterExpr {
}
};
let expr = AlterExpr {
catalog_name: Some(catalog),
schema_name: Some(schema),
table_name: table,
catalog_name,
schema_name,
table_name,
kind: Some(kind),
};
tests-integration/tests/grpc.rs
+16 -14
View File
@@ -15,7 +15,7 @@ use api::v1::alter_expr::Kind;
use api::v1::column::SemanticType;
use api::v1::{
admin_result, column, AddColumn, AddColumns, AlterExpr, Column, ColumnDataType, ColumnDef,
CreateExpr, InsertExpr, MutateResult,
CreateTableExpr, InsertExpr, MutateResult, TableId,
};
use client::admin::Admin;
use client::{Client, Database, ObjectResult};
@@ -151,7 +151,7 @@ pub async fn test_insert_and_select(store_type: StorageType) {
name: "test_column".to_string(),
datatype: ColumnDataType::Int64.into(),
is_nullable: true,
default_constraint: None,
default_constraint: vec![],
};
let kind = Kind::AddColumns(AddColumns {
add_columns: vec![AddColumn {
@@ -161,8 +161,8 @@ pub async fn test_insert_and_select(store_type: StorageType) {
});
let expr = AlterExpr {
table_name: "test_table".to_string(),
catalog_name: None,
schema_name: None,
catalog_name: "".to_string(),
schema_name: "".to_string(),
kind: Some(kind),
};
let result = admin.alter(expr).await.unwrap();
@@ -222,44 +222,46 @@ async fn insert_and_assert(db: &Database) {
}
}
fn testing_create_expr() -> CreateExpr {
fn testing_create_expr() -> CreateTableExpr {
let column_defs = vec![
ColumnDef {
name: "host".to_string(),
datatype: ColumnDataType::String as i32,
is_nullable: false,
default_constraint: None,
default_constraint: vec![],
},
ColumnDef {
name: "cpu".to_string(),
datatype: ColumnDataType::Float64 as i32,
is_nullable: true,
default_constraint: None,
default_constraint: vec![],
},
ColumnDef {
name: "memory".to_string(),
datatype: ColumnDataType::Float64 as i32,
is_nullable: true,
default_constraint: None,
default_constraint: vec![],
},
ColumnDef {
name: "ts".to_string(),
datatype: ColumnDataType::TimestampMillisecond as i32, // timestamp
is_nullable: true,
default_constraint: None,
default_constraint: vec![],
},
];
CreateExpr {
catalog_name: None,
schema_name: None,
CreateTableExpr {
catalog_name: "".to_string(),
schema_name: "".to_string(),
table_name: "demo".to_string(),
desc: Some("blabla".to_string()),
desc: "blabla little magic fairy".to_string(),
column_defs,
time_index: "ts".to_string(),
primary_keys: vec!["host".to_string()],
create_if_not_exists: true,
table_options: Default::default(),
table_id: Some(MIN_USER_TABLE_ID),
table_id: Some(TableId {
id: MIN_USER_TABLE_ID,
}),
region_ids: vec![0],
}
}