feat(create): create table in volcano

- bind column ref

Author: loloxwg

rust-toolchain

@@ -0,0 +1 @@
+nightly

src/binder/expr.rs

@@ -1,11 +1,78 @@
+use crate::binder::BindError;
 use anyhow::Result;
-use sqlparser::ast::Expr;
+use itertools::Itertools;
+use sqlparser::ast::{Expr, Ident};
+use std::slice;
 
 use super::Binder;
 use crate::expression::ScalarExpression;
 
 impl Binder {
     pub(crate) fn bind_expr(&mut self, expr: &Expr) -> Result<ScalarExpression> {
-        todo!()
+        match expr {
+            Expr::Identifier(ident) => {
+                self.bind_column_ref_from_identifiers(slice::from_ref(ident))
+            }
+            _ => {
+                todo!()
+            }
+        }
+    }
+
+    pub fn bind_column_ref_from_identifiers(
+        &mut self,
+        idents: &[Ident],
+    ) -> Result<ScalarExpression> {
+        let idents = idents
+            .iter()
+            .map(|ident| Ident::new(ident.value.to_lowercase()))
+            .collect_vec();
+        let (_schema_name, table_name, column_name) = match idents.as_slice() {
+            [column] => (None, None, &column.value),
+            [table, column] => (None, Some(&table.value), &column.value),
+            [schema, table, column] => (Some(&schema.value), Some(&table.value), &column.value),
+            _ => {
+                return Err(BindError::InvalidColumn(
+                    idents
+                        .iter()
+                        .map(|ident| ident.value.clone())
+                        .join(".")
+                        .to_string(),
+                )
+                .into())
+            }
+        };
+
+        if let Some(table) = table_name {
+            let table_catalog = self
+                .context
+                .catalog
+                .get_table_by_name(table)
+                .ok_or_else(|| BindError::InvalidTable(table.to_string()))?;
+
+            let column_catalog = table_catalog
+                .get_column_by_name(column_name)
+                .ok_or_else(|| BindError::InvalidColumn(column_name.to_string()))?;
+            Ok(ScalarExpression::ColumnRef(column_catalog.clone()))
+        } else {
+            // handle col syntax
+            let mut got_column = None;
+            for table_catalog in self.context.catalog.tables.values() {
+                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());
+                    }
+                    got_column = Some(column_catalog);
+                }
+            }
+            if got_column.is_none() {
+                if let Some(expr) = self.context.aliases.get(column_name) {
+                    return Ok(expr.clone());
+                }
+            }
+            let column_catalog =
+                got_column.ok_or_else(|| BindError::InvalidColumn(column_name.to_string()))?;
+            Ok(ScalarExpression::ColumnRef(column_catalog.clone()))
+        }
     }
 }

src/catalog/root.rs

@@ -5,8 +5,8 @@ use crate::types::TableId;
 
 #[derive(Debug, Clone)]
 pub struct RootCatalog {
-    table_idxs: BTreeMap<String, TableId>,
-    tables: BTreeMap<TableId, TableCatalog>,
+    pub table_idxs: BTreeMap<String, TableId>,
+    pub tables: BTreeMap<TableId, TableCatalog>,
 }
 
 impl Default for RootCatalog {

src/catalog/table.rs

@@ -22,6 +22,11 @@ impl TableCatalog {
         self.column_idxs.get(name).cloned()
     }
 
+    pub(crate) fn get_column_by_name(&self, name: &str) -> Option<&ColumnCatalog> {
+        let id = self.column_idxs.get(name)?;
+        self.columns.get(id)
+    }
+
     pub(crate) fn contains_column(&self, name: &str) -> bool {
         self.column_idxs.contains_key(name)
     }

src/db.rs

@@ -164,6 +164,18 @@ mod test {
             let batch = database.run("select * from t1").await?;
             println!("{:#?}", batch);
 
+            Ok(())
+        })
+    }
+    #[test]
+    fn test_sql() -> anyhow::Result<()> {
+        let database = Database::new_on_mem();
+
+        tokio_test::block_on(async move {
+            let _batch = database.run("create table t1 (a int)").await?;
+            let batch = database.run("select a from t1").await?;
+            println!("{:#?}", batch);
+
             Ok(())
         })
     }

src/execution_v1/physical_plan/mod.rs

@@ -1,11 +1,14 @@
+use crate::execution_v1::physical_plan::physical_create_table::PhysicalCreateTable;
 use crate::execution_v1::physical_plan::physical_projection::PhysicalProjection;
 use crate::execution_v1::physical_plan::physical_table_scan::PhysicalTableScan;
 
-pub(crate) mod physical_table_scan;
-pub(crate) mod physical_projection;
+pub(crate) mod physical_create_table;
 pub(crate) mod physical_plan_builder;
+pub(crate) mod physical_projection;
+pub(crate) mod physical_table_scan;
 
 pub enum PhysicalOperator {
+    CreateTable(PhysicalCreateTable),
     TableScan(PhysicalTableScan),
-    Projection(PhysicalProjection)
+    Projection(PhysicalProjection),
 }

src/execution_v1/physical_plan/physical_create_table.rs

@@ -0,0 +1,10 @@
+use crate::catalog::ColumnDesc;
+use crate::execution_v1::physical_plan::PhysicalOperator;
+use crate::expression::ScalarExpression;
+
+pub struct PhysicalCreateTable {
+    /// Table name to insert to
+    pub table_name: String,
+    /// List of columns of the table
+    pub columns: Vec<(String, bool, ColumnDesc)>,
+}

src/execution_v1/physical_plan/physical_plan_builder.rs

@@ -1,13 +1,14 @@
-use std::sync::Arc;
-use anyhow::anyhow;
-use anyhow::Result;
+use crate::execution_v1::physical_plan::physical_create_table::PhysicalCreateTable;
 use crate::execution_v1::physical_plan::physical_projection::PhysicalProjection;
 use crate::execution_v1::physical_plan::physical_table_scan::PhysicalTableScan;
 use crate::execution_v1::physical_plan::PhysicalOperator;
+use crate::planner::logical_create_table_plan::LogicalCreateTablePlan;
 use crate::planner::logical_select_plan::LogicalSelectPlan;
-use crate::planner::LogicalPlan;
-use crate::planner::operator::Operator;
 use crate::planner::operator::scan::ScanOperator;
+use crate::planner::operator::Operator;
+use crate::planner::LogicalPlan;
+use anyhow::anyhow;
+use anyhow::Result;
 
 pub struct PhysicalPlanBuilder {
     plan_id: u32,
@@ -27,10 +28,22 @@ impl PhysicalPlanBuilder {
     pub fn build_plan(&mut self, plan: &LogicalPlan) -> Result<PhysicalOperator> {
         match plan {
             LogicalPlan::Select(select) => self.build_select_logical_plan(select),
-            LogicalPlan::CreateTable(_) => todo!(),
+            LogicalPlan::CreateTable(create_table) => {
+                self.build_create_table_logic_plan(create_table)
+            }
         }
     }
 
+    fn build_create_table_logic_plan(
+        &mut self,
+        plan: &LogicalCreateTablePlan,
+    ) -> Result<PhysicalOperator> {
+        Ok(PhysicalOperator::CreateTable(PhysicalCreateTable {
+            table_name: plan.table_name.to_string(),
+            columns: plan.columns.clone(),
+        }))
+    }
+
     fn build_select_logical_plan(&mut self, plan: &LogicalSelectPlan) -> Result<PhysicalOperator> {
         match plan.operator.as_ref() {
             Operator::Project(op) => {

src/execution_v1/volcano_executor/create_table.rs

@@ -0,0 +1,30 @@
+use crate::catalog::ColumnCatalog;
+use crate::execution_v1::physical_plan::physical_create_table::PhysicalCreateTable;
+use crate::execution_v1::ExecutorError;
+use crate::storage::Storage;
+use arrow::datatypes::{Schema, SchemaRef};
+use arrow::record_batch::RecordBatch;
+use futures_async_stream::try_stream;
+use std::sync::Arc;
+
+pub struct CreateTable {}
+
+impl CreateTable {
+    #[try_stream(boxed, ok = RecordBatch , error = ExecutorError)]
+    pub async fn execute(plan: PhysicalCreateTable, storage: impl Storage) {
+        let mut columns = Vec::new();
+        plan.columns.iter().for_each(|c| {
+            columns.push(ColumnCatalog::new(c.0.clone(), c.1, c.2.clone()));
+        });
+        let table_name = plan.table_name.clone();
+        // columns->batch record
+        let mut data = Vec::new();
+
+        columns.iter().for_each(|c| {
+            let batch = RecordBatch::new_empty(Arc::new(Schema::new(vec![c.to_field()])));
+            data.push(batch);
+        });
+
+        storage.create_table(plan.table_name.as_str(), data);
+    }
+}

src/execution_v1/volcano_executor/mod.rs

@@ -1,15 +1,17 @@
-mod table_scan;
+mod create_table;
 mod projection;
+mod table_scan;
 
-use arrow::record_batch::RecordBatch;
-use futures::stream::BoxStream;
-use futures::TryStreamExt;
-use crate::execution_v1::ExecutorError;
 use crate::execution_v1::physical_plan::physical_projection::PhysicalProjection;
 use crate::execution_v1::physical_plan::PhysicalOperator;
+use crate::execution_v1::volcano_executor::create_table::CreateTable;
 use crate::execution_v1::volcano_executor::projection::Projection;
 use crate::execution_v1::volcano_executor::table_scan::TableScan;
+use crate::execution_v1::ExecutorError;
 use crate::storage::StorageImpl;
+use arrow::record_batch::RecordBatch;
+use futures::stream::BoxStream;
+use futures::TryStreamExt;
 
 pub type BoxedExecutor = BoxStream<'static, Result<RecordBatch, ExecutorError>>;
 
@@ -33,6 +35,12 @@ impl VolcanoExecutor {
                 let input = self.build(*input);
                 Projection::execute(exprs, input)
             }
+            PhysicalOperator::CreateTable(op) => match &self.storage {
+                StorageImpl::InMemoryStorage(storage) => CreateTable::execute(op, storage.clone()),
+            },
+            _ => {
+                unimplemented!()
+            }
         }
     }