refactor: changes CreateTableRequest::schema to RawSchema (#1018)

* refactor: changes CreateTableRequest::schema to RawSchema

* refactor(grpc-expr): create_table_schema returns RawSchema

src/datanode/src/error.rs

@@ -177,12 +177,6 @@ pub enum Error {
     #[snafu(display("Not support SQL, error: {}", msg))]
     NotSupportSql { msg: String },
 
-    #[snafu(display("Failed to create schema when creating table, source: {}", source))]
-    CreateSchema {
-        #[snafu(backtrace)]
-        source: datatypes::error::Error,
-    },
-
     #[snafu(display("Failed to convert datafusion schema, source: {}", source))]
     ConvertSchema {
         #[snafu(backtrace)]
@@ -370,9 +364,9 @@ impl ErrorExt for Error {
             | Error::CreateExprToRequest { source }
             | Error::InsertData { source } => source.status_code(),
 
-            Error::CreateSchema { source, .. }
-            | Error::ConvertSchema { source, .. }
-            | Error::VectorComputation { source } => source.status_code(),
+            Error::ConvertSchema { source, .. } | Error::VectorComputation { source } => {
+                source.status_code()
+            }
 
             Error::ColumnValuesNumberMismatch { .. }
             | Error::InvalidSql { .. }

src/datanode/src/server/grpc.rs

@@ -86,13 +86,11 @@ impl Instance {
 
 #[cfg(test)]
 mod tests {
-    use std::sync::Arc;
-
     use api::v1::{column_def, ColumnDataType, ColumnDef, TableId};
     use common_catalog::consts::MIN_USER_TABLE_ID;
     use common_grpc_expr::create_table_schema;
     use datatypes::prelude::ConcreteDataType;
-    use datatypes::schema::{ColumnDefaultConstraint, ColumnSchema, SchemaBuilder, SchemaRef};
+    use datatypes::schema::{ColumnDefaultConstraint, ColumnSchema, RawSchema};
     use datatypes::value::Value;
 
     use super::*;
@@ -224,7 +222,7 @@ mod tests {
         }
     }
 
-    fn expected_table_schema() -> SchemaRef {
+    fn expected_table_schema() -> RawSchema {
         let column_schemas = vec![
             ColumnSchema::new("host", ConcreteDataType::string_datatype(), false),
             ColumnSchema::new(
@@ -236,11 +234,7 @@ mod tests {
             ColumnSchema::new("cpu", ConcreteDataType::float32_datatype(), true),
             ColumnSchema::new("memory", ConcreteDataType::float64_datatype(), true),
         ];
-        Arc::new(
-            SchemaBuilder::try_from(column_schemas)
-                .unwrap()
-                .build()
-                .unwrap(),
-        )
+
+        RawSchema::new(column_schemas)
     }
 }

src/datanode/src/sql/create.rs

@@ -13,13 +13,12 @@
 // limitations under the License.
 
 use std::collections::HashMap;
-use std::sync::Arc;
 
 use catalog::{RegisterSchemaRequest, RegisterTableRequest};
 use common_query::Output;
 use common_telemetry::tracing::info;
 use common_telemetry::tracing::log::error;
-use datatypes::schema::SchemaBuilder;
+use datatypes::schema::RawSchema;
 use session::context::QueryContextRef;
 use snafu::{ensure, OptionExt, ResultExt};
 use sql::ast::{ColumnOption, TableConstraint};
@@ -31,8 +30,8 @@ use table::metadata::TableId;
 use table::requests::*;
 
 use crate::error::{
-    self, CatalogNotFoundSnafu, CatalogSnafu, ConstraintNotSupportedSnafu, CreateSchemaSnafu,
-    CreateTableSnafu, IllegalPrimaryKeysDefSnafu, InsertSystemCatalogSnafu, KeyColumnNotFoundSnafu,
+    self, CatalogNotFoundSnafu, CatalogSnafu, ConstraintNotSupportedSnafu, CreateTableSnafu,
+    IllegalPrimaryKeysDefSnafu, InsertSystemCatalogSnafu, KeyColumnNotFoundSnafu,
     RegisterSchemaSnafu, Result, SchemaExistsSnafu, SchemaNotFoundSnafu,
 };
 use crate::sql::SqlHandler;
@@ -239,13 +238,7 @@ impl SqlHandler {
             })
             .collect::<Result<Vec<_>>>()?;
 
-        let schema = Arc::new(
-            SchemaBuilder::try_from(columns_schemas)
-                .context(CreateSchemaSnafu)?
-                .build()
-                .context(CreateSchemaSnafu)?,
-        );
-
+        let schema = RawSchema::new(columns_schemas);
         let request = CreateTableRequest {
             id: table_id,
             catalog_name: table_ref.catalog.to_string(),
@@ -267,6 +260,7 @@ mod tests {
     use std::assert_matches::assert_matches;
 
     use datatypes::prelude::ConcreteDataType;
+    use datatypes::schema::Schema;
     use sql::dialect::GenericDialect;
     use sql::parser::ParserContext;
     use sql::statements::statement::Statement;
@@ -320,8 +314,8 @@ mod tests {
         assert_eq!(42, c.id);
         assert!(!c.create_if_not_exists);
         assert_eq!(vec![0], c.primary_key_indices);
-        assert_eq!(1, c.schema.timestamp_index().unwrap());
-        assert_eq!(4, c.schema.column_schemas().len());
+        assert_eq!(1, c.schema.timestamp_index.unwrap());
+        assert_eq!(4, c.schema.column_schemas.len());
     }
 
     #[tokio::test]
@@ -371,7 +365,7 @@ mod tests {
             .create_to_request(42, parsed_stmt, &TableReference::bare("demo_table"))
             .unwrap();
         assert!(c.primary_key_indices.is_empty());
-        assert_eq!(c.schema.timestamp_index(), Some(1));
+        assert_eq!(c.schema.timestamp_index, Some(1));
     }
 
     /// Constraints specified, not column cannot be found.
@@ -438,40 +432,25 @@ mod tests {
         assert_eq!("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!(4, request.schema.column_schemas.len());
 
         assert_eq!(vec![0], request.primary_key_indices);
+        let schema = Schema::try_from(request.schema).unwrap();
         assert_eq!(
             ConcreteDataType::string_datatype(),
-            request
-                .schema
-                .column_schema_by_name("host")
-                .unwrap()
-                .data_type
+            schema.column_schema_by_name("host").unwrap().data_type
         );
         assert_eq!(
             ConcreteDataType::timestamp_millisecond_datatype(),
-            request
-                .schema
-                .column_schema_by_name("ts")
-                .unwrap()
-                .data_type
+            schema.column_schema_by_name("ts").unwrap().data_type
         );
         assert_eq!(
             ConcreteDataType::float64_datatype(),
-            request
-                .schema
-                .column_schema_by_name("cpu")
-                .unwrap()
-                .data_type
+            schema.column_schema_by_name("cpu").unwrap().data_type
         );
         assert_eq!(
             ConcreteDataType::float64_datatype(),
-            request
-                .schema
-                .column_schema_by_name("memory")
-                .unwrap()
-                .data_type
+            schema.column_schema_by_name("memory").unwrap().data_type
         );
     }
 }

src/datanode/src/tests/test_util.rs

@@ -19,7 +19,7 @@ use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, MIN_USER
 use common_query::Output;
 use common_recordbatch::util;
 use datatypes::data_type::ConcreteDataType;
-use datatypes::schema::{ColumnSchema, SchemaBuilder};
+use datatypes::schema::{ColumnSchema, RawSchema};
 use mito::config::EngineConfig;
 use mito::table::test_util::{new_test_object_store, MockEngine, MockMitoEngine};
 use query::QueryEngineFactory;
@@ -104,12 +104,7 @@ pub(crate) async fn create_test_table(
                 schema_name: "public".to_string(),
                 table_name: table_name.to_string(),
                 desc: Some(" a test table".to_string()),
-                schema: Arc::new(
-                    SchemaBuilder::try_from(column_schemas)
-                        .unwrap()
-                        .build()
-                        .expect("ts is expected to be timestamp column"),
-                ),
+                schema: RawSchema::new(column_schemas),
                 create_if_not_exists: true,
                 primary_key_indices: vec![0], // "host" is in primary keys
                 table_options: HashMap::new(),