bugfix(planner): Fix correlated subquery with joins

query/src/sql/planner/binder/bind_context.rs

@@ -68,6 +68,13 @@ impl BindContext {
         }
     }
 
+    /// Create a new BindContext with self's parent as its parent
+    pub fn replace(&self) -> Self {
+        let mut bind_context = BindContext::new();
+        bind_context.parent = self.parent.clone();
+        bind_context
+    }
+
     /// Generate a new BindContext and take current BindContext as its parent.
     pub fn push(self) -> Self {
         Self::with_parent(Box::new(self))

query/src/sql/planner/binder/join.rs

@@ -49,13 +49,12 @@ impl<'a> Binder {
     ) -> Result<(SExpr, BindContext)> {
         let (left_child, left_context) =
             self.bind_table_reference(bind_context, &join.left).await?;
-        let (right_child, right_context) = self
-            .bind_table_reference(&left_context, &join.right)
-            .await?;
+        let (right_child, right_context) =
+            self.bind_table_reference(bind_context, &join.right).await?;
 
         check_duplicate_join_tables(&left_context, &right_context)?;
 
-        let mut bind_context = BindContext::new();
+        let mut bind_context = bind_context.replace();
         for column in left_context.all_column_bindings() {
             bind_context.add_column_binding(column.clone());
         }
@@ -69,12 +68,12 @@ impl<'a> Binder {
             {
                 return Err(ErrorCode::SemanticError(
                     "outer join should contain join conditions".to_string(),
-                ))
+                ));
             }
             JoinOperator::CrossJoin if join.condition != JoinCondition::None => {
                 return Err(ErrorCode::SemanticError(
                     "cross join should not contain join conditions".to_string(),
-                ))
+                ));
             }
             _ => (),
         };

query/tests/it/sql/optimizer/heuristic/mod.rs

@@ -14,6 +14,7 @@
 
 mod join;
 mod select;
+mod subquery;
 
 use std::io::Write;
 use std::sync::Arc;

query/tests/it/sql/optimizer/heuristic/subquery.rs

@@ -0,0 +1,40 @@
+// Copyright 2022 Datafuse Labs.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+use common_base::base::tokio;
+use common_exception::Result;
+use databend_query::sql::optimizer::DEFAULT_REWRITE_RULES;
+use goldenfile::Mint;
+
+use super::run_suites;
+use super::Suite;
+use crate::tests::create_query_context;
+
+#[tokio::test]
+pub async fn test_optimizer_subquery() -> Result<()> {
+    let mut mint = Mint::new("tests/it/sql/optimizer/heuristic/testdata/");
+    let mut file = mint.new_goldenfile("subquery.test")?;
+
+    let ctx = create_query_context().await?;
+
+    let suites = vec![
+        Suite {
+            comment: "# Correlated subquery with joins".to_string(),
+            query: "select t.number from numbers(1) as t, numbers(1) as t1 where t.number = (select count(*) from numbers(1) as t2, numbers(1) as t3 where t.number = t2.number)"
+                .to_string(),
+            rules: DEFAULT_REWRITE_RULES.clone(),
+        }
+    ];
+
+    run_suites(ctx, &mut file, &suites).await
+}

query/tests/it/sql/optimizer/heuristic/testdata/subquery.test

@@ -0,0 +1,19 @@
+# Correlated subquery with joins
+select t.number from numbers(1) as t, numbers(1) as t1 where t.number = (select count(*) from numbers(1) as t2, numbers(1) as t3 where t.number = t2.number)
+----
+Project: [number]
+    Filter: [number = subquery_4]
+        CrossApply
+            PhysicalHashJoin: build keys: [], probe keys: []
+                PhysicalScan: default.system.numbers
+                PhysicalScan: default.system.numbers
+            Max1Row
+                Project: [COUNT(*)]
+                    EvalScalar: [COUNT(*)]
+                        Aggregate: group items: [], aggregate functions: [COUNT(*)]
+                            Filter: [number = number]
+                                PhysicalHashJoin: build keys: [], probe keys: []
+                                    PhysicalScan: default.system.numbers
+                                    PhysicalScan: default.system.numbers
+
+