feat(flow): execute the sql process to arrive at the logic plan

src/binder/create.rs

@@ -9,7 +9,7 @@ use sqlparser::ast::{ColumnDef, ObjectName};
 use std::collections::HashSet;
 
 impl Binder {
-    pub(super) fn bind_create_table(
+    pub(crate) fn bind_create_table(
         &mut self,
         name: ObjectName,
         columns: &[ColumnDef],
@@ -18,14 +18,6 @@ impl Binder {
 
         let (_, table_name) = split_name(&name)?;
 
-        let table = self
-            .context
-            .catalog
-            .get_table_by_name(table_name)
-            .ok_or_else(|| {
-                anyhow::Error::msg(format!("table {} not found", table_name.to_string()))
-            })?;
-
         // check duplicated column names
         let mut set = HashSet::new();
         for col in columns.iter() {
@@ -53,3 +45,43 @@ impl Binder {
         Ok(plan)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::binder::BinderContext;
+    use crate::catalog::Root;
+    use crate::types::{DataTypeExt, DataTypeKind};
+    use sqlparser::ast::CharacterLength;
+    use std::sync::Arc;
+
+    #[test]
+    fn test_create_bind() {
+        let sql = "create table t1 (id int , name varchar(10))";
+        let mut binder = Binder::new(BinderContext::new(Arc::new(Root::new())));
+        let stmt = crate::parser::parse_sql(sql).unwrap();
+        let plan1 = binder.bind(&stmt[0]).unwrap();
+
+        let character_length = CharacterLength {
+            length: 10,
+            unit: None,
+        };
+        let plan2 = LogicalPlan::CreateTable(LogicalCreateTablePlan {
+            table_name: "t1".to_string(),
+            columns: vec![
+                (
+                    "id".to_string(),
+                    DataTypeKind::Int(None).nullable().to_column(),
+                ),
+                (
+                    "name".to_string(),
+                    DataTypeKind::Varchar(Option::from(character_length))
+                        .nullable()
+                        .to_column(),
+                ),
+            ],
+        });
+
+        assert_eq!(plan1, plan2);
+    }
+}

src/binder/mod.rs

@@ -11,7 +11,7 @@ use crate::catalog::DEFAULT_SCHEMA_NAME;
 use crate::types::TableId;
 use anyhow::Result;
 use sqlparser::ast::{Ident, ObjectName, Statement};
-
+#[derive(Clone)]
 pub struct BinderContext {
     catalog: CatalogRef,
     bind_table: HashMap<String, TableId>,

src/binder/select.rs

@@ -25,7 +25,7 @@ use sqlparser::ast::{
 };
 
 impl Binder {
-    pub(super) fn bind_query(&mut self, query: &Query) -> Result<LogicalSelectPlan> {
+    pub(crate) fn bind_query(&mut self, query: &Query) -> Result<LogicalSelectPlan> {
         if let Some(_with) = &query.with {
             // TODO support with clause.
         }

src/catalog/root.rs

@@ -9,7 +9,7 @@ pub struct Root {
 
 impl Root {
     #[allow(dead_code)]
-    pub(crate) fn new() -> Self {
+    pub fn new() -> Self {
         Root {
             table_idxs: Default::default(),
             tables: Default::default(),

src/main.rs

@@ -4,29 +4,31 @@ use kip_sql::planner::{display::display_plan, logical_plan_builder::PlanBuilder}
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
-    println!(":) Welcome to the KIPSQL, Please input sql.");
-
-    let mut input = String::new();
-    io::stdin().read_line(&mut input)?;
-
-    let output = handle_query(input);
-
-    println!("KIPSQL Output: {output}");
+    let builder = PlanBuilder::new();
 
-    Ok(())
+    loop {
+        println!(":) Welcome to the KIPSQL, Please input sql.");
+        let mut input = String::new();
+        io::stdin().read_line(&mut input)?;
+        let output = handle_query(input, &builder);
+        println!("KIPSQL Output: {output}");
+    }
 }
 
-fn handle_query(sql: String) -> String {
-    let builder = PlanBuilder::new();
-    let plan = builder.build_sql(&sql).unwrap();
+fn handle_query(sql: String, builder: &PlanBuilder) -> String {
+    let plan = builder.build_sql(&sql);
+    if plan.is_err() {
+        return plan.err().unwrap().to_string() + " ERROR";
+    }
+    return "OK".to_string();
 
-    let logical_graph = display_plan(&plan);
-    tracing::info!("logical plan {}", logical_graph);
+    //let logical_graph = display_plan(&plan);
+    //tracing::info!("logical plan {}", logical_graph);
 
     // todo optimize.
     // let mut optimize = Optimizer::new();
     // let _optmized_plan = optimize.optimize(&plan).unwrap();
 
     // todo executor.
-    logical_graph
+    //let logical_graph = display_plan(&plan);
 }

src/planner/logical_create_table_plan.rs

@@ -1,5 +1,5 @@
 use crate::catalog::ColumnDesc;
-
+#[derive(Debug, PartialEq, Clone)]
 pub struct LogicalCreateTablePlan {
     pub table_name: String,
     pub columns: Vec<(String, ColumnDesc)>,

src/planner/logical_plan_builder.rs

@@ -1,13 +1,25 @@
+use crate::binder::{Binder, BinderContext};
+use crate::catalog::Root;
 use crate::parser;
 use anyhow::Result;
+use serde::de::Unexpected::Option;
+use sqlparser::ast::Statement;
+use std::sync::Arc;
+
+use crate::planner::logical_select_plan::LogicalSelectPlan;
 
 use super::LogicalPlan;
 
-pub struct PlanBuilder {}
+#[derive(Clone)]
+pub struct PlanBuilder {
+    context: BinderContext,
+}
 
 impl PlanBuilder {
     pub fn new() -> Self {
-        PlanBuilder {}
+        PlanBuilder {
+            context: BinderContext::new(Arc::new(Root::new())),
+        }
     }
 
     /// Build a logical plan.
@@ -20,25 +32,24 @@ impl PlanBuilder {
     pub fn build_sql<'a>(&self, sql: &'a str) -> Result<LogicalPlan> {
         let stmts = parser::parse_sql(sql)?;
 
+        println!("stmt:{:#}", stmts[0]);
+
         // TODO: add plan-cache for fast return.
 
-        // let mut binder = Binder::new(self.context.clone());
+        let mut binder = Binder::new(self.context.clone());
 
-        // let mut plan = match &stmts[0] {
-        //     Statement::Query(query) => binder.bind_query(query)?,
-        //     _ => unreachable!(),
-        // };
+        let logical_plan = binder.bind(&stmts[0])?;
 
-        // tracing::info!("optimize before {:?}", plan);
+        println!("logical_plan:{:?}", logical_plan);
 
         // let mut optimizer = Optimizer::new(self.context.clone());
 
         // plan = optimizer.optimize(plan)?;
 
         // tracing::info!("optimize after {:?}", plan);
 
-        // Ok(plan)
+        Ok(logical_plan)
 
-        todo!()
+        //todo!()
     }
 }

src/planner/logical_select_plan.rs

@@ -4,10 +4,9 @@ use super::operator::OperatorRef;
 use anyhow::Result;
 
 /// LogicalSelectPlan is a tree of operators that represent a logical query plan.
-
+#[derive(Debug, PartialEq, Clone)]
 pub struct LogicalSelectPlan {
     pub operator: OperatorRef,
-
     pub children: Vec<Arc<LogicalSelectPlan>>,
 }
 

src/planner/mod.rs

@@ -7,7 +7,9 @@ pub mod operator;
 use self::{
     logical_create_table_plan::LogicalCreateTablePlan, logical_select_plan::LogicalSelectPlan,
 };
+use thiserror::__private::DisplayAsDisplay;
 
+#[derive(Debug, PartialEq, Clone)]
 pub enum LogicalPlan {
     Select(LogicalSelectPlan),
     CreateTable(LogicalCreateTablePlan),

src/planner/operator/aggregate.rs

@@ -5,7 +5,7 @@ use crate::{
     planner::{logical_select_plan::LogicalSelectPlan, operator::Operator},
 };
 
-#[derive(Debug)]
+#[derive(Debug, PartialEq, Clone)]
 pub struct AggregateOperator {
     pub groupby_exprs: Vec<ScalarExpression>,
 

src/planner/operator/filter.rs

@@ -4,7 +4,7 @@ use crate::{expression::ScalarExpression, planner::logical_select_plan::LogicalS
 
 use super::Operator;
 
-#[derive(Debug)]
+#[derive(Debug, PartialEq, Clone)]
 pub struct FilterOperator {
     pub predicate: ScalarExpression,
     having: bool,

src/planner/operator/join.rs

@@ -4,7 +4,7 @@ use crate::{expression::ScalarExpression, planner::logical_select_plan::LogicalS
 
 use super::Operator;
 
-#[derive(Debug)]
+#[derive(Debug, PartialEq, Clone)]
 pub enum JoinType {
     Inner,
     LeftOuter,
@@ -17,7 +17,7 @@ pub enum JoinType {
     RightAnti,
 }
 
-#[derive(Debug)]
+#[derive(Debug, PartialEq, Clone)]
 pub struct JoinOperator {
     pub on: Option<ScalarExpression>,
     pub join_type: JoinType,

src/planner/operator/limit.rs

@@ -4,7 +4,7 @@ use crate::planner::logical_select_plan::LogicalSelectPlan;
 
 use super::Operator;
 
-#[derive(Debug)]
+#[derive(Debug, PartialEq, Clone)]
 pub struct LimitOperator {
     pub offset: usize,
     pub count: usize,

src/planner/operator/mod.rs

@@ -15,7 +15,7 @@ use self::{
 
 pub type OperatorRef = Arc<Operator>;
 
-#[derive(Debug)]
+#[derive(Debug, PartialEq, Clone)]
 pub enum Operator {
     Dummy,
     Aggregate(AggregateOperator),

src/planner/operator/project.rs

@@ -6,7 +6,7 @@ use crate::{expression::ScalarExpression, planner::logical_select_plan::LogicalS
 
 use super::Operator;
 
-#[derive(Debug)]
+#[derive(Debug, PartialEq, Clone)]
 pub struct ProjectOperator {
     pub columns: Vec<ScalarExpression>,
 }

src/planner/operator/scan.rs

@@ -8,7 +8,7 @@ use crate::{
 
 use super::{sort::SortField, Operator};
 
-#[derive(Debug, Clone)]
+#[derive(Debug, PartialEq, Clone)]
 pub struct ScanOperator {
     pub table_ref_id: TableId,
     pub columns: Vec<ColumnRefId>,

src/planner/operator/sort.rs

@@ -1,6 +1,6 @@
 use crate::expression::ScalarExpression;
 
-#[derive(Debug, Clone)]
+#[derive(Debug, PartialEq, Clone)]
 pub struct SortField {
     expr: ScalarExpression,
     desc: bool,
@@ -17,7 +17,7 @@ impl SortField {
     }
 }
 
-#[derive(Debug)]
+#[derive(Debug, PartialEq, Clone)]
 pub struct SortOperator {
     pub sort_fields: Vec<SortField>,
     /// Support push down limit to sort plan.