Introduce KipDB for persistence (#54)

* feat: `Tuple` implement serialization

* feat: implement `TableCodec`

* refactor: remove `TableCatalog::table_id`

* refactor: remove `Storage::tables`

* refactor: `Binder` and `Storage` are asynchronous

* feat: Introduce `KipDB` for persistence

* fix: repair `quit` command

Author: KKould

Cargo.toml

@@ -18,6 +18,7 @@ chrono = "0.4.26"
 tokio = { version = "1.28.2", features = ["full"] }
 serde = { version = "1", features = ["derive", "rc"] }
 serde_json = "1"
+bincode = "1.3.3"
 async-trait = "0.1.68"
 integer-encoding = "3.0.4"
 strum_macros = "0.24"
@@ -28,13 +29,17 @@ futures = "0.3.25"
 ahash = "0.8.3"
 lazy_static = "1.4.0"
 comfy-table = "7.0.1"
+bytes = "*"
+kip_db = "0.1.2-alpha.10"
+async-recursion = "1.0.5"
 
 [dev-dependencies]
 tokio-test = "0.4.2"
 ctor = "0.2.0"
 env_logger = "0.10"
 paste = "^1.0"
 rstest = "0.17"
+tempfile = "3.0.7"
 
 [workspace]
 members = [

src/binder/aggregate.rs

@@ -32,28 +32,28 @@ impl Binder {
         Ok(())
     }
 
-    pub fn extract_group_by_aggregate(
+    pub async fn extract_group_by_aggregate(
         &mut self,
         select_list: &mut [ScalarExpression],
         groupby: &[Expr],
     ) -> Result<(), BindError> {
-        self.validate_groupby_illegal_column(select_list, groupby)?;
+        self.validate_groupby_illegal_column(select_list, groupby).await?;
 
         for gb in groupby {
-            let mut expr = self.bind_expr(gb)?;
+            let mut expr = self.bind_expr(gb).await?;
             self.visit_group_by_expr(select_list, &mut expr);
         }
         Ok(())
     }
 
-    pub fn extract_having_orderby_aggregate(
+    pub async fn extract_having_orderby_aggregate(
         &mut self,
         having: &Option<Expr>,
         orderbys: &[OrderByExpr],
     ) -> Result<(Option<ScalarExpression>, Option<Vec<SortField>>), BindError> {
         // Extract having expression.
         let return_having = if let Some(having) = having {
-            let mut having = self.bind_expr(having)?;
+            let mut having = self.bind_expr(having).await?;
             self.visit_column_agg_expr(&mut having);
             Some(having)
         } else {
@@ -69,7 +69,7 @@ impl Binder {
                     asc,
                     nulls_first,
                 } = orderby;
-                let mut expr = self.bind_expr(expr)?;
+                let mut expr = self.bind_expr(expr).await?;
                 self.visit_column_agg_expr(&mut expr);
 
                 return_orderby.push(SortField::new(
@@ -135,14 +135,14 @@ impl Binder {
     /// e.g. SELECT a,count(b) FROM t GROUP BY a. it's ok.
     ///      SELECT a,b FROM t GROUP BY a.        it's error.
     ///      SELECT a,count(b) FROM t GROUP BY b. it's error.
-    fn validate_groupby_illegal_column(
+    async fn validate_groupby_illegal_column(
         &mut self,
         select_items: &[ScalarExpression],
         groupby: &[Expr],
     ) -> Result<(), BindError> {
         let mut group_raw_exprs = vec![];
         for expr in groupby {
-            let expr = self.bind_expr(expr)?;
+            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 {

src/binder/create_table.rs

@@ -1,10 +1,11 @@
 use std::collections::HashSet;
 use std::sync::Arc;
+use itertools::Itertools;
 use sqlparser::ast::{ColumnDef, ObjectName};
 
 use super::Binder;
 use crate::binder::{BindError, lower_case_name, split_name};
-use crate::catalog::{ColumnCatalog, ColumnRef};
+use crate::catalog::ColumnCatalog;
 use crate::planner::LogicalPlan;
 use crate::planner::operator::create_table::CreateTableOperator;
 use crate::planner::operator::Operator;
@@ -16,7 +17,8 @@ impl Binder {
         columns: &[ColumnDef],
     ) -> Result<LogicalPlan, BindError> {
         let name = lower_case_name(&name);
-        let (_, table_name) = split_name(&name)?;
+        let (_, name) = split_name(&name)?;
+        let table_name = Arc::new(name.to_string());
 
         // check duplicated column names
         let mut set = HashSet::new();
@@ -27,15 +29,15 @@ impl Binder {
             }
         }
 
-        let columns: Vec<ColumnRef> = columns
+        let columns = columns
             .iter()
-            .map(|col| Arc::new(ColumnCatalog::from(col.clone())))
-            .collect();
+            .map(|col| ColumnCatalog::from(col.clone()))
+            .collect_vec();
 
         let plan = LogicalPlan {
             operator: Operator::CreateTable(
                 CreateTableOperator {
-                    table_name: table_name.to_string(),
+                    table_name,
                     columns
                 }
             ),
@@ -47,21 +49,26 @@ impl Binder {
 
 #[cfg(test)]
 mod tests {
+    use tempfile::TempDir;
     use super::*;
     use crate::binder::BinderContext;
-    use crate::catalog::{ColumnDesc, RootCatalog};
+    use crate::catalog::ColumnDesc;
+    use crate::storage::kip::KipStorage;
     use crate::types::LogicalType;
 
-    #[test]
-    fn test_create_bind() {
+    #[tokio::test]
+    async fn test_create_bind() {
+        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 binder = Binder::new(BinderContext::new(RootCatalog::new()));
+        let binder = Binder::new(BinderContext::new(storage));
         let stmt = crate::parser::parse_sql(sql).unwrap();
-        let plan1 = binder.bind(&stmt[0]).unwrap();
+        let plan1 = binder.bind(&stmt[0]).await.unwrap();
 
         match plan1.operator {
             Operator::CreateTable(op) => {
-                assert_eq!(op.table_name, "t1".to_string());
+                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));

src/binder/expr.rs

@@ -3,34 +3,37 @@ use itertools::Itertools;
 use sqlparser::ast::{BinaryOperator, Expr, Function, FunctionArg, FunctionArgExpr, Ident};
 use std::slice;
 use std::sync::Arc;
+use async_recursion::async_recursion;
 use crate::expression::agg::AggKind;
 
 use super::Binder;
 use crate::expression::ScalarExpression;
+use crate::storage::Storage;
 use crate::types::LogicalType;
 
 impl Binder {
-    pub(crate) fn bind_expr(&mut self, expr: &Expr) -> Result<ScalarExpression, BindError> {
+    #[async_recursion]
+    pub(crate) async fn bind_expr(&mut self, expr: &Expr) -> Result<ScalarExpression, BindError> {
         match expr {
             Expr::Identifier(ident) => {
-                self.bind_column_ref_from_identifiers(slice::from_ref(ident), None)
+                self.bind_column_ref_from_identifiers(slice::from_ref(ident), None).await
             }
             Expr::CompoundIdentifier(idents) => {
-                self.bind_column_ref_from_identifiers(idents, None)
+                self.bind_column_ref_from_identifiers(idents, None).await
             }
             Expr::BinaryOp { left, right, op} => {
-                self.bind_binary_op_internal(left, right, op)
+                self.bind_binary_op_internal(left, right, op).await
             }
             Expr::Value(v) => Ok(ScalarExpression::Constant(Arc::new(v.into()))),
-            Expr::Function(func) => self.bind_agg_call(func),
-            Expr::Nested(expr) => self.bind_expr(expr),
+            Expr::Function(func) => self.bind_agg_call(func).await,
+            Expr::Nested(expr) => self.bind_expr(expr).await,
             _ => {
                 todo!()
             }
         }
     }
 
-    pub fn bind_column_ref_from_identifiers(
+    pub async fn bind_column_ref_from_identifiers(
         &mut self,
         idents: &[Ident],
         bind_table_name: Option<&String>,
@@ -58,8 +61,9 @@ impl Binder {
         if let Some(table) = table_name.or(bind_table_name) {
             let table_catalog = self
                 .context
-                .catalog
-                .get_table_by_name(table)
+                .storage
+                .table_catalog(table)
+                .await
                 .ok_or_else(|| BindError::InvalidTable(table.to_string()))?;
 
             let column_catalog = table_catalog
@@ -69,7 +73,7 @@ impl Binder {
         } else {
             // handle col syntax
             let mut got_column = None;
-            for (_, table_catalog) in self.context.catalog.tables() {
+            for (_, (table_catalog, _)) in &self.context.bind_table {
                 if let Some(column_catalog) = table_catalog.get_column_by_name(column_name) {
                     if got_column.is_some() {
                         return Err(BindError::InvalidColumn(column_name.to_string()).into());
@@ -88,14 +92,14 @@ impl Binder {
         }
     }
 
-    fn bind_binary_op_internal(
+    async fn bind_binary_op_internal(
         &mut self,
         left: &Expr,
         right: &Expr,
         op: &BinaryOperator,
     ) -> Result<ScalarExpression, BindError> {
-        let left_expr = Box::new(self.bind_expr(left)?);
-        let right_expr = Box::new(self.bind_expr(right)?);
+        let left_expr = Box::new(self.bind_expr(left).await?);
+        let right_expr = Box::new(self.bind_expr(right).await?);
 
         let ty = match op {
             BinaryOperator::Plus | BinaryOperator::Minus | BinaryOperator::Multiply |
@@ -121,20 +125,19 @@ impl Binder {
         })
     }
 
-    fn bind_agg_call(&mut self, func: &Function) -> Result<ScalarExpression, BindError> {
-        let args: Vec<ScalarExpression> = func.args
-            .iter()
-            .map(|arg| {
-                let arg_expr = match arg {
-                    FunctionArg::Named { arg, .. } => arg,
-                    FunctionArg::Unnamed(arg) => arg,
-                };
-                match arg_expr {
-                    FunctionArgExpr::Expr(expr) => self.bind_expr(expr),
-                    _ => todo!()
-                }
-            })
-            .try_collect()?;
+    async fn bind_agg_call(&mut self, func: &Function) -> Result<ScalarExpression, BindError> {
+        let mut args = Vec::with_capacity(func.args.len());
+
+        for arg in func.args.iter() {
+            let arg_expr = match arg {
+                FunctionArg::Named { arg, .. } => arg,
+                FunctionArg::Unnamed(arg) => arg,
+            };
+            match arg_expr {
+                FunctionArgExpr::Expr(expr) => args.push(self.bind_expr(expr).await?),
+                _ => todo!()
+            }
+        }
         let ty = args[0].return_type();
 
         Ok(match func.name.to_string().to_lowercase().as_str() {

src/binder/insert.rs

@@ -1,65 +1,64 @@
 use std::slice;
+use std::sync::Arc;
 use sqlparser::ast::{Expr, Ident, ObjectName};
-use itertools::Itertools;
 use crate::binder::{Binder, BindError, lower_case_name, split_name};
 use crate::catalog::ColumnRef;
 use crate::expression::ScalarExpression;
 use crate::planner::LogicalPlan;
 use crate::planner::operator::insert::InsertOperator;
 use crate::planner::operator::Operator;
 use crate::planner::operator::values::ValuesOperator;
+use crate::storage::Storage;
 use crate::types::value::ValueRef;
 
 impl Binder {
-    pub(crate) fn bind_insert(
+    pub(crate) async fn bind_insert(
         &mut self,
         name: ObjectName,
         idents: &[Ident],
-        rows: &Vec<Vec<Expr>>
+        expr_rows: &Vec<Vec<Expr>>
     ) -> Result<LogicalPlan, BindError> {
         let name = lower_case_name(&name);
-        let (_, table_name) = split_name(&name)?;
+        let (_, name) = split_name(&name)?;
+        let table_name = Arc::new(name.to_string());
 
-        if let Some(table) = self.context.catalog.get_table_by_name(table_name) {
-            let table_id = table.id;
+        if let Some(table) = self.context.storage.table_catalog(&table_name).await {
             let mut columns = Vec::new();
 
             if idents.is_empty() {
-                columns = table.all_columns()
-                    .into_iter()
-                    .map(|(_, catalog)| catalog.clone())
-                    .collect_vec();
+                columns = table.all_columns();
             } else {
                 let bind_table_name = Some(table_name.to_string());
                 for ident in idents {
                     match self.bind_column_ref_from_identifiers(
                         slice::from_ref(ident),
                         bind_table_name.as_ref()
-                    )? {
+                    ).await? {
                         ScalarExpression::ColumnRef(catalog) => columns.push(catalog),
                         _ => unreachable!()
                     }
                 }
             }
+            let mut rows = Vec::with_capacity(expr_rows.len());
 
-            let rows = rows
-                .into_iter()
-                .map(|row| {
-                    row.into_iter()
-                        .map(|expr| match self.bind_expr(expr)? {
-                            ScalarExpression::Constant(value) => Ok::<ValueRef, BindError>(value),
-                            _ => unreachable!(),
-                        })
-                        .try_collect()
-                })
-                .try_collect()?;
+            for expr_row in expr_rows {
+                let mut row = Vec::with_capacity(expr_row.len());
 
+                for expr in expr_row {
+                    match self.bind_expr(expr).await? {
+                        ScalarExpression::Constant(value) => row.push(value),
+                        _ => unreachable!(),
+                    }
+                }
+
+                rows.push(row);
+            }
             let values_plan = self.bind_values(rows, columns);
 
             Ok(LogicalPlan {
                 operator: Operator::Insert(
                     InsertOperator {
-                        table_id,
+                        table_name,
                     }
                 ),
                 childrens: vec![values_plan],