fix(value): DataValue::to_primary_key has an order reversal problem… (#59)

KKould · web-flow · commit 9a05ed82fe6e · 2023-09-16T18:13:09.000+08:00
* fix(value): `DataValue::to_primary_key` has an order reversal problem when the signed integer is negative.

e.g. will happen -9 &lt; -100

* feat: supports `UnaryOperator`

* feat: supports `Insert Overwrite`

* invalid: Throw the failure of data conversion in the `DataValue::cast`
diff --git a/README.md b/README.md
@@ -75,6 +75,7 @@ Storage Support:
   - [x] Limit
 - DML
   - [x] Insert
+  - [x] Insert Overwrite
   - [x] Update
   - [x] Delete
 - DataTypes
diff --git a/src/binder/expr.rs b/src/binder/expr.rs
@@ -1,6 +1,6 @@
 use crate::binder::BindError;
 use itertools::Itertools;
-use sqlparser::ast::{BinaryOperator, Expr, Function, FunctionArg, FunctionArgExpr, Ident};
+use sqlparser::ast::{BinaryOperator, Expr, Function, FunctionArg, FunctionArgExpr, Ident, UnaryOperator};
 use std::slice;
 use std::sync::Arc;
 use async_recursion::async_recursion;
@@ -28,6 +28,7 @@ impl<S: Storage> Binder<S> {
             Expr::Value(v) => Ok(ScalarExpression::Constant(Arc::new(v.into()))),
             Expr::Function(func) => self.bind_agg_call(func).await,
             Expr::Nested(expr) => self.bind_expr(expr).await,
+            Expr::UnaryOp { expr, op } => self.bind_unary_op_internal(expr, op).await,
             _ => {
                 todo!()
             }
@@ -126,6 +127,25 @@ impl<S: Storage> Binder<S> {
         })
     }
 
+    async fn bind_unary_op_internal(
+        &mut self,
+        expr: &Expr,
+        op: &UnaryOperator,
+    ) -> Result<ScalarExpression, BindError> {
+        let expr = Box::new(self.bind_expr(expr).await?);
+        let ty = if let UnaryOperator::Not = op {
+            LogicalType::Boolean
+        } else {
+            expr.return_type()
+        };
+
+        Ok(ScalarExpression::Unary {
+            op: (op.clone()).into(),
+            expr,
+            ty,
+        })
+    }
+
     async fn bind_agg_call(&mut self, func: &Function) -> Result<ScalarExpression, BindError> {
         let mut args = Vec::with_capacity(func.args.len());
 
diff --git a/src/binder/insert.rs b/src/binder/insert.rs
@@ -4,6 +4,7 @@ use sqlparser::ast::{Expr, Ident, ObjectName};
 use crate::binder::{Binder, BindError, lower_case_name, split_name};
 use crate::catalog::ColumnRef;
 use crate::expression::ScalarExpression;
+use crate::expression::value_compute::unary_op;
 use crate::planner::LogicalPlan;
 use crate::planner::operator::insert::InsertOperator;
 use crate::planner::operator::Operator;
@@ -16,7 +17,8 @@ impl<S: Storage> Binder<S> {
         &mut self,
         name: ObjectName,
         idents: &[Ident],
-        expr_rows: &Vec<Vec<Expr>>
+        expr_rows: &Vec<Vec<Expr>>,
+        is_overwrite: bool
     ) -> Result<LogicalPlan, BindError> {
         let name = lower_case_name(&name);
         let (_, name) = split_name(&name)?;
@@ -45,13 +47,20 @@ impl<S: Storage> Binder<S> {
                 let mut row = Vec::with_capacity(expr_row.len());
 
                 for (i, expr) in expr_row.into_iter().enumerate() {
-                    match self.bind_expr(expr).await? {
+                    match &self.bind_expr(expr).await? {
                         ScalarExpression::Constant(value) => {
-                            let cast_value = DataValue::clone(&value)
+                            let cast_value = DataValue::clone(value)
                                 .cast(columns[i].datatype())?;
 
                             row.push(Arc::new(cast_value))
                         },
+                        ScalarExpression::Unary { expr, op, .. } => {
+                            if let ScalarExpression::Constant(value) = expr.as_ref() {
+                                row.push(Arc::new(unary_op(value, op)?))
+                            } else {
+                                unreachable!()
+                            }
+                        }
                         _ => unreachable!(),
                     }
                 }
@@ -64,6 +73,7 @@ impl<S: Storage> Binder<S> {
                 operator: Operator::Insert(
                     InsertOperator {
                         table_name,
+                        is_overwrite,
                     }
                 ),
                 childrens: vec![values_plan],
diff --git a/src/binder/mod.rs b/src/binder/mod.rs
@@ -90,9 +90,14 @@ impl<S: Storage> Binder<S> {
                     _ => todo!()
                 }
             }
-            Statement::Insert { table_name, columns, source, .. } => {
+            Statement::Insert { table_name, columns, source, overwrite, .. } => {
                 if let SetExpr::Values(values) = source.body.as_ref() {
-                    self.bind_insert(table_name.to_owned(), columns, &values.rows).await?
+                    self.bind_insert(
+                        table_name.to_owned(),
+                        columns,
+                        &values.rows,
+                        *overwrite
+                    ).await?
                 } else {
                     todo!()
                 }
diff --git a/src/db.rs b/src/db.rs
@@ -184,7 +184,7 @@ mod test {
         let kipsql = Database::with_kipdb(temp_dir.path()).await?;
         let _ = kipsql.run("create table t1 (a int primary key, b int, k int)").await?;
         let _ = kipsql.run("create table t2 (c int primary key, d int unsigned null, e datetime)").await?;
-        let _ = kipsql.run("insert into t1 (a, b, k) values (1, 1, 1), (4, 2, 2), (5, 2, 2)").await?;
+        let _ = kipsql.run("insert into t1 (a, b, k) values (-99, 1, 1), (-1, 2, 2), (5, 2, 2)").await?;
         let _ = kipsql.run("insert into t2 (d, c, e) values (2, 1, '2021-05-20 21:00:00'), (3, 4, '2023-09-10 00:00:00')").await?;
 
         println!("full t1:");
@@ -278,11 +278,17 @@ mod test {
         println!("{}", create_table(&tuples_distinct_t1));
 
         println!("update t1 with filter:");
-        let _ = kipsql.run("update t1 set a = 0 where b > 1").await?;
+        let _ = kipsql.run("update t1 set b = 0 where b > 1").await?;
         println!("after t1:");
         let update_after_full_t1 = kipsql.run("select * from t1").await?;
         println!("{}", create_table(&update_after_full_t1));
 
+        println!("insert overwrite t1:");
+        let _ = kipsql.run("insert overwrite t1 (a, b, k) values (-1, 1, 1)").await?;
+        println!("after t1:");
+        let insert_overwrite_after_full_t1 = kipsql.run("select * from t1").await?;
+        println!("{}", create_table(&insert_overwrite_after_full_t1));
+
         println!("delete t1 with filter:");
         let _ = kipsql.run("delete from t1 where b > 1").await?;
         println!("after t1:");
diff --git a/src/execution/executor/dml/insert.rs b/src/execution/executor/dml/insert.rs
@@ -14,13 +14,15 @@ use crate::types::value::{DataValue, ValueRef};
 pub struct Insert {
     table_name: TableName,
     input: BoxedExecutor,
+    is_overwrite: bool
 }
 
 impl From<(InsertOperator, BoxedExecutor)> for Insert {
-    fn from((InsertOperator { table_name }, input): (InsertOperator, BoxedExecutor)) -> Self {
+    fn from((InsertOperator { table_name, is_overwrite }, input): (InsertOperator, BoxedExecutor)) -> Self {
         Insert {
             table_name,
-            input
+            input,
+            is_overwrite,
         }
     }
 }
@@ -34,7 +36,7 @@ impl<S: Storage> Executor<S> for Insert {
 impl Insert {
     #[try_stream(boxed, ok = Tuple, error = ExecutorError)]
     pub async fn _execute<S: Storage>(self, storage: S) {
-        let Insert { table_name, input } = self;
+        let Insert { table_name, input, is_overwrite } = self;
         let mut primary_key_index = None;
 
         if let (Some(table_catalog), Some(mut table)) =
@@ -75,7 +77,7 @@ impl Insert {
                     tuple.values.push(value)
                 }
 
-                table.append(tuple)?;
+                table.append(tuple, is_overwrite)?;
             }
             table.commit().await?;
         }
diff --git a/src/execution/executor/dml/update.rs b/src/execution/executor/dml/update.rs
@@ -48,21 +48,21 @@ impl Update {
             #[for_await]
             for tuple in input {
                 let mut tuple = tuple?;
+                let mut is_overwrite = true;
 
                 for (i, column) in tuple.columns.iter().enumerate() {
                     if let Some(value) = value_map.get(&column.id) {
                         if column.desc.is_primary {
                             if let Some(old_key) = tuple.id.replace(value.clone()) {
-                                if value != &old_key {
-                                    table.delete(old_key)?;
-                                }
+                                table.delete(old_key)?;
+                                is_overwrite = false;
                             }
                         }
                         tuple.values[i] = value.clone();
                     }
                 }
 
-                table.append(tuple)?;
+                table.append(tuple, is_overwrite)?;
             }
             table.commit().await?;
         }
diff --git a/src/expression/evaluator.rs b/src/expression/evaluator.rs
@@ -1,6 +1,6 @@
 use std::sync::Arc;
 use itertools::Itertools;
-use crate::expression::value_compute::binary_op;
+use crate::expression::value_compute::{binary_op, unary_op};
 use crate::expression::ScalarExpression;
 use crate::types::errors::TypeError;
 use crate::types::tuple::Tuple;
@@ -35,7 +35,11 @@ impl ScalarExpression {
             ScalarExpression::IsNull{ expr } => {
                 Ok(Arc::new(DataValue::Boolean(Some(expr.nullable()))))
             }
-            ScalarExpression::Unary{ .. } => todo!(),
+            ScalarExpression::Unary{ expr, op, .. } => {
+                let value = expr.eval_column(tuple)?;
+
+                Ok(Arc::new(unary_op(&value, op)?))
+            },
             ScalarExpression::AggCall{ .. } => todo!()
         }
     }
diff --git a/src/expression/mod.rs b/src/expression/mod.rs
@@ -239,13 +239,16 @@ pub enum UnaryOperator {
     Plus,
     Minus,
     Not,
-    True,
-    False,
 }
 
 impl From<SqlUnaryOperator> for UnaryOperator {
     fn from(value: SqlUnaryOperator) -> Self {
-        todo!()
+        match value {
+            SqlUnaryOperator::Plus =>  UnaryOperator::Plus,
+            SqlUnaryOperator::Minus => UnaryOperator::Minus,
+            SqlUnaryOperator::Not => UnaryOperator::Not,
+            _ => unimplemented!("not support!")
+        }
     }
 }
 
@@ -311,7 +314,7 @@ impl From<SqlBinaryOperator> for BinaryOperator {
             SqlBinaryOperator::And => BinaryOperator::And,
             SqlBinaryOperator::Or => BinaryOperator::Or,
             SqlBinaryOperator::Xor => BinaryOperator::Xor,
-            _ => todo!()
+            _ => unimplemented!("not support!")
         }
     }
 }
diff --git a/src/expression/value_compute.rs b/src/expression/value_compute.rs
@@ -1,4 +1,4 @@
-use crate::expression::BinaryOperator;
+use crate::expression::{BinaryOperator, UnaryOperator};
 use crate::types::errors::TypeError;
 use crate::types::LogicalType;
 use crate::types::value::DataValue;
@@ -59,6 +59,52 @@ fn unpack_date(value: DataValue) -> Option<i64> {
     }
 }
 
+pub fn unary_op(
+    value: &DataValue,
+    op: &UnaryOperator,
+) -> Result<DataValue, TypeError> {
+    let mut value_type = value.logical_type();
+    let mut value = value.clone();
+
+    if value_type.is_numeric() && matches!(op, UnaryOperator::Plus | UnaryOperator::Minus) {
+        if value_type.is_unsigned_numeric() {
+            match value_type {
+                LogicalType::UTinyint => value_type = LogicalType::Tinyint,
+                LogicalType::USmallint => value_type = LogicalType::Smallint,
+                LogicalType::UInteger => value_type = LogicalType::Integer,
+                LogicalType::UBigint => value_type = LogicalType::Bigint,
+                _ => unreachable!()
+            };
+            value = value.cast(&value_type)?;
+        }
+
+        let result = match op {
+            UnaryOperator::Plus => value,
+            UnaryOperator::Minus => {
+                match value {
+                    DataValue::Float32(option) => DataValue::Float32(option.map(|v| -v)),
+                    DataValue::Float64(option) => DataValue::Float64(option.map(|v| -v)),
+                    DataValue::Int8(option) => DataValue::Int8(option.map(|v| -v)),
+                    DataValue::Int16(option) => DataValue::Int16(option.map(|v| -v)),
+                    DataValue::Int32(option) => DataValue::Int32(option.map(|v| -v)),
+                    DataValue::Int64(option) => DataValue::Int64(option.map(|v| -v)),
+                    _ => unreachable!()
+                }
+            }
+            _ => unreachable!()
+        };
+
+        Ok(result)
+    } else if matches!((value_type, op), (LogicalType::Boolean, UnaryOperator::Not)) {
+        match value {
+            DataValue::Boolean(option) => Ok(DataValue::Boolean(option.map(|v| !v))),
+            _ => unreachable!()
+        }
+    } else {
+        Err(TypeError::InvalidType)
+    }
+}
+
 /// Tips:
 /// - Null values operate as null values
 pub fn binary_op(
diff --git a/src/planner/operator/insert.rs b/src/planner/operator/insert.rs
@@ -3,4 +3,5 @@ use crate::catalog::TableName;
 #[derive(Debug, PartialEq, Clone)]
 pub struct InsertOperator {
     pub table_name: TableName,
+    pub is_overwrite: bool,
 }
diff --git a/src/storage/kip.rs b/src/storage/kip.rs
@@ -163,8 +163,12 @@ impl Table for KipTable {
         })
     }
 
-    fn append(&mut self, tuple: Tuple) -> Result<(), StorageError> {
+    fn append(&mut self, tuple: Tuple, is_overwrite: bool) -> Result<(), StorageError> {
         let (key, value) = self.table_codec.encode_tuple(&tuple)?;
+
+        if !is_overwrite && self.tx.get(&key)?.is_some() {
+            return Err(StorageError::DuplicatePrimaryKey);
+        }
         self.tx.set(key, value);
 
         Ok(())
diff --git a/src/storage/memory.rs b/src/storage/memory.rs
@@ -156,14 +156,17 @@ impl Table for MemTable {
         }
     }
 
-    fn append(&mut self, tuple: Tuple) -> Result<(), StorageError> {
+    fn append(&mut self, tuple: Tuple, is_overwrite: bool) -> Result<(), StorageError> {
         let tuples = unsafe {
             self.tuples
                 .as_ptr()
                 .as_mut()
         }.unwrap();
 
         if let Some(original_tuple) = tuples.iter_mut().find(|t| t.id == tuple.id) {
+            if !is_overwrite {
+                return Err(StorageError::DuplicatePrimaryKey);
+            }
             *original_tuple = tuple;
         } else {
             tuples.push(tuple);
@@ -255,15 +258,15 @@ pub(crate) mod test {
                 Arc::new(DataValue::Int32(Some(1))),
                 Arc::new(DataValue::Boolean(Some(true)))
             ],
-        })?;
+        }, false)?;
         table.append(Tuple {
             id: Some(Arc::new(DataValue::Int32(Some(2)))),
             columns: columns.clone(),
             values: vec![
                 Arc::new(DataValue::Int32(Some(2))),
                 Arc::new(DataValue::Boolean(Some(false)))
             ],
-        })?;
+        }, false)?;
 
         Ok(())
     }
diff --git a/src/storage/mod.rs b/src/storage/mod.rs
@@ -44,7 +44,7 @@ pub trait Table: Sync + Send + 'static {
         projection: Projections,
     ) -> Result<Self::TransactionType<'_>, StorageError>;
 
-    fn append(&mut self, tuple: Tuple) -> Result<(), StorageError>;
+    fn append(&mut self, tuple: Tuple, is_overwrite: bool) -> Result<(), StorageError>;
 
     fn delete(&mut self, tuple_id: TupleId) -> Result<(), StorageError>;
 
@@ -68,6 +68,9 @@ pub enum StorageError {
 
     #[error("type error")]
     TypeError(#[from] TypeError),
+
+    #[error("The same primary key data already exists")]
+    DuplicatePrimaryKey,
 }
 
 impl From<KernelError> for StorageError {
diff --git a/src/types/errors.rs b/src/types/errors.rs
diff --git a/src/types/mod.rs b/src/types/mod.rs
diff --git a/src/types/value.rs b/src/types/value.rs

Original file line number	Diff line number	Diff line change
`@@ -90,9 +90,14 @@ impl<S: Storage> Binder<S> {`
`90`	`90`	`_ => todo!()`
`91`	`91`	`}`
`92`	`92`	`}`
`93`		`- Statement::Insert { table_name, columns, source, .. } => {`
	`93`	`+ Statement::Insert { table_name, columns, source, overwrite, .. } => {`
`94`	`94`	`if let SetExpr::Values(values) = source.body.as_ref() {`
`95`		`- self.bind_insert(table_name.to_owned(), columns, &values.rows).await?`
	`95`	`+ self.bind_insert(`
	`96`	`+ table_name.to_owned(),`
	`97`	`+ columns,`
	`98`	`+ &values.rows,`
	`99`	`+ *overwrite`
	`100`	`+ ).await?`
`96`	`101`	`} else {`
`97`	`102`	`todo!()`
`98`	`103`	`}`
Original file line number	Diff line number	Diff line change
`@@ -14,13 +14,15 @@ use crate::types::value::{DataValue, ValueRef};`
`14`	`14`	`pub struct Insert {`
`15`	`15`	`table_name: TableName,`
`16`	`16`	`input: BoxedExecutor,`
	`17`	`+ is_overwrite: bool`
`17`	`18`	`}`
`18`	`19`
`19`	`20`	`impl From<(InsertOperator, BoxedExecutor)> for Insert {`
`20`		`- fn from((InsertOperator { table_name }, input): (InsertOperator, BoxedExecutor)) -> Self {`
	`21`	`+ fn from((InsertOperator { table_name, is_overwrite }, input): (InsertOperator, BoxedExecutor)) -> Self {`
`21`	`22`	`Insert {`
`22`	`23`	`table_name,`
`23`		`- input`
	`24`	`+ input,`
	`25`	`+ is_overwrite,`
`24`	`26`	`}`
`25`	`27`	`}`
`26`	`28`	`}`
`@@ -34,7 +36,7 @@ impl<S: Storage> Executor<S> for Insert {`
`34`	`36`	`impl Insert {`
`35`	`37`	`#[try_stream(boxed, ok = Tuple, error = ExecutorError)]`
`36`	`38`	`pub async fn _execute<S: Storage>(self, storage: S) {`
`37`		`- let Insert { table_name, input } = self;`
	`39`	`+ let Insert { table_name, input, is_overwrite } = self;`
`38`	`40`	`let mut primary_key_index = None;`
`39`	`41`
`40`	`42`	`if let (Some(table_catalog), Some(mut table)) =`
`@@ -75,7 +77,7 @@ impl Insert {`
`75`	`77`	`tuple.values.push(value)`
`76`	`78`	`}`
`77`	`79`
`78`		`- table.append(tuple)?;`
	`80`	`+ table.append(tuple, is_overwrite)?;`
`79`	`81`	`}`
`80`	`82`	`table.commit().await?;`
`81`	`83`	`}`
Original file line number	Diff line number	Diff line change
`@@ -48,21 +48,21 @@ impl Update {`
`48`	`48`	`#[for_await]`
`49`	`49`	`for tuple in input {`
`50`	`50`	`let mut tuple = tuple?;`
	`51`	`+ let mut is_overwrite = true;`
`51`	`52`
`52`	`53`	`for (i, column) in tuple.columns.iter().enumerate() {`
`53`	`54`	`if let Some(value) = value_map.get(&column.id) {`
`54`	`55`	`if column.desc.is_primary {`
`55`	`56`	`if let Some(old_key) = tuple.id.replace(value.clone()) {`
`56`		`- if value != &old_key {`
`57`		`- table.delete(old_key)?;`
`58`		`- }`
	`57`	`+ table.delete(old_key)?;`
	`58`	`+ is_overwrite = false;`
`59`	`59`	`}`
`60`	`60`	`}`
`61`	`61`	`tuple.values[i] = value.clone();`
`62`	`62`	`}`
`63`	`63`	`}`
`64`	`64`
`65`		`- table.append(tuple)?;`
	`65`	`+ table.append(tuple, is_overwrite)?;`
`66`	`66`	`}`
`67`	`67`	`table.commit().await?;`
`68`	`68`	`}`
Original file line number	Diff line number	Diff line change
`@@ -239,13 +239,16 @@ pub enum UnaryOperator {`
`239`	`239`	`Plus,`
`240`	`240`	`Minus,`
`241`	`241`	`Not,`
`242`		`- True,`
`243`		`- False,`
`244`	`242`	`}`
`245`	`243`
`246`	`244`	`impl From<SqlUnaryOperator> for UnaryOperator {`
`247`	`245`	`fn from(value: SqlUnaryOperator) -> Self {`
`248`		`- todo!()`
	`246`	`+ match value {`
	`247`	`+ SqlUnaryOperator::Plus => UnaryOperator::Plus,`
	`248`	`+ SqlUnaryOperator::Minus => UnaryOperator::Minus,`
	`249`	`+ SqlUnaryOperator::Not => UnaryOperator::Not,`
	`250`	`+ _ => unimplemented!("not support!")`
	`251`	`+ }`
`249`	`252`	`}`
`250`	`253`	`}`
`251`	`254`
`@@ -311,7 +314,7 @@ impl From<SqlBinaryOperator> for BinaryOperator {`
`311`	`314`	`SqlBinaryOperator::And => BinaryOperator::And,`
`312`	`315`	`SqlBinaryOperator::Or => BinaryOperator::Or,`
`313`	`316`	`SqlBinaryOperator::Xor => BinaryOperator::Xor,`
`314`		`- _ => todo!()`
	`317`	`+ _ => unimplemented!("not support!")`
`315`	`318`	`}`
`316`	`319`	`}`
`317`	`320`	`}`