feat(hash_join): implement basic Inner/Left/Right/Full Join operations (#43)

* feat(hash_join): implement basic Inner/Left/Right/Full Join operations

found a problem that needs to be fixed:
- when the `Projection` operator executes eval_column, the reading data is abnormal due to the duplication of the ColumnIdx subscript of the Catalog
So `Binder::bind_project` need to support join

* fix(hash_join): fix filter on right join

* feat(projection): `Projection` supports multi-table mapping

By marshaling Column and rewriting `ScalarExpression` to convert `ColumnRef` to `InputRef`

* fix(insert): fix the misalignment between the specified field and the data when inserting

* refactor(physical_plan_builder): for subsequent Agg compatibility, cancel converting InputRef to ColumnRef

* fix(create_table): fix `test_create_bind`

* refactor(binder): when selecting Join in `Binder::bind`, perform corresponding Select nullable processing to avoid rewrite

Author: KKould

Cargo.toml

@@ -40,6 +40,7 @@ async-channel = "1.8.0"
 async-backtrace = "0.2.6"
 futures = "0.3.25"
 futures-lite = "1.12.0"
+ahash = "0.8.3"
 
 [dev-dependencies]
 tokio-test = "0.4.2"

src/binder/aggregate.rs

@@ -90,11 +90,7 @@ impl Binder {
             ScalarExpression::AggCall {
                 ty: return_type, ..
             } => {
-                let index = if self.context.agg_calls.len() == 0 {
-                    0
-                } else {
-                    self.context.agg_calls.len() + 1
-                };
+                let index = self.context.agg_calls.len();
                 let input_ref = ScalarExpression::InputRef {
                     index,
                     ty: return_type.clone(),

src/binder/create_table.rs

@@ -52,38 +52,28 @@ impl Binder {
 mod tests {
     use super::*;
     use crate::binder::BinderContext;
-    use crate::catalog::{ColumnCatalog, ColumnDesc, RootCatalog};
-    use crate::planner::LogicalPlan;
+    use crate::catalog::{ColumnDesc, RootCatalog};
     use crate::types::LogicalType;
 
     #[test]
     fn test_create_bind() {
-        let sql = "create table t1 (id int , name varchar(10))";
+        let sql = "create table t1 (id int , name varchar(10) null)";
         let binder = Binder::new(BinderContext::new(RootCatalog::new()));
         let stmt = crate::parser::parse_sql(sql).unwrap();
         let plan1 = binder.bind(&stmt[0]).unwrap();
 
-        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![],
-        };
+        match plan1.operator {
+            Operator::CreateTable(op) => {
+                assert_eq!(op.table_name, "t1".to_string());
+                assert_eq!(op.columns[0].name, "id".to_string());
+                assert_eq!(op.columns[0].nullable, false);
+                assert_eq!(op.columns[0].desc, ColumnDesc::new(LogicalType::Integer, false));
+                assert_eq!(op.columns[1].name, "name".to_string());
+                assert_eq!(op.columns[1].nullable, true);
+                assert_eq!(op.columns[1].desc, ColumnDesc::new(LogicalType::Varchar, false));
+            }
+            _ => unreachable!()
+        }
 
-        assert_eq!(plan1, plan2);
     }
 }

src/binder/expr.rs

@@ -68,7 +68,7 @@ impl Binder {
         } else {
             // handle col syntax
             let mut got_column = None;
-            for table_catalog in &self.context.catalog.tables {
+            for (_, table_catalog) in self.context.catalog.tables() {
                 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());

src/binder/mod.rs

@@ -4,20 +4,21 @@ pub mod expr;
 mod select;
 mod insert;
 
-use std::collections::HashMap;
+use std::collections::BTreeMap;
 
 use anyhow::Result;
 use sqlparser::ast::{Ident, ObjectName, SetExpr, Statement};
 
 use crate::catalog::{RootCatalog, DEFAULT_SCHEMA_NAME, CatalogError};
 use crate::expression::ScalarExpression;
 use crate::planner::LogicalPlan;
-use crate::types::TableIdx;
-#[derive(Clone)]
+use crate::planner::operator::join::JoinType;
+use crate::types::TableId;
+#[derive(Debug, Clone)]
 pub struct BinderContext {
-    catalog: RootCatalog,
-    bind_table: HashMap<String, TableIdx>,
-    aliases: HashMap<String, ScalarExpression>,
+    pub(crate) catalog: RootCatalog,
+    pub(crate) bind_table: BTreeMap<String, (TableId, Option<JoinType>)>,
+    aliases: BTreeMap<String, ScalarExpression>,
     group_by_exprs: Vec<ScalarExpression>,
     agg_calls: Vec<ScalarExpression>,
     index: u16,