refactor(logical_plan): unified LogicalPlan classification (#41)

* refactor(logical_plan): unified `LogicalPlan` classification

* fix(limit): fix limit field problem and bind_limit error after merging

* style(physical): `PhysicalOperator` -> `PhysicalPlan`

* style(logical_plan): remove 'Arc'

* fix(create_table): fix `create_table.rs` test_crud_sql

Author: KKould

src/binder/aggregate.rs

@@ -5,20 +5,20 @@ use sqlparser::ast::{Expr, OrderByExpr};
 use crate::{
     expression::ScalarExpression,
     planner::{
-        logical_select_plan::LogicalSelectPlan,
         operator::{aggregate::AggregateOperator, sort::SortField},
     },
 };
+use crate::planner::LogicalPlan;
 
 use super::Binder;
 
 impl Binder {
     pub fn bind_aggregate(
         &mut self,
-        children: LogicalSelectPlan,
+        children: LogicalPlan,
         agg_calls: Vec<ScalarExpression>,
         groupby_exprs: Vec<ScalarExpression>,
-    ) -> LogicalSelectPlan {
+    ) -> LogicalPlan {
         AggregateOperator::new(children, agg_calls, groupby_exprs)
     }
 

src/binder/create_table.rs

@@ -5,15 +5,16 @@ use sqlparser::ast::{ColumnDef, ObjectName};
 use super::Binder;
 use crate::binder::{lower_case_name, split_name};
 use crate::catalog::ColumnCatalog;
-use crate::planner::logical_create_table_plan::LogicalCreateTablePlan;
+use crate::planner::LogicalPlan;
 use crate::planner::operator::create_table::CreateTableOperator;
+use crate::planner::operator::Operator;
 
 impl Binder {
     pub(crate) fn bind_create_table(
         &mut self,
         name: &ObjectName,
         columns: &[ColumnDef],
-    ) -> Result<LogicalCreateTablePlan> {
+    ) -> Result<LogicalPlan> {
         let name = lower_case_name(&name);
         let (_, table_name) = split_name(&name)?;
 
@@ -34,11 +35,14 @@ impl Binder {
             .map(|col| ColumnCatalog::from(col.clone()))
             .collect();
 
-        let plan = LogicalCreateTablePlan {
-            operator: CreateTableOperator {
-                table_name: table_name.to_string(),
-                columns
-            },
+        let plan = LogicalPlan {
+            operator: Operator::CreateTable(
+                CreateTableOperator {
+                    table_name: table_name.to_string(),
+                    columns
+                }
+            ),
+            childrens: vec![],
         };
         Ok(plan)
     }
@@ -59,23 +63,26 @@ mod tests {
         let stmt = crate::parser::parse_sql(sql).unwrap();
         let plan1 = binder.bind(&stmt[0]).unwrap();
 
-        let plan2 = LogicalPlan::CreateTable(LogicalCreateTablePlan {
-            operator: CreateTableOperator {
-                table_name: "t1".to_string(),
-                columns: vec![
-                    ColumnCatalog::new(
-                        "id".to_string(),
-                        false,
-                        ColumnDesc::new(LogicalType::Integer, false)
-                    ),
-                    ColumnCatalog::new(
-                        "name".to_string(),
-                        false,
-                        ColumnDesc::new(LogicalType::Varchar, false)
-                    )
-                ],
-            },
-        });
+        let plan2 = LogicalPlan {
+            operator: Operator::CreateTable(
+                CreateTableOperator {
+                    table_name: "t1".to_string(),
+                    columns: vec![
+                        ColumnCatalog::new(
+                            "id".to_string(),
+                            false,
+                            ColumnDesc::new(LogicalType::Integer, false)
+                        ),
+                        ColumnCatalog::new(
+                            "name".to_string(),
+                            false,
+                            ColumnDesc::new(LogicalType::Varchar, false)
+                        )
+                    ],
+                }
+            ),
+            childrens: vec![],
+        };
 
         assert_eq!(plan1, plan2);
     }

src/binder/insert.rs

@@ -1,12 +1,11 @@
 use std::slice;
-use std::sync::Arc;
 use sqlparser::ast::{Expr, Ident, ObjectName};
 use anyhow::{Error, Result};
 use itertools::Itertools;
 use crate::binder::{Binder, lower_case_name, split_name};
 use crate::catalog::ColumnCatalog;
 use crate::expression::ScalarExpression;
-use crate::planner::logical_insert_plan::LogicalInsertPlan;
+use crate::planner::LogicalPlan;
 use crate::planner::operator::insert::InsertOperator;
 use crate::planner::operator::Operator;
 use crate::planner::operator::values::ValuesOperator;
@@ -21,7 +20,7 @@ impl Binder {
         name: ObjectName,
         idents: &[Ident],
         rows: &Vec<Vec<Expr>>
-    ) -> Result<LogicalInsertPlan> {
+    ) -> Result<LogicalPlan> {
         let name = lower_case_name(&name);
         let (_, table_name) = split_name(&name)?;
 
@@ -56,15 +55,13 @@ impl Binder {
 
             let values_plan = self.bind_values(rows, col_catalogs.clone());
 
-            Ok(LogicalInsertPlan {
-                operator: Arc::new(
-                    Operator::Insert(
-                        InsertOperator {
-                            table: table_name.to_string(),
-                        }
-                    )
+            Ok(LogicalPlan {
+                operator: Operator::Insert(
+                    InsertOperator {
+                        table: table_name.to_string(),
+                    }
                 ),
-                children: vec![Arc::new(values_plan)],
+                childrens: vec![values_plan],
             })
         } else {
             Err(anyhow::Error::msg(format!(
@@ -78,13 +75,13 @@ impl Binder {
         &mut self,
         rows: Vec<Vec<DataValue>>,
         col_catalogs: Vec<ColumnCatalog>
-    ) -> LogicalInsertPlan {
-        LogicalInsertPlan {
-            operator: Arc::new(Operator::Values(ValuesOperator {
+    ) -> LogicalPlan {
+        LogicalPlan {
+            operator: Operator::Values(ValuesOperator {
                 rows,
                 col_catalogs,
-            })),
-            children: vec![],
+            }),
+            childrens: vec![],
         }
     }
 }

src/binder/mod.rs

@@ -61,18 +61,11 @@ impl Binder {
 
     pub fn bind(mut self, stmt: &Statement) -> Result<LogicalPlan> {
         let plan = match stmt {
-            Statement::Query(query) => {
-                let plan = self.bind_query(query)?;
-                LogicalPlan::Select(plan)
-            }
-            Statement::CreateTable { name, columns, .. } => {
-                let plan = self.bind_create_table(name, &columns)?;
-                LogicalPlan::CreateTable(plan)
-            }
+            Statement::Query(query) => self.bind_query(query)?,
+            Statement::CreateTable { name, columns, .. } => self.bind_create_table(name, &columns)?,
             Statement::Insert { table_name, columns, source, .. } => {
                 if let SetExpr::Values(values) = source.body.as_ref() {
-                    let plan = self.bind_insert(table_name.to_owned(), columns, &values.rows)?;
-                    LogicalPlan::Insert(plan)
+                    self.bind_insert(table_name.to_owned(), columns, &values.rows)?
                 } else {
                     todo!()
                 }

src/binder/select.rs

@@ -1,10 +1,9 @@
-use std::{borrow::Borrow, sync::Arc};
+use std::borrow::Borrow;
 
 use crate::{
     catalog::ColumnRefId,
     expression::ScalarExpression,
     planner::{
-        logical_select_plan::LogicalSelectPlan,
         operator::{
             filter::FilterOperator, join::JoinOperator as LJoinOperator, limit::LimitOperator,
             project::ProjectOperator, Operator,
@@ -23,10 +22,11 @@ use sqlparser::ast::{
     Expr, Ident, Join, JoinConstraint, JoinOperator, Offset, OrderByExpr, Query, Select,
     SelectItem, SetExpr, TableFactor, TableWithJoins,
 };
+use crate::planner::LogicalPlan;
 use crate::planner::operator::sort::{SortField, SortOperator};
 
 impl Binder {
-    pub(crate) fn bind_query(&mut self, query: &Query) -> Result<LogicalSelectPlan> {
+    pub(crate) fn bind_query(&mut self, query: &Query) -> Result<LogicalPlan> {
         if let Some(_with) = &query.with {
             // TODO support with clause.
         }
@@ -51,7 +51,7 @@ impl Binder {
         &mut self,
         select: &Select,
         orderby: &[OrderByExpr],
-    ) -> Result<LogicalSelectPlan> {
+    ) -> Result<LogicalPlan> {
         let mut plan = self.bind_table_ref(&select.from)?;
 
         // Resolve scalar function call.
@@ -99,12 +99,12 @@ impl Binder {
         Ok(plan)
     }
 
-    fn bind_table_ref(&mut self, from: &[TableWithJoins]) -> Result<LogicalSelectPlan> {
+    fn bind_table_ref(&mut self, from: &[TableWithJoins]) -> Result<LogicalPlan> {
         assert!(from.len() < 2, "not support yet.");
         if from.is_empty() {
-            return Ok(LogicalSelectPlan {
-                operator: Arc::new(Operator::Dummy),
-                children: vec![],
+            return Ok(LogicalPlan {
+                operator: Operator::Dummy,
+                childrens: vec![],
             });
         }
 
@@ -120,7 +120,7 @@ impl Binder {
         Ok(plan)
     }
 
-    fn bind_single_table_ref(&mut self, table: &TableFactor) -> Result<LogicalSelectPlan> {
+    fn bind_single_table_ref(&mut self, table: &TableFactor) -> Result<LogicalPlan> {
         let plan = match table {
             TableFactor::Table { name, alias, .. } => {
                 let obj_name = name
@@ -201,23 +201,22 @@ impl Binder {
         let mut exprs = vec![];
         for ref_id in self.context.bind_table.values().cloned().collect_vec() {
             let table = self.context.catalog.get_table(ref_id).unwrap();
-            for (col_id, col) in &table.all_columns() {
-                let column_ref_id = ColumnRefId::from_table(ref_id, *col_id);
+            for (col_id, col) in table.all_columns() {
+                let column_ref_id = ColumnRefId::from_table(ref_id, col_id);
                 // self.record_regular_table_column(
                 //     &table.name(),
                 //     col.name(),
                 //     *col_id,
                 //     col.desc().clone(),
                 // );
-                let expr = ScalarExpression::ColumnRef((*col).clone());
-                exprs.push(expr);
+                exprs.push(ScalarExpression::ColumnRef(col.clone()));
             }
         }
 
         Ok(exprs)
     }
 
-    fn bind_join(&mut self, left: LogicalSelectPlan, join: &Join) -> Result<LogicalSelectPlan> {
+    fn bind_join(&mut self, left: LogicalPlan, join: &Join) -> Result<LogicalPlan> {
         let Join {
             relation,
             join_operator,
@@ -250,9 +249,9 @@ impl Binder {
 
     fn bind_where(
         &mut self,
-        children: LogicalSelectPlan,
+        children: LogicalPlan,
         predicate: &Expr,
-    ) -> Result<LogicalSelectPlan> {
+    ) -> Result<LogicalPlan> {
         Ok(FilterOperator::new(
             self.bind_expr(predicate)?,
             children,
@@ -262,46 +261,46 @@ impl Binder {
 
     fn bind_having(
         &mut self,
-        children: LogicalSelectPlan,
+        children: LogicalPlan,
         having: ScalarExpression,
-    ) -> Result<LogicalSelectPlan> {
+    ) -> Result<LogicalPlan> {
         self.validate_having_orderby(&having)?;
         Ok(FilterOperator::new(having, children, true))
     }
 
     fn bind_project(
         &mut self,
-        children: LogicalSelectPlan,
+        children: LogicalPlan,
         select_list: Vec<ScalarExpression>,
-    ) -> LogicalSelectPlan {
-        LogicalSelectPlan {
-            operator: Arc::new(Operator::Project(ProjectOperator {
+    ) -> LogicalPlan {
+        LogicalPlan {
+            operator: Operator::Project(ProjectOperator {
                 columns: select_list,
-            })),
-            children: vec![Arc::new(children)],
+            }),
+            childrens: vec![children],
         }
     }
 
     fn bind_sort(
         &mut self,
-        children: LogicalSelectPlan,
+        children: LogicalPlan,
         sort_fields: Vec<SortField>,
-    ) -> LogicalSelectPlan {
-        LogicalSelectPlan {
-            operator: Arc::new(Operator::Sort(SortOperator {
+    ) -> LogicalPlan {
+        LogicalPlan {
+            operator: Operator::Sort(SortOperator {
                 sort_fields,
                 limit: None,
-            })),
-            children: vec![Arc::new(children)],
+            }),
+            childrens: vec![children],
         }
     }
 
     fn bind_limit(
         &mut self,
-        children: LogicalSelectPlan,
+        children: LogicalPlan,
         limit_expr: &Option<Expr>,
         offset_expr: &Option<Offset>,
-    ) -> Result<LogicalSelectPlan> {
+    ) -> Result<LogicalPlan> {
         let mut limit = 0;
         let mut offset = 0;
         if let Some(expr) = limit_expr {

src/catalog/column.rs

@@ -74,8 +74,4 @@ impl ColumnDesc {
             is_primary,
         }
     }
-
-    pub(crate) fn get_datatype(&self) -> LogicalType {
-        self.column_datatype.clone()
-    }
 }