refactor: convert InSubquery to predicate

Signed-off-by: Kould <[email protected]>

Author: KKould

src/query/sql/src/planner/binder/bind_table_reference/bind_table_function.rs

@@ -143,7 +143,7 @@ impl Binder {
             } else {
                 None
             };
-            let (table_index, source_table_index) = self.metadata.write().add_table(
+            let (table_index, _) = self.metadata.write().add_table(
                 CATALOG_DEFAULT.to_string(),
                 "system".to_string(),
                 table.clone(),
@@ -154,14 +154,8 @@ impl Binder {
                 None,
             );
 
-            let (s_expr, mut bind_context) = self.bind_base_table(
-                bind_context,
-                "system",
-                table_index,
-                source_table_index,
-                None,
-                sample,
-            )?;
+            let (s_expr, mut bind_context) =
+                self.bind_base_table(bind_context, "system", table_index, None, None, sample)?;
             if let Some(alias) = alias {
                 bind_context.apply_table_alias(alias, &self.name_resolution_ctx)?;
             }

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

@@ -29,6 +29,12 @@ pub struct ColumnBinding {
     /// Table index of this `ColumnBinding` in current context
     pub table_index: Option<IndexType>,
     /// Source table index of this `ColumnBinding` in current context
+    ///
+    /// Example:
+    /// `SELECT n.* FROM notes AS n WHERE n.type = 'D' AND n.id IN (SELECT m.id FROM notes AS m WHERE m.commit_id = 7)`
+    /// In this case, the table index and column index of `n.id` and `m.id` are inconsistent,
+    /// but logically they both belong to the `notes` table.
+    /// Therefore, the source table index is used to allow `n` and `m` to determine whether they are both `notes` in this case.
     pub source_table_index: Option<IndexType>,
     /// Column name of this `ColumnBinding` in current context
     pub column_name: String,

src/query/sql/src/planner/optimizer/decorrelate/subquery_rewriter.rs

@@ -37,7 +37,6 @@ use crate::plans::CastExpr;
 use crate::plans::ComparisonOp;
 use crate::plans::ConstantExpr;
 use crate::plans::EvalScalar;
-use crate::plans::Exchange;
 use crate::plans::Filter;
 use crate::plans::FunctionCall;
 use crate::plans::Join;
@@ -274,10 +273,10 @@ impl SubqueryRewriter {
                     from_count_func: false,
                 };
                 let (s_expr, result) = if prop.outer_columns.is_empty() {
-                    if let Some((scalar_expr, s_expr)) =
-                        self.try_eliminate_in_subquery(s_expr, &subquery)?
+                    if let Some(scalar_expr) =
+                        self.try_eliminate_in_subquery(s_expr, &subquery, is_conjunctive_predicate)?
                     {
-                        return Ok((scalar_expr, s_expr));
+                        return Ok((scalar_expr, s_expr.clone()));
                     }
                     self.try_rewrite_uncorrelated_subquery(
                         s_expr,
@@ -465,22 +464,23 @@ impl SubqueryRewriter {
         &self,
         left: &SExpr,
         subquery: &SubqueryExpr,
-    ) -> Result<Option<(ScalarExpr, SExpr)>> {
+        is_conjunctive_predicate: bool,
+    ) -> Result<Option<ScalarExpr>> {
         #[derive(Debug)]
         struct SubqueryBodyVisitor {
             replacer: BindingReplacer,
             failed: bool,
-            new_s_expr: Option<SExpr>,
+            predicates: Vec<ScalarExpr>,
         }
 
         impl SubqueryBodyVisitor {
-            fn new(new_column_bindings: HashMap<IndexType, ColumnBinding>, expr: SExpr) -> Self {
+            fn new(new_column_bindings: HashMap<IndexType, ColumnBinding>) -> Self {
                 Self {
                     replacer: BindingReplacer {
                         new_column_bindings,
                     },
                     failed: false,
-                    new_s_expr: Some(expr),
+                    predicates: Vec::new(),
                 }
             }
 
@@ -491,43 +491,15 @@ impl SubqueryRewriter {
                 if self.failed {
                     return Ok(());
                 }
-                if !matches!(
-                    s_expr.plan(),
-                    RelOperator::Scan(_)
-                        | RelOperator::Filter(_)
-                        | RelOperator::EvalScalar(_)
-                        | RelOperator::Sort(_)
-                        | RelOperator::Exchange(_)
-                ) {
-                    self.failed = true;
-                    return Ok(());
-                }
                 match s_expr.plan() {
-                    RelOperator::Scan(_) | RelOperator::Sort(_) => (),
-                    RelOperator::EvalScalar(eval_scalar) => {
-                        let mut eval_scalar = eval_scalar.clone();
-
-                        for scalar_item in eval_scalar.items.iter_mut() {
-                            self.replacer.visit(&mut scalar_item.scalar)?;
-                        }
-                        self.build_s_expr(RelOperator::EvalScalar(eval_scalar));
-                    }
+                    RelOperator::Scan(_) | RelOperator::Sort(_) | RelOperator::EvalScalar(_) => (),
                     RelOperator::Filter(filter) => {
                         let mut filter = filter.clone();
 
                         for scalar_expr in filter.predicates.iter_mut() {
                             self.replacer.visit(scalar_expr)?;
                         }
-                        self.build_s_expr(RelOperator::Filter(filter));
-                    }
-                    RelOperator::Exchange(exchange) => {
-                        let mut exchange = exchange.clone();
-                        if let Exchange::Hash(scalar_exprs) = &mut exchange {
-                            for scalar_expr in scalar_exprs {
-                                self.replacer.visit(scalar_expr)?;
-                            }
-                        }
-                        self.build_s_expr(RelOperator::Exchange(exchange));
+                        self.predicates.append(&mut filter.predicates);
                     }
                     _ => {
                         self.failed = true;
@@ -537,15 +509,23 @@ impl SubqueryRewriter {
                 Ok(())
             }
 
-            fn build_s_expr(&mut self, op: RelOperator) {
-                self.new_s_expr = Some(match self.new_s_expr.take() {
-                    None => SExpr::create_leaf(Arc::new(op)),
-                    Some(child_expr) => SExpr::create_unary(Arc::new(op), Arc::new(child_expr)),
-                });
-            }
-
-            fn result(self) -> Option<SExpr> {
-                (!self.failed).then_some(self.new_s_expr).flatten()
+            fn result(self) -> Option<ScalarExpr> {
+                (!self.failed).then(|| {
+                    self.predicates.into_iter().fold(
+                        ScalarExpr::ConstantExpr(ConstantExpr {
+                            span: None,
+                            value: Scalar::Boolean(true),
+                        }),
+                        |acc, expr| {
+                            ScalarExpr::FunctionCall(FunctionCall {
+                                span: None,
+                                func_name: "and".to_string(),
+                                params: vec![],
+                                arguments: vec![acc, expr],
+                            })
+                        },
+                    )
+                })
             }
         }
 
@@ -570,7 +550,7 @@ impl SubqueryRewriter {
             return Ok(None);
         };
         // `xx in null` != `where xx = null`
-        if left_column.column.data_type.is_nullable() {
+        if left_column.column.data_type.is_nullable() || !is_conjunctive_predicate {
             return Ok(None);
         }
         let (Some(left_table_index), Some(right_table_index)) = (
@@ -641,17 +621,9 @@ impl SubqueryRewriter {
             new_column_bindings
         };
         // restore ColumnBinding of same table in Subquery to table in Left
-        let mut body_visitor = SubqueryBodyVisitor::new(new_column_bindings, left.clone());
+        let mut body_visitor = SubqueryBodyVisitor::new(new_column_bindings);
         body_visitor.visit(subquery.subquery.as_ref())?;
-        Ok(body_visitor.result().map(|s_expr| {
-            (
-                ScalarExpr::ConstantExpr(ConstantExpr {
-                    span: None,
-                    value: Scalar::Boolean(true),
-                }),
-                s_expr,
-            )
-        }))
+        Ok(body_visitor.result())
     }
 
     fn try_rewrite_uncorrelated_subquery(

src/query/sql/tests/optimizer/mod.rs

@@ -14,3 +14,4 @@
 
 mod filter;
 mod histogram;
+mod subquery;

src/query/sql/tests/optimizer/subquery/eliminate_in_subquery_test.rs

@@ -0,0 +1,22 @@
+// Copyright 2021 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 databend_common_exception::Result;
+
+#[test]
+fn test_eliminate_simple_in_subquery() -> Result<()> {
+    // TODO: How to build a SExpr and Metadata better
+
+    Ok(())
+}

src/query/sql/tests/optimizer/subquery/mod.rs

@@ -0,0 +1,15 @@
+// Copyright 2021 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.
+
+mod eliminate_in_subquery_test;

tests/sqllogictests/suites/mode/standalone/explain/explain.test

@@ -1330,18 +1330,18 @@ statement ok
 drop table t2;
 
 statement ok
-create OR REPLACE table t1(a int not null, b int, c varchar(20));
+create OR REPLACE table t1(a int, b int not null, c varchar(20));
 
 statement ok
-create OR REPLACE table t2(a int not null, b int, c varchar(20));
+create OR REPLACE table t2(a int, b int not null, c varchar(20));
 
 # eliminate subquery
 query T
-explain select t3.* from t1 as t3 where t3.c = 'D' and t3.a in (select t4.a from t1 as t4 where t4.b = 7);
+explain select t3.* from t1 as t3 where t3.c = 'D' and t3.b in (select t4.b from t1 as t4 where t4.a = 7);
 ----
 Filter
 ├── output columns: [t3.a (#0), t3.b (#1), t3.c (#2)]
-├── filters: [is_true(t1.b (#1) = 7), is_true(t3.c (#2) = 'D')]
+├── filters: [is_true(t1.a (#0) = 7), is_true(t3.c (#2) = 'D')]
 ├── estimated rows: 0.00
 └── TableScan
     ├── table: default.default.t1
@@ -1350,10 +1350,9 @@ Filter
     ├── read size: 0
     ├── partitions total: 0
     ├── partitions scanned: 0
-    ├── push downs: [filters: [and_filters(t1.b (#1) = 7, t1.c (#2) = 'D')], limit: NONE]
+    ├── push downs: [filters: [and_filters(t1.a (#0) = 7, t1.c (#2) = 'D')], limit: NONE]
     └── estimated rows: 0.00
 
-
 # scalar subquery and sort plan contains count() agg function.
 query T
 explain select * from t2 where c > (select c from t1 where t1.a = t2.a group by c order by count(a));