Using the RBO method, the related subquery is implemented

wszhdshys · wszhdshys · commit 818a9c5b0e01 · 2025-07-23T23:07:44.000+08:00
diff --git a/src/binder/expr.rs b/src/binder/expr.rs
@@ -369,10 +369,13 @@ impl<'a, T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'a, '_, T
             try_default!(&full_name.0, full_name.1);
         }
         if let Some(table) = full_name.0.or(bind_table_name) {
-            let (source,is_parent) = self.context.bind_source::<A>(self.parent, &table, false)?;
+            let (source, is_parent) = self.context.bind_source::<A>(self.parent, &table, false)?;
 
             if is_parent {
-                self.parent_table_col.entry(Arc::new(table.clone())).or_insert(HashSet::new()).insert(full_name.1.clone());
+                self.parent_table_col
+                    .entry(Arc::new(table.clone()))
+                    .or_default()
+                    .insert(full_name.1.clone());
             }
 
             let schema_buf = self.table_schema_buf.entry(Arc::new(table)).or_default();
diff --git a/src/binder/mod.rs b/src/binder/mod.rs
@@ -25,7 +25,7 @@ use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::Arc;
 
 use crate::catalog::view::View;
-use crate::catalog::{ColumnCatalog, ColumnRef, TableCatalog, TableName};
+use crate::catalog::{ColumnRef, TableCatalog, TableName};
 use crate::db::{ScalaFunctions, TableFunctions};
 use crate::errors::DatabaseError;
 use crate::expression::ScalarExpression;
@@ -275,9 +275,9 @@ impl<'a, T: Transaction> BinderContext<'a, T> {
         Ok(source)
     }
 
-    pub fn bind_source<'b: 'a, A: AsRef<[(&'static str, DataValue)]> >(
+    pub fn bind_source<'b: 'a, A: AsRef<[(&'static str, DataValue)]>>(
         &self,
-        parent: Option<&'a Binder<'a,'b,T,A>>,
+        parent: Option<&'a Binder<'a, 'b, T, A>>,
         table_name: &str,
         is_parent: bool,
     ) -> Result<(&'b Source, bool), DatabaseError> {
@@ -287,10 +287,10 @@ impl<'a, T: Transaction> BinderContext<'a, T> {
         }) {
             Ok((source.1, is_parent))
         } else if let Some(binder) = parent {
-            binder.context.bind_source(binder.parent, table_name,true)
+            binder.context.bind_source(binder.parent, table_name, true)
         } else {
             Err(DatabaseError::InvalidTable(table_name.into()))
-        } 
+        }
     }
 
     // Tips: The order of this index is based on Aggregate being bound first.
diff --git a/src/binder/select.rs b/src/binder/select.rs
@@ -36,7 +36,7 @@ use crate::types::value::Utf8Type;
 use crate::types::{ColumnId, LogicalType};
 use itertools::Itertools;
 use sqlparser::ast::{
-    CharLengthUnits, Distinct, Expr, Ident, Join, JoinConstraint, JoinOperator, ObjectName, Offset,
+    CharLengthUnits, Distinct, Expr, Ident, Join, JoinConstraint, JoinOperator, Offset,
     OrderByExpr, Query, Select, SelectInto, SelectItem, SetExpr, SetOperator, SetQuantifier,
     TableAlias, TableFactor, TableWithJoins,
 };
@@ -101,8 +101,6 @@ impl<'a: 'b, 'b, T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'
         };
         let mut select_list = self.normalize_select_item(&select.projection, &plan)?;
 
-        plan = self.bind_parent(plan)?;
-
         if let Some(predicate) = &select.selection {
             plan = self.bind_where(plan, predicate)?;
         }
@@ -536,22 +534,6 @@ impl<'a: 'b, 'b, T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'
         Ok(())
     }
 
-    fn bind_parent(
-        &mut self,
-        mut plan: LogicalPlan,
-    ) -> Result<LogicalPlan, DatabaseError> {
-        for (table,columns) in self.parent_table_col.clone().into_iter() {
-            let parent = self._bind_single_table_ref(None,table.as_str(),None)?;
-            plan = LJoinOperator::build(
-                plan,
-                parent,
-                JoinCondition::None,
-                JoinType::Full,
-            );
-        }
-        Ok(plan)
-    }
-
     fn bind_join(
         &mut self,
         mut left: LogicalPlan,
@@ -613,8 +595,6 @@ impl<'a: 'b, 'b, T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'
 
         let predicate = self.bind_expr(predicate)?;
 
-        children = self.bind_parent(children)?;
-
         if let Some(sub_queries) = self.context.sub_queries_at_now() {
             for sub_query in sub_queries {
                 let mut on_keys: Vec<(ScalarExpression, ScalarExpression)> = vec![];
diff --git a/src/db.rs b/src/db.rs
@@ -166,13 +166,18 @@ impl<S: Storage> State<S> {
 
         let best_plan = Self::default_optimizer(source_plan)
             .find_best(Some(&transaction.meta_loader(meta_cache)))?;
-        // println!("best_plan plan: {:#?}", best_plan);
+        //println!("best_plan plan: {:#?}", best_plan);
 
         Ok(best_plan)
     }
 
     pub(crate) fn default_optimizer(source_plan: LogicalPlan) -> HepOptimizer {
         HepOptimizer::new(source_plan)
+            .batch(
+                "Correlated Subquery".to_string(),
+                HepBatchStrategy::once_topdown(),
+                vec![NormalizationRuleImpl::CorrelateSubquery],
+            )
             .batch(
                 "Column Pruning".to_string(),
                 HepBatchStrategy::once_topdown(),
diff --git a/src/execution/dql/filter.rs b/src/execution/dql/filter.rs
@@ -35,12 +35,15 @@ impl<'a, T: Transaction + 'a> ReadExecutor<'a, T> for Filter {
 
                 let schema = input.output_schema().clone();
 
+                //println!("{:#?}114514'\n'1919810{:#?}",predicate,schema);
+
                 let mut coroutine = build_read(input, cache, transaction);
 
                 while let CoroutineState::Yielded(tuple) = Pin::new(&mut coroutine).resume(()) {
                     let tuple = throw!(tuple);
-
+                    //println!("-> Coroutine returned: {:?}", tuple);
                     if throw!(throw!(predicate.eval(Some((&tuple, &schema)))).is_true()) {
+                        //println!("-> throw!: {:?}", tuple);
                         yield Ok(tuple);
                     }
                 }
diff --git a/src/execution/dql/join/nested_loop_join.rs b/src/execution/dql/join/nested_loop_join.rs
@@ -505,12 +505,12 @@ mod test {
 
         let filter = ScalarExpression::Binary {
             op: crate::expression::BinaryOperator::Gt,
-            left_expr: Box::new(ScalarExpression::ColumnRef(
-                ColumnRef::from(ColumnCatalog::new("c1".to_owned(), true, desc.clone())),
-            )),
-            right_expr: Box::new(ScalarExpression::ColumnRef(
-                ColumnRef::from(ColumnCatalog::new("c4".to_owned(), true, desc.clone())),
-            )),
+            left_expr: Box::new(ScalarExpression::ColumnRef(ColumnRef::from(
+                ColumnCatalog::new("c1".to_owned(), true, desc.clone()),
+            ))),
+            right_expr: Box::new(ScalarExpression::ColumnRef(ColumnRef::from(
+                ColumnCatalog::new("c4".to_owned(), true, desc.clone()),
+            ))),
             evaluator: Some(BinaryEvaluatorBox(Arc::new(Int32GtBinaryEvaluator))),
             ty: LogicalType::Boolean,
         };
diff --git a/src/expression/mod.rs b/src/expression/mod.rs
@@ -1105,37 +1105,33 @@ mod test {
         )?;
         fn_assert(
             &mut cursor,
-            ScalarExpression::ColumnRef(
-                ColumnRef::from(ColumnCatalog::direct_new(
-                    ColumnSummary {
-                        name: "c3".to_string(),
-                        relation: ColumnRelation::Table {
-                            column_id: c3_column_id,
-                            table_name: Arc::new("t1".to_string()),
-                            is_temp: false,
-                        },
+            ScalarExpression::ColumnRef(ColumnRef::from(ColumnCatalog::direct_new(
+                ColumnSummary {
+                    name: "c3".to_string(),
+                    relation: ColumnRelation::Table {
+                        column_id: c3_column_id,
+                        table_name: Arc::new("t1".to_string()),
+                        is_temp: false,
                     },
-                    false,
-                    ColumnDesc::new(LogicalType::Integer, None, false, None)?,
-                    false,
-                )),
-            ),
+                },
+                false,
+                ColumnDesc::new(LogicalType::Integer, None, false, None)?,
+                false,
+            ))),
             Some((&transaction, &table_cache)),
             &mut reference_tables,
         )?;
         fn_assert(
             &mut cursor,
-            ScalarExpression::ColumnRef(
-                ColumnRef::from(ColumnCatalog::direct_new(
-                    ColumnSummary {
-                        name: "c4".to_string(),
-                        relation: ColumnRelation::None,
-                    },
-                    false,
-                    ColumnDesc::new(LogicalType::Boolean, None, false, None)?,
-                    false,
-                )),
-            ),
+            ScalarExpression::ColumnRef(ColumnRef::from(ColumnCatalog::direct_new(
+                ColumnSummary {
+                    name: "c4".to_string(),
+                    relation: ColumnRelation::None,
+                },
+                false,
+                ColumnDesc::new(LogicalType::Boolean, None, false, None)?,
+                false,
+            ))),
             Some((&transaction, &table_cache)),
             &mut reference_tables,
         )?;
diff --git a/src/optimizer/heuristic/graph.rs b/src/optimizer/heuristic/graph.rs
@@ -79,7 +79,7 @@ impl HepGraph {
         source_id: HepNodeId,
         children_option: Option<HepNodeId>,
         new_node: Operator,
-    ) {
+    ) -> HepNodeId {
         let new_index = self.graph.add_node(new_node);
         let mut order = self.graph.edges(source_id).count();
 
@@ -95,6 +95,7 @@ impl HepGraph {
 
         self.graph.add_edge(source_id, new_index, order);
         self.version += 1;
+        new_index
     }
 
     pub fn replace_node(&mut self, source_id: HepNodeId, new_node: Operator) {
diff --git a/src/optimizer/rule/normalization/correlated_subquery.rs b/src/optimizer/rule/normalization/correlated_subquery.rs
diff --git a/src/optimizer/rule/normalization/mod.rs b/src/optimizer/rule/normalization/mod.rs
diff --git a/src/optimizer/rule/normalization/simplification.rs b/src/optimizer/rule/normalization/simplification.rs
diff --git a/tests/slt/correlated_subquery.slt b/tests/slt/correlated_subquery.slt

Original file line number	Diff line number	Diff line change
`@@ -369,10 +369,13 @@ impl<'a, T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'a, '_, T`
`369`	`369`	`try_default!(&full_name.0, full_name.1);`
`370`	`370`	`}`
`371`	`371`	`if let Some(table) = full_name.0.or(bind_table_name) {`
`372`		`- let (source,is_parent) = self.context.bind_source::<A>(self.parent, &table, false)?;`
	`372`	`+ let (source, is_parent) = self.context.bind_source::<A>(self.parent, &table, false)?;`
`373`	`373`
`374`	`374`	`if is_parent {`
`375`		`- self.parent_table_col.entry(Arc::new(table.clone())).or_insert(HashSet::new()).insert(full_name.1.clone());`
	`375`	`+ self.parent_table_col`
	`376`	`+ .entry(Arc::new(table.clone()))`
	`377`	`+ .or_default()`
	`378`	`+ .insert(full_name.1.clone());`
`376`	`379`	`}`
`377`	`380`
`378`	`381`	`let schema_buf = self.table_schema_buf.entry(Arc::new(table)).or_default();`
Original file line number	Diff line number	Diff line change
`@@ -35,12 +35,15 @@ impl<'a, T: Transaction + 'a> ReadExecutor<'a, T> for Filter {`
`35`	`35`
`36`	`36`	`let schema = input.output_schema().clone();`
`37`	`37`
	`38`	`+ //println!("{:#?}114514'\n'1919810{:#?}",predicate,schema);`
	`39`	`+`
`38`	`40`	`let mut coroutine = build_read(input, cache, transaction);`
`39`	`41`
`40`	`42`	`while let CoroutineState::Yielded(tuple) = Pin::new(&mut coroutine).resume(()) {`
`41`	`43`	`let tuple = throw!(tuple);`
`42`		`-`
	`44`	`+ //println!("-> Coroutine returned: {:?}", tuple);`
`43`	`45`	`if throw!(throw!(predicate.eval(Some((&tuple, &schema)))).is_true()) {`
	`46`	`+ //println!("-> throw!: {:?}", tuple);`
`44`	`47`	`yield Ok(tuple);`
`45`	`48`	`}`
`46`	`49`	`}`