feat(optimizer): support logical filter expression simplify rule

src/frontend/planner_test/tests/testdata/output/temporal_filter.yaml

@@ -383,17 +383,15 @@
       │   └─StreamDynamicFilter { predicate: ($expr1 > now), output_watermarks: [$expr1], output: [t1.ts, $expr1, t1._row_id], cleaned_by_watermark: true }
       │     ├─StreamProject { exprs: [t1.ts, AddWithTimeZone(t1.ts, '01:00:00':Interval, 'UTC':Varchar) as $expr1, t1._row_id] }
       │     │ └─StreamFilter { predicate: Not((t1.ts > '2023-12-18 00:00:00+00:00':Timestamptz)) }
-      │     │   └─StreamShare { id: 2 }
-      │     │     └─StreamFilter { predicate: (Not((t1.ts > '2023-12-18 00:00:00+00:00':Timestamptz)) OR (t1.ts > '2023-12-18 00:00:00+00:00':Timestamptz)) }
-      │     │       └─StreamTableScan { table: t1, columns: [t1.ts, t1._row_id], pk: [t1._row_id], dist: UpstreamHashShard(t1._row_id) }
+      │     │   └─StreamShare { id: 1 }
+      │     │     └─StreamTableScan { table: t1, columns: [t1.ts, t1._row_id], pk: [t1._row_id], dist: UpstreamHashShard(t1._row_id) }
       │     └─StreamExchange { dist: Broadcast }
       │       └─StreamNow { output: [now] }
       └─StreamExchange { dist: HashShard(t1._row_id, 1:Int32) }
         └─StreamProject { exprs: [t1.ts, t1._row_id, 1:Int32] }
           └─StreamFilter { predicate: (t1.ts > '2023-12-18 00:00:00+00:00':Timestamptz) }
-            └─StreamShare { id: 2 }
-              └─StreamFilter { predicate: (Not((t1.ts > '2023-12-18 00:00:00+00:00':Timestamptz)) OR (t1.ts > '2023-12-18 00:00:00+00:00':Timestamptz)) }
-                └─StreamTableScan { table: t1, columns: [t1.ts, t1._row_id], pk: [t1._row_id], dist: UpstreamHashShard(t1._row_id) }
+            └─StreamShare { id: 1 }
+              └─StreamTableScan { table: t1, columns: [t1.ts, t1._row_id], pk: [t1._row_id], dist: UpstreamHashShard(t1._row_id) }
 - name: Temporal filter with or is null
   sql: |
     create table t1 (ts timestamp with time zone);
@@ -406,17 +404,15 @@
       │   └─StreamDynamicFilter { predicate: ($expr1 > now), output_watermarks: [$expr1], output: [t1.ts, $expr1, t1._row_id], cleaned_by_watermark: true }
       │     ├─StreamProject { exprs: [t1.ts, AddWithTimeZone(t1.ts, '01:00:00':Interval, 'UTC':Varchar) as $expr1, t1._row_id] }
       │     │ └─StreamFilter { predicate: Not(IsNull(t1.ts)) }
-      │     │   └─StreamShare { id: 2 }
-      │     │     └─StreamFilter { predicate: (Not(IsNull(t1.ts)) OR IsNull(t1.ts)) }
-      │     │       └─StreamTableScan { table: t1, columns: [t1.ts, t1._row_id], pk: [t1._row_id], dist: UpstreamHashShard(t1._row_id) }
+      │     │   └─StreamShare { id: 1 }
+      │     │     └─StreamTableScan { table: t1, columns: [t1.ts, t1._row_id], pk: [t1._row_id], dist: UpstreamHashShard(t1._row_id) }
       │     └─StreamExchange { dist: Broadcast }
       │       └─StreamNow { output: [now] }
       └─StreamExchange { dist: HashShard(t1._row_id, 1:Int32) }
         └─StreamProject { exprs: [t1.ts, t1._row_id, 1:Int32] }
           └─StreamFilter { predicate: IsNull(t1.ts) }
-            └─StreamShare { id: 2 }
-              └─StreamFilter { predicate: (Not(IsNull(t1.ts)) OR IsNull(t1.ts)) }
-                └─StreamTableScan { table: t1, columns: [t1.ts, t1._row_id], pk: [t1._row_id], dist: UpstreamHashShard(t1._row_id) }
+            └─StreamShare { id: 1 }
+              └─StreamTableScan { table: t1, columns: [t1.ts, t1._row_id], pk: [t1._row_id], dist: UpstreamHashShard(t1._row_id) }
 - name: Temporal filter with or predicate
   sql: |
     create table t1 (ts timestamp with time zone);
@@ -458,7 +454,7 @@
       │     │     │   └─StreamDynamicFilter { predicate: (t.t > $expr1), output_watermarks: [t.t], output: [t.t, t.a, t._row_id], cleaned_by_watermark: true }
       │     │     │     ├─StreamFilter { predicate: (Not((t.a > 1:Int32)) OR (t.a > 1:Int32)) AND Not(IsNull(t.t)) AND Not((t.a < 1:Int32)) }
       │     │     │     │ └─StreamShare { id: 2 }
-      │     │     │     │   └─StreamFilter { predicate: (Not((t.a > 1:Int32)) OR (t.a > 1:Int32)) AND (((Not(IsNull(t.t)) AND Not((t.a < 1:Int32))) OR IsNull(t.t)) OR (t.a < 1:Int32)) }
+      │     │     │     │   └─StreamFilter { predicate: (((Not(IsNull(t.t)) AND Not((t.a < 1:Int32))) OR IsNull(t.t)) OR (t.a < 1:Int32)) }
       │     │     │     │     └─StreamTableScan { table: t, columns: [t.t, t.a, t._row_id], pk: [t._row_id], dist: UpstreamHashShard(t._row_id) }
       │     │     │     └─StreamExchange { dist: Broadcast }
       │     │     │       └─StreamProject { exprs: [SubtractWithTimeZone(now, '01:00:00':Interval, 'UTC':Varchar) as $expr1], output_watermarks: [$expr1] }
@@ -467,7 +463,7 @@
       │     │       └─StreamProject { exprs: [t.t, t.a, t._row_id, 1:Int32] }
       │     │         └─StreamFilter { predicate: (Not((t.a > 1:Int32)) OR (t.a > 1:Int32)) AND (IsNull(t.t) OR (t.a < 1:Int32)) }
       │     │           └─StreamShare { id: 2 }
-      │     │             └─StreamFilter { predicate: (Not((t.a > 1:Int32)) OR (t.a > 1:Int32)) AND (((Not(IsNull(t.t)) AND Not((t.a < 1:Int32))) OR IsNull(t.t)) OR (t.a < 1:Int32)) }
+      │     │             └─StreamFilter { predicate: (((Not(IsNull(t.t)) AND Not((t.a < 1:Int32))) OR IsNull(t.t)) OR (t.a < 1:Int32)) }
       │     │               └─StreamTableScan { table: t, columns: [t.t, t.a, t._row_id], pk: [t._row_id], dist: UpstreamHashShard(t._row_id) }
       │     └─StreamExchange { dist: Broadcast }
       │       └─StreamProject { exprs: [SubtractWithTimeZone(now, '01:00:00':Interval, 'UTC':Varchar) as $expr2], output_watermarks: [$expr2] }
@@ -482,7 +478,7 @@
                 │   └─StreamDynamicFilter { predicate: (t.t > $expr1), output_watermarks: [t.t], output: [t.t, t.a, t._row_id], cleaned_by_watermark: true }
                 │     ├─StreamFilter { predicate: (Not((t.a > 1:Int32)) OR (t.a > 1:Int32)) AND Not(IsNull(t.t)) AND Not((t.a < 1:Int32)) }
                 │     │ └─StreamShare { id: 2 }
-                │     │   └─StreamFilter { predicate: (Not((t.a > 1:Int32)) OR (t.a > 1:Int32)) AND (((Not(IsNull(t.t)) AND Not((t.a < 1:Int32))) OR IsNull(t.t)) OR (t.a < 1:Int32)) }
+                │     │   └─StreamFilter { predicate: (((Not(IsNull(t.t)) AND Not((t.a < 1:Int32))) OR IsNull(t.t)) OR (t.a < 1:Int32)) }
                 │     │     └─StreamTableScan { table: t, columns: [t.t, t.a, t._row_id], pk: [t._row_id], dist: UpstreamHashShard(t._row_id) }
                 │     └─StreamExchange { dist: Broadcast }
                 │       └─StreamProject { exprs: [SubtractWithTimeZone(now, '01:00:00':Interval, 'UTC':Varchar) as $expr1], output_watermarks: [$expr1] }
@@ -491,5 +487,5 @@
                   └─StreamProject { exprs: [t.t, t.a, t._row_id, 1:Int32] }
                     └─StreamFilter { predicate: (Not((t.a > 1:Int32)) OR (t.a > 1:Int32)) AND (IsNull(t.t) OR (t.a < 1:Int32)) }
                       └─StreamShare { id: 2 }
-                        └─StreamFilter { predicate: (Not((t.a > 1:Int32)) OR (t.a > 1:Int32)) AND (((Not(IsNull(t.t)) AND Not((t.a < 1:Int32))) OR IsNull(t.t)) OR (t.a < 1:Int32)) }
+                        └─StreamFilter { predicate: (((Not(IsNull(t.t)) AND Not((t.a < 1:Int32))) OR IsNull(t.t)) OR (t.a < 1:Int32)) }
                           └─StreamTableScan { table: t, columns: [t.t, t.a, t._row_id], pk: [t._row_id], dist: UpstreamHashShard(t._row_id) }

src/frontend/src/optimizer/logical_optimization.rs

@@ -416,6 +416,14 @@ static COMMON_SUB_EXPR_EXTRACT: LazyLock<OptimizationStage> = LazyLock::new(|| {
     )
 });
 
+static SIMPLIFY_FILTER_EXPRESSION: LazyLock<OptimizationStage> = LazyLock::new(|| {
+    OptimizationStage::new(
+        "Simplify Filter Expression",
+        vec![SimplifyFilterExpressionRule::create()],
+        ApplyOrder::TopDown,
+    )
+});
+
 impl LogicalOptimizer {
     pub fn predicate_pushdown(
         plan: PlanRef,
@@ -624,6 +632,8 @@ impl LogicalOptimizer {
 
         plan = plan.optimize_by_rules(&COMMON_SUB_EXPR_EXTRACT);
 
+        plan = plan.optimize_by_rules(&SIMPLIFY_FILTER_EXPRESSION);
+
         #[cfg(debug_assertions)]
         InputRefValidator.validate(plan.clone());
 
@@ -720,6 +730,8 @@ impl LogicalOptimizer {
 
         plan = plan.optimize_by_rules(&DAG_TO_TREE);
 
+        plan = plan.optimize_by_rules(&SIMPLIFY_FILTER_EXPRESSION);
+
         #[cfg(debug_assertions)]
         InputRefValidator.validate(plan.clone());
 

src/frontend/src/optimizer/plan_node/expr_rewritable.rs

@@ -18,7 +18,7 @@ use super::*;
 use crate::expr::ExprRewriter;
 
 /// Rewrites expressions in a `PlanRef`. Due to `Share` operator,
-/// the `ExprRewriter` needs to be idempotent i.e. applying it more than once
+/// the `ExprRewriter` needs to be idempotent i.e., applying it more than once
 /// to the same `ExprImpl` will be a noop on subsequent applications.
 /// `rewrite_exprs` should only return a plan with the given node modified.
 /// To rewrite recursively, call `rewrite_exprs_recursive` on [`RewriteExprsRecursive`].

src/frontend/src/optimizer/rule/mod.rs

@@ -90,6 +90,7 @@ pub use top_n_on_index_rule::*;
 mod stream;
 pub use stream::bushy_tree_join_ordering_rule::*;
 pub use stream::filter_with_now_to_join_rule::*;
+pub use stream::simplify_filter_expression_rule::*;
 pub use stream::split_now_and_rule::*;
 pub use stream::split_now_or_rule::*;
 pub use stream::stream_project_merge_rule::*;
@@ -204,6 +205,7 @@ macro_rules! for_all_rules {
             , { AlwaysFalseFilterRule }
             , { BushyTreeJoinOrderingRule }
             , { StreamProjectMergeRule }
+            , { SimplifyFilterExpressionRule }
             , { JoinProjectTransposeRule }
             , { LimitPushDownRule }
             , { PullUpHopRule }

src/frontend/src/optimizer/rule/stream/mod.rs

@@ -14,6 +14,7 @@
 
 pub(crate) mod bushy_tree_join_ordering_rule;
 pub(crate) mod filter_with_now_to_join_rule;
+pub(crate) mod simplify_filter_expression_rule;
 pub(crate) mod split_now_and_rule;
 pub(crate) mod split_now_or_rule;
 pub(crate) mod stream_project_merge_rule;

src/frontend/src/optimizer/rule/stream/simplify_filter_expression_rule.rs

@@ -0,0 +1,102 @@
+// Copyright 2024 RisingWave 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 risingwave_common::types::DataType;
+
+use crate::expr::{
+    Expr, ExprImpl, ExprRewriter, ExprType,
+};
+use crate::optimizer::plan_node::{
+    ExprRewritable, LogicalFilter, LogicalShare, PlanTreeNodeUnary,
+};
+use crate::optimizer::rule::{BoxedRule, Rule};
+use crate::optimizer::PlanRef;
+
+pub struct SimplifyFilterExpressionRule {}
+impl Rule for SimplifyFilterExpressionRule {
+    /// Currently the rule is only applied to the potential filters in `LogicalShare`
+    fn apply(&self, plan: PlanRef) -> Option<PlanRef> {
+        let filter: &LogicalFilter = plan.as_logical_filter()?;
+        let mut rewriter = SimplifyFilterExpressionRewriter {};
+        let logical_share_plan = filter.input();
+        let share: &LogicalShare = logical_share_plan.as_logical_share()?;
+        let input = share.input().rewrite_exprs(&mut rewriter);
+        share.replace_input(input);
+        Some(LogicalFilter::create(
+            share.clone().into(),
+            filter.predicate().clone(),
+        ))
+    }
+}
+
+impl SimplifyFilterExpressionRule {
+    pub fn create() -> BoxedRule {
+        Box::new(SimplifyFilterExpressionRule {})
+    }
+}
+
+/// If ever `Not (e)` and `(e)` appear together
+fn check_pattern(e1: ExprImpl, e2: ExprImpl) -> bool {
+    let ExprImpl::FunctionCall(e1_func) = e1.clone() else {
+        return false;
+    };
+    let ExprImpl::FunctionCall(e2_func) = e2.clone() else {
+        return false;
+    };
+    if e1_func.func_type() != ExprType::Not && e2_func.func_type() != ExprType::Not {
+        return false;
+    }
+    if e1_func.func_type() != ExprType::Not {
+        if e2_func.inputs().len() != 1 {
+            return false;
+        }
+        e1 == e2_func.inputs()[0].clone()
+    } else {
+        if e1_func.inputs().len() != 1 {
+            return false;
+        }
+        e2 == e1_func.inputs()[0].clone()
+    }
+}
+
+struct SimplifyFilterExpressionRewriter {}
+
+impl ExprRewriter for SimplifyFilterExpressionRewriter {
+    /// The pattern we aim to optimize, e.g.,
+    /// 1. (NOT (e)) OR (e) => True | (NOT (e)) AND (e) => False
+    /// 2. (NOT (e1) AND NOT (e2)) OR (e1 OR e2) => True # TODO
+    fn rewrite_expr(&mut self, expr: ExprImpl) -> ExprImpl {
+        let ExprImpl::FunctionCall(func_call) = expr.clone() else {
+            return expr;
+        };
+        if func_call.func_type() != ExprType::Or && func_call.func_type() != ExprType::And {
+            return expr;
+        }
+        assert_eq!(func_call.return_type(), DataType::Boolean);
+        // Currently just optimize the first rule
+        if func_call.inputs().len() != 2 {
+            return expr;
+        }
+        let inputs = func_call.inputs();
+        if check_pattern(inputs[0].clone(), inputs[1].clone()) {
+            match func_call.func_type() {
+                ExprType::Or => ExprImpl::literal_bool(true),
+                ExprType::And => ExprImpl::literal_bool(false),
+                _ => expr,
+            }
+        } else {
+            expr
+        }
+    }
+}