feat: Add RuleFilterFlattenOr (#18621)

* feat: Add `RuleFilterFlattenOr` is used to fold the outer `or` of the predicate in Filter into `or_filters`

* chore: codefmt

* fix: fix `or_filters` compatibility with statistics calculation and optimization rules

* chore: codefmt

* chore: codefmt

* chore: codefmt

Author: KKould

src/query/expression/src/expression.rs

@@ -771,9 +771,9 @@ impl<Index: ColumnIndex> Expr<Index> {
         }
     }
 
-    pub fn visit_func(&self, func_name: &str, visitor: &mut impl FnMut(&FunctionCall<Index>)) {
+    pub fn visit_func(&self, func_names: &[&str], visitor: &mut impl FnMut(&FunctionCall<Index>)) {
         struct Visitor<'a, Index: ColumnIndex, F: FnMut(&FunctionCall<Index>)> {
-            name: String,
+            names: &'a [&'a str],
             visitor: &'a mut F,
             _marker: std::marker::PhantomData<Index>,
         }
@@ -785,15 +785,15 @@ impl<Index: ColumnIndex> Expr<Index> {
                 &mut self,
                 call: &FunctionCall<Index>,
             ) -> Result<Option<Expr<Index>>, Self::Error> {
-                if call.function.signature.name == self.name {
+                if self.names.contains(&call.function.signature.name.as_str()) {
                     (self.visitor)(call);
                 }
                 Self::visit_function_call(call, self)
             }
         }
 
         visit_expr(self, &mut Visitor {
-            name: func_name.to_string(),
+            names: func_names,
             visitor,
             _marker: std::marker::PhantomData,
         })

src/query/expression/src/utils/filter_helper.rs

@@ -113,7 +113,7 @@ impl FilterHelpers {
                     // for the equality columns set,let's call `ecs`
                     // if any side of `or` of the name columns is not in `ecs`, it's valid
                     // otherwise, it's invalid
-                    expr.visit_func("or", &mut |call| {
+                    expr.visit_func(&["or", "or_filters"], &mut |call| {
                         for arg in call.args.iter() {
                             let mut ecs = HashSet::new();
                             arg.find_function_literals("eq", &mut |col_name, _scalar, _| {

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

@@ -12,15 +12,19 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::borrow::Cow;
 use std::collections::HashSet;
 
 use databend_common_exception::Result;
 use databend_common_expression::types::DataType;
+use databend_common_expression::Scalar;
 
 use crate::optimizer::ir::RelationalProperty;
 use crate::plans::walk_expr;
 use crate::plans::BoundColumnRef;
 use crate::plans::CastExpr;
+use crate::plans::ConstantExpr;
+use crate::plans::FunctionCall;
 use crate::plans::ScalarExpr;
 use crate::plans::Visitor;
 
@@ -111,13 +115,32 @@ pub enum JoinPredicate<'a> {
     Left(&'a ScalarExpr),
     Right(&'a ScalarExpr),
     Both {
-        left: &'a ScalarExpr,
-        right: &'a ScalarExpr,
+        left: Box<Cow<'a, ScalarExpr>>,
+        right: Box<Cow<'a, ScalarExpr>>,
         is_equal_op: bool,
     },
     Other(&'a ScalarExpr),
 }
 
+fn fold_or_arguments(iter: impl Iterator<Item = ScalarExpr>) -> ScalarExpr {
+    iter.fold(
+        ConstantExpr {
+            span: None,
+            value: Scalar::Boolean(false),
+        }
+        .into(),
+        |acc, arg| {
+            FunctionCall {
+                span: None,
+                func_name: "or".to_string(),
+                params: vec![],
+                arguments: vec![acc, arg.clone()],
+            }
+            .into()
+        },
+    )
+}
+
 impl<'a> JoinPredicate<'a> {
     pub fn new(
         scalar: &'a ScalarExpr,
@@ -128,7 +151,35 @@ impl<'a> JoinPredicate<'a> {
             return Self::ALL(scalar);
         }
 
+        if satisfied_by(scalar, left_prop) {
+            return Self::Left(scalar);
+        }
+
+        if satisfied_by(scalar, right_prop) {
+            return Self::Right(scalar);
+        }
+
         if let ScalarExpr::FunctionCall(func) = scalar {
+            if func.func_name == "or_filters" && func.arguments.len() > 1 {
+                let mut left_exprs = Vec::new();
+                let mut right_exprs = Vec::new();
+
+                for expr in func.arguments.iter() {
+                    if satisfied_by(expr, left_prop) {
+                        left_exprs.push(expr.clone());
+                    } else if satisfied_by(expr, right_prop) {
+                        right_exprs.push(expr.clone());
+                    } else {
+                        return Self::Other(scalar);
+                    }
+                }
+                return Self::Both {
+                    left: Box::new(Cow::Owned(fold_or_arguments(left_exprs.into_iter()))),
+                    right: Box::new(Cow::Owned(fold_or_arguments(right_exprs.into_iter()))),
+                    is_equal_op: false,
+                };
+            }
+
             if func.arguments.len() > 2 {
                 return Self::Other(scalar);
             }
@@ -140,30 +191,22 @@ impl<'a> JoinPredicate<'a> {
 
                 if satisfied_by(left, left_prop) && satisfied_by(right, right_prop) {
                     return Self::Both {
-                        left,
-                        right,
+                        left: Box::new(Cow::Borrowed(left)),
+                        right: Box::new(Cow::Borrowed(right)),
                         is_equal_op,
                     };
                 }
 
                 if satisfied_by(right, left_prop) && satisfied_by(left, right_prop) {
                     return Self::Both {
-                        left: right,
-                        right: left,
+                        left: Box::new(Cow::Borrowed(right)),
+                        right: Box::new(Cow::Borrowed(left)),
                         is_equal_op,
                     };
                 }
             }
         }
 
-        if satisfied_by(scalar, left_prop) {
-            return Self::Left(scalar);
-        }
-
-        if satisfied_by(scalar, right_prop) {
-            return Self::Right(scalar);
-        }
-
         Self::Other(scalar)
     }
 }

src/query/sql/src/planner/optimizer/ir/stats/selectivity.rs

@@ -97,10 +97,16 @@ impl<'a> SelectivityEstimator<'a> {
                 left_selectivity.min(right_selectivity)
             }
 
-            ScalarExpr::FunctionCall(func) if func.func_name == "or" => {
-                let left_selectivity = self.compute_selectivity(&func.arguments[0], false)?;
-                let right_selectivity = self.compute_selectivity(&func.arguments[1], false)?;
-                left_selectivity + right_selectivity - left_selectivity * right_selectivity
+            ScalarExpr::FunctionCall(func)
+                if matches!(func.func_name.as_str(), "or" | "or_filters") =>
+            {
+                func.arguments
+                    .iter()
+                    .map(|arg| self.compute_selectivity(arg, false))
+                    .try_fold(0.0, |acc, p| {
+                        let p = p?;
+                        Result::Ok(acc + p - acc * p)
+                    })?
             }
 
             ScalarExpr::FunctionCall(func) if func.func_name == "not" => {

src/query/sql/src/planner/optimizer/optimizers/operator/decorrelate/decorrelate.rs

@@ -156,8 +156,8 @@ impl SubqueryDecorrelatorOptimizer {
                     ..
                 } => {
                     if is_equal_op {
-                        left_conditions.push(left.clone());
-                        right_conditions.push(right.clone());
+                        left_conditions.push(left.into_owned());
+                        right_conditions.push(right.into_owned());
                     } else {
                         non_equi_conditions.push(pred.clone());
                     }

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

@@ -79,7 +79,9 @@ fn predicate_scalar(scalar: &ScalarExpr) -> PredicateScalar {
             }
             PredicateScalar::And(and_args)
         }
-        ScalarExpr::FunctionCall(func) if func.func_name == "or" => {
+        ScalarExpr::FunctionCall(func)
+            if matches!(func.func_name.as_str(), "or" | "or_filters") =>
+        {
             let mut or_args = vec![];
             for argument in func.arguments.iter() {
                 // Recursively flatten the OR expressions.

src/query/sql/src/planner/optimizer/optimizers/rule/factory.rs

@@ -24,6 +24,7 @@ use crate::optimizer::optimizers::rule::RuleEliminateEvalScalar;
 use crate::optimizer::optimizers::rule::RuleEliminateFilter;
 use crate::optimizer::optimizers::rule::RuleEliminateSort;
 use crate::optimizer::optimizers::rule::RuleEliminateUnion;
+use crate::optimizer::optimizers::rule::RuleFilterFlattenOr;
 use crate::optimizer::optimizers::rule::RuleFilterNulls;
 use crate::optimizer::optimizers::rule::RuleFoldCountAggregate;
 use crate::optimizer::optimizers::rule::RuleGroupingSetsToUnion;
@@ -73,6 +74,7 @@ impl RuleFactory {
             RuleID::FilterNulls => Ok(Box::new(RuleFilterNulls::new(
                 ctx.get_enable_distributed_optimization(),
             ))),
+            RuleID::FilterFlattenOr => Ok(Box::new(RuleFilterFlattenOr::new())),
             RuleID::PushDownFilterUnion => Ok(Box::new(RulePushDownFilterUnion::new())),
             RuleID::PushDownFilterEvalScalar => Ok(Box::new(RulePushDownFilterEvalScalar::new())),
             RuleID::PushDownFilterJoin => Ok(Box::new(RulePushDownFilterJoin::new(metadata))),

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

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 mod rule_eliminate_filter;
+mod rule_filter_flatten_or;
 mod rule_filter_nulls;
 mod rule_merge_filter;
 mod rule_merge_filter_into_mutation;
@@ -30,6 +31,7 @@ mod rule_push_down_sort_filter_scan;
 mod rule_push_down_sort_scan;
 
 pub use rule_eliminate_filter::RuleEliminateFilter;
+pub use rule_filter_flatten_or::RuleFilterFlattenOr;
 pub use rule_filter_nulls::RuleFilterNulls;
 pub use rule_merge_filter::RuleMergeFilter;
 pub use rule_merge_filter_into_mutation::RuleMergeFilterIntoMutation;

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

@@ -0,0 +1,123 @@
+// 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 std::sync::Arc;
+
+use databend_common_exception::Result;
+use databend_common_expression::Scalar;
+
+use crate::optimizer::ir::Matcher;
+use crate::optimizer::ir::SExpr;
+use crate::optimizer::optimizers::rule::Rule;
+use crate::optimizer::optimizers::rule::RuleID;
+use crate::optimizer::optimizers::rule::TransformResult;
+use crate::plans::ConstantExpr;
+use crate::plans::Filter;
+use crate::plans::FunctionCall;
+use crate::plans::RelOp;
+use crate::plans::ScalarExpr;
+
+pub struct RuleFilterFlattenOr {
+    id: RuleID,
+    matchers: Vec<Matcher>,
+}
+
+impl RuleFilterFlattenOr {
+    pub fn new() -> Self {
+        Self {
+            id: RuleID::FilterFlattenOr,
+            // Filter
+            //  \
+            //   *
+            matchers: vec![Matcher::MatchOp {
+                op_type: RelOp::Filter,
+                children: vec![Matcher::Leaf],
+            }],
+        }
+    }
+}
+
+impl Default for RuleFilterFlattenOr {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl Rule for RuleFilterFlattenOr {
+    fn id(&self) -> RuleID {
+        self.id
+    }
+
+    fn apply(&self, s_expr: &SExpr, state: &mut TransformResult) -> Result<()> {
+        let mut filter: Filter = s_expr.plan().clone().try_into()?;
+        let mut has_replace = false;
+
+        for predicate in filter.predicates.iter_mut() {
+            let mut or_exprs = Vec::new();
+            flatten_or_expr(predicate, &mut or_exprs);
+
+            if or_exprs.len() > 2 {
+                let replace_expr = FunctionCall {
+                    span: None,
+                    func_name: "or_filters".to_string(),
+                    params: vec![],
+                    arguments: or_exprs,
+                }
+                .into();
+                if predicate == &replace_expr {
+                    continue;
+                }
+                *predicate = replace_expr;
+                has_replace = true
+            }
+        }
+        if !has_replace {
+            state.add_result(s_expr.clone());
+            return Ok(());
+        }
+        let mut res =
+            SExpr::create_unary(Arc::new(filter.into()), Arc::new(s_expr.child(0)?.clone()));
+        res.set_applied_rule(&self.id());
+        state.add_result(res);
+
+        Ok(())
+    }
+
+    fn matchers(&self) -> &[Matcher] {
+        &self.matchers
+    }
+}
+
+fn flatten_or_expr(expr: &ScalarExpr, or_exprs: &mut Vec<ScalarExpr>) {
+    match expr {
+        ScalarExpr::FunctionCall(func)
+            if matches!(func.func_name.as_str(), "or" | "or_filters") =>
+        {
+            for argument in func.arguments.iter() {
+                flatten_or_expr(argument, or_exprs);
+            }
+        }
+        ScalarExpr::ConstantExpr(ConstantExpr { value, span }) if value.is_null() => {
+            // predicates cannot directly pass null
+            or_exprs.push(
+                ConstantExpr {
+                    span: *span,
+                    value: Scalar::Boolean(false),
+                }
+                .into(),
+            )
+        }
+        _ => or_exprs.push(expr.clone()),
+    }
+}

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

@@ -267,8 +267,8 @@ pub fn try_push_down_filter_join(s_expr: &SExpr, metadata: MetadataRef) -> Resul
             }
             JoinPredicate::Both { left, right, .. } => {
                 join.equi_conditions.push(JoinEquiCondition::new(
-                    left.clone(),
-                    right.clone(),
+                    left.into_owned(),
+                    right.into_owned(),
                     false,
                 ));
             }