Merge branch 'KipData:main' into main

Author: GE-fighting

src/binder/expr.rs

@@ -95,10 +95,22 @@ impl Binder {
     ) -> Result<ScalarExpression> {
         let left_expr = Box::new(self.bind_expr(left)?);
         let right_expr = Box::new(self.bind_expr(right)?);
-        let ty = LogicalType::max_logical_type(
-            &left_expr.return_type(),
-            &right_expr.return_type()
-        )?;
+
+        let ty = match op {
+            BinaryOperator::Plus | BinaryOperator::Minus | BinaryOperator::Multiply |
+            BinaryOperator::Divide | BinaryOperator::Modulo => {
+                LogicalType::max_logical_type(
+                    &left_expr.return_type(),
+                    &right_expr.return_type()
+                )?
+            }
+            BinaryOperator::Gt | BinaryOperator::Lt | BinaryOperator::GtEq |
+            BinaryOperator::LtEq | BinaryOperator::Eq | BinaryOperator::NotEq |
+            BinaryOperator::And | BinaryOperator::Or | BinaryOperator::Xor => {
+                LogicalType::Boolean
+            },
+            _ => todo!()
+        };
 
         Ok(ScalarExpression::Binary {
             op: (op.clone()).into(),

src/binder/select.rs

@@ -324,13 +324,8 @@ mod tests {
     use super::*;
     use crate::binder::BinderContext;
     use crate::catalog::{ColumnCatalog, ColumnDesc, RootCatalog};
-    use crate::expression::agg::AggKind;
-    use crate::expression::BinaryOperator::{Gt, Minus};
-    use crate::expression::ScalarExpression::{AggCall, Binary, ColumnRef, Constant, InputRef};
     use crate::planner::LogicalPlan;
-    use crate::planner::operator::aggregate::AggregateOperator;
     use crate::types::LogicalType::Integer;
-    use crate::types::value::DataValue::Int32;
 
     fn test_root_catalog() -> Result<RootCatalog> {
         let mut root = RootCatalog::new();

src/db.rs

@@ -45,45 +45,18 @@ impl Database {
         ///     Limit(1)
         ///       Project(a,b)
         let logical_plan = binder.bind(&stmts[0])?;
-        println!("logic plan: {:#?}", logical_plan);
+        // println!("logic plan: {:#?}", logical_plan);
 
         let mut builder = PhysicalPlanBuilder::new();
         let operator = builder.build_plan(&logical_plan)?;
-        println!("operator: {:#?}", operator);
+        // println!("operator: {:#?}", operator);
 
         let storage = StorageImpl::InMemoryStorage(self.storage.clone());
         let executor = VolcanoExecutor::new(storage);
 
         let mut stream = executor.build(operator);
 
         Ok(VolcanoExecutor::try_collect(&mut stream).await?)
-
-        // // let physical_planner = PhysicalPlaner::default();
-        // // let executor_builder = ExecutorBuilder::new(self.env.clone());
-        //
-        // // let physical_plan = physical_planner.plan(logical_plan)?;
-        // // let executor = executor_builder.build(physical_plan)?;
-        // // futures::executor::block_on(executor).unwrap();
-        //
-        // /// THE FOLLOWING CODE IS FOR TESTING ONLY
-        // /// THE FINAL CODE WILL BE IN executor MODULE
-        // if let LogicalPlan::CreateTable(plan) = logical_plan {
-        //     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);
-        //     });
-        //
-        //     self.storage
-        //         .create_table(IdGenerator::build(), table_name.as_str(), data)?;
-        // }
     }
 }
 
@@ -126,6 +99,7 @@ mod test {
     use arrow::compute::concat_batches;
     use arrow::datatypes::Schema;
     use arrow::record_batch::RecordBatch;
+    use arrow::util::pretty::print_batches;
     use itertools::Itertools;
     use crate::catalog::{ColumnCatalog, ColumnDesc};
     use crate::db::Database;
@@ -173,19 +147,22 @@ mod test {
             Ok(())
         })
     }
+
     #[test]
     fn test_crud_sql() -> anyhow::Result<()> {
-        let database = Database::new_on_mem();
+        let kipsql = Database::new_on_mem();
 
         tokio_test::block_on(async move {
-            let _ = database.run("create table t1 (a int, b boolean)").await?;
-            let _ = database.run("insert into t1 values (1, true), (2, false)").await?;
-            let vec_batch = database.run("select * from t1").await?;
-
-            let table = database.storage
-                .get_catalog()
-                .get_table(0).unwrap().clone();
-            println!("{:#?}", concat_batches(&table.schema(), &vec_batch));
+            let _ = kipsql.run("create table t1 (a int, b int)").await?;
+            let _ = kipsql.run("insert into t1 values (1, 2), (3, 4)").await?;
+
+            println!("full:");
+            let vec_batch_full_fields = kipsql.run("select * from t1").await?;
+            print_batches(&vec_batch_full_fields)?;
+
+            println!("projection_and_filter:");
+            let vec_batch_projection_a = kipsql.run("select a from t1 where a != 3 and a < b ").await?;
+            print_batches(&vec_batch_projection_a)?;
 
             Ok(())
         })

src/execution_v1/physical_plan/mod.rs

@@ -1,4 +1,5 @@
 use crate::execution_v1::physical_plan::physical_create_table::PhysicalCreateTable;
+use crate::execution_v1::physical_plan::physical_filter::PhysicalFilter;
 use crate::execution_v1::physical_plan::physical_insert::PhysicalInsert;
 use crate::execution_v1::physical_plan::physical_projection::PhysicalProjection;
 use crate::execution_v1::physical_plan::physical_table_scan::PhysicalTableScan;
@@ -10,12 +11,14 @@ pub(crate) mod physical_projection;
 pub(crate) mod physical_table_scan;
 pub(crate) mod physical_insert;
 pub(crate) mod physical_values;
+pub(crate) mod physical_filter;
 
 #[derive(Debug)]
 pub enum PhysicalOperator {
     Insert(PhysicalInsert),
     CreateTable(PhysicalCreateTable),
     TableScan(PhysicalTableScan),
     Projection(PhysicalProjection),
+    Filter(PhysicalFilter),
     Values(PhysicalValues)
 }

src/execution_v1/physical_plan/physical_filter.rs

@@ -0,0 +1,9 @@
+use crate::execution_v1::physical_plan::PhysicalOperator;
+use crate::expression::ScalarExpression;
+
+#[derive(Debug)]
+pub struct PhysicalFilter {
+    pub(crate) plan_id: u32,
+    pub(crate) predicate: ScalarExpression,
+    pub(crate) input: Box<PhysicalOperator>
+}

src/execution_v1/physical_plan/physical_plan_builder.rs

@@ -9,9 +9,11 @@ use crate::planner::operator::Operator;
 use crate::planner::LogicalPlan;
 use anyhow::anyhow;
 use anyhow::Result;
+use crate::execution_v1::physical_plan::physical_filter::PhysicalFilter;
 use crate::execution_v1::physical_plan::physical_insert::PhysicalInsert;
 use crate::execution_v1::physical_plan::physical_values::PhysicalValues;
 use crate::planner::logical_insert_plan::LogicalInsertPlan;
+use crate::planner::operator::filter::FilterOperator;
 use crate::planner::operator::insert::InsertOperator;
 use crate::planner::operator::project::ProjectOperator;
 use crate::planner::operator::values::ValuesOperator;
@@ -84,6 +86,7 @@ impl PhysicalPlanBuilder {
         match plan.operator.as_ref() {
             Operator::Project(op) => self.build_physical_select_projection(plan, op),
             Operator::Scan(scan) => Ok(self.build_physical_scan(scan.clone())),
+            Operator::Filter(op) => self.build_physical_filter(plan, op),
             _ => Err(anyhow!(format!(
                 "Unsupported physical plan: {:?}",
                 plan.operator
@@ -104,4 +107,14 @@ impl PhysicalPlanBuilder {
     fn build_physical_scan(&mut self, base: ScanOperator) -> PhysicalOperator {
         PhysicalOperator::TableScan(PhysicalTableScan { plan_id: self.next_plan_id(), base })
     }
+
+    fn build_physical_filter(&mut self, plan: &LogicalSelectPlan, base: &FilterOperator) -> Result<PhysicalOperator> {
+        let input = self.build_select_logical_plan(plan.child(0)?)?;
+
+        Ok(PhysicalOperator::Filter(PhysicalFilter {
+            plan_id: 0,
+            predicate: base.predicate.clone(),
+            input: Box::new(input),
+        }))
+    }
 }

src/execution_v1/volcano_executor/filter.rs

@@ -0,0 +1,22 @@
+use arrow::array::as_boolean_array;
+use arrow::compute::filter_record_batch;
+use arrow::record_batch::RecordBatch;
+use futures_async_stream::try_stream;
+use crate::execution_v1::volcano_executor::BoxedExecutor;
+use crate::execution_v1::ExecutorError;
+use crate::expression::ScalarExpression;
+
+pub struct Filter { }
+
+impl Filter {
+    #[try_stream(boxed, ok = RecordBatch, error = ExecutorError)]
+    pub async fn execute(predicate: ScalarExpression, input: BoxedExecutor) {
+        #[for_await]
+        for batch in input {
+            let batch = batch?;
+            let predicate = predicate.eval_column(&batch)?;
+
+            yield filter_record_batch(&batch, as_boolean_array(&predicate))?;
+        }
+    }
+}

src/execution_v1/volcano_executor/mod.rs

@@ -3,6 +3,7 @@ mod projection;
 mod table_scan;
 mod insert;
 mod values;
+mod filter;
 
 use crate::execution_v1::physical_plan::physical_projection::PhysicalProjection;
 use crate::execution_v1::physical_plan::PhysicalOperator;
@@ -14,7 +15,9 @@ use crate::storage::StorageImpl;
 use arrow::record_batch::RecordBatch;
 use futures::stream::BoxStream;
 use futures::TryStreamExt;
+use crate::execution_v1::physical_plan::physical_filter::PhysicalFilter;
 use crate::execution_v1::physical_plan::physical_insert::PhysicalInsert;
+use crate::execution_v1::volcano_executor::filter::Filter;
 use crate::execution_v1::volcano_executor::insert::Insert;
 use crate::execution_v1::volcano_executor::values::Values;
 
@@ -38,6 +41,7 @@ impl VolcanoExecutor {
             }
             PhysicalOperator::Projection(PhysicalProjection { input, exprs, .. }) => {
                 let input = self.build(*input);
+
                 Projection::execute(exprs, input)
             }
             PhysicalOperator::CreateTable(op) => match &self.storage {
@@ -51,7 +55,12 @@ impl VolcanoExecutor {
                         Insert::execute(table_name, input, storage.clone()),
                 }
             }
-            PhysicalOperator::Values(op) => Values::execute(op)
+            PhysicalOperator::Values(op) => Values::execute(op),
+            PhysicalOperator::Filter(PhysicalFilter { predicate, input, .. }) => {
+                let input = self.build(*input);
+
+                Filter::execute(predicate, input)
+            }
         }
     }
 

src/execution_v1/volcano_executor/table_scan.rs

@@ -13,7 +13,7 @@ impl TableScan {
     pub async fn execute(plan: PhysicalTableScan, storage: impl Storage) {
         // TODO: sort_fields, pre_where, limit
         let ScanOperator { table_ref_id,  .. } = plan.base;
-        println!("ref id: {}", table_ref_id);
+
         let table = storage.get_table(table_ref_id)?;
         let mut transaction = table.read(
             None,

src/expression/cast.rs

@@ -0,0 +1,20 @@
+use arrow::array::{Array, BooleanArray};
+use anyhow::Result;
+
+/// Downcast an Arrow Array to a concrete type
+macro_rules! downcast_value {
+    ($Value:expr, $Type:ident) => {{
+        use std::any::type_name;
+
+        if let Some(value) = $Value.as_any().downcast_ref::<$Type>() {
+            Ok(value)
+        } else {
+            Err(anyhow::anyhow!("could not cast value to {}", type_name::<$Type>()))
+        }
+    }};
+}
+
+/// Downcast ArrayRef to BooleanArray
+pub fn as_boolean_array(array: &dyn Array) -> Result<&BooleanArray> {
+    downcast_value!(array, BooleanArray)
+}