feat: support default value when inserting data (#854)

lizhemingi · web-flow · commit 0959c1d16b2a · 2023-01-13T14:49:05.000+08:00
diff --git a/src/datanode/src/error.rs b/src/datanode/src/error.rs
@@ -295,6 +295,26 @@ pub enum Error {
 
     #[snafu(display("Cannot find requested database: {}-{}", catalog, schema))]
     DatabaseNotFound { catalog: String, schema: String },
+
+    #[snafu(display(
+        "Failed to build default value, column: {}, source: {}",
+        column,
+        source
+    ))]
+    ColumnDefaultValue {
+        column: String,
+        #[snafu(backtrace)]
+        source: datatypes::error::Error,
+    },
+
+    #[snafu(display(
+        "No valid default value can be build automatically, column: {}",
+        column,
+    ))]
+    ColumnNoneDefaultValue {
+        column: String,
+        backtrace: Backtrace,
+    },
 }
 
 pub type Result<T> = std::result::Result<T, Error>;
@@ -363,6 +383,8 @@ impl ErrorExt for Error {
             Error::BumpTableId { source, .. } => source.status_code(),
             Error::MissingNodeId { .. } => StatusCode::InvalidArguments,
             Error::MissingMetasrvOpts { .. } => StatusCode::InvalidArguments,
+            Error::ColumnDefaultValue { source, .. } => source.status_code(),
+            Error::ColumnNoneDefaultValue { .. } => StatusCode::InvalidArguments,
         }
     }
 
diff --git a/src/datanode/src/sql/insert.rs b/src/datanode/src/sql/insert.rs
@@ -15,7 +15,7 @@
 use catalog::CatalogManagerRef;
 use common_query::Output;
 use datatypes::data_type::DataType;
-use datatypes::prelude::ConcreteDataType;
+use datatypes::schema::ColumnSchema;
 use datatypes::vectors::MutableVector;
 use snafu::{ensure, OptionExt, ResultExt};
 use sql::ast::Value as SqlValue;
@@ -25,11 +25,14 @@ use table::engine::TableReference;
 use table::requests::*;
 
 use crate::error::{
-    CatalogSnafu, ColumnNotFoundSnafu, ColumnValuesNumberMismatchSnafu, InsertSnafu,
-    ParseSqlValueSnafu, Result, TableNotFoundSnafu,
+    CatalogSnafu, ColumnDefaultValueSnafu, ColumnNoneDefaultValueSnafu, ColumnNotFoundSnafu,
+    ColumnValuesNumberMismatchSnafu, InsertSnafu, ParseSqlSnafu, ParseSqlValueSnafu, Result,
+    TableNotFoundSnafu,
 };
 use crate::sql::{SqlHandler, SqlRequest};
 
+const DEFAULT_PLACEHOLDER_VALUE: &str = "default";
+
 impl SqlHandler {
     pub(crate) async fn insert(&self, req: InsertRequest) -> Result<Output> {
         // FIXME(dennis): table_ref is used in InsertSnafu and the req is consumed
@@ -72,17 +75,13 @@ impl SqlHandler {
         };
         let rows_num = values.len();
 
-        let mut columns_builders: Vec<(&String, &ConcreteDataType, Box<dyn MutableVector>)> =
+        let mut columns_builders: Vec<(&ColumnSchema, Box<dyn MutableVector>)> =
             Vec::with_capacity(columns_num);
 
         if columns.is_empty() {
             for column_schema in schema.column_schemas() {
                 let data_type = &column_schema.data_type;
-                columns_builders.push((
-                    &column_schema.name,
-                    data_type,
-                    data_type.create_mutable_vector(rows_num),
-                ));
+                columns_builders.push((column_schema, data_type.create_mutable_vector(rows_num)));
             }
         } else {
             for column_name in columns {
@@ -94,11 +93,7 @@ impl SqlHandler {
                         }
                     })?;
                 let data_type = &column_schema.data_type;
-                columns_builders.push((
-                    column_name,
-                    data_type,
-                    data_type.create_mutable_vector(rows_num),
-                ));
+                columns_builders.push((column_schema, data_type.create_mutable_vector(rows_num)));
             }
         }
 
@@ -112,10 +107,8 @@ impl SqlHandler {
                 }
             );
 
-            for (sql_val, (column_name, data_type, builder)) in
-                row.iter().zip(columns_builders.iter_mut())
-            {
-                add_row_to_vector(column_name, data_type, sql_val, builder)?;
+            for (sql_val, (column_schema, builder)) in row.iter().zip(columns_builders.iter_mut()) {
+                add_row_to_vector(column_schema, sql_val, builder)?;
             }
         }
 
@@ -125,21 +118,34 @@ impl SqlHandler {
             table_name: table_ref.table.to_string(),
             columns_values: columns_builders
                 .into_iter()
-                .map(|(c, _, mut b)| (c.to_owned(), b.to_vector()))
+                .map(|(cs, mut b)| (cs.name.to_string(), b.to_vector()))
                 .collect(),
         }))
     }
 }
 
 fn add_row_to_vector(
-    column_name: &str,
-    data_type: &ConcreteDataType,
+    column_schema: &ColumnSchema,
     sql_val: &SqlValue,
     builder: &mut Box<dyn MutableVector>,
 ) -> Result<()> {
-    let value = statements::sql_value_to_value(column_name, data_type, sql_val)
-        .context(ParseSqlValueSnafu)?;
+    let value = if replace_default(sql_val) {
+        column_schema
+            .create_default()
+            .context(ColumnDefaultValueSnafu {
+                column: column_schema.name.to_string(),
+            })?
+            .context(ColumnNoneDefaultValueSnafu {
+                column: column_schema.name.to_string(),
+            })?
+    } else {
+        statements::sql_value_to_value(&column_schema.name, &column_schema.data_type, sql_val)
+            .context(ParseSqlSnafu)?
+    };
     builder.push_value_ref(value.as_value_ref()).unwrap();
-
     Ok(())
 }
+
+fn replace_default(sql_val: &SqlValue) -> bool {
+    matches!(sql_val, SqlValue::Placeholder(s) if s.to_lowercase() == DEFAULT_PLACEHOLDER_VALUE)
+}
diff --git a/src/datatypes/src/schema/column_schema.rs b/src/datatypes/src/schema/column_schema.rs
@@ -150,6 +150,14 @@ impl ColumnSchema {
         }
         mutable_vector.to_vector()
     }
+
+    /// Creates a default value for this column.
+    ///
+    /// If the column is `NOT NULL` but doesn't has `DEFAULT` value supplied, returns `Ok(None)`.
+    pub fn create_default(&self) -> Result<Option<Value>> {
+        self.create_default_vector(1)
+            .map(|vec_ref_option| vec_ref_option.map(|vec_ref| vec_ref.get(0)))
+    }
 }
 
 impl TryFrom<&Field> for ColumnSchema {
@@ -337,4 +345,34 @@ mod tests {
         let expect: VectorRef = Arc::new(Int32Vector::from_slice(&[0, 0, 0, 0]));
         assert_eq!(expect, vector);
     }
+
+    #[test]
+    fn test_column_schema_single_create_default_null() {
+        // Implicit default null.
+        let column_schema = ColumnSchema::new("test", ConcreteDataType::int32_datatype(), true);
+        let v = column_schema.create_default().unwrap().unwrap();
+        assert!(v.is_null());
+
+        // Explicit default null.
+        let column_schema = ColumnSchema::new("test", ConcreteDataType::int32_datatype(), true)
+            .with_default_constraint(Some(ColumnDefaultConstraint::null_value()))
+            .unwrap();
+        let v = column_schema.create_default().unwrap().unwrap();
+        assert!(v.is_null());
+    }
+
+    #[test]
+    fn test_column_schema_single_create_default_not_null() {
+        let column_schema = ColumnSchema::new("test", ConcreteDataType::int32_datatype(), true)
+            .with_default_constraint(Some(ColumnDefaultConstraint::Value(Value::Int32(6))))
+            .unwrap();
+        let v = column_schema.create_default().unwrap().unwrap();
+        assert_eq!(v, Value::Int32(6));
+    }
+
+    #[test]
+    fn test_column_schema_single_no_default() {
+        let column_schema = ColumnSchema::new("test", ConcreteDataType::int32_datatype(), false);
+        assert!(column_schema.create_default().unwrap().is_none());
+    }
 }
diff --git a/src/frontend/src/error.rs b/src/frontend/src/error.rs
@@ -378,6 +378,26 @@ pub enum Error {
         #[snafu(backtrace)]
         source: common_grpc_expr::error::Error,
     },
+
+    #[snafu(display(
+        "Failed to build default value, column: {}, source: {}",
+        column,
+        source
+    ))]
+    ColumnDefaultValue {
+        column: String,
+        #[snafu(backtrace)]
+        source: datatypes::error::Error,
+    },
+
+    #[snafu(display(
+        "No valid default value can be build automatically, column: {}",
+        column,
+    ))]
+    ColumnNoneDefaultValue {
+        column: String,
+        backtrace: Backtrace,
+    },
 }
 
 pub type Result<T> = std::result::Result<T, Error>;
@@ -457,6 +477,8 @@ impl ErrorExt for Error {
             Error::EncodeSubstraitLogicalPlan { source } => source.status_code(),
             Error::BuildVector { source, .. } => source.status_code(),
             Error::InvokeDatanode { source } => source.status_code(),
+            Error::ColumnDefaultValue { source, .. } => source.status_code(),
+            Error::ColumnNoneDefaultValue { .. } => StatusCode::InvalidArguments,
         }
     }
 
diff --git a/src/frontend/src/sql.rs b/src/frontend/src/sql.rs
@@ -14,15 +14,18 @@
 
 use common_error::snafu::ensure;
 use datatypes::data_type::DataType;
-use datatypes::prelude::{ConcreteDataType, MutableVector};
+use datatypes::prelude::MutableVector;
+use datatypes::schema::ColumnSchema;
 use snafu::{OptionExt, ResultExt};
 use sql::ast::Value as SqlValue;
 use sql::statements;
 use sql::statements::insert::Insert;
 use table::requests::InsertRequest;
 use table::TableRef;
 
-use crate::error::{self, BuildVectorSnafu, Result};
+use crate::error::{self, Result};
+
+const DEFAULT_PLACEHOLDER_VALUE: &str = "default";
 
 // TODO(fys): Extract the common logic in datanode and frontend in the future.
 #[allow(dead_code)]
@@ -40,17 +43,13 @@ pub(crate) fn insert_to_request(table: &TableRef, stmt: Insert) -> Result<Insert
     };
     let rows_num = values.len();
 
-    let mut columns_builders: Vec<(&String, &ConcreteDataType, Box<dyn MutableVector>)> =
+    let mut columns_builders: Vec<(&ColumnSchema, Box<dyn MutableVector>)> =
         Vec::with_capacity(columns_num);
 
     if columns.is_empty() {
         for column_schema in schema.column_schemas() {
             let data_type = &column_schema.data_type;
-            columns_builders.push((
-                &column_schema.name,
-                data_type,
-                data_type.create_mutable_vector(rows_num),
-            ));
+            columns_builders.push((column_schema, data_type.create_mutable_vector(rows_num)));
         }
     } else {
         for column_name in columns {
@@ -61,11 +60,7 @@ pub(crate) fn insert_to_request(table: &TableRef, stmt: Insert) -> Result<Insert
                 }
             })?;
             let data_type = &column_schema.data_type;
-            columns_builders.push((
-                column_name,
-                data_type,
-                data_type.create_mutable_vector(rows_num),
-            ));
+            columns_builders.push((column_schema, data_type.create_mutable_vector(rows_num)));
         }
     }
 
@@ -78,10 +73,8 @@ pub(crate) fn insert_to_request(table: &TableRef, stmt: Insert) -> Result<Insert
             }
         );
 
-        for (sql_val, (column_name, data_type, builder)) in
-            row.iter().zip(columns_builders.iter_mut())
-        {
-            add_row_to_vector(column_name, data_type, sql_val, builder)?;
+        for (sql_val, (column_schema, builder)) in row.iter().zip(columns_builders.iter_mut()) {
+            add_row_to_vector(column_schema, sql_val, builder)?;
         }
     }
 
@@ -91,21 +84,33 @@ pub(crate) fn insert_to_request(table: &TableRef, stmt: Insert) -> Result<Insert
         table_name,
         columns_values: columns_builders
             .into_iter()
-            .map(|(c, _, mut b)| (c.to_owned(), b.to_vector()))
+            .map(|(cs, mut b)| (cs.name.to_string(), b.to_vector()))
             .collect(),
     })
 }
 
 fn add_row_to_vector(
-    column_name: &str,
-    data_type: &ConcreteDataType,
+    column_schema: &ColumnSchema,
     sql_val: &SqlValue,
     builder: &mut Box<dyn MutableVector>,
 ) -> Result<()> {
-    let value = statements::sql_value_to_value(column_name, data_type, sql_val)
-        .context(error::ParseSqlSnafu)?;
-    builder
-        .push_value_ref(value.as_value_ref())
-        .context(BuildVectorSnafu { value })?;
+    let value = if replace_default(sql_val) {
+        column_schema
+            .create_default()
+            .context(error::ColumnDefaultValueSnafu {
+                column: column_schema.name.to_string(),
+            })?
+            .context(error::ColumnNoneDefaultValueSnafu {
+                column: column_schema.name.to_string(),
+            })?
+    } else {
+        statements::sql_value_to_value(&column_schema.name, &column_schema.data_type, sql_val)
+            .context(error::ParseSqlSnafu)?
+    };
+    builder.push_value_ref(value.as_value_ref()).unwrap();
     Ok(())
 }
+
+fn replace_default(sql_val: &SqlValue) -> bool {
+    matches!(sql_val, SqlValue::Placeholder(s) if s.to_lowercase() == DEFAULT_PLACEHOLDER_VALUE)
+}
diff --git a/src/sql/src/error.rs b/src/sql/src/error.rs
@@ -131,6 +131,9 @@ pub enum Error {
         #[snafu(backtrace)]
         source: api::error::Error,
     },
+
+    #[snafu(display("Invalid sql value: {}", value))]
+    InvalidSqlValue { value: String, backtrace: Backtrace },
 }
 
 impl ErrorExt for Error {
@@ -156,6 +159,7 @@ impl ErrorExt for Error {
             UnsupportedAlterTableStatement { .. } => StatusCode::InvalidSyntax,
             SerializeColumnDefaultConstraint { source, .. } => source.status_code(),
             ConvertToGrpcDataType { source, .. } => source.status_code(),
+            InvalidSqlValue { .. } => StatusCode::InvalidArguments,
         }
     }
 
diff --git a/src/sql/src/statements.rs b/src/sql/src/statements.rs
@@ -39,8 +39,9 @@ use crate::ast::{
     Value as SqlValue,
 };
 use crate::error::{
-    self, ColumnTypeMismatchSnafu, ConvertToGrpcDataTypeSnafu, ParseSqlValueSnafu, Result,
-    SerializeColumnDefaultConstraintSnafu, UnsupportedDefaultValueSnafu,
+    self, ColumnTypeMismatchSnafu, ConvertToGrpcDataTypeSnafu, InvalidSqlValueSnafu,
+    ParseSqlValueSnafu, Result, SerializeColumnDefaultConstraintSnafu,
+    UnsupportedDefaultValueSnafu,
 };
 
 // TODO(LFC): Get rid of this function, use session context aware version of "table_idents_to_full_name" instead.
@@ -222,6 +223,7 @@ pub fn sql_value_to_value(
             parse_string_to_value(column_name, s.to_owned(), data_type)?
         }
         SqlValue::HexStringLiteral(s) => parse_hex_string(s)?,
+        SqlValue::Placeholder(s) => return InvalidSqlValueSnafu { value: s }.fail(),
         _ => todo!("Other sql value"),
     })
 }
@@ -720,4 +722,14 @@ mod tests {
         assert!(!column_schema.is_nullable());
         assert!(!column_schema.is_time_index());
     }
+
+    #[test]
+    pub fn test_parse_placeholder_value() {
+        assert!(sql_value_to_value(
+            "test",
+            &ConcreteDataType::string_datatype(),
+            &SqlValue::Placeholder("default".into())
+        )
+        .is_err());
+    }
 }
diff --git a/src/sql/src/statements/insert.rs b/src/sql/src/statements/insert.rs

Original file line number	Diff line number	Diff line change
`@@ -131,6 +131,9 @@ pub enum Error {`
`131`	`131`	`#[snafu(backtrace)]`
`132`	`132`	`source: api::error::Error,`
`133`	`133`	`},`
	`134`	`+`
	`135`	`+ #[snafu(display("Invalid sql value: {}", value))]`
	`136`	`+ InvalidSqlValue { value: String, backtrace: Backtrace },`
`134`	`137`	`}`
`135`	`138`
`136`	`139`	`impl ErrorExt for Error {`
`@@ -156,6 +159,7 @@ impl ErrorExt for Error {`
`156`	`159`	`UnsupportedAlterTableStatement { .. } => StatusCode::InvalidSyntax,`
`157`	`160`	`SerializeColumnDefaultConstraint { source, .. } => source.status_code(),`
`158`	`161`	`ConvertToGrpcDataType { source, .. } => source.status_code(),`
	`162`	`+ InvalidSqlValue { .. } => StatusCode::InvalidArguments,`
`159`	`163`	`}`
`160`	`164`	`}`
`161`	`165`