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

risingwave_frontend/expr/
mod.rs
