chore(query): refactor the cardinality of anti-join

Author: zhang2014

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

@@ -263,14 +263,14 @@ pub async fn optimize_query(opt_ctx: Arc<OptimizerContext>, s_expr: SExpr) -> Re
         ]))
         // 10. Apply DPhyp algorithm for cost-based join reordering
         .add(DPhpyOptimizer::new(opt_ctx.clone()))
-        .add(RecursiveRuleOptimizer::new(
-            opt_ctx.clone(),
-            [RuleID::PushDownAntiJoin].as_slice(),
-        ))
         // 11. After join reorder, Convert some single join to inner join.
         .add(SingleToInnerOptimizer::new())
         // 12. Deduplicate join conditions.
         .add(DeduplicateJoinConditionOptimizer::new())
+        .add(RecursiveRuleOptimizer::new(
+            opt_ctx.clone(),
+            [RuleID::PushDownAntiJoin].as_slice(),
+        ))
         // 13. Apply join commutativity to further optimize join ordering
         .add_if(
             opt_ctx.get_enable_join_reorder(),

src/query/sql/src/planner/optimizer/optimizers/operator/filter/infer_filter.rs

@@ -732,18 +732,6 @@ impl<'a> InferFilterOptimizer<'a> {
                                 arguments: vec![
                                     self.exprs[equal_indexes[i]].clone(),
                                     self.exprs[equal_indexes[j]].clone(),
-                                    // ScalarExpr::FunctionCall(FunctionCall {
-                                    //     span: None,
-                                    //     func_name: "infer_predicate".to_string(),
-                                    //     params: vec![],
-                                    //     arguments: vec![self.exprs[equal_indexes[i]].clone()],
-                                    // }),
-                                    // ScalarExpr::FunctionCall(FunctionCall {
-                                    //     span: None,
-                                    //     func_name: "infer_predicate".to_string(),
-                                    //     params: vec![],
-                                    //     arguments: vec![self.exprs[equal_indexes[j]].clone()],
-                                    // }),
                                 ],
                             }));
                         }

src/query/sql/src/planner/optimizer/optimizers/rule/filter_rules/rule_push_down_filter_join.rs

@@ -21,6 +21,8 @@ use crate::optimizer::ir::Matcher;
 use crate::optimizer::ir::RelExpr;
 use crate::optimizer::ir::SExpr;
 use crate::optimizer::optimizers::operator::EquivalentConstantsVisitor;
+use crate::optimizer::optimizers::operator::InferFilterOptimizer;
+use crate::optimizer::optimizers::operator::JoinProperty;
 use crate::optimizer::optimizers::rule::can_filter_null;
 use crate::optimizer::optimizers::rule::constant::false_constant;
 use crate::optimizer::optimizers::rule::constant::is_falsy;
@@ -240,9 +242,10 @@ pub fn try_push_down_filter_join(s_expr: &SExpr, metadata: MetadataRef) -> Resul
                 right_push_down = vec![];
             }
         }
-        // let join_prop = JoinProperty::new(&left_prop.output_columns, &right_prop.output_columns);
-        // let mut infer_filter = InferFilterOptimizer::new(Some(join_prop));
-        // push_down_predicates = infer_filter.optimize(push_down_predicates)?;
+
+        let join_prop = JoinProperty::new(&left_prop.output_columns, &right_prop.output_columns);
+        let mut infer_filter = InferFilterOptimizer::new(Some(join_prop));
+        push_down_predicates = infer_filter.optimize(push_down_predicates)?;
     }
 
     let mut all_push_down = vec![];

src/query/sql/src/planner/optimizer/optimizers/rule/join_rules/rule_push_down_anti_join.rs

@@ -68,15 +68,6 @@ impl RulePushdownAntiJoin {
                 .collect::<Vec<_>>();
 
             if equi_conditions.iter().all(left_predicate) {
-                // let mut new_equi_conditions = Vec::with_capacity(equi_conditions.len());
-
-                // for (idx, (inferred, predicate)) in equi_conditions.into_iter().enumerate() {
-                //     if !inferred || matches!(predicate, JoinPredicate::Left(_)) {
-                //         new_equi_conditions.push(join.equi_conditions[idx].clone());
-                //     }
-                // }
-
-                // join.equi_conditions = new_equi_conditions;
                 let right_prop = right_rel_expr.derive_relational_prop()?;
                 let mut union_output_columns = ColumnSet::new();
                 union_output_columns.extend(right_prop.output_columns.clone());
@@ -99,15 +90,6 @@ impl RulePushdownAntiJoin {
                     return replace_inner_join(left, new_inner_join);
                 }
             } else if equi_conditions.iter().all(right_predicate) {
-                // let mut new_equi_conditions = Vec::with_capacity(equi_conditions.len());
-
-                // for (idx, (inferred, predicate)) in equi_conditions.into_iter().enumerate() {
-                //     if !inferred || matches!(predicate, JoinPredicate::Left(_)) {
-                //         new_equi_conditions.push(join.equi_conditions[idx].clone());
-                //     }
-                // }
-
-                // join.equi_conditions = new_equi_conditions;
                 let right_prop = right_rel_expr.derive_relational_prop()?;
                 let mut union_output_columns = ColumnSet::new();
                 union_output_columns.extend(right_prop.output_columns.clone());
@@ -186,46 +168,6 @@ fn extract_inner_join(expr: &SExpr) -> Result<Option<SExpr>> {
     }
 }
 
-// struct ColumnMappingRewriter<'a> {
-//     mapping: &'a HashMap<usize, ColumnBinding>,
-// }
-//
-// impl VisitorMut<'_> for ColumnMappingRewriter {
-//     fn visit_bound_column_ref(&mut self, col: &mut BoundColumnRef) -> Result<()> {
-//         if let Some(&new_index) = self.mapping.get(&col.column.index) {
-//             col.column = new_index.clone();
-//         }
-//         Ok(())
-//     }
-// }
-//
-// fn replace_by_equivalence(
-//     expr: &ScalarExpr,
-//     mapping: &HashMap<IndexType, ColumnBinding>,
-// ) -> Result<ScalarExpr> {
-//     if mapping.is_empty() {
-//         return Ok(expr.clone());
-//     }
-//
-//     let mut new_expr = expr.clone();
-//     let mut rewriter = ColumnMappingRewriter { mapping };
-//     rewriter.visit(&mut new_expr)?;
-//     Ok(new_expr)
-// }
-//
-// fn collect_mapping(join: &Join) -> Result<Vec<(ColumnBinding, ColumnBinding)>> {
-//     for equi_condition in &join.equi_conditions {
-//         match equi_condition.left
-//     }
-// }
-//
-// fn all_left(
-//     conditions: &[JoinEquiCondition],
-//     inner_join: Join,
-//     left: Arc<RelationalProperty>,
-// ) -> Result<bool> {
-// }
-
 fn left_predicate(tuple: &JoinPredicate) -> bool {
     matches!(&tuple, JoinPredicate::Left(_))
 }