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

risingwave_frontend/expr/
mod.rs
