risingwave_frontend/expr/
mod.rs

1// Copyright 2022 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use enum_as_inner::EnumAsInner;
16use fixedbitset::FixedBitSet;
17use futures::FutureExt;
18use paste::paste;
19use risingwave_common::array::ListValue;
20use risingwave_common::types::{
21    DataType, Datum, JsonbVal, MapType, Scalar, ScalarImpl, ToOwnedDatum,
22};
23use risingwave_expr::aggregate::PbAggKind;
24use risingwave_expr::expr::build_from_prost;
25use risingwave_pb::expr::expr_node::RexNode;
26use risingwave_pb::expr::{ExprNode, ProjectSetSelectItem};
27use user_defined_function::UserDefinedFunctionDisplay;
28
29use crate::error::{ErrorCode, Result as RwResult};
30use crate::session::current;
31
32mod agg_call;
33mod correlated_input_ref;
34mod function_call;
35mod function_call_with_lambda;
36mod input_ref;
37mod literal;
38mod now;
39mod parameter;
40mod pure;
41mod subquery;
42mod table_function;
43mod user_defined_function;
44mod window_function;
45
46mod order_by_expr;
47pub use order_by_expr::{OrderBy, OrderByExpr};
48
49mod expr_mutator;
50mod expr_rewriter;
51mod expr_visitor;
52pub mod function_impl;
53mod secret_ref;
54mod session_timezone;
55mod type_inference;
56mod utils;
57
58pub use agg_call::AggCall;
59pub use correlated_input_ref::{CorrelatedId, CorrelatedInputRef, InputRefScopeRewriter, ScopeId};
60pub use expr_mutator::ExprMutator;
61pub use expr_rewriter::{ExprRewriter, default_rewrite_expr};
62pub use expr_visitor::{ExprVisitor, default_visit_expr};
63pub use function_call::{FunctionCall, FunctionCallDisplay, is_row_function};
64pub use function_call_with_lambda::FunctionCallWithLambda;
65pub use input_ref::{InputRef, InputRefDisplay, input_ref_to_column_indices};
66pub use literal::Literal;
67pub use now::{InlineNowProcTime, Now, NowProcTimeFinder};
68pub use parameter::Parameter;
69pub use pure::*;
70pub use risingwave_pb::expr::expr_node::Type as ExprType;
71pub use secret_ref::SecretRef;
72pub use session_timezone::{SessionTimezone, TimestamptzExprFinder};
73pub use subquery::{Subquery, SubqueryKind};
74pub(crate) use table_function::{MSSQL_QUERY_ARG_COUNT, expr_impl_to_string_fn};
75pub use table_function::{TableFunction, TableFunctionType};
76pub use type_inference::*;
77pub use user_defined_function::UserDefinedFunction;
78pub use utils::*;
79pub use window_function::WindowFunction;
80
81pub(crate) const EXPR_DEPTH_THRESHOLD: usize = 30;
82pub(crate) const EXPR_TOO_DEEP_NOTICE: &str = "Some expression is too complicated. \
83Consider simplifying or splitting the query if you encounter any issues.";
84
85pub(crate) fn reject_impure(expr: impl Into<ExprImpl>, context: &str) -> RwResult<()> {
86    if let Some(impure_expr_desc) = impure_expr_desc(&expr.into()) {
87        let msg = format!(
88            "using an impure expression ({impure_expr_desc}) in {context} \
89             on a retract stream may lead to inconsistent results"
90        );
91        if current::config()
92            .is_some_and(|c| c.read().streaming_unsafe_allow_unmaterialized_impure_expr())
93        {
94            current::notice_to_user(msg);
95        } else {
96            return Err(ErrorCode::NotSupported(
97                msg,
98                "rewrite the query to extract the impure expression into the select list, \
99                 or set `streaming_unsafe_allow_unmaterialized_impure_expr` to allow \
100                 the behavior at the risk of inconsistent results or panics during execution"
101                    .into(),
102            )
103            .into());
104        }
105    }
106    Ok(())
107}
108
109/// the trait of bound expressions
110pub trait Expr: Into<ExprImpl> {
111    /// Get the return type of the expr
112    fn return_type(&self) -> DataType;
113
114    /// Try to serialize the expression, returning an error if it's impossible.
115    fn try_to_expr_proto(&self) -> Result<ExprNode, String>;
116
117    /// Serialize the expression. Panic if it's impossible.
118    fn to_expr_proto(&self) -> ExprNode {
119        self.try_to_expr_proto()
120            .expect("failed to serialize expression to protobuf")
121    }
122
123    /// Serialize the expression. Returns an error if this will result in an impure expression on a
124    /// retract stream, which may lead to inconsistent results.
125    fn to_expr_proto_checked_pure(
126        &self,
127        retract: bool,
128        context: &str,
129    ) -> crate::error::Result<ExprNode>
130    where
131        Self: Clone,
132    {
133        if retract {
134            reject_impure(self.clone(), context)?;
135        }
136        self.try_to_expr_proto()
137            .map_err(|e| ErrorCode::InternalError(e).into())
138    }
139}
140
141macro_rules! impl_expr_impl {
142    ($($t:ident,)*) => {
143        #[derive(Clone, Eq, PartialEq, Hash, EnumAsInner)]
144        pub enum ExprImpl {
145            $($t(Box<$t>),)*
146        }
147
148        impl ExprImpl {
149            pub fn variant_name(&self) -> &'static str {
150                match self {
151                    $(ExprImpl::$t(_) => stringify!($t),)*
152                }
153            }
154        }
155
156        $(
157        impl From<$t> for ExprImpl {
158            fn from(o: $t) -> ExprImpl {
159                ExprImpl::$t(Box::new(o))
160            }
161        })*
162
163        impl Expr for ExprImpl {
164            fn return_type(&self) -> DataType {
165                match self {
166                    $(ExprImpl::$t(expr) => expr.return_type(),)*
167                }
168            }
169
170            fn try_to_expr_proto(&self) -> Result<ExprNode, String> {
171                match self {
172                    $(ExprImpl::$t(expr) => expr.try_to_expr_proto(),)*
173                }
174            }
175        }
176    };
177}
178
179impl_expr_impl!(
180    // BoundColumnRef, might be used in binder.
181    CorrelatedInputRef,
182    InputRef,
183    Literal,
184    FunctionCall,
185    FunctionCallWithLambda,
186    AggCall,
187    Subquery,
188    TableFunction,
189    WindowFunction,
190    UserDefinedFunction,
191    Parameter,
192    Now,
193    SecretRef,
194);
195
196impl ExprImpl {
197    /// A literal int value.
198    #[inline(always)]
199    pub fn literal_int(v: i32) -> Self {
200        Literal::new(Some(v.to_scalar_value()), DataType::Int32).into()
201    }
202
203    /// A literal bigint value
204    #[inline(always)]
205    pub fn literal_bigint(v: i64) -> Self {
206        Literal::new(Some(v.to_scalar_value()), DataType::Int64).into()
207    }
208
209    /// A literal float64 value.
210    #[inline(always)]
211    pub fn literal_f64(v: f64) -> Self {
212        Literal::new(Some(v.into()), DataType::Float64).into()
213    }
214
215    /// A literal boolean value.
216    #[inline(always)]
217    pub fn literal_bool(v: bool) -> Self {
218        Literal::new(Some(v.to_scalar_value()), DataType::Boolean).into()
219    }
220
221    /// A literal varchar value.
222    #[inline(always)]
223    pub fn literal_varchar(v: String) -> Self {
224        Literal::new(Some(v.into()), DataType::Varchar).into()
225    }
226
227    /// A literal null value.
228    #[inline(always)]
229    pub fn literal_null(element_type: DataType) -> Self {
230        Literal::new(None, element_type).into()
231    }
232
233    /// A literal jsonb value.
234    #[inline(always)]
235    pub fn literal_jsonb(v: JsonbVal) -> Self {
236        Literal::new(Some(v.into()), DataType::Jsonb).into()
237    }
238
239    /// A literal list value.
240    #[inline(always)]
241    pub fn literal_list(v: ListValue, element_type: DataType) -> Self {
242        Literal::new(Some(v.to_scalar_value()), DataType::list(element_type)).into()
243    }
244
245    /// Takes the expression, leaving a literal null of the same type in its place.
246    pub fn take(&mut self) -> Self {
247        std::mem::replace(self, Self::literal_null(self.return_type()))
248    }
249
250    /// A `count(*)` aggregate function.
251    #[inline(always)]
252    pub fn count_star() -> Self {
253        AggCall::new(
254            PbAggKind::Count.into(),
255            vec![],
256            false,
257            OrderBy::any(),
258            Condition::true_cond(),
259            vec![],
260        )
261        .unwrap()
262        .into()
263    }
264
265    /// Create a new expression by merging the given expressions by `And`.
266    ///
267    /// If `exprs` is empty, return a literal `true`.
268    pub fn and(exprs: impl IntoIterator<Item = ExprImpl>) -> Self {
269        merge_expr_by_logical(exprs, ExprType::And, ExprImpl::literal_bool(true))
270    }
271
272    /// Create a new expression by merging the given expressions by `Or`.
273    ///
274    /// If `exprs` is empty, return a literal `false`.
275    pub fn or(exprs: impl IntoIterator<Item = ExprImpl>) -> Self {
276        merge_expr_by_logical(exprs, ExprType::Or, ExprImpl::literal_bool(false))
277    }
278
279    /// Collect all `InputRef`s' indexes in the expression.
280    ///
281    /// # Panics
282    /// Panics if `input_ref >= input_col_num`.
283    pub fn collect_input_refs(&self, input_col_num: usize) -> FixedBitSet {
284        collect_input_refs(input_col_num, [self])
285    }
286
287    /// Check if the expression has no side effects and output is deterministic
288    pub fn is_pure(&self) -> bool {
289        is_pure(self)
290    }
291
292    pub fn is_impure(&self) -> bool {
293        is_impure(self)
294    }
295
296    pub fn needs_materialization(&self) -> bool {
297        needs_materialization(self)
298    }
299
300    /// Count `Now`s in the expression.
301    pub fn count_nows(&self) -> usize {
302        let mut visitor = CountNow::default();
303        visitor.visit_expr(self);
304        visitor.count()
305    }
306
307    /// Check whether self is literal NULL.
308    pub fn is_null(&self) -> bool {
309        matches!(self, ExprImpl::Literal(literal) if literal.get_data().is_none())
310    }
311
312    /// Check whether self is a literal NULL or literal string.
313    pub fn is_untyped(&self) -> bool {
314        matches!(self, ExprImpl::Literal(literal) if literal.is_untyped())
315            || matches!(self, ExprImpl::Parameter(parameter) if !parameter.has_infer())
316    }
317
318    /// Shorthand to create cast expr to `target` type in implicit context.
319    pub fn cast_implicit(mut self, target: &DataType) -> Result<ExprImpl, CastError> {
320        FunctionCall::cast_mut(&mut self, target, CastContext::Implicit)?;
321        Ok(self)
322    }
323
324    /// Shorthand to create cast expr to `target` type in assign context.
325    pub fn cast_assign(mut self, target: &DataType) -> Result<ExprImpl, CastError> {
326        FunctionCall::cast_mut(&mut self, target, CastContext::Assign)?;
327        Ok(self)
328    }
329
330    /// Shorthand to create cast expr to `target` type in explicit context.
331    pub fn cast_explicit(mut self, target: &DataType) -> Result<ExprImpl, CastError> {
332        FunctionCall::cast_mut(&mut self, target, CastContext::Explicit)?;
333        Ok(self)
334    }
335
336    /// Shorthand to inplace cast expr to `target` type in implicit context.
337    pub fn cast_implicit_mut(&mut self, target: &DataType) -> Result<(), CastError> {
338        FunctionCall::cast_mut(self, target, CastContext::Implicit)
339    }
340
341    /// Shorthand to inplace cast expr to `target` type in explicit context.
342    pub fn cast_explicit_mut(&mut self, target: &DataType) -> Result<(), CastError> {
343        FunctionCall::cast_mut(self, target, CastContext::Explicit)
344    }
345
346    /// Casting to Regclass type means getting the oid of expr.
347    /// See <https://www.postgresql.org/docs/current/datatype-oid.html>
348    pub fn cast_to_regclass(self) -> Result<ExprImpl, CastError> {
349        match self.return_type() {
350            DataType::Varchar => Ok(ExprImpl::FunctionCall(Box::new(
351                FunctionCall::new_unchecked(ExprType::CastRegclass, vec![self], DataType::Int32),
352            ))),
353            DataType::Int32 => Ok(self),
354            dt if dt.is_int() => Ok(self.cast_explicit(&DataType::Int32)?),
355            _ => bail_cast_error!("unsupported input type"),
356        }
357    }
358
359    /// Shorthand to inplace cast expr to `regclass` type.
360    pub fn cast_to_regclass_mut(&mut self) -> Result<(), CastError> {
361        let owned = std::mem::replace(self, ExprImpl::literal_bool(false));
362        *self = owned.cast_to_regclass()?;
363        Ok(())
364    }
365
366    /// Ensure the return type of this expression is an array of some type.
367    pub fn ensure_array_type(&self) -> Result<(), ErrorCode> {
368        if self.is_untyped() {
369            return Err(ErrorCode::BindError(
370                "could not determine polymorphic type because input has type unknown".into(),
371            ));
372        }
373        match self.return_type() {
374            DataType::List(_) => Ok(()),
375            t => Err(ErrorCode::BindError(format!("expects array but got {t}"))),
376        }
377    }
378
379    /// Ensure the return type of this expression is a map of some type.
380    pub fn try_into_map_type(&self) -> Result<MapType, ErrorCode> {
381        if self.is_untyped() {
382            return Err(ErrorCode::BindError(
383                "could not determine polymorphic type because input has type unknown".into(),
384            ));
385        }
386        match self.return_type() {
387            DataType::Map(m) => Ok(m),
388            t => Err(ErrorCode::BindError(format!("expects map but got {t}"))),
389        }
390    }
391
392    /// Shorthand to enforce implicit cast to boolean
393    pub fn enforce_bool_clause(self, clause: &str) -> RwResult<ExprImpl> {
394        if self.is_untyped() {
395            let inner = self.cast_implicit(&DataType::Boolean)?;
396            return Ok(inner);
397        }
398        let return_type = self.return_type();
399        if return_type != DataType::Boolean {
400            bail!(
401                "argument of {} must be boolean, not type {:?}",
402                clause,
403                return_type
404            )
405        }
406        Ok(self)
407    }
408
409    /// Create "cast" expr to string (`varchar`) type. This is different from a real cast, as
410    /// boolean is converted to a single char rather than full word.
411    ///
412    /// Choose between `cast_output` and `cast_{assign,explicit}(Varchar)` based on `PostgreSQL`'s
413    /// behavior on bools. For example, `concat(':', true)` is `:t` but `':' || true` is `:true`.
414    /// All other types have the same behavior when formatting to output and casting to string.
415    ///
416    /// References in `PostgreSQL`:
417    /// * [cast](https://github.com/postgres/postgres/blob/a3ff08e0b08dbfeb777ccfa8f13ebaa95d064c04/src/include/catalog/pg_cast.dat#L437-L444)
418    /// * [impl](https://github.com/postgres/postgres/blob/27b77ecf9f4d5be211900eda54d8155ada50d696/src/backend/utils/adt/bool.c#L204-L209)
419    pub fn cast_output(self) -> RwResult<ExprImpl> {
420        if self.return_type() == DataType::Boolean {
421            return Ok(FunctionCall::new(ExprType::BoolOut, vec![self])?.into());
422        }
423        // Use normal cast for other types. Both `assign` and `explicit` can pass the castability
424        // check and there is no difference.
425        self.cast_assign(&DataType::Varchar)
426            .map_err(|err| err.into())
427    }
428
429    /// Evaluate the expression on the given input.
430    ///
431    /// TODO: This is a naive implementation. We should avoid proto ser/de.
432    /// Tracking issue: <https://github.com/risingwavelabs/risingwave/issues/3479>
433    pub async fn eval_row(&self, input: &OwnedRow) -> RwResult<Datum> {
434        let backend_expr = build_from_prost(&self.to_expr_proto())?;
435        Ok(backend_expr.eval_row(input).await?)
436    }
437
438    /// Try to evaluate an expression if it's a constant expression by `ExprImpl::is_const`.
439    ///
440    /// Returns...
441    /// - `None` if it's not a constant expression,
442    /// - `Some(Ok(_))` if constant evaluation succeeds,
443    /// - `Some(Err(_))` if there's an error while evaluating a constant expression.
444    pub fn try_fold_const(&self) -> Option<RwResult<Datum>> {
445        if self.is_const() {
446            self.eval_row(&OwnedRow::empty())
447                .now_or_never()
448                .expect("constant expression should not be async")
449                .into()
450        } else {
451            None
452        }
453    }
454
455    /// Similar to `ExprImpl::try_fold_const`, but panics if the expression is not constant.
456    pub fn fold_const(&self) -> RwResult<Datum> {
457        self.try_fold_const().expect("expression is not constant")
458    }
459}
460
461/// Implement helper functions which recursively checks whether an variant is included in the
462/// expression. e.g., `has_subquery(&self) -> bool`
463///
464/// It will not traverse inside subqueries.
465macro_rules! impl_has_variant {
466    ( $($variant:ty),* ) => {
467        paste! {
468            impl ExprImpl {
469                $(
470                    pub fn [<has_ $variant:snake>](&self) -> bool {
471                        struct Has { has: bool }
472
473                        impl ExprVisitor for Has {
474                            fn [<visit_ $variant:snake>](&mut self, _: &$variant) {
475                                self.has = true;
476                            }
477                        }
478
479                        let mut visitor = Has { has: false };
480                        visitor.visit_expr(self);
481                        visitor.has
482                    }
483                )*
484            }
485        }
486    };
487}
488
489impl_has_variant! {InputRef, Literal, FunctionCall, FunctionCallWithLambda, AggCall, Subquery, TableFunction, WindowFunction, UserDefinedFunction, Now}
490
491/// Inequality condition between two input columns with clearer semantics.
492/// Represents: `left_col <op> right_col` where op is one of `<`, `<=`, `>`, `>=`.
493#[derive(Debug, Clone, PartialEq, Eq, Hash)]
494pub struct InequalityInputPair {
495    /// Index of the left side column (from left input).
496    pub left_idx: usize,
497    /// Index of the right side column (from right input, NOT offset by `left_cols_num`).
498    pub right_idx: usize,
499    /// Comparison operator: `left_col <op> right_col`.
500    pub op: ExprType,
501}
502
503impl InequalityInputPair {
504    pub fn new(left_idx: usize, right_idx: usize, op: ExprType) -> Self {
505        debug_assert!(matches!(
506            op,
507            ExprType::LessThan
508                | ExprType::LessThanOrEqual
509                | ExprType::GreaterThan
510                | ExprType::GreaterThanOrEqual
511        ));
512        Self {
513            left_idx,
514            right_idx,
515            op,
516        }
517    }
518
519    /// Returns true if the left side has larger values based on the operator.
520    /// For `>` and `>=`, left side is larger.
521    /// State cleanup applies to the side with larger values.
522    pub fn left_side_is_larger(&self) -> bool {
523        matches!(
524            self.op,
525            ExprType::GreaterThan | ExprType::GreaterThanOrEqual
526        )
527    }
528}
529
530impl ExprImpl {
531    /// This function is not meant to be called. In most cases you would want
532    /// [`ExprImpl::has_correlated_input_ref_by_scope`].
533    ///
534    /// When an expr only contains [`CorrelatedInputRef`]s owned by scopes inside it, the whole
535    /// expr is still considered to be uncorrelated, and can be checked with
536    /// [`ExprImpl::has_subquery`] as well.
537    /// See examples on [`crate::binder::BoundQuery::is_correlated_by_scope`] for details.
538    ///
539    /// This is a placeholder to trigger a compiler error when a trivial implementation checking for
540    /// enum variant is generated by accident. It cannot be called either because you cannot pass
541    /// `Infallible` to it.
542    pub fn has_correlated_input_ref(&self, _: std::convert::Infallible) -> bool {
543        unreachable!()
544    }
545
546    /// Used to check whether the expression has [`CorrelatedInputRef`] owned by `scope`.
547    ///
548    /// This is the core logic that supports [`crate::binder::BoundQuery::is_correlated_by_scope`].
549    // We need to traverse inside subqueries.
550    pub fn has_correlated_input_ref_by_scope(&self, scope: ScopeId) -> bool {
551        struct Has {
552            scope: ScopeId,
553            has: bool,
554        }
555
556        impl ExprVisitor for Has {
557            fn visit_correlated_input_ref(&mut self, correlated_input_ref: &CorrelatedInputRef) {
558                if correlated_input_ref.scope() == Some(self.scope) {
559                    self.has = true;
560                }
561            }
562
563            fn visit_subquery(&mut self, subquery: &Subquery) {
564                self.has |= subquery.is_correlated_by_scope(self.scope);
565            }
566        }
567
568        let mut visitor = Has { scope, has: false };
569        visitor.visit_expr(self);
570        visitor.has
571    }
572
573    pub fn has_correlated_input_ref_by_correlated_id(&self, correlated_id: CorrelatedId) -> bool {
574        struct Has {
575            correlated_id: CorrelatedId,
576            has: bool,
577        }
578
579        impl ExprVisitor for Has {
580            fn visit_correlated_input_ref(&mut self, correlated_input_ref: &CorrelatedInputRef) {
581                if correlated_input_ref.correlated_id() == self.correlated_id {
582                    self.has = true;
583                }
584            }
585
586            fn visit_subquery(&mut self, subquery: &Subquery) {
587                self.has |= subquery.is_correlated_by_correlated_id(self.correlated_id);
588            }
589        }
590
591        let mut visitor = Has {
592            correlated_id,
593            has: false,
594        };
595        visitor.visit_expr(self);
596        visitor.has
597    }
598
599    /// Collect `CorrelatedInputRef`s in `ExprImpl` owned by `scope`, return their indices, and
600    /// assign absolute `correlated_id` for them.
601    pub fn collect_correlated_indices_by_scope_and_assign_id(
602        &mut self,
603        scope: ScopeId,
604        correlated_id: CorrelatedId,
605    ) -> Vec<usize> {
606        struct Collector {
607            scope: ScopeId,
608            correlated_indices: Vec<usize>,
609            correlated_id: CorrelatedId,
610        }
611
612        impl ExprMutator for Collector {
613            fn visit_correlated_input_ref(
614                &mut self,
615                correlated_input_ref: &mut CorrelatedInputRef,
616            ) {
617                if correlated_input_ref.scope() == Some(self.scope) {
618                    self.correlated_indices.push(correlated_input_ref.index());
619                    correlated_input_ref.set_correlated_id(self.correlated_id);
620                }
621            }
622
623            fn visit_subquery(&mut self, subquery: &mut Subquery) {
624                self.correlated_indices.extend(
625                    subquery.collect_correlated_indices_by_scope_and_assign_id(
626                        self.scope,
627                        self.correlated_id,
628                    ),
629                );
630            }
631        }
632
633        let mut collector = Collector {
634            scope,
635            correlated_indices: vec![],
636            correlated_id,
637        };
638        collector.visit_expr(self);
639        collector.correlated_indices.sort();
640        collector.correlated_indices.dedup();
641        collector.correlated_indices
642    }
643
644    pub fn only_literal_and_func(&self) -> bool {
645        {
646            struct HasOthers {
647                has_others: bool,
648            }
649
650            impl ExprVisitor for HasOthers {
651                fn visit_expr(&mut self, expr: &ExprImpl) {
652                    match expr {
653                        ExprImpl::CorrelatedInputRef(_)
654                        | ExprImpl::InputRef(_)
655                        | ExprImpl::AggCall(_)
656                        | ExprImpl::Subquery(_)
657                        | ExprImpl::TableFunction(_)
658                        | ExprImpl::WindowFunction(_)
659                        | ExprImpl::UserDefinedFunction(_)
660                        | ExprImpl::Parameter(_)
661                        | ExprImpl::Now(_)
662                        | ExprImpl::SecretRef(_) => self.has_others = true,
663                        ExprImpl::Literal(_inner) => {}
664                        ExprImpl::FunctionCall(inner) => {
665                            if !self.is_short_circuit(inner) {
666                                // only if the current `func_call` is *not* a short-circuit
667                                // expression, e.g., true or (...) | false and (...),
668                                // shall we proceed to visit it.
669                                self.visit_function_call(inner)
670                            }
671                        }
672                        ExprImpl::FunctionCallWithLambda(inner) => {
673                            self.visit_function_call_with_lambda(inner)
674                        }
675                    }
676                }
677            }
678
679            impl HasOthers {
680                fn is_short_circuit(&self, func_call: &FunctionCall) -> bool {
681                    /// evaluate the first parameter of `Or` or `And` function call
682                    fn eval_first(e: &ExprImpl, expect: bool) -> bool {
683                        if let ExprImpl::Literal(l) = e {
684                            *l.get_data() == Some(ScalarImpl::Bool(expect))
685                        } else {
686                            false
687                        }
688                    }
689
690                    match func_call.func_type {
691                        ExprType::Or => eval_first(&func_call.inputs()[0], true),
692                        ExprType::And => eval_first(&func_call.inputs()[0], false),
693                        _ => false,
694                    }
695                }
696            }
697
698            let mut visitor = HasOthers { has_others: false };
699            visitor.visit_expr(self);
700            !visitor.has_others
701        }
702    }
703
704    /// Checks whether this is a constant expr that can be evaluated over a dummy chunk.
705    ///
706    /// The expression tree should only consist of literals and **pure** function calls.
707    pub fn is_const(&self) -> bool {
708        self.only_literal_and_func() && self.is_pure()
709    }
710
711    /// Returns the `InputRefs` of an Equality predicate if it matches
712    /// ordered by the canonical ordering (lower, higher), else returns None
713    pub fn as_eq_cond(&self) -> Option<(InputRef, InputRef)> {
714        if let ExprImpl::FunctionCall(function_call) = self
715            && function_call.func_type() == ExprType::Equal
716            && let (_, ExprImpl::InputRef(x), ExprImpl::InputRef(y)) =
717                function_call.clone().decompose_as_binary()
718        {
719            if x.index() < y.index() {
720                Some((*x, *y))
721            } else {
722                Some((*y, *x))
723            }
724        } else {
725            None
726        }
727    }
728
729    pub fn as_is_not_distinct_from_cond(&self) -> Option<(InputRef, InputRef)> {
730        if let ExprImpl::FunctionCall(function_call) = self
731            && function_call.func_type() == ExprType::IsNotDistinctFrom
732            && let (_, ExprImpl::InputRef(x), ExprImpl::InputRef(y)) =
733                function_call.clone().decompose_as_binary()
734        {
735            if x.index() < y.index() {
736                Some((*x, *y))
737            } else {
738                Some((*y, *x))
739            }
740        } else {
741            None
742        }
743    }
744
745    pub fn reverse_comparison(comparison: ExprType) -> ExprType {
746        match comparison {
747            ExprType::LessThan => ExprType::GreaterThan,
748            ExprType::LessThanOrEqual => ExprType::GreaterThanOrEqual,
749            ExprType::GreaterThan => ExprType::LessThan,
750            ExprType::GreaterThanOrEqual => ExprType::LessThanOrEqual,
751            ExprType::Equal | ExprType::IsNotDistinctFrom => comparison,
752            _ => unreachable!(),
753        }
754    }
755
756    pub fn as_comparison_cond(&self) -> Option<(InputRef, ExprType, InputRef)> {
757        if let ExprImpl::FunctionCall(function_call) = self {
758            match function_call.func_type() {
759                ty @ (ExprType::LessThan
760                | ExprType::LessThanOrEqual
761                | ExprType::GreaterThan
762                | ExprType::GreaterThanOrEqual) => {
763                    let (_, op1, op2) = function_call.clone().decompose_as_binary();
764                    if let (ExprImpl::InputRef(x), ExprImpl::InputRef(y)) = (op1, op2) {
765                        if x.index < y.index {
766                            Some((*x, ty, *y))
767                        } else {
768                            Some((*y, Self::reverse_comparison(ty), *x))
769                        }
770                    } else {
771                        None
772                    }
773                }
774                _ => None,
775            }
776        } else {
777            None
778        }
779    }
780
781    /// Accepts expressions of the form `input_expr cmp now_expr` or `now_expr cmp input_expr`,
782    /// where `input_expr` contains an `InputRef` and contains no `now()`, and `now_expr`
783    /// contains a `now()` but no `InputRef`.
784    ///
785    /// Canonicalizes to the first ordering and returns `(input_expr, cmp, now_expr)`
786    pub fn as_now_comparison_cond(&self) -> Option<(ExprImpl, ExprType, ExprImpl)> {
787        if let ExprImpl::FunctionCall(function_call) = self {
788            match function_call.func_type() {
789                ty @ (ExprType::Equal
790                | ExprType::LessThan
791                | ExprType::LessThanOrEqual
792                | ExprType::GreaterThan
793                | ExprType::GreaterThanOrEqual) => {
794                    let (_, op1, op2) = function_call.clone().decompose_as_binary();
795                    if !op1.has_now()
796                        && op1.has_input_ref()
797                        && op2.has_now()
798                        && !op2.has_input_ref()
799                    {
800                        Some((op1, ty, op2))
801                    } else if op1.has_now()
802                        && !op1.has_input_ref()
803                        && !op2.has_now()
804                        && op2.has_input_ref()
805                    {
806                        Some((op2, Self::reverse_comparison(ty), op1))
807                    } else {
808                        None
809                    }
810                }
811                _ => None,
812            }
813        } else {
814            None
815        }
816    }
817
818    /// Returns the `InputRef` and offset of a predicate if it matches
819    /// the form `InputRef [+- const_expr]`, else returns None.
820    ///
821    /// Deprecated: Only used by `as_input_comparison_cond`.
822    #[expect(dead_code)]
823    fn as_input_offset(&self) -> Option<(usize, Option<(ExprType, ExprImpl)>)> {
824        match self {
825            ExprImpl::InputRef(input_ref) => Some((input_ref.index(), None)),
826            ExprImpl::FunctionCall(function_call) => {
827                let expr_type = function_call.func_type();
828                match expr_type {
829                    ExprType::Add | ExprType::Subtract => {
830                        let (_, lhs, rhs) = function_call.clone().decompose_as_binary();
831                        if let ExprImpl::InputRef(input_ref) = &lhs
832                            && rhs.is_const()
833                        {
834                            // Currently we will return `None` for non-literal because the result of the expression might be '1 day'. However, there will definitely exist false positives such as '1 second + 1 second'.
835                            // We will treat the expression as an input offset when rhs is `null`.
836                            if rhs.return_type() == DataType::Interval
837                                && rhs.as_literal().is_none_or(|literal| {
838                                    literal.get_data().as_ref().is_some_and(|scalar| {
839                                        let interval = scalar.as_interval();
840                                        interval.months() != 0 || interval.days() != 0
841                                    })
842                                })
843                            {
844                                None
845                            } else {
846                                Some((input_ref.index(), Some((expr_type, rhs))))
847                            }
848                        } else {
849                            None
850                        }
851                    }
852                    _ => None,
853                }
854            }
855            _ => None,
856        }
857    }
858
859    pub fn as_eq_const(&self) -> Option<(InputRef, ExprImpl)> {
860        if let ExprImpl::FunctionCall(function_call) = self
861            && function_call.func_type() == ExprType::Equal
862        {
863            match function_call.clone().decompose_as_binary() {
864                (_, ExprImpl::InputRef(x), y) if y.is_const() => Some((*x, y)),
865                (_, x, ExprImpl::InputRef(y)) if x.is_const() => Some((*y, x)),
866                _ => None,
867            }
868        } else {
869            None
870        }
871    }
872
873    pub fn as_eq_correlated_input_ref(&self) -> Option<(InputRef, CorrelatedInputRef)> {
874        if let ExprImpl::FunctionCall(function_call) = self
875            && function_call.func_type() == ExprType::Equal
876        {
877            match function_call.clone().decompose_as_binary() {
878                (_, ExprImpl::InputRef(x), ExprImpl::CorrelatedInputRef(y)) => Some((*x, *y)),
879                (_, ExprImpl::CorrelatedInputRef(x), ExprImpl::InputRef(y)) => Some((*y, *x)),
880                _ => None,
881            }
882        } else {
883            None
884        }
885    }
886
887    pub fn as_is_null(&self) -> Option<InputRef> {
888        if let ExprImpl::FunctionCall(function_call) = self
889            && function_call.func_type() == ExprType::IsNull
890        {
891            match function_call.clone().decompose_as_unary() {
892                (_, ExprImpl::InputRef(x)) => Some(*x),
893                _ => None,
894            }
895        } else {
896            None
897        }
898    }
899
900    pub fn as_comparison_const(&self) -> Option<(InputRef, ExprType, ExprImpl)> {
901        fn reverse_comparison(comparison: ExprType) -> ExprType {
902            match comparison {
903                ExprType::LessThan => ExprType::GreaterThan,
904                ExprType::LessThanOrEqual => ExprType::GreaterThanOrEqual,
905                ExprType::GreaterThan => ExprType::LessThan,
906                ExprType::GreaterThanOrEqual => ExprType::LessThanOrEqual,
907                _ => unreachable!(),
908            }
909        }
910
911        if let ExprImpl::FunctionCall(function_call) = self {
912            match function_call.func_type() {
913                ty @ (ExprType::LessThan
914                | ExprType::LessThanOrEqual
915                | ExprType::GreaterThan
916                | ExprType::GreaterThanOrEqual) => {
917                    let (_, op1, op2) = function_call.clone().decompose_as_binary();
918                    match (op1, op2) {
919                        (ExprImpl::InputRef(x), y) if y.is_const() => Some((*x, ty, y)),
920                        (x, ExprImpl::InputRef(y)) if x.is_const() => {
921                            Some((*y, reverse_comparison(ty), x))
922                        }
923                        _ => None,
924                    }
925                }
926                _ => None,
927            }
928        } else {
929            None
930        }
931    }
932
933    pub fn as_in_const_list(&self) -> Option<(InputRef, Vec<ExprImpl>)> {
934        if let ExprImpl::FunctionCall(function_call) = self
935            && function_call.func_type() == ExprType::In
936        {
937            let (input, list) = function_call.inputs().split_first()?;
938            let input_ref = match input {
939                ExprImpl::InputRef(i) => i.as_ref().clone(),
940                _ => return None,
941            };
942            if !list.iter().all(ExprImpl::is_const) {
943                return None;
944            }
945
946            Some((input_ref, list.to_vec()))
947        } else {
948            None
949        }
950    }
951
952    pub fn as_some_eq_const_list(&self) -> Option<(InputRef, Vec<ExprImpl>)> {
953        if let ExprImpl::FunctionCall(function_call) = self
954            && function_call.func_type() == ExprType::Some
955        {
956            let (_, inner) = function_call.clone().decompose_as_unary();
957            let ExprImpl::FunctionCall(inner_call) = inner else {
958                return None;
959            };
960            if inner_call.func_type() != ExprType::Equal {
961                return None;
962            }
963
964            let (_, left, right) = inner_call.decompose_as_binary();
965            let (input_ref, list_expr) = match (left, right) {
966                (ExprImpl::InputRef(input_ref), list_expr) if list_expr.is_const() => {
967                    (input_ref.as_ref().clone(), list_expr)
968                }
969                (list_expr, ExprImpl::InputRef(input_ref)) if list_expr.is_const() => {
970                    (input_ref.as_ref().clone(), list_expr)
971                }
972                _ => return None,
973            };
974
975            let literal = list_expr.as_literal()?;
976            let DataType::List(list_type) = literal.return_type() else {
977                return None;
978            };
979            let list = literal
980                .get_data()
981                .as_ref()?
982                .as_list()
983                .iter()
984                .map(|datum| {
985                    ExprImpl::Literal(Box::new(Literal::new(
986                        datum.to_owned_datum(),
987                        list_type.elem().clone(),
988                    )))
989                })
990                .collect();
991
992            Some((input_ref, list))
993        } else {
994            None
995        }
996    }
997
998    pub fn as_or_disjunctions(&self) -> Option<Vec<ExprImpl>> {
999        if let ExprImpl::FunctionCall(function_call) = self
1000            && function_call.func_type() == ExprType::Or
1001        {
1002            Some(to_disjunctions(self.clone()))
1003        } else {
1004            None
1005        }
1006    }
1007
1008    pub fn to_project_set_select_item_proto(&self) -> ProjectSetSelectItem {
1009        use risingwave_pb::expr::project_set_select_item::SelectItem::*;
1010
1011        ProjectSetSelectItem {
1012            select_item: Some(match self {
1013                ExprImpl::TableFunction(tf) => TableFunction(tf.to_protobuf()),
1014                expr => Expr(expr.to_expr_proto()),
1015            }),
1016        }
1017    }
1018
1019    /// Serialize the expression. Returns an error if this will result in an impure expression on a
1020    /// retract stream, which may lead to inconsistent results.
1021    pub fn to_project_set_select_item_proto_checked_pure(
1022        &self,
1023        retract: bool,
1024    ) -> crate::error::Result<ProjectSetSelectItem> {
1025        use risingwave_pb::expr::project_set_select_item::SelectItem::*;
1026
1027        Ok(ProjectSetSelectItem {
1028            select_item: Some(match self {
1029                ExprImpl::TableFunction(tf) => TableFunction(tf.to_protobuf_checked_pure(retract)?),
1030                expr => Expr(expr.to_expr_proto_checked_pure(retract, "SELECT list")?),
1031            }),
1032        })
1033    }
1034
1035    pub fn from_expr_proto(proto: &ExprNode) -> RwResult<Self> {
1036        let rex_node = proto.get_rex_node()?;
1037        let ret_type = proto.get_return_type()?.into();
1038
1039        Ok(match rex_node {
1040            RexNode::InputRef(column_index) => Self::InputRef(Box::new(InputRef::from_expr_proto(
1041                *column_index as _,
1042                ret_type,
1043            )?)),
1044            RexNode::Constant(_) => Self::Literal(Box::new(Literal::from_expr_proto(proto)?)),
1045            RexNode::Udf(udf) => Self::UserDefinedFunction(Box::new(
1046                UserDefinedFunction::from_expr_proto(udf, ret_type)?,
1047            )),
1048            RexNode::FuncCall(function_call) => {
1049                Self::FunctionCall(Box::new(FunctionCall::from_expr_proto(
1050                    function_call,
1051                    proto.get_function_type()?, // only interpret if it's a function call
1052                    ret_type,
1053                )?))
1054            }
1055            RexNode::Now(_) => Self::Now(Box::new(Now {})),
1056            RexNode::SecretRef(sr) => Self::SecretRef(Box::new(SecretRef {
1057                secret_id: sr.secret_id.into(),
1058                ref_as: risingwave_pb::secret::secret_ref::RefAsType::try_from(sr.ref_as)
1059                    .unwrap_or(risingwave_pb::secret::secret_ref::RefAsType::Text),
1060                secret_name: format!("<secret:{}>", sr.secret_id),
1061            })),
1062        })
1063    }
1064}
1065
1066impl From<Condition> for ExprImpl {
1067    fn from(c: Condition) -> Self {
1068        ExprImpl::and(c.conjunctions)
1069    }
1070}
1071
1072/// A custom Debug implementation that is more concise and suitable to use with
1073/// [`std::fmt::Formatter::debug_list`] in plan nodes. If the verbose output is preferred, it is
1074/// still available via `{:#?}`.
1075impl std::fmt::Debug for ExprImpl {
1076    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1077        if f.alternate() {
1078            return match self {
1079                Self::InputRef(arg0) => f.debug_tuple("InputRef").field(arg0).finish(),
1080                Self::Literal(arg0) => f.debug_tuple("Literal").field(arg0).finish(),
1081                Self::FunctionCall(arg0) => f.debug_tuple("FunctionCall").field(arg0).finish(),
1082                Self::FunctionCallWithLambda(arg0) => {
1083                    f.debug_tuple("FunctionCallWithLambda").field(arg0).finish()
1084                }
1085                Self::AggCall(arg0) => f.debug_tuple("AggCall").field(arg0).finish(),
1086                Self::Subquery(arg0) => f.debug_tuple("Subquery").field(arg0).finish(),
1087                Self::CorrelatedInputRef(arg0) => {
1088                    f.debug_tuple("CorrelatedInputRef").field(arg0).finish()
1089                }
1090                Self::TableFunction(arg0) => f.debug_tuple("TableFunction").field(arg0).finish(),
1091                Self::WindowFunction(arg0) => f.debug_tuple("WindowFunction").field(arg0).finish(),
1092                Self::UserDefinedFunction(arg0) => {
1093                    f.debug_tuple("UserDefinedFunction").field(arg0).finish()
1094                }
1095                Self::Parameter(arg0) => f.debug_tuple("Parameter").field(arg0).finish(),
1096                Self::Now(_) => f.debug_tuple("Now").finish(),
1097                Self::SecretRef(arg0) => f.debug_tuple("SecretRef").field(arg0).finish(),
1098            };
1099        }
1100        match self {
1101            Self::InputRef(x) => write!(f, "{:?}", x),
1102            Self::Literal(x) => write!(f, "{:?}", x),
1103            Self::FunctionCall(x) => write!(f, "{:?}", x),
1104            Self::FunctionCallWithLambda(x) => write!(f, "{:?}", x),
1105            Self::AggCall(x) => write!(f, "{:?}", x),
1106            Self::Subquery(x) => write!(f, "{:?}", x),
1107            Self::CorrelatedInputRef(x) => write!(f, "{:?}", x),
1108            Self::TableFunction(x) => write!(f, "{:?}", x),
1109            Self::WindowFunction(x) => write!(f, "{:?}", x),
1110            Self::UserDefinedFunction(x) => write!(f, "{:?}", x),
1111            Self::Parameter(x) => write!(f, "{:?}", x),
1112            Self::Now(x) => write!(f, "{:?}", x),
1113            Self::SecretRef(x) => write!(f, "Secret({})", x.secret_name),
1114        }
1115    }
1116}
1117
1118pub struct ExprDisplay<'a> {
1119    pub expr: &'a ExprImpl,
1120    pub input_schema: &'a Schema,
1121}
1122
1123impl std::fmt::Debug for ExprDisplay<'_> {
1124    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1125        let that = self.expr;
1126        match that {
1127            ExprImpl::InputRef(x) => write!(
1128                f,
1129                "{:?}",
1130                InputRefDisplay {
1131                    input_ref: x,
1132                    input_schema: self.input_schema
1133                }
1134            ),
1135            ExprImpl::Literal(x) => write!(f, "{:?}", x),
1136            ExprImpl::FunctionCall(x) => write!(
1137                f,
1138                "{:?}",
1139                FunctionCallDisplay {
1140                    function_call: x,
1141                    input_schema: self.input_schema
1142                }
1143            ),
1144            ExprImpl::FunctionCallWithLambda(x) => write!(
1145                f,
1146                "{:?}",
1147                FunctionCallDisplay {
1148                    function_call: &x.to_full_function_call(),
1149                    input_schema: self.input_schema
1150                }
1151            ),
1152            ExprImpl::AggCall(x) => write!(f, "{:?}", x),
1153            ExprImpl::Subquery(x) => write!(f, "{:?}", x),
1154            ExprImpl::CorrelatedInputRef(x) => write!(f, "{:?}", x),
1155            ExprImpl::TableFunction(x) => {
1156                // TODO: TableFunctionCallVerboseDisplay
1157                write!(f, "{:?}", x)
1158            }
1159            ExprImpl::WindowFunction(x) => {
1160                // TODO: WindowFunctionCallVerboseDisplay
1161                write!(f, "{:?}", x)
1162            }
1163            ExprImpl::UserDefinedFunction(x) => {
1164                write!(
1165                    f,
1166                    "{:?}",
1167                    UserDefinedFunctionDisplay {
1168                        func_call: x,
1169                        input_schema: self.input_schema
1170                    }
1171                )
1172            }
1173            ExprImpl::Parameter(x) => write!(f, "{:?}", x),
1174            ExprImpl::Now(x) => write!(f, "{:?}", x),
1175            ExprImpl::SecretRef(x) => write!(f, "Secret({})", x.secret_name),
1176        }
1177    }
1178}
1179
1180impl std::fmt::Display for ExprDisplay<'_> {
1181    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1182        (self as &dyn std::fmt::Debug).fmt(f)
1183    }
1184}
1185
1186#[cfg(test)]
1187/// Asserts that the expression is an [`InputRef`] with the given index.
1188macro_rules! assert_eq_input_ref {
1189    ($e:expr, $index:expr) => {
1190        match $e {
1191            ExprImpl::InputRef(i) => assert_eq!(i.index(), $index),
1192            _ => assert!(false, "Expected input ref, found {:?}", $e),
1193        }
1194    };
1195}
1196
1197#[cfg(test)]
1198pub(crate) use assert_eq_input_ref;
1199use risingwave_common::bail;
1200use risingwave_common::catalog::Schema;
1201use risingwave_common::row::OwnedRow;
1202
1203use crate::utils::Condition;
1204
1205#[cfg(test)]
1206mod tests {
1207    use risingwave_pb::secret::secret_ref::RefAsType;
1208
1209    use super::*;
1210
1211    #[test]
1212    fn test_expr_debug_alternate() {
1213        let mut e = InputRef::new(1, DataType::Boolean).into();
1214        e = FunctionCall::new(ExprType::Not, vec![e]).unwrap().into();
1215        let s = format!("{:#?}", e);
1216        assert!(s.contains("return_type: Boolean"))
1217    }
1218
1219    #[test]
1220    fn test_secret_ref_display_contains_name() {
1221        let expr = ExprImpl::SecretRef(Box::new(SecretRef {
1222            secret_id: 42.into(),
1223            ref_as: RefAsType::Text,
1224            secret_name: "test_secret".to_owned(),
1225        }));
1226
1227        assert_eq!(format!("{expr:?}"), "Secret(test_secret)");
1228        assert_eq!(
1229            format!(
1230                "{:?}",
1231                ExprDisplay {
1232                    expr: &expr,
1233                    input_schema: Schema::empty(),
1234                }
1235            ),
1236            "Secret(test_secret)"
1237        );
1238        assert!(format!("{expr:#?}").contains("test_secret"));
1239    }
1240}
risingwave_frontend/expr/mod.rs

risingwave_frontend/expr/
mod.rs
