feat(filter): implement the Filter physical operator, support similar select c1 from t1 where c1 > 1

Author: KKould

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/db.rs

@@ -180,7 +180,7 @@ mod test {
         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 vec_batch = database.run("select * from t1 where a = 1 or b = false").await?;
 
             let table = database.storage
                 .get_catalog()

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/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)
+}

src/expression/mod.rs

@@ -10,6 +10,7 @@ use crate::types::LogicalType;
 pub mod agg;
 mod evaluator;
 mod array_compute;
+mod cast;
 
 /// ScalarExpression represnet all scalar expression in SQL.
 /// SELECT a+1, b FROM t1.