Feat/dml delete and Remove PhysicalPlan (#56)

* feat: implement dml delete

* fix: data is inserted after deletion and the search is empty

* perf: kipdb version up

* ci

* refactor: Remove `PhysicalPlan` and convert `LogicPlan` to `Executor`

* feat: remove the coupling between DataBase and Storage

* docs: add Storage support statement and code usage instructions

* feat: Implement ddl `Drop`

* feat: Implement ddl `Truncate`

* feat: Implement dql `Count`

* feat: Implement dql `Sum`

* feat: Implement dql `Avg`\`Min`\`Max`

* feat: Implement `GroupBy` clause on dql

* fix(rbo): losing `InputRef` when pushing down `Project`

* refactor: primary key id is actively specified by the user

* fix: 'Alias' fails under aggregate function

* feat: Support `DateTime` type

Author: KKould

.github/workflows/ci.yml

@@ -1,28 +1,55 @@
-name: Rust
+name: CI
 
 on:
   push:
-    branches: [main]
   pull_request:
-    branches: [main]
 
 env:
   CARGO_TERM_COLOR: always
+  CARGO_REGISTRIES_MY_REGISTRY_INDEX: https://github.com/rust-lang/crates.io-index
 
 jobs:
-  build:
+  check:
+    runs-on: ubuntu-20.04
+    steps:
+      - uses: actions/checkout@v2
+      - uses: actions-rs/toolchain@v1
+        with:
+          profile: minimal
+          toolchain: nightly
+          components: rustfmt, clippy
+      - name: Check code format
+        uses: actions-rs/cargo@v1
+        with:
+          command: fmt
+          args: --all -- --check
 
-    runs-on: ubuntu-latest
 
+  build:
+    runs-on: ubuntu-20.04
     steps:
       - uses: actions/checkout@v2
-      - name: Setup Rust
-        uses: actions-rs/toolchain@v1
+      - uses: actions-rs/toolchain@v1
         with:
+          profile: minimal
           toolchain: nightly-2023-04-07
-          override: true
+      - uses: actions/checkout@v2
+      - name: Build
+        uses: actions-rs/cargo@v1
+        with:
+          command: build
 
-      - name: Run tests
+  test:
+    runs-on: ubuntu-20.04
+    steps:
+      - uses: actions/checkout@v2
+      - uses: actions-rs/toolchain@v1
+        with:
+          profile: minimal
+          toolchain: nightly-2023-04-07
+      - uses: actions/checkout@v2
+      - name: Test
         uses: actions-rs/cargo@v1
         with:
-          command: test
+          command: test
+          args: --release --no-fail-fast

Cargo.toml

@@ -30,7 +30,7 @@ ahash = "0.8.3"
 lazy_static = "1.4.0"
 comfy-table = "7.0.1"
 bytes = "*"
-kip_db = "0.1.2-alpha.10"
+kip_db = "0.1.2-alpha.15"
 async-recursion = "1.0.5"
 
 [dev-dependencies]

README.md

@@ -11,39 +11,62 @@ Install rust toolchain first.
 cargo run
 ```
 test command
-```mysql
+```sql
 create table t1 (a int, b int);
 
 insert into t1 (a, b) values (1, 1), (5, 3), (5, 2);
 
+update t1 set a = 0 where b > 1;
+
+delete from t1 where b > 1;
+
 select * from t1;
 
 select * from t1 order by a asc nulls first
+
+select count(distinct a) from t1;
+
+truncate table t1;
+
+drop table t1;
 ```
+Using KipSQL in code
+```rust
+let kipsql = Database::with_kipdb("./data").await?;
+
+let tupes = db.run("select * from t1").await?;
+```
+Storage Support:
+- KipDB
+- Memory
 
 ![demo](./static/images/demo.png)
 
 ### Features
 - DDL
   - Create
-    - [x] CreateTable
-    - [ ] CreateIndex
+    - [x] Table
+    - [ ] Index
   - Drop
+    - [x] Table
+    - [ ] Index
+  - [x] Truncate
 - DQL
   - [x] Select
   - [x] Where
   - [ ] Distinct
-  - [ ] Aggregation: count()/sum()/avg()/min()/max()
+  - [x] Alias
+  - [x] Aggregation: count()/sum()/avg()/min()/max()
   - [ ] Subquery
   - [x] Join: Inner/Left/Right/Full Cross(x)
-  - [ ] Group By
-  - [ ] Having
+  - [x] Group By
+  - [x] Having
   - [x] Order By
   - [x] Limit
 - DML
   - [x] Insert
   - [x] Update
-  - [ ] Delete
+  - [x] Delete
 - DataTypes
   - Invalid
   - SqlNull
@@ -59,6 +82,7 @@ select * from t1 order by a asc nulls first
   - Float
   - Double
   - Varchar
+  - DateTime
 - Optimizer rules
   - Limit Project Transpose
   - Eliminate Limits
@@ -67,18 +91,6 @@ select * from t1 order by a asc nulls first
   - Combine Filters
   - Column Pruning
   - Collapse Project
-- Executors
-  - [x] CreateTable
-  - [x] SeqScan
-  - [ ] IndexScan
-  - [x] Filter
-  - [x] Project
-  - [x] Limit
-  - [x] Hash Join
-  - [x] Insert
-  - [x] Values
-  - [x] Update
-  - [ ] Delete
 
 ### Thanks For
 - [Fedomn/sqlrs](https://github.com/Fedomn/sqlrs): 主要参考资料，Optimizer、Executor均参考自sqlrs的设计

src/binder/aggregate.rs

@@ -1,3 +1,5 @@
+use std::collections::HashSet;
+use ahash::RandomState;
 use itertools::Itertools;
 use sqlparser::ast::{Expr, OrderByExpr};
 
@@ -7,12 +9,13 @@ use crate::{
         operator::{aggregate::AggregateOperator, sort::SortField},
     },
 };
-use crate::binder::BindError;
+use crate::binder::{BindError, InputRefType};
 use crate::planner::LogicalPlan;
+use crate::storage::Storage;
 
 use super::Binder;
 
-impl Binder {
+impl<S: Storage> Binder<S> {
     pub fn bind_aggregate(
         &mut self,
         children: LogicalPlan,
@@ -90,7 +93,7 @@ impl Binder {
             ScalarExpression::AggCall {
                 ty: return_type, ..
             } => {
-                let index = self.context.agg_calls.len();
+                let index = self.context.input_ref_index(InputRefType::AggCall);
                 let input_ref = ScalarExpression::InputRef {
                     index,
                     ty: return_type.clone(),
@@ -99,12 +102,14 @@ impl Binder {
                     ScalarExpression::AggCall {
                         kind,
                         args,
-                        ty: return_type,
+                        ty,
+                        distinct
                     } => {
                         self.context.agg_calls.push(ScalarExpression::AggCall {
+                            distinct,
                             kind,
                             args,
-                            ty: return_type,
+                            ty,
                         });
                     }
                     _ => unreachable!(),
@@ -114,6 +119,7 @@ impl Binder {
             ScalarExpression::TypeCast { expr, .. } => self.visit_column_agg_expr(expr),
             ScalarExpression::IsNull { expr } => self.visit_column_agg_expr(expr),
             ScalarExpression::Unary { expr, .. } => self.visit_column_agg_expr(expr),
+            ScalarExpression::Alias { expr, .. } => self.visit_column_agg_expr(expr),
             ScalarExpression::Binary {
                 left_expr,
                 right_expr,
@@ -122,11 +128,9 @@ impl Binder {
                 self.visit_column_agg_expr(left_expr);
                 self.visit_column_agg_expr(right_expr);
             }
-
             ScalarExpression::Constant(_)
             | ScalarExpression::ColumnRef { .. }
-            | ScalarExpression::InputRef { .. }
-            | ScalarExpression::Alias { .. } => {}
+            | ScalarExpression::InputRef { .. } => {}
         }
     }
 
@@ -143,6 +147,7 @@ impl Binder {
         let mut group_raw_exprs = vec![];
         for expr in groupby {
             let expr = self.bind_expr(expr).await?;
+
             if let ScalarExpression::Alias { alias, .. } = expr {
                 let alias_expr = select_items.iter().find(|column| {
                     if let ScalarExpression::Alias {
@@ -162,22 +167,30 @@ impl Binder {
                 group_raw_exprs.push(expr);
             }
         }
+        let mut group_raw_set: HashSet<&ScalarExpression, RandomState> = HashSet::from_iter(group_raw_exprs.iter());
 
-        for column in select_items {
-            let expr = &column;
-            if expr.has_agg_call() {
+        for expr in select_items {
+            if expr.has_agg_call(&self.context) {
                 continue;
             }
+            group_raw_set.remove(expr);
 
             if !group_raw_exprs.iter().contains(expr) {
                 return Err(BindError::AggMiss(
                     format!(
-                        "{} must appear in the GROUP BY clause or be used in an aggregate function",
+                        "{:?} must appear in the GROUP BY clause or be used in an aggregate function",
                         expr
                     )
                 ));
             }
         }
+
+        if !group_raw_set.is_empty() {
+            return Err(BindError::AggMiss(
+                format!("In the GROUP BY clause the field must be in the select clause")
+            ));
+        }
+
         Ok(())
     }
 
@@ -197,14 +210,10 @@ impl Binder {
                     false
                 }
             }) {
-                let index = if self.context.group_by_exprs.len() == 0 {
-                    0
-                } else {
-                    self.context.group_by_exprs.len() + 1
-                };
-
+                let index = self.context.input_ref_index(InputRefType::GroupBy);
                 let mut select_item = &mut select_list[i];
                 let return_type = select_item.return_type();
+
                 self.context.group_by_exprs.push(std::mem::replace(
                     &mut select_item,
                     ScalarExpression::InputRef {
@@ -223,11 +232,8 @@ impl Binder {
                     self.context.group_by_exprs.push(expr.clone())
                 }
                 _ => {
-                    let index = if self.context.group_by_exprs.len() == 0 {
-                        0
-                    } else {
-                        self.context.group_by_exprs.len() + 1
-                    };
+                    let index = self.context.input_ref_index(InputRefType::GroupBy);
+
                     self.context.group_by_exprs.push(std::mem::replace(
                         expr,
                         ScalarExpression::InputRef {
@@ -256,7 +262,7 @@ impl Binder {
 
                 Err(BindError::AggMiss(
                     format!(
-                        "column {} must appear in the GROUP BY clause or be used in an aggregate function",
+                        "column {:?} must appear in the GROUP BY clause or be used in an aggregate function",
                         expr
                     )
                 ))
@@ -268,7 +274,7 @@ impl Binder {
 
                 Err(BindError::AggMiss(
                     format!(
-                        "column {} must appear in the GROUP BY clause or be used in an aggregate function",
+                        "column {:?} must appear in the GROUP BY clause or be used in an aggregate function",
                         expr
                     )
                 ))

src/binder/create_table.rs

@@ -9,8 +9,10 @@ use crate::catalog::ColumnCatalog;
 use crate::planner::LogicalPlan;
 use crate::planner::operator::create_table::CreateTableOperator;
 use crate::planner::operator::Operator;
+use crate::storage::Storage;
+use crate::types::LogicalType;
 
-impl Binder {
+impl<S: Storage> Binder<S> {
     pub(crate) fn bind_create_table(
         &mut self,
         name: &ObjectName,
@@ -34,6 +36,14 @@ impl Binder {
             .map(|col| ColumnCatalog::from(col.clone()))
             .collect_vec();
 
+        if let Some(col) = columns.iter().find(|col| col.desc.is_primary) {
+            if !matches!(col.datatype(), LogicalType::Integer) {
+                return Err(BindError::InvalidColumn("Primary key types only support signed integers".to_string()));
+            }
+        } else {
+            return Err(BindError::InvalidTable("At least one primary key field exists".to_string()));
+        }
+
         let plan = LogicalPlan {
             operator: Operator::CreateTable(
                 CreateTableOperator {
@@ -61,7 +71,7 @@ mod tests {
         let temp_dir = TempDir::new().expect("unable to create temporary working directory");
         let storage = KipStorage::new(temp_dir.path()).await.unwrap();
 
-        let sql = "create table t1 (id int , name varchar(10) null)";
+        let sql = "create table t1 (id int primary key, name varchar(10) null)";
         let binder = Binder::new(BinderContext::new(storage));
         let stmt = crate::parser::parse_sql(sql).unwrap();
         let plan1 = binder.bind(&stmt[0]).await.unwrap();
@@ -71,7 +81,7 @@ mod tests {
                 assert_eq!(op.table_name, Arc::new("t1".to_string()));
                 assert_eq!(op.columns[0].name, "id".to_string());
                 assert_eq!(op.columns[0].nullable, false);
-                assert_eq!(op.columns[0].desc, ColumnDesc::new(LogicalType::Integer, false));
+                assert_eq!(op.columns[0].desc, ColumnDesc::new(LogicalType::Integer, true));
                 assert_eq!(op.columns[1].name, "name".to_string());
                 assert_eq!(op.columns[1].nullable, true);
                 assert_eq!(op.columns[1].desc, ColumnDesc::new(LogicalType::Varchar, false));