fix(optimizer): consider impure expression in predicate push down of Project

src/frontend/src/expr/mod.rs

@@ -32,6 +32,7 @@ mod parameter;
 mod subquery;
 mod table_function;
 mod user_defined_function;
+mod volatility;
 mod window_function;
 
 mod order_by_expr;
@@ -63,6 +64,7 @@ pub use type_inference::{
 };
 pub use user_defined_function::UserDefinedFunction;
 pub use utils::*;
+pub use volatility::*;
 pub use window_function::{WindowFunction, WindowFunctionType};
 
 /// the trait of bound expressions

src/frontend/src/expr/volatility.rs

@@ -0,0 +1,34 @@
+use super::{ExprImpl, ExprVisitor};
+
+/// describe the "volatility" for all expressions. see <https://www.postgresql.org/docs/current/xfunc-volatility.html> and <https://github.com/risingwavelabs/risingwave/issues/9030> for more information.
+#[derive(Clone, Debug, Default, PartialEq, Eq)]
+pub enum Volatility {
+    Volatile,
+    // we do not distinguish the VOLATILE and STABLE function
+    // STABLE,
+    #[default]
+    Immutable,
+}
+
+struct VolatilityAnalyzer {}
+
+impl ExprVisitor<Volatility> for VolatilityAnalyzer {
+    fn merge(a: Volatility, b: Volatility) -> Volatility {
+        if a == Volatility::Volatile || b == Volatility::Volatile {
+            return Volatility::Volatile;
+        }
+        Volatility::Immutable
+    }
+
+    fn visit_user_defined_function(
+        &mut self,
+        _func_call: &super::UserDefinedFunction,
+    ) -> Volatility {
+        Volatility::Volatile
+    }
+}
+
+pub fn derive_volatility(expr: &ExprImpl) -> Volatility {
+    let mut a = VolatilityAnalyzer {};
+    a.visit_expr(expr)
+}

src/frontend/src/optimizer/plan_node/logical_project.rs

@@ -210,9 +210,10 @@ impl PredicatePushdown for LogicalProject {
         let mut subst = Substitute {
             mapping: self.exprs().clone(),
         };
-        let predicate = predicate.rewrite_expr(&mut subst);
+        let (pushed_cond, remained_cond) = predicate.split_immutable_expr();
+        let pushed_cond = pushed_cond.rewrite_expr(&mut subst);
 
-        gen_filter_and_pushdown(self, Condition::true_cond(), predicate, ctx)
+        gen_filter_and_pushdown(self, remained_cond, pushed_cond, ctx)
     }
 }
 

src/frontend/src/utils/condition.rs

@@ -26,9 +26,9 @@ use risingwave_common::types::{DataType, ScalarImpl};
 use risingwave_common::util::scan_range::{is_full_range, ScanRange};
 
 use crate::expr::{
-    factorization_expr, fold_boolean_constant, push_down_not, to_conjunctions,
+    derive_volatility, factorization_expr, fold_boolean_constant, push_down_not, to_conjunctions,
     try_get_bool_constant, ExprDisplay, ExprImpl, ExprMutator, ExprRewriter, ExprType, ExprVisitor,
-    FunctionCall, InequalityInputPair, InputRef,
+    FunctionCall, InequalityInputPair, InputRef, Volatility,
 };
 use crate::utils::condition::cast_compare::{ResultForCmp, ResultForEq};
 
@@ -144,6 +144,20 @@ impl Condition {
         .unwrap()
     }
 
+    /// Split the condition into 2 groups: immutable(pure/deterministic) and volatility functions
+    pub fn split_immutable_expr(self) -> (Self, Self) {
+        self.group_by::<_, 2>(|expr| {
+            if derive_volatility(expr) == Volatility::Immutable {
+                0
+            } else {
+                1
+            }
+        })
+        .into_iter()
+        .next_tuple()
+        .unwrap()
+    }
+
     pub fn collect_input_refs(&self, input_col_num: usize) -> FixedBitSet {
         let mut input_bits = FixedBitSet::with_capacity(input_col_num);
         for expr in &self.conjunctions {