mz_expr/

scalar.rs

// Copyright Materialize, Inc. and contributors. All rights reserved.
//
// Use of this software is governed by the Business Source License
// included in the LICENSE file.
//
// As of the Change Date specified in that file, in accordance with
// the Business Source License, use of this software will be governed
// by the Apache License, Version 2.0.

use std::collections::BTreeSet;
use std::ops::BitOrAssign;
use std::sync::Arc;
use std::{fmt, mem};

use itertools::Itertools;
use mz_ore::cast::CastFrom;
use mz_ore::iter::IteratorExt;
use mz_ore::soft_assert_or_log;
use mz_ore::stack::RecursionLimitError;
use mz_ore::str::StrExt;
use mz_ore::treat_as_equal::TreatAsEqual;
use mz_ore::vec::swap_remove_multiple;
use mz_pgrepr::TypeFromOidError;
use mz_proto::{IntoRustIfSome, ProtoType, RustType, TryFromProtoError};
use mz_repr::adt::array::InvalidArrayError;
use mz_repr::adt::date::DateError;
use mz_repr::adt::range::InvalidRangeError;
use mz_repr::adt::regex::RegexCompilationError;
use mz_repr::adt::timestamp::TimestampError;
use mz_repr::strconv::{ParseError, ParseHexError};
use mz_repr::{Datum, ReprColumnType, ReprScalarType, Row, RowArena, SqlColumnType};

#[cfg(any(test, feature = "proptest"))]
use proptest::prelude::*;
#[cfg(any(test, feature = "proptest"))]
use proptest_derive::Arbitrary;
use serde::{Deserialize, Serialize};

use crate::explain::{HumanizedExplain, HumanizerMode};
pub use crate::scalar::columns::Columns;
pub use crate::scalar::eval::Eval;
use crate::scalar::func::variadic::{And, Or};
use crate::scalar::func::{BinaryFunc, UnaryFunc, UnmaterializableFunc, VariadicFunc};
pub use crate::scalar::optimizable::OptimizableExpr;
use crate::scalar::proto_eval_error::proto_incompatible_array_dimensions::ProtoDims;
use crate::visit::{Visit, VisitChildren};

pub mod columns;
pub mod eval;
pub mod func;
pub mod like_pattern;
pub mod optimizable;
mod reduce;

include!(concat!(env!("OUT_DIR"), "/mz_expr.scalar.rs"));

#[derive(Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
pub enum MirScalarExpr {
    /// A column of the input row
    Column(usize, TreatAsEqual<Option<Arc<str>>>),
    /// A literal value.
    /// (Stored as a row, because we can't own a Datum)
    Literal(Result<Row, EvalError>, ReprColumnType),
    /// A call to an unmaterializable function.
    ///
    /// These functions cannot be evaluated by `MirScalarExpr::eval`. They must
    /// be transformed away by a higher layer.
    CallUnmaterializable(UnmaterializableFunc),
    /// A function call that takes one expression as an argument.
    CallUnary {
        func: UnaryFunc,
        expr: Box<MirScalarExpr>,
    },
    /// A function call that takes two expressions as arguments.
    CallBinary {
        func: BinaryFunc,
        expr1: Box<MirScalarExpr>,
        expr2: Box<MirScalarExpr>,
    },
    /// A function call that takes an arbitrary number of arguments.
    CallVariadic {
        func: VariadicFunc,
        exprs: Vec<MirScalarExpr>,
    },
    /// Conditionally evaluated expressions.
    ///
    /// It is important that `then` and `els` only be evaluated if
    /// `cond` is true or not, respectively. This is the only way
    /// users can guard execution (other logical operator do not
    /// short-circuit) and we need to preserve that.
    If {
        cond: Box<MirScalarExpr>,
        then: Box<MirScalarExpr>,
        els: Box<MirScalarExpr>,
    },
}

// We need a custom Debug because we don't want to show `None` for name information.
// Sadly, the `derivative` crate doesn't support this use case.
impl std::fmt::Debug for MirScalarExpr {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            MirScalarExpr::Column(i, TreatAsEqual(Some(name))) => {
                write!(f, "Column({i}, {name:?})")
            }
            MirScalarExpr::Column(i, TreatAsEqual(None)) => write!(f, "Column({i})"),
            MirScalarExpr::Literal(lit, typ) => write!(f, "Literal({lit:?}, {typ:?})"),
            MirScalarExpr::CallUnmaterializable(func) => {
                write!(f, "CallUnmaterializable({func:?})")
            }
            MirScalarExpr::CallUnary { func, expr } => {
                write!(f, "CallUnary({func:?}, {expr:?})")
            }
            MirScalarExpr::CallBinary { func, expr1, expr2 } => {
                write!(f, "CallBinary({func:?}, {expr1:?}, {expr2:?})")
            }
            MirScalarExpr::CallVariadic { func, exprs } => {
                write!(f, "CallVariadic({func:?}, {exprs:?})")
            }
            MirScalarExpr::If { cond, then, els } => {
                write!(f, "If({cond:?}, {then:?}, {els:?})")
            }
        }
    }
}

impl MirScalarExpr {
    pub fn columns(is: &[usize]) -> Vec<MirScalarExpr> {
        is.iter().map(|i| MirScalarExpr::column(*i)).collect()
    }

    pub fn column(column: usize) -> Self {
        MirScalarExpr::Column(column, TreatAsEqual(None))
    }

    pub fn named_column(column: usize, name: Arc<str>) -> Self {
        MirScalarExpr::Column(column, TreatAsEqual(Some(name)))
    }

    pub fn literal(res: Result<Datum, EvalError>, typ: ReprScalarType) -> Self {
        let typ = ReprColumnType {
            scalar_type: typ,
            nullable: matches!(res, Ok(Datum::Null)),
        };
        let row = res.map(|datum| Row::pack_slice(&[datum]));
        MirScalarExpr::Literal(row, typ)
    }

    pub fn literal_ok(datum: Datum, typ: ReprScalarType) -> Self {
        MirScalarExpr::literal(Ok(datum), typ)
    }

    /// Constructs a `MirScalarExpr::Literal` from a pre-packed `Row`
    /// containing a single datum and a `ReprScalarType`. Nullability is
    /// derived by inspecting the first datum in the row.
    pub fn literal_from_single_element_row(row: Row, typ: ReprScalarType) -> Self {
        soft_assert_or_log!(
            row.iter().count() == 1,
            "literal_from_row called with a Row containing {} datums",
            row.iter().count()
        );
        let nullable = row.unpack_first() == Datum::Null;
        let typ = ReprColumnType {
            scalar_type: typ,
            nullable,
        };
        MirScalarExpr::Literal(Ok(row), typ)
    }

    pub fn literal_null(typ: ReprScalarType) -> Self {
        MirScalarExpr::literal_ok(Datum::Null, typ)
    }

    pub fn literal_false() -> Self {
        MirScalarExpr::literal_ok(Datum::False, ReprScalarType::Bool)
    }

    pub fn literal_true() -> Self {
        MirScalarExpr::literal_ok(Datum::True, ReprScalarType::Bool)
    }

    pub fn call_unary<U: Into<UnaryFunc>>(self, func: U) -> Self {
        MirScalarExpr::CallUnary {
            func: func.into(),
            expr: Box::new(self),
        }
    }

    pub fn call_binary<B: Into<BinaryFunc>>(self, other: Self, func: B) -> Self {
        MirScalarExpr::CallBinary {
            func: func.into(),
            expr1: Box::new(self),
            expr2: Box::new(other),
        }
    }

    /// Call function `func` on `exprs`.
    pub fn call_variadic<V: Into<VariadicFunc>>(func: V, exprs: Vec<Self>) -> Self {
        MirScalarExpr::CallVariadic {
            func: func.into(),
            exprs,
        }
    }

    pub fn if_then_else(self, t: Self, f: Self) -> Self {
        MirScalarExpr::If {
            cond: Box::new(self),
            then: Box::new(t),
            els: Box::new(f),
        }
    }

    pub fn or(self, other: Self) -> Self {
        MirScalarExpr::call_variadic(Or, vec![self, other])
    }

    pub fn and(self, other: Self) -> Self {
        MirScalarExpr::call_variadic(And, vec![self, other])
    }

    pub fn not(self) -> Self {
        self.call_unary(UnaryFunc::Not(func::Not))
    }

    pub fn call_is_null(self) -> Self {
        self.call_unary(UnaryFunc::IsNull(func::IsNull))
    }

    /// Match AND or OR on self and get the args. If no match, then interpret self as if it were
    /// wrapped in a 1-arg AND/OR.
    pub fn and_or_args(&self, func_to_match: VariadicFunc) -> Vec<MirScalarExpr> {
        assert!(func_to_match == Or.into() || func_to_match == And.into());
        match self {
            MirScalarExpr::CallVariadic { func, exprs } if *func == func_to_match => exprs.clone(),
            _ => vec![self.clone()],
        }
    }

    /// Try to match a literal equality involving the given expression on one side.
    /// Return the (non-null) literal and a bool that indicates whether an inversion was needed.
    ///
    /// More specifically:
    /// If `self` is an equality with a `null` literal on any side, then the match fails!
    /// Otherwise: for a given `expr`, if `self` is `<expr> = <literal>` or `<literal> = <expr>`
    /// then return `Some((<literal>, false))`. In addition to just trying to match `<expr>` as it
    /// is, we also try to remove an invertible function call (such as a cast). If the match
    /// succeeds with the inversion, then return `Some((<inverted-literal>, true))`. For more
    /// details on the inversion, see `invert_casts_on_expr_eq_literal_inner`.
    pub fn expr_eq_literal(&self, expr: &MirScalarExpr) -> Option<(Row, bool)> {
        if let MirScalarExpr::CallBinary {
            func: BinaryFunc::Eq(_),
            expr1,
            expr2,
        } = self
        {
            if expr1.is_literal_null() || expr2.is_literal_null() {
                return None;
            }
            if let Some(Ok(lit)) = expr1.as_literal_owned() {
                return Self::expr_eq_literal_inner(expr, lit, expr1, expr2);
            }
            if let Some(Ok(lit)) = expr2.as_literal_owned() {
                return Self::expr_eq_literal_inner(expr, lit, expr2, expr1);
            }
        }
        None
    }

    fn expr_eq_literal_inner(
        expr_to_match: &MirScalarExpr,
        literal: Row,
        literal_expr: &MirScalarExpr,
        other_side: &MirScalarExpr,
    ) -> Option<(Row, bool)> {
        if other_side == expr_to_match {
            return Some((literal, false));
        } else {
            // expr didn't exactly match. See if we can match it by inverse-casting.
            let (cast_removed, inv_cast_lit) =
                Self::invert_casts_on_expr_eq_literal_inner(other_side, literal_expr);
            if &cast_removed == expr_to_match {
                if let Some(Ok(inv_cast_lit_row)) = inv_cast_lit.as_literal_owned() {
                    return Some((inv_cast_lit_row, true));
                }
            }
        }
        None
    }

    /// If `self` is `<expr> = <literal>` or `<literal> = <expr>` then
    /// return `<expr>`. It also tries to remove a cast (or other invertible function call) from
    /// `<expr>` before returning it, see `invert_casts_on_expr_eq_literal_inner`.
    pub fn any_expr_eq_literal(&self) -> Option<MirScalarExpr> {
        if let MirScalarExpr::CallBinary {
            func: BinaryFunc::Eq(_),
            expr1,
            expr2,
        } = self
        {
            if expr1.is_literal() {
                let (expr, _literal) = Self::invert_casts_on_expr_eq_literal_inner(expr2, expr1);
                return Some(expr);
            }
            if expr2.is_literal() {
                let (expr, _literal) = Self::invert_casts_on_expr_eq_literal_inner(expr1, expr2);
                return Some(expr);
            }
        }
        None
    }

    /// If the given `MirScalarExpr` is a literal equality where one side is an invertible function
    /// call, then calls the inverse function on both sides of the equality and returns the modified
    /// version of the given `MirScalarExpr`. Otherwise, it returns the original expression.
    /// For more details, see `invert_casts_on_expr_eq_literal_inner`.
    pub fn invert_casts_on_expr_eq_literal(&self) -> MirScalarExpr {
        if let MirScalarExpr::CallBinary {
            func: BinaryFunc::Eq(_),
            expr1,
            expr2,
        } = self
        {
            if expr1.is_literal() {
                let (expr, literal) = Self::invert_casts_on_expr_eq_literal_inner(expr2, expr1);
                return literal.call_binary(expr, func::Eq);
            }
            if expr2.is_literal() {
                let (expr, literal) = Self::invert_casts_on_expr_eq_literal_inner(expr1, expr2);
                return literal.call_binary(expr, func::Eq);
            }
            // Note: The above return statements should be consistent in whether they put the
            // literal in expr1 or expr2, for the deduplication in CanonicalizeMfp to work.
        }
        self.clone()
    }

    /// Given an `<expr>` and a `<literal>` that were taken out from `<expr> = <literal>` or
    /// `<literal> = <expr>`, it tries to simplify the equality by applying the inverse function of
    /// the outermost function call of `<expr>` (if exists):
    ///
    /// `<literal> = func(<inner_expr>)`, where `func` is invertible
    ///  -->
    /// `<func^-1(literal)> = <inner_expr>`
    /// if `func^-1(literal)` doesn't error out, and both `func` and `func^-1` preserve uniqueness.
    ///
    /// The return value is the `<inner_expr>` and the literal value that we get by applying the
    /// inverse function.
    fn invert_casts_on_expr_eq_literal_inner(
        expr: &MirScalarExpr,
        literal: &MirScalarExpr,
    ) -> (MirScalarExpr, MirScalarExpr) {
        assert!(matches!(literal, MirScalarExpr::Literal(..)));

        let temp_storage = &RowArena::new();
        let eval = |e: &MirScalarExpr| {
            MirScalarExpr::literal(e.eval(&[], temp_storage), e.typ(&[]).scalar_type)
        };

        if let MirScalarExpr::CallUnary {
            func,
            expr: inner_expr,
        } = expr
        {
            if let Some(inverse_func) = func.inverse() {
                // We don't want to remove a function call that doesn't preserve uniqueness, e.g.,
                // if `f` is a float, we don't want to inverse-cast `f::INT = 0`, because the
                // inserted int-to-float cast wouldn't be able to invert the rounding.
                // Also, we don't want to insert a function call that doesn't preserve
                // uniqueness. E.g., if `a` has an integer type, we don't want to do
                // a surprise rounding for `WHERE a = 3.14`.
                if func.preserves_uniqueness() && inverse_func.preserves_uniqueness() {
                    let lit_inv = eval(&MirScalarExpr::CallUnary {
                        func: inverse_func,
                        expr: Box::new(literal.clone()),
                    });
                    // The evaluation can error out, e.g., when casting a too large int32 to int16.
                    // This case is handled by `impossible_literal_equality_because_types`.
                    if !lit_inv.is_literal_err() {
                        return (*inner_expr.clone(), lit_inv);
                    }
                }
            }
        }
        (expr.clone(), literal.clone())
    }

    /// Tries to remove a cast (or other invertible function) in the same way as
    /// `invert_casts_on_expr_eq_literal`, but if calling the inverse function fails on the literal,
    /// then it deems the equality to be impossible. For example if `a` is a smallint column, then
    /// it catches `a::integer = 1000000` to be an always false predicate (where the `::integer`
    /// could have been inserted implicitly).
    pub fn impossible_literal_equality_because_types(&self) -> bool {
        if let MirScalarExpr::CallBinary {
            func: BinaryFunc::Eq(_),
            expr1,
            expr2,
        } = self
        {
            if expr1.is_literal() {
                return Self::impossible_literal_equality_because_types_inner(expr1, expr2);
            }
            if expr2.is_literal() {
                return Self::impossible_literal_equality_because_types_inner(expr2, expr1);
            }
        }
        false
    }

    fn impossible_literal_equality_because_types_inner(
        literal: &MirScalarExpr,
        other_side: &MirScalarExpr,
    ) -> bool {
        assert!(matches!(literal, MirScalarExpr::Literal(..)));

        let temp_storage = &RowArena::new();
        let eval = |e: &MirScalarExpr| {
            MirScalarExpr::literal(e.eval(&[], temp_storage), e.typ(&[]).scalar_type)
        };

        if let MirScalarExpr::CallUnary { func, .. } = other_side {
            if let Some(inverse_func) = func.inverse() {
                if inverse_func.preserves_uniqueness()
                    && eval(&MirScalarExpr::CallUnary {
                        func: inverse_func,
                        expr: Box::new(literal.clone()),
                    })
                    .is_literal_err()
                {
                    return true;
                }
            }
        }

        false
    }

    /// Determines if `self` is
    /// `<expr> < <literal>` or
    /// `<expr> > <literal>` or
    /// `<literal> < <expr>` or
    /// `<literal> > <expr>` or
    /// `<expr> <= <literal>` or
    /// `<expr> >= <literal>` or
    /// `<literal> <= <expr>` or
    /// `<literal> >= <expr>`.
    pub fn any_expr_ineq_literal(&self) -> bool {
        match self {
            MirScalarExpr::CallBinary {
                func:
                    BinaryFunc::Lt(_) | BinaryFunc::Lte(_) | BinaryFunc::Gt(_) | BinaryFunc::Gte(_),
                expr1,
                expr2,
            } => expr1.is_literal() || expr2.is_literal(),
            _ => false,
        }
    }

    pub fn take(&mut self) -> Self {
        mem::replace(self, MirScalarExpr::literal_null(ReprScalarType::String))
    }

    /// If the expression is a literal, this returns the literal's Datum or the literal's EvalError.
    /// Otherwise, it returns None.
    pub fn as_literal(&self) -> Option<Result<Datum<'_>, &EvalError>> {
        if let MirScalarExpr::Literal(lit, _column_type) = self {
            Some(lit.as_ref().map(|row| row.unpack_first()))
        } else {
            None
        }
    }

    /// Flattens the two failure modes of `as_literal` into one layer of Option: returns the
    /// literal's Datum only if the expression is a literal, and it's not a literal error.
    pub fn as_literal_non_error(&self) -> Option<Datum<'_>> {
        self.as_literal().map(|eval_err| eval_err.ok()).flatten()
    }

    pub fn as_literal_owned(&self) -> Option<Result<Row, EvalError>> {
        if let MirScalarExpr::Literal(lit, _column_type) = self {
            Some(lit.clone())
        } else {
            None
        }
    }

    /// Returns a reference to the `Row` if the expression is a non-NULL `Ok` literal.
    pub fn as_literal_non_null_row(&self) -> Option<&Row> {
        if let MirScalarExpr::Literal(Ok(row), _) = self {
            if !row.unpack_first().is_null() {
                return Some(row);
            }
        }
        None
    }

    pub fn as_literal_str(&self) -> Option<&str> {
        match self.as_literal() {
            Some(Ok(Datum::String(s))) => Some(s),
            _ => None,
        }
    }

    pub fn as_literal_int64(&self) -> Option<i64> {
        match self.as_literal() {
            Some(Ok(Datum::Int64(i))) => Some(i),
            _ => None,
        }
    }

    pub fn as_literal_err(&self) -> Option<&EvalError> {
        self.as_literal().and_then(|lit| lit.err())
    }

    pub fn is_literal(&self) -> bool {
        matches!(self, MirScalarExpr::Literal(_, _))
    }

    pub fn is_literal_true(&self) -> bool {
        Some(Ok(Datum::True)) == self.as_literal()
    }

    pub fn is_literal_false(&self) -> bool {
        Some(Ok(Datum::False)) == self.as_literal()
    }

    pub fn is_literal_null(&self) -> bool {
        Some(Ok(Datum::Null)) == self.as_literal()
    }

    pub fn is_literal_ok(&self) -> bool {
        matches!(self, MirScalarExpr::Literal(Ok(_), _typ))
    }

    pub fn is_literal_err(&self) -> bool {
        matches!(self, MirScalarExpr::Literal(Err(_), _typ))
    }

    pub fn is_error_if_null(&self) -> bool {
        matches!(
            self,
            Self::CallVariadic {
                func: VariadicFunc::ErrorIfNull(_),
                ..
            }
        )
    }

    /// If `self` expresses a temporal filter, normalize it to start with `mz_now()` and return
    /// references.
    ///
    /// A temporal filter is an expression of the form `mz_now() <BINOP> <EXPR>`,
    /// for a restricted set of `BINOP` and `EXPR` that do not themselves contain `mz_now()`.
    /// Expressions may conform to this once their expressions are swapped.
    ///
    /// If the expression is not a temporal filter, it will be unchanged, and the reason for why
    /// it's not a temporal filter is returned as a string.
    pub fn as_mut_temporal_filter(&mut self) -> Result<(&BinaryFunc, &mut MirScalarExpr), String> {
        if !self.contains_temporal() {
            return Err("Does not involve mz_now()".to_string());
        }
        // Supported temporal predicates are exclusively binary operators.
        if let MirScalarExpr::CallBinary { func, expr1, expr2 } = self {
            // Attempt to put `LogicalTimestamp` in the first argument position.
            if !expr1.contains_temporal()
                && **expr2 == MirScalarExpr::CallUnmaterializable(UnmaterializableFunc::MzNow)
            {
                let new_func = match func {
                    BinaryFunc::Eq(_) => func::Eq.into(),
                    BinaryFunc::Lt(_) => func::Gt.into(),
                    BinaryFunc::Lte(_) => func::Gte.into(),
                    BinaryFunc::Gt(_) => func::Lt.into(),
                    BinaryFunc::Gte(_) => func::Lte.into(),
                    x => {
                        return Err(format!("Unsupported binary temporal operation: {:?}", x));
                    }
                };
                std::mem::swap(expr1, expr2);
                *func = new_func;
            }

            // Error if MLT is referenced in an unsupported position.
            if expr2.contains_temporal()
                || **expr1 != MirScalarExpr::CallUnmaterializable(UnmaterializableFunc::MzNow)
            {
                let mode = HumanizedExplain::new(false); // no redaction
                let bad_expr = MirScalarExpr::CallBinary {
                    func: func.clone(),
                    expr1: expr1.clone(),
                    expr2: expr2.clone(),
                };
                return Err(format!(
                    "Unsupported temporal predicate. Note: `mz_now()` must be directly compared to a mz_timestamp-castable expression. Expression found: {}",
                    mode.expr(&bad_expr, None),
                ));
            }

            Ok((&*func, expr2))
        } else {
            let mode = HumanizedExplain::new(false); // no redaction
            Err(format!(
                "Unsupported temporal predicate. Note: `mz_now()` must be directly compared to a non-temporal expression of mz_timestamp-castable type. Expression found: {}",
                mode.expr(self, None),
            ))
        }
    }

    #[deprecated = "Use `might_error` instead"]
    pub fn contains_error_if_null(&self) -> bool {
        let mut worklist = vec![self];
        while let Some(expr) = worklist.pop() {
            if expr.is_error_if_null() {
                return true;
            }
            worklist.extend(expr.children());
        }
        false
    }

    pub fn contains_err(&self) -> bool {
        let mut worklist = vec![self];
        while let Some(expr) = worklist.pop() {
            if expr.is_literal_err() {
                return true;
            }
            worklist.extend(expr.children());
        }
        false
    }

    /// A very crude approximation for scalar expressions that might produce an
    /// error.
    ///
    /// Currently, this is restricted only to expressions that either contain a
    /// literal error or a [`VariadicFunc::ErrorIfNull`] call.
    pub fn might_error(&self) -> bool {
        let mut worklist = vec![self];
        while let Some(expr) = worklist.pop() {
            if expr.is_literal_err() || expr.is_error_if_null() {
                return true;
            }
            worklist.extend(expr.children());
        }
        false
    }

    /// Reduces a complex expression where possible.
    ///
    /// This function uses nullability information present in `column_types`,
    /// and the result may only continue to be a correct transformation as
    /// long as this information continues to hold (nullability may not hold
    /// as expressions migrate around).
    ///
    /// (If you'd like to not use nullability information here, then you can
    /// tweak the nullabilities in `column_types` before passing it to this
    /// function, see e.g. in `EquivalenceClasses::minimize`.)
    ///
    /// Also performs partial canonicalization on the expression.
    ///
    /// ```rust
    /// use mz_expr::MirScalarExpr;
    /// use mz_repr::{ReprColumnType, Datum, SqlScalarType};
    ///
    /// let expr_0 = MirScalarExpr::column(0);
    /// let expr_t = MirScalarExpr::literal_true();
    /// let expr_f = MirScalarExpr::literal_false();
    ///
    /// let mut test =
    /// expr_t
    ///     .clone()
    ///     .and(expr_f.clone())
    ///     .if_then_else(expr_0, expr_t.clone());
    ///
    /// let input_type = vec![ReprColumnType::from(&SqlScalarType::Int32.nullable(false))];
    /// test.reduce(&input_type);
    /// assert_eq!(test, expr_t);
    /// ```
    /// Reduce the expression to a simpler form.
    pub fn reduce(&mut self, column_types: &[ReprColumnType]) {
        reduce::reduce(self, column_types);
    }

    /// Decompose an IsNull expression into a disjunction of
    /// simpler expressions.
    ///
    /// Assumes that `self` is the expression inside of an IsNull.
    /// Returns `Some(expressions)` if the outer IsNull is to be
    /// replaced by some other expression. Note: if it returns
    /// None, it might still have mutated *self.
    fn decompose_is_null(&mut self) -> Option<MirScalarExpr> {
        // TODO: allow simplification of unmaterializable functions

        match self {
            MirScalarExpr::CallUnary {
                func,
                expr: inner_expr,
            } => {
                if !func.introduces_nulls() {
                    if func.propagates_nulls() {
                        *self = inner_expr.take();
                        return self.decompose_is_null();
                    } else {
                        // We can simplify to `false`, because the function simply can't produce
                        // nulls at all. This is because
                        // - !propagates_nulls means that the input type of the Rust function is not
                        //   nullable, so the automatic null propagation won't kick in;
                        // - !introduces_nulls means that the output type of the Rust function is
                        //   not nullable, so the Rust function can't produce a null manually either.
                        //
                        // Note that we can't do this same optimization for binary and variadic
                        // functions. This is because for binary and variadic functions the value of
                        // propagates_nulls and introduces_nulls is not derived solely from the
                        // input/output type nullabilities, but instead depends on what the Rust
                        // function does. For example, list concatenation neither introduces nor
                        // propagates nulls, but it can produce a null:
                        // - It does not introduce nulls, because if both input lists are not null,
                        //   then it will produce a list.
                        // - It does not propagate nulls, because giving a null as just one of the
                        //   arguments returns the other argument instead of null.
                        // - It does produce a null if both arguments are null.
                        return Some(MirScalarExpr::literal_false());
                    }
                }
            }
            MirScalarExpr::CallBinary { func, expr1, expr2 } => {
                // (<expr1> <op> <expr2>) IS NULL can often be simplified to
                // (<expr1> IS NULL) OR (<expr2> IS NULL).
                if func.propagates_nulls() && !func.introduces_nulls() {
                    let expr1 = expr1.take().call_is_null();
                    let expr2 = expr2.take().call_is_null();
                    return Some(expr1.or(expr2));
                }
            }
            MirScalarExpr::CallVariadic { func, exprs } => {
                if func.propagates_nulls() && !func.introduces_nulls() {
                    let exprs = exprs.into_iter().map(|e| e.take().call_is_null()).collect();
                    return Some(MirScalarExpr::call_variadic(Or, exprs));
                }
            }
            _ => {}
        }

        None
    }

    /// Flattens a chain of calls to associative variadic functions
    /// (For example: ORs or ANDs)
    pub fn flatten_associative(&mut self) {
        match self {
            MirScalarExpr::CallVariadic {
                exprs: outer_operands,
                func: outer_func,
            } if outer_func.is_associative() => {
                *outer_operands = outer_operands
                    .into_iter()
                    .flat_map(|o| {
                        if let MirScalarExpr::CallVariadic {
                            exprs: inner_operands,
                            func: inner_func,
                        } = o
                        {
                            if *inner_func == *outer_func {
                                mem::take(inner_operands)
                            } else {
                                vec![o.take()]
                            }
                        } else {
                            vec![o.take()]
                        }
                    })
                    .collect();
            }
            _ => {}
        }
    }

    /* #region AND/OR canonicalization and transformations  */

    /// Canonicalizes AND/OR, and does some straightforward simplifications
    fn reduce_and_canonicalize_and_or(&mut self) {
        // We do this until fixed point, because after undistribute_and_or calls us, it relies on
        // the property that self is not an 1-arg AND/OR. Just one application of our loop body
        // can't ensure this, because the application itself might create a 1-arg AND/OR.
        let mut old_self = MirScalarExpr::column(0);
        while old_self != *self {
            old_self = self.clone();
            match self {
                MirScalarExpr::CallVariadic {
                    func: func @ (VariadicFunc::And(_) | VariadicFunc::Or(_)),
                    exprs,
                } => {
                    // Canonically order elements so that various deduplications work better,
                    // e.g., in undistribute_and_or.
                    // Also, extract_equal_or_both_null_inner depends on the args being sorted.
                    exprs.sort();

                    // x AND/OR x --> x
                    exprs.dedup(); // this also needs the above sorting

                    if exprs.len() == 1 {
                        // AND/OR of 1 argument evaluates to that argument
                        *self = exprs.swap_remove(0);
                    } else if exprs.len() == 0 {
                        // AND/OR of 0 arguments evaluates to true/false
                        *self = func.unit_of_and_or();
                    } else if exprs.iter().any(|e| *e == func.zero_of_and_or()) {
                        // short-circuiting
                        *self = func.zero_of_and_or();
                    } else {
                        // a AND true --> a
                        // a OR false --> a
                        exprs.retain(|e| *e != func.unit_of_and_or());
                    }
                }
                _ => {}
            }
        }
    }

    /// Transforms !(a && b) into !a || !b, and !(a || b) into !a && !b
    fn demorgans(&mut self) {
        if let MirScalarExpr::CallUnary {
            expr: inner,
            func: UnaryFunc::Not(func::Not),
        } = self
        {
            inner.flatten_associative();
            match &mut **inner {
                MirScalarExpr::CallVariadic {
                    func: inner_func @ (VariadicFunc::And(_) | VariadicFunc::Or(_)),
                    exprs,
                } => {
                    *inner_func = inner_func.switch_and_or();
                    *exprs = exprs.into_iter().map(|e| e.take().not()).collect();
                    *self = (*inner).take(); // Removes the outer not
                }
                _ => {}
            }
        }
    }

    /// AND/OR undistribution (factoring out) to apply at each `MirScalarExpr`.
    ///
    /// This method attempts to apply one of the [distribution laws][distributivity]
    /// (in a direction opposite to the their name):
    /// ```text
    /// (a && b) || (a && c) --> a && (b || c)  // Undistribute-OR
    /// (a || b) && (a || c) --> a || (b && c)  // Undistribute-AND
    /// ```
    /// or one of their corresponding two [absorption law][absorption] special
    /// cases:
    /// ```text
    /// a || (a && c)  -->  a  // Absorb-OR
    /// a && (a || c)  -->  a  // Absorb-AND
    /// ```
    ///
    /// The method also works with more than 2 arguments at the top, e.g.
    /// ```text
    /// (a && b) || (a && c) || (a && d)  -->  a && (b || c || d)
    /// ```
    /// It can also factor out only a subset of the top arguments, e.g.
    /// ```text
    /// (a && b) || (a && c) || (d && e)  -->  (a && (b || c)) || (d && e)
    /// ```
    ///
    /// Note that sometimes there are two overlapping possibilities to factor
    /// out from, e.g.
    /// ```text
    /// (a && b) || (a && c) || (d && c)
    /// ```
    /// Here we can factor out `a` from from the 1. and 2. terms, or we can
    /// factor out `c` from the 2. and 3. terms. One of these might lead to
    /// more/better undistribution opportunities later, but we just pick one
    /// locally, because recursively trying out all of them would lead to
    /// exponential run time.
    ///
    /// The local heuristic is that we prefer a candidate that leads to an
    /// absorption, or if there is no such one then we simply pick the first. In
    /// case of multiple absorption candidates, it doesn't matter which one we
    /// pick, because applying an absorption cannot adversely effect the
    /// possibility of applying other absorptions.
    ///
    /// # Assumption
    ///
    /// Assumes that nested chains of AND/OR applications are flattened (this
    /// can be enforced with [`Self::flatten_associative`]).
    ///
    /// # Examples
    ///
    /// Absorb-OR:
    /// ```text
    /// a || (a && c) || (a && d)
    /// -->
    /// a && (true || c || d)
    /// -->
    /// a && true
    /// -->
    /// a
    /// ```
    /// Here only the first step is performed by this method. The rest is done
    /// by [`Self::reduce_and_canonicalize_and_or`] called after us in
    /// `reduce()`.
    ///
    /// [distributivity]: https://en.wikipedia.org/wiki/Distributive_property
    /// [absorption]: https://en.wikipedia.org/wiki/Absorption_law
    fn undistribute_and_or(&mut self) {
        // It wouldn't be strictly necessary to wrap this fn in this loop, because `reduce()` calls
        // us in a loop anyway. However, `reduce()` tries to do many other things, so the loop here
        // improves performance when there are several undistributions to apply in sequence, which
        // can occur in `CanonicalizeMfp` when undoing the DNF.
        let mut old_self = MirScalarExpr::column(0);
        while old_self != *self {
            old_self = self.clone();
            self.reduce_and_canonicalize_and_or(); // We don't want to deal with 1-arg AND/OR at the top
            if let MirScalarExpr::CallVariadic {
                exprs: outer_operands,
                func: outer_func @ (VariadicFunc::Or(_) | VariadicFunc::And(_)),
            } = self
            {
                let inner_func = outer_func.switch_and_or();

                // Make sure that each outer operand is a call to inner_func, by wrapping in a 1-arg
                // call if necessary.
                outer_operands.iter_mut().for_each(|o| {
                    if !matches!(o, MirScalarExpr::CallVariadic {func: f, ..} if *f == inner_func) {
                        *o = MirScalarExpr::CallVariadic {
                            func: inner_func.clone(),
                            exprs: vec![o.take()],
                        };
                    }
                });

                let mut inner_operands_refs: Vec<&mut Vec<MirScalarExpr>> = outer_operands
                    .iter_mut()
                    .map(|o| match o {
                        MirScalarExpr::CallVariadic { func: f, exprs } if *f == inner_func => exprs,
                        _ => unreachable!(), // the wrapping made sure that we'll get a match
                    })
                    .collect();

                // Find inner operands to undistribute, i.e., which are in _all_ of the outer operands.
                let mut intersection = inner_operands_refs
                    .iter()
                    .map(|v| (*v).clone())
                    .reduce(|ops1, ops2| ops1.into_iter().filter(|e| ops2.contains(e)).collect())
                    .unwrap();
                intersection.sort();
                intersection.dedup();

                if !intersection.is_empty() {
                    // Factor out the intersection from all the top-level args.

                    // Remove the intersection from each inner operand vector.
                    inner_operands_refs
                        .iter_mut()
                        .for_each(|ops| (**ops).retain(|o| !intersection.contains(o)));

                    // Simplify terms that now have only 0 or 1 args due to removing the intersection.
                    outer_operands
                        .iter_mut()
                        .for_each(|o| o.reduce_and_canonicalize_and_or());

                    // Add the intersection at the beginning
                    *self = MirScalarExpr::CallVariadic {
                        func: inner_func,
                        exprs: intersection.into_iter().chain_one(self.clone()).collect(),
                    };
                } else {
                    // If the intersection was empty, that means that there is nothing we can factor out
                    // from _all_ the top-level args. However, we might still find something to factor
                    // out from a subset of the top-level args. To find such an opportunity, we look for
                    // duplicates across all inner args, e.g. if we have
                    // `(...) OR (... AND `a` AND ...) OR (...) OR (... AND `a` AND ...)`
                    // then we'll find that `a` occurs in more than one top-level arg, so
                    // `indexes_to_undistribute` will point us to the 2. and 4. top-level args.

                    // Create (inner_operand, index) pairs, where the index is the position in
                    // outer_operands
                    let all_inner_operands = inner_operands_refs
                        .iter()
                        .enumerate()
                        .flat_map(|(i, inner_vec)| inner_vec.iter().map(move |a| ((*a).clone(), i)))
                        .sorted()
                        .collect_vec();

                    // Find inner operand expressions that occur in more than one top-level arg.
                    // Each inner vector in `undistribution_opportunities` will belong to one such inner
                    // operand expression, and it is a set of indexes pointing to top-level args where
                    // that inner operand occurs.
                    let undistribution_opportunities = all_inner_operands
                        .iter()
                        .chunk_by(|(a, _i)| a)
                        .into_iter()
                        .map(|(_a, g)| g.map(|(_a, i)| *i).sorted().dedup().collect_vec())
                        .filter(|g| g.len() > 1)
                        .collect_vec();

                    // Choose one of the inner vectors from `undistribution_opportunities`.
                    let indexes_to_undistribute = undistribution_opportunities
                        .iter()
                        // Let's prefer index sets that directly lead to an absorption.
                        .find(|index_set| {
                            index_set
                                .iter()
                                .any(|i| inner_operands_refs.get(*i).unwrap().len() == 1)
                        })
                        // If we didn't find any absorption, then any index set will do.
                        .or_else(|| undistribution_opportunities.first())
                        .cloned();

                    // In any case, undo the 1-arg wrapping that we did at the beginning.
                    outer_operands
                        .iter_mut()
                        .for_each(|o| o.reduce_and_canonicalize_and_or());

                    if let Some(indexes_to_undistribute) = indexes_to_undistribute {
                        // Found something to undistribute from a subset of the outer operands.
                        // We temporarily remove these from outer_operands, call ourselves on it, and
                        // then push back the result.
                        let mut undistribute_from = MirScalarExpr::CallVariadic {
                            func: outer_func.clone(),
                            exprs: swap_remove_multiple(outer_operands, indexes_to_undistribute),
                        };
                        // By construction, the recursive call is guaranteed to hit
                        // the `!intersection.is_empty()` branch.
                        undistribute_from.undistribute_and_or();
                        // Append the undistributed result to outer operands that were not included in
                        // indexes_to_undistribute.
                        outer_operands.push(undistribute_from);
                    }
                }
            }
        }
    }

    /* #endregion */

    /// Adds any columns that *must* be non-Null for `self` to be non-Null.
    pub fn non_null_requirements(&self, columns: &mut BTreeSet<usize>) {
        match self {
            MirScalarExpr::Column(col, _name) => {
                columns.insert(*col);
            }
            MirScalarExpr::Literal(..) => {}
            MirScalarExpr::CallUnmaterializable(_) => (),
            MirScalarExpr::CallUnary { func, expr } => {
                if func.propagates_nulls() {
                    expr.non_null_requirements(columns);
                }
            }
            MirScalarExpr::CallBinary { func, expr1, expr2 } => {
                if func.propagates_nulls() {
                    expr1.non_null_requirements(columns);
                    expr2.non_null_requirements(columns);
                }
            }
            MirScalarExpr::CallVariadic { func, exprs } => {
                if func.propagates_nulls() {
                    for expr in exprs {
                        expr.non_null_requirements(columns);
                    }
                }
            }
            MirScalarExpr::If { .. } => (),
        }
    }

    pub fn sql_typ(&self, column_types: &[SqlColumnType]) -> SqlColumnType {
        let repr_column_types = column_types.iter().map(ReprColumnType::from).collect_vec();
        SqlColumnType::from_repr(&self.typ(&repr_column_types))
    }

    pub fn typ(&self, column_types: &[ReprColumnType]) -> ReprColumnType {
        match self {
            MirScalarExpr::Column(i, _name) => column_types[*i].clone(),
            MirScalarExpr::Literal(_, typ) => typ.clone(),
            MirScalarExpr::CallUnmaterializable(func) => func.output_type(),
            MirScalarExpr::CallUnary { expr, func } => func.output_type(expr.typ(column_types)),
            MirScalarExpr::CallBinary { expr1, expr2, func } => {
                func.output_type(&[expr1.typ(column_types), expr2.typ(column_types)])
            }
            MirScalarExpr::CallVariadic { exprs, func } => {
                func.output_type(exprs.iter().map(|e| e.typ(column_types)).collect())
            }
            MirScalarExpr::If { cond: _, then, els } => {
                let then_type = then.typ(column_types);
                let else_type = els.typ(column_types);
                then_type.union(&else_type).unwrap()
            }
        }
    }

    /// True iff the expression contains
    /// `UnmaterializableFunc::MzNow`.
    pub fn contains_temporal(&self) -> bool {
        let mut contains = false;
        self.visit_pre(|e| {
            if let MirScalarExpr::CallUnmaterializable(UnmaterializableFunc::MzNow) = e {
                contains = true;
            }
        });
        contains
    }

    /// True iff the expression contains an `UnmaterializableFunc`.
    pub fn contains_unmaterializable(&self) -> bool {
        let mut contains = false;
        self.visit_pre(|e| {
            if let MirScalarExpr::CallUnmaterializable(_) = e {
                contains = true;
            }
        });
        contains
    }

    /// True iff the expression contains a call that constant folding must not
    /// evaluate: an `UnmaterializableFunc`, or `jq`, whose programs can loop or
    /// recurse without bound and would hang or abort the optimizing process.
    pub fn contains_unfoldable(&self) -> bool {
        let mut contains = false;
        self.visit_pre(|e| match e {
            MirScalarExpr::CallUnmaterializable(_)
            | MirScalarExpr::CallUnary {
                func: UnaryFunc::Jq(_),
                ..
            } => contains = true,
            _ => {}
        });
        contains
    }

    /// True iff the expression contains an `UnmaterializableFunc` that is not in the `exceptions`
    /// list.
    pub fn contains_unmaterializable_except(&self, exceptions: &[UnmaterializableFunc]) -> bool {
        let mut contains = false;
        self.visit_pre(|e| match e {
            MirScalarExpr::CallUnmaterializable(f) if !exceptions.contains(f) => contains = true,
            _ => (),
        });
        contains
    }

    /// True iff the expression contains a `Column`.
    pub fn contains_column(&self) -> bool {
        let mut contains = false;
        self.visit_pre(|e| {
            if let MirScalarExpr::Column(_col, _name) = e {
                contains = true;
            }
        });
        contains
    }

    /// True iff the expression contains a `Dummy`.
    pub fn contains_dummy(&self) -> bool {
        let mut contains = false;
        self.visit_pre(|e| {
            if let MirScalarExpr::Literal(row, _) = e {
                if let Ok(row) = row {
                    contains |= row.iter().any(|d| d.contains_dummy());
                }
            }
        });
        contains
    }

    /// The size of the expression as a tree.
    pub fn size(&self) -> usize {
        let mut size = 0;
        self.visit_pre(&mut |_: &MirScalarExpr| {
            size += 1;
        });
        size
    }
}

/// Fails once `temp_storage` holds more than the budget it was built with.
///
/// Checked after each function call rather than inside the arena, because `RowArena`'s pushes are
/// infallible: refusing one would hand back a truncated value. Between calls is the innermost point
/// that can return an error, so a budgeted arena reaches at most its budget plus whatever the call
/// that crossed it allocated. Functions that can predict their own size cut that overshoot by
/// consulting [`crate::func::max_string_func_result_bytes`] first.
///
/// An unbudgeted arena, which is every arena in a dataflow, costs one branch on a `None`.
fn check_temp_storage_budget(temp_storage: &RowArena) -> Result<(), EvalError> {
    if temp_storage.over_budget() {
        return Err(EvalError::TempStorageBudgetExceeded);
    }
    Ok(())
}

impl Eval for MirScalarExpr {
    fn eval<'a>(
        &'a self,
        datums: &[Datum<'a>],
        temp_storage: &'a RowArena,
    ) -> Result<Datum<'a>, EvalError> {
        match self {
            MirScalarExpr::Column(index, _name) => Ok(datums[*index]),
            MirScalarExpr::Literal(res, _column_type) => match res {
                Ok(row) => Ok(row.unpack_first()),
                Err(e) => Err(e.clone()),
            },
            // Unmaterializable functions must be transformed away before
            // evaluation. Their purpose is as a placeholder for data that is
            // not known at plan time but can be inlined before runtime.
            MirScalarExpr::CallUnmaterializable(x) => Err(EvalError::Internal(
                format!("cannot evaluate unmaterializable function: {:?}", x).into(),
            )),
            MirScalarExpr::CallUnary { func, expr } => {
                let datum = func.eval(datums, temp_storage, expr.as_ref())?;
                check_temp_storage_budget(temp_storage)?;
                Ok(datum)
            }
            MirScalarExpr::CallBinary { func, expr1, expr2 } => {
                let datum = func.eval(datums, temp_storage, &[expr1.as_ref(), expr2.as_ref()])?;
                check_temp_storage_budget(temp_storage)?;
                Ok(datum)
            }
            MirScalarExpr::CallVariadic { func, exprs } => {
                let datum = func.eval(datums, temp_storage, exprs.as_slice())?;
                check_temp_storage_budget(temp_storage)?;
                Ok(datum)
            }
            MirScalarExpr::If { cond, then, els } => match cond.eval(datums, temp_storage)? {
                Datum::True => then.eval(datums, temp_storage),
                Datum::False | Datum::Null => els.eval(datums, temp_storage),
                d => Err(EvalError::Internal(
                    format!("if condition evaluated to non-boolean datum: {:?}", d).into(),
                )),
            },
        }
    }

    fn could_error(&self) -> bool {
        match self {
            MirScalarExpr::Column(_col, _name) => false,
            MirScalarExpr::Literal(row, ..) => row.is_err(),
            MirScalarExpr::CallUnmaterializable(_) => true,
            MirScalarExpr::CallUnary { func, expr } => func.could_error() || expr.could_error(),
            MirScalarExpr::CallBinary { func, expr1, expr2 } => {
                func.could_error() || expr1.could_error() || expr2.could_error()
            }
            MirScalarExpr::CallVariadic { func, exprs } => {
                func.could_error() || exprs.iter().any(|e| e.could_error())
            }
            MirScalarExpr::If { cond, then, els } => {
                cond.could_error() || then.could_error() || els.could_error()
            }
        }
    }
}

impl Columns for MirScalarExpr {
    fn column(c: usize) -> Self {
        MirScalarExpr::column(c)
    }

    fn is_column(&self) -> bool {
        matches!(self, MirScalarExpr::Column(_col, _name))
    }

    fn as_column(&self) -> Option<usize> {
        if let MirScalarExpr::Column(c, _) = self {
            Some(*c)
        } else {
            None
        }
    }

    fn as_column_mut(&mut self) -> Option<&mut usize> {
        if let MirScalarExpr::Column(c, _) = self {
            Some(c)
        } else {
            None
        }
    }

    fn support_into(&self, support: &mut BTreeSet<usize>) {
        self.visit_pre(|e| {
            if let MirScalarExpr::Column(i, _) = e {
                support.insert(*i);
            }
        });
    }

    fn visit_columns<F>(&mut self, mut action: F)
    where
        F: FnMut(&mut usize),
    {
        self.visit_pre_mut(|e| {
            if let MirScalarExpr::Column(col, _) = e {
                action(col);
            }
        });
    }
}

impl VisitChildren<Self> for MirScalarExpr {
    fn visit_children<F>(&self, mut f: F)
    where
        F: FnMut(&Self),
    {
        use MirScalarExpr::*;
        match self {
            Column(_, _) | Literal(_, _) | CallUnmaterializable(_) => (),
            CallUnary { expr, .. } => {
                f(expr);
            }
            CallBinary { expr1, expr2, .. } => {
                f(expr1);
                f(expr2);
            }
            CallVariadic { exprs, .. } => {
                for expr in exprs {
                    f(expr);
                }
            }
            If { cond, then, els } => {
                f(cond);
                f(then);
                f(els);
            }
        }
    }

    fn visit_mut_children<F>(&mut self, mut f: F)
    where
        F: FnMut(&mut Self),
    {
        use MirScalarExpr::*;
        match self {
            Column(_, _) | Literal(_, _) | CallUnmaterializable(_) => (),
            CallUnary { expr, .. } => {
                f(expr);
            }
            CallBinary { expr1, expr2, .. } => {
                f(expr1);
                f(expr2);
            }
            CallVariadic { exprs, .. } => {
                for expr in exprs {
                    f(expr);
                }
            }
            If { cond, then, els } => {
                f(cond);
                f(then);
                f(els);
            }
        }
    }

    fn try_visit_children<F, E>(&self, mut f: F) -> Result<(), E>
    where
        F: FnMut(&Self) -> Result<(), E>,
    {
        use MirScalarExpr::*;
        match self {
            Column(_, _) | Literal(_, _) | CallUnmaterializable(_) => (),
            CallUnary { expr, .. } => {
                f(expr)?;
            }
            CallBinary { expr1, expr2, .. } => {
                f(expr1)?;
                f(expr2)?;
            }
            CallVariadic { exprs, .. } => {
                for expr in exprs {
                    f(expr)?;
                }
            }
            If { cond, then, els } => {
                f(cond)?;
                f(then)?;
                f(els)?;
            }
        }
        Ok(())
    }

    fn try_visit_mut_children<F, E>(&mut self, mut f: F) -> Result<(), E>
    where
        F: FnMut(&mut Self) -> Result<(), E>,
    {
        use MirScalarExpr::*;
        match self {
            Column(_, _) | Literal(_, _) | CallUnmaterializable(_) => (),
            CallUnary { expr, .. } => {
                f(expr)?;
            }
            CallBinary { expr1, expr2, .. } => {
                f(expr1)?;
                f(expr2)?;
            }
            CallVariadic { exprs, .. } => {
                for expr in exprs {
                    f(expr)?;
                }
            }
            If { cond, then, els } => {
                f(cond)?;
                f(then)?;
                f(els)?;
            }
        }
        Ok(())
    }

    fn children<'a>(&'a self) -> impl DoubleEndedIterator<Item = &'a Self>
    where
        Self: 'a,
    {
        self.children()
    }

    fn children_mut<'a>(&'a mut self) -> impl DoubleEndedIterator<Item = &'a mut Self>
    where
        Self: 'a,
    {
        self.children_mut()
    }
}

impl MirScalarExpr {
    /// Reports whether this expression contains a non-strict variadic call with an
    /// operand that could error, the shape of the open bug CLU-137.
    ///
    /// `And`, `Or` and `ErrorIfNull` do not evaluate every operand: `Or::eval`
    /// returns `true` the moment it sees a true operand and drops any error it
    /// collected, `And::eval` does the same for `false`, and `ErrorIfNull`
    /// evaluates its message operand only when the first operand is NULL. Yet
    /// `reduce_call_variadic`'s generic error propagation replaces the whole call
    /// with any operand's literal error, wherever it sits. So `reduce` can turn a
    /// row the expression should evaluate into an error, and, because that literal
    /// is typed non-nullable, it can go on to license a nullability-dependent
    /// rewrite and yield a different *value* rather than an error.
    ///
    /// Fuzz oracles that compare evaluation across `reduce` use this to skip the
    /// shape rather than rediscover CLU-137 on every run. It lives here, next to
    /// the fold it describes, so the several oracles that need it cannot drift
    /// apart on which functions count as non-strict.
    ///
    /// Deliberately conservative: it asks whether an operand *could* error rather
    /// than whether it already holds a literal error, because `reduce` folds a
    /// column-free fallible operand (`1 / 0`) to a literal error first and absorbs
    /// it after.
    pub fn could_hit_nonstrict_error_fold(&self) -> bool {
        let mut hit = false;
        self.visit_pre(|e| {
            if let MirScalarExpr::CallVariadic { func, exprs } = e {
                let non_strict = matches!(
                    func,
                    VariadicFunc::And(_) | VariadicFunc::Or(_) | VariadicFunc::ErrorIfNull(_)
                );
                if non_strict && exprs.iter().any(|operand| operand.could_error()) {
                    hit = true;
                }
            }
        });
        hit
    }

    /// Iterates through references to child expressions.
    pub fn children(&self) -> impl DoubleEndedIterator<Item = &Self> {
        let mut first = None;
        let mut second = None;
        let mut third = None;
        let mut variadic = None;

        use MirScalarExpr::*;
        match self {
            Column(_, _) | Literal(_, _) | CallUnmaterializable(_) => (),
            CallUnary { expr, .. } => {
                first = Some(&**expr);
            }
            CallBinary { expr1, expr2, .. } => {
                first = Some(&**expr1);
                second = Some(&**expr2);
            }
            CallVariadic { exprs, .. } => {
                variadic = Some(exprs);
            }
            If { cond, then, els } => {
                first = Some(&**cond);
                second = Some(&**then);
                third = Some(&**els);
            }
        }

        first
            .into_iter()
            .chain(second)
            .chain(third)
            .chain(variadic.into_iter().flatten())
    }

    /// Iterates through mutable references to child expressions.
    pub fn children_mut(&mut self) -> impl DoubleEndedIterator<Item = &mut Self> {
        let mut first = None;
        let mut second = None;
        let mut third = None;
        let mut variadic = None;

        use MirScalarExpr::*;
        match self {
            Column(_, _) | Literal(_, _) | CallUnmaterializable(_) => (),
            CallUnary { expr, .. } => {
                first = Some(&mut **expr);
            }
            CallBinary { expr1, expr2, .. } => {
                first = Some(&mut **expr1);
                second = Some(&mut **expr2);
            }
            CallVariadic { exprs, .. } => {
                variadic = Some(exprs);
            }
            If { cond, then, els } => {
                first = Some(&mut **cond);
                second = Some(&mut **then);
                third = Some(&mut **els);
            }
        }

        first
            .into_iter()
            .chain(second)
            .chain(third)
            .chain(variadic.into_iter().flatten())
    }

    /// Visits all subexpressions in DFS preorder.
    pub fn visit_pre<F>(&self, mut f: F)
    where
        F: FnMut(&Self),
    {
        let mut worklist = vec![self];
        while let Some(e) = worklist.pop() {
            f(e);
            worklist.extend(e.children().rev());
        }
    }

    /// Iterative pre-order visitor.
    pub fn visit_pre_mut<F: FnMut(&mut Self)>(&mut self, mut f: F) {
        let mut worklist = vec![self];
        while let Some(expr) = worklist.pop() {
            f(expr);
            worklist.extend(expr.children_mut().rev());
        }
    }
}

/// Filter characteristics that are used for ordering join inputs.
/// This can be created for a `Vec<MirScalarExpr>`, which represents an AND of predicates.
///
/// The fields are ordered based on heuristic assumptions about their typical selectivity, so that
/// Ord gives the right ordering for join inputs. Bigger is better, i.e., will tend to come earlier
/// than other inputs.
#[derive(
    Eq,
    PartialEq,
    Ord,
    PartialOrd,
    Debug,
    Clone,
    Serialize,
    Deserialize,
    Hash
)]
pub struct FilterCharacteristics {
    // `<expr> = <literal>` appears in the filter.
    // Excludes cases where NOT appears anywhere above the literal equality.
    literal_equality: bool,
    // (Assuming a random string of lower-case characters, `LIKE 'a%'` has a selectivity of 1/26.)
    like: bool,
    is_null: bool,
    // Number of Vec elements that involve inequality predicates. (A BETWEEN is represented as two
    // inequality predicates.)
    // Excludes cases where NOT appears around the literal inequality.
    // Note that for inequality predicates, some databases assume 1/3 selectivity in the absence of
    // concrete statistics.
    literal_inequality: usize,
    /// Any filter, except ones involving `IS NOT NULL`, because those are too common.
    /// Can be true by itself, or any other field being true can also make this true.
    /// `NOT LIKE` is only in this category.
    /// `!=` is only in this category.
    /// `NOT (a = b)` is turned into `!=` by `reduce` before us!
    any_filter: bool,
}

impl BitOrAssign for FilterCharacteristics {
    fn bitor_assign(&mut self, rhs: Self) {
        self.literal_equality |= rhs.literal_equality;
        self.like |= rhs.like;
        self.is_null |= rhs.is_null;
        self.literal_inequality += rhs.literal_inequality;
        self.any_filter |= rhs.any_filter;
    }
}

impl FilterCharacteristics {
    pub fn none() -> FilterCharacteristics {
        FilterCharacteristics {
            literal_equality: false,
            like: false,
            is_null: false,
            literal_inequality: 0,
            any_filter: false,
        }
    }

    pub fn explain(&self) -> String {
        let mut e = "".to_owned();
        if self.literal_equality {
            e.push_str("e");
        }
        if self.like {
            e.push_str("l");
        }
        if self.is_null {
            e.push_str("n");
        }
        for _ in 0..self.literal_inequality {
            e.push_str("i");
        }
        if self.any_filter {
            e.push_str("f");
        }
        e
    }

    pub fn filter_characteristics(
        filters: &Vec<MirScalarExpr>,
    ) -> Result<FilterCharacteristics, RecursionLimitError> {
        let mut literal_equality = false;
        let mut like = false;
        let mut is_null = false;
        let mut literal_inequality = 0;
        let mut any_filter = false;
        filters.iter().try_for_each(|f| {
            let mut literal_inequality_in_current_filter = false;
            let mut is_not_null_in_current_filter = false;
            f.visit_pre_with_context(
                false,
                &mut |not_in_parent_chain, expr| {
                    not_in_parent_chain
                        || matches!(
                            expr,
                            MirScalarExpr::CallUnary {
                                func: UnaryFunc::Not(func::Not),
                                ..
                            }
                        )
                },
                &mut |not_in_parent_chain, expr| {
                    if !not_in_parent_chain {
                        if expr.any_expr_eq_literal().is_some() {
                            literal_equality = true;
                        }
                        if expr.any_expr_ineq_literal() {
                            literal_inequality_in_current_filter = true;
                        }
                        if matches!(
                            expr,
                            MirScalarExpr::CallUnary {
                                func: UnaryFunc::IsLikeMatch(_),
                                ..
                            }
                        ) {
                            like = true;
                        }
                    };
                    if matches!(
                        expr,
                        MirScalarExpr::CallUnary {
                            func: UnaryFunc::IsNull(crate::func::IsNull),
                            ..
                        }
                    ) {
                        if *not_in_parent_chain {
                            is_not_null_in_current_filter = true;
                        } else {
                            is_null = true;
                        }
                    }
                },
            );
            if literal_inequality_in_current_filter {
                literal_inequality += 1;
            }
            if !is_not_null_in_current_filter {
                // We want to ignore `IS NOT NULL` for `any_filter`.
                any_filter = true;
            }
            Ok(())
        })?;
        Ok(FilterCharacteristics {
            literal_equality,
            like,
            is_null,
            literal_inequality,
            any_filter,
        })
    }

    pub fn add_literal_equality(&mut self) {
        self.literal_equality = true;
    }

    pub fn worst_case_scaling_factor(&self) -> f64 {
        let mut factor = 1.0;

        if self.literal_equality {
            factor *= 0.1;
        }

        if self.is_null {
            factor *= 0.1;
        }

        if self.literal_inequality >= 2 {
            factor *= 0.25;
        } else if self.literal_inequality == 1 {
            factor *= 0.33;
        }

        // catch various negated filters, treat them pessimistically
        if !(self.literal_equality || self.is_null || self.literal_inequality > 0)
            && self.any_filter
        {
            factor *= 0.9;
        }

        factor
    }
}

#[derive(
    Ord,
    PartialOrd,
    Copy,
    Clone,
    Debug,
    Eq,
    PartialEq,
    Serialize,
    Deserialize,
    Hash
)]
#[cfg_attr(any(test, feature = "proptest"), derive(Arbitrary))]
pub enum DomainLimit {
    None,
    Inclusive(i64),
    Exclusive(i64),
}

impl RustType<ProtoDomainLimit> for DomainLimit {
    fn into_proto(&self) -> ProtoDomainLimit {
        use proto_domain_limit::Kind::*;
        let kind = match self {
            DomainLimit::None => None(()),
            DomainLimit::Inclusive(v) => Inclusive(*v),
            DomainLimit::Exclusive(v) => Exclusive(*v),
        };
        ProtoDomainLimit { kind: Some(kind) }
    }

    fn from_proto(proto: ProtoDomainLimit) -> Result<Self, TryFromProtoError> {
        use proto_domain_limit::Kind::*;
        if let Some(kind) = proto.kind {
            match kind {
                None(()) => Ok(DomainLimit::None),
                Inclusive(v) => Ok(DomainLimit::Inclusive(v)),
                Exclusive(v) => Ok(DomainLimit::Exclusive(v)),
            }
        } else {
            Err(TryFromProtoError::missing_field("ProtoDomainLimit::kind"))
        }
    }
}

#[derive(
    Ord,
    PartialOrd,
    Clone,
    Debug,
    Eq,
    PartialEq,
    Serialize,
    Deserialize,
    Hash
)]
#[cfg_attr(any(test, feature = "proptest"), derive(Arbitrary))]
pub enum EvalError {
    CharacterNotValidForEncoding(i32),
    CharacterTooLargeForEncoding(i32),
    DateBinOutOfRange(Box<str>),
    DivisionByZero,
    Unsupported {
        feature: Box<str>,
        discussion_no: Option<usize>,
    },
    FloatOverflow,
    FloatUnderflow,
    NumericFieldOverflow,
    Float32OutOfRange(Box<str>),
    Float64OutOfRange(Box<str>),
    Int16OutOfRange(Box<str>),
    Int32OutOfRange(Box<str>),
    Int64OutOfRange(Box<str>),
    UInt16OutOfRange(Box<str>),
    UInt32OutOfRange(Box<str>),
    UInt64OutOfRange(Box<str>),
    MzTimestampOutOfRange(Box<str>),
    MzTimestampStepOverflow,
    OidOutOfRange(Box<str>),
    IntervalOutOfRange(Box<str>),
    TimestampCannotBeNan,
    TimestampOutOfRange,
    DateOutOfRange,
    CharOutOfRange,
    IndexOutOfRange {
        provided: i32,
        // The last valid index position, i.e. `v.len() - 1`
        valid_end: i32,
    },
    InvalidBase64Equals,
    InvalidBase64Symbol(char),
    InvalidBase64EndSequence,
    InvalidTimezone(Box<str>),
    InvalidTimezoneInterval,
    InvalidTimezoneConversion,
    InvalidIanaTimezoneId(Box<str>),
    InvalidLayer {
        max_layer: usize,
        val: i64,
    },
    InvalidArray(InvalidArrayError),
    InvalidEncodingName(Box<str>),
    InvalidHashAlgorithm(Box<str>),
    InvalidByteSequence {
        byte_sequence: Box<str>,
        encoding_name: Box<str>,
    },
    InvalidJsonbCast {
        from: Box<str>,
        to: Box<str>,
    },
    InvalidRegex(Box<str>),
    InvalidRegexFlag(char),
    InvalidParameterValue(Box<str>),
    InvalidDatePart(Box<str>),
    KeyCannotBeNull,
    NegSqrt,
    NegLimit,
    NullCharacterNotPermitted,
    UnknownUnits(Box<str>),
    UnsupportedUnits(Box<str>, Box<str>),
    UnterminatedLikeEscapeSequence,
    Parse(ParseError),
    ParseHex(ParseHexError),
    Internal(Box<str>),
    InfinityOutOfDomain(Box<str>),
    NegativeOutOfDomain(Box<str>),
    ZeroOutOfDomain(Box<str>),
    OutOfDomain(DomainLimit, DomainLimit, Box<str>),
    ComplexOutOfRange(Box<str>),
    MultipleRowsFromSubquery,
    NegativeRowsFromSubquery,
    Undefined(Box<str>),
    LikePatternTooLong,
    LikeEscapeTooLong,
    StringValueTooLong {
        target_type: Box<str>,
        length: usize,
    },
    MultidimensionalArrayRemovalNotSupported,
    IncompatibleArrayDimensions {
        dims: Option<(usize, usize)>,
    },
    TypeFromOid(Box<str>),
    InvalidRange(InvalidRangeError),
    InvalidRoleId(Box<str>),
    InvalidPrivileges(Box<str>),
    InvalidCatalogJson(Box<str>),
    LetRecLimitExceeded(Box<str>),
    MultiDimensionalArraySearch,
    MustNotBeNull(Box<str>),
    InvalidIdentifier {
        ident: Box<str>,
        detail: Option<Box<str>>,
    },
    ArrayFillWrongArraySubscripts,
    // TODO: propagate this check more widely throughout the expr crate
    MaxArraySizeExceeded(usize),
    DateDiffOverflow {
        unit: Box<str>,
        a: Box<str>,
        b: Box<str>,
    },
    // The error for ErrorIfNull; this should not be used in other contexts as a generic error
    // printer.
    IfNullError(Box<str>),
    LengthTooLarge,
    // A budgeted `RowArena` (`mz_repr::RowArena::with_budget`) exceeded its budget while an
    // expression was being evaluated. Only a budgeted arena raises this, so it never arises in a
    // dataflow, only on the webhook `CHECK` path that runs user expressions in `environmentd`.
    TempStorageBudgetExceeded,
    AclArrayNullElement,
    MzAclArrayNullElement,
    PrettyError(Box<str>),
    RedactError(Box<str>),
}

impl fmt::Display for EvalError {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        match self {
            EvalError::CharacterNotValidForEncoding(v) => {
                write!(f, "requested character not valid for encoding: {v}")
            }
            EvalError::CharacterTooLargeForEncoding(v) => {
                write!(f, "requested character too large for encoding: {v}")
            }
            EvalError::DateBinOutOfRange(message) => f.write_str(message),
            EvalError::DivisionByZero => f.write_str("division by zero"),
            EvalError::Unsupported {
                feature,
                discussion_no,
            } => {
                write!(f, "{} not yet supported", feature)?;
                if let Some(discussion_no) = discussion_no {
                    write!(
                        f,
                        ", see https://github.com/MaterializeInc/materialize/discussions/{} for more details",
                        discussion_no
                    )?;
                }
                Ok(())
            }
            EvalError::FloatOverflow => f.write_str("value out of range: overflow"),
            EvalError::FloatUnderflow => f.write_str("value out of range: underflow"),
            EvalError::NumericFieldOverflow => f.write_str("numeric field overflow"),
            EvalError::Float32OutOfRange(val) => write!(f, "{} real out of range", val.quoted()),
            EvalError::Float64OutOfRange(val) => {
                write!(f, "{} double precision out of range", val.quoted())
            }
            EvalError::Int16OutOfRange(val) => write!(f, "{} smallint out of range", val.quoted()),
            EvalError::Int32OutOfRange(val) => write!(f, "{} integer out of range", val.quoted()),
            EvalError::Int64OutOfRange(val) => write!(f, "{} bigint out of range", val.quoted()),
            EvalError::UInt16OutOfRange(val) => write!(f, "{} uint2 out of range", val.quoted()),
            EvalError::UInt32OutOfRange(val) => write!(f, "{} uint4 out of range", val.quoted()),
            EvalError::UInt64OutOfRange(val) => write!(f, "{} uint8 out of range", val.quoted()),
            EvalError::MzTimestampOutOfRange(val) => {
                write!(f, "{} mz_timestamp out of range", val.quoted())
            }
            EvalError::MzTimestampStepOverflow => f.write_str("step mz_timestamp overflow"),
            EvalError::OidOutOfRange(val) => write!(f, "{} OID out of range", val.quoted()),
            EvalError::IntervalOutOfRange(val) => {
                write!(f, "{} interval out of range", val.quoted())
            }
            EvalError::TimestampCannotBeNan => f.write_str("timestamp cannot be NaN"),
            EvalError::TimestampOutOfRange => f.write_str("timestamp out of range"),
            EvalError::DateOutOfRange => f.write_str("date out of range"),
            EvalError::CharOutOfRange => f.write_str("\"char\" out of range"),
            EvalError::IndexOutOfRange {
                provided,
                valid_end,
            } => write!(f, "index {provided} out of valid range, 0..{valid_end}",),
            EvalError::InvalidBase64Equals => {
                f.write_str("unexpected \"=\" while decoding base64 sequence")
            }
            EvalError::InvalidBase64Symbol(c) => write!(
                f,
                "invalid symbol \"{}\" found while decoding base64 sequence",
                c.escape_default()
            ),
            EvalError::InvalidBase64EndSequence => f.write_str("invalid base64 end sequence"),
            EvalError::InvalidJsonbCast { from, to } => {
                write!(f, "cannot cast jsonb {} to type {}", from, to)
            }
            EvalError::InvalidTimezone(tz) => write!(f, "invalid time zone '{}'", tz),
            EvalError::InvalidTimezoneInterval => {
                f.write_str("timezone interval must not contain months or years")
            }
            EvalError::InvalidTimezoneConversion => f.write_str("invalid timezone conversion"),
            EvalError::InvalidIanaTimezoneId(tz) => {
                write!(f, "invalid IANA Time Zone Database identifier: '{}'", tz)
            }
            EvalError::InvalidLayer { max_layer, val } => write!(
                f,
                "invalid layer: {}; must use value within [1, {}]",
                val, max_layer
            ),
            EvalError::InvalidArray(e) => e.fmt(f),
            EvalError::InvalidEncodingName(name) => write!(f, "invalid encoding name '{}'", name),
            EvalError::InvalidHashAlgorithm(alg) => write!(f, "invalid hash algorithm '{}'", alg),
            EvalError::InvalidByteSequence {
                byte_sequence,
                encoding_name,
            } => write!(
                f,
                "invalid byte sequence '{}' for encoding '{}'",
                byte_sequence, encoding_name
            ),
            EvalError::InvalidDatePart(part) => write!(f, "invalid datepart {}", part.quoted()),
            EvalError::KeyCannotBeNull => f.write_str("key cannot be null"),
            EvalError::NegSqrt => f.write_str("cannot take square root of a negative number"),
            EvalError::NegLimit => f.write_str("LIMIT must not be negative"),
            EvalError::NullCharacterNotPermitted => f.write_str("null character not permitted"),
            EvalError::InvalidRegex(e) => write!(f, "invalid regular expression: {}", e),
            EvalError::InvalidRegexFlag(c) => write!(f, "invalid regular expression flag: {}", c),
            EvalError::InvalidParameterValue(s) => f.write_str(s),
            EvalError::UnknownUnits(units) => write!(f, "unit '{}' not recognized", units),
            EvalError::UnsupportedUnits(units, typ) => {
                write!(f, "unit '{}' not supported for type {}", units, typ)
            }
            EvalError::UnterminatedLikeEscapeSequence => {
                f.write_str("unterminated escape sequence in LIKE")
            }
            EvalError::Parse(e) => e.fmt(f),
            EvalError::PrettyError(e) => e.fmt(f),
            EvalError::RedactError(e) => e.fmt(f),
            EvalError::ParseHex(e) => e.fmt(f),
            EvalError::Internal(s) => write!(f, "internal error: {}", s),
            EvalError::InfinityOutOfDomain(s) => {
                write!(f, "function {} is only defined for finite arguments", s)
            }
            EvalError::NegativeOutOfDomain(s) => {
                write!(f, "function {} is not defined for negative numbers", s)
            }
            EvalError::ZeroOutOfDomain(s) => {
                write!(f, "function {} is not defined for zero", s)
            }
            EvalError::OutOfDomain(lower, upper, s) => {
                use DomainLimit::*;
                write!(f, "function {s} is defined for numbers ")?;
                match (lower, upper) {
                    (Inclusive(n), None) => write!(f, "greater than or equal to {n}"),
                    (Exclusive(n), None) => write!(f, "greater than {n}"),
                    (None, Inclusive(n)) => write!(f, "less than or equal to {n}"),
                    (None, Exclusive(n)) => write!(f, "less than {n}"),
                    (Inclusive(lo), Inclusive(hi)) => write!(f, "between {lo} and {hi} inclusive"),
                    (Exclusive(lo), Exclusive(hi)) => write!(f, "between {lo} and {hi} exclusive"),
                    (Inclusive(lo), Exclusive(hi)) => {
                        write!(f, "between {lo} inclusive and {hi} exclusive")
                    }
                    (Exclusive(lo), Inclusive(hi)) => {
                        write!(f, "between {lo} exclusive and {hi} inclusive")
                    }
                    // No caller constructs an unbounded domain, but a corrupted or
                    // forged `ProtoEvalError` decodes into one. Render it instead of
                    // panicking: `DataflowErrorSer::Display` decodes errors straight
                    // out of a persist shard and Displays them on the index peek
                    // path, so a panicking arm here wedges the dataflow on every
                    // retry rather than producing a bad error message once.
                    (None, None) => write!(f, "in an unspecified range"),
                }
            }
            EvalError::ComplexOutOfRange(s) => {
                write!(f, "function {} cannot return complex numbers", s)
            }
            EvalError::MultipleRowsFromSubquery => {
                write!(f, "more than one record produced in subquery")
            }
            EvalError::NegativeRowsFromSubquery => {
                write!(f, "negative number of rows produced in subquery")
            }
            EvalError::Undefined(s) => {
                write!(f, "{} is undefined", s)
            }
            EvalError::LikePatternTooLong => {
                write!(f, "LIKE pattern exceeds maximum length")
            }
            EvalError::LikeEscapeTooLong => {
                write!(f, "invalid escape string")
            }
            EvalError::StringValueTooLong {
                target_type,
                length,
            } => {
                write!(f, "value too long for type {}({})", target_type, length)
            }
            EvalError::MultidimensionalArrayRemovalNotSupported => {
                write!(
                    f,
                    "removing elements from multidimensional arrays is not supported"
                )
            }
            EvalError::IncompatibleArrayDimensions { dims: _ } => {
                write!(f, "cannot concatenate incompatible arrays")
            }
            EvalError::TypeFromOid(msg) => write!(f, "{msg}"),
            EvalError::InvalidRange(e) => e.fmt(f),
            EvalError::InvalidRoleId(msg) => write!(f, "{msg}"),
            EvalError::InvalidPrivileges(privilege) => {
                write!(f, "unrecognized privilege type: {privilege}")
            }
            EvalError::InvalidCatalogJson(msg) => {
                write!(f, "invalid catalog JSON: {msg}")
            }
            EvalError::LetRecLimitExceeded(max_iters) => {
                write!(
                    f,
                    "Recursive query exceeded the recursion limit {}. (Use RETURN AT RECURSION LIMIT to not error, but return the current state as the final result when reaching the limit.)",
                    max_iters
                )
            }
            EvalError::MultiDimensionalArraySearch => write!(
                f,
                "searching for elements in multidimensional arrays is not supported"
            ),
            EvalError::MustNotBeNull(v) => write!(f, "{v} must not be null"),
            EvalError::InvalidIdentifier { ident, .. } => {
                write!(f, "string is not a valid identifier: {}", ident.quoted())
            }
            EvalError::ArrayFillWrongArraySubscripts => {
                f.write_str("wrong number of array subscripts")
            }
            EvalError::MaxArraySizeExceeded(max_size) => {
                write!(
                    f,
                    "array size exceeds the maximum allowed ({max_size} bytes)"
                )
            }
            EvalError::DateDiffOverflow { unit, a, b } => {
                write!(f, "datediff overflow, {unit} of {a}, {b}")
            }
            EvalError::IfNullError(s) => f.write_str(s),
            EvalError::LengthTooLarge => write!(f, "requested length too large"),
            EvalError::TempStorageBudgetExceeded => {
                write!(f, "expression exceeded its temporary storage limit")
            }
            EvalError::AclArrayNullElement => write!(f, "ACL arrays must not contain null values"),
            EvalError::MzAclArrayNullElement => {
                write!(f, "MZ_ACL arrays must not contain null values")
            }
        }
    }
}

impl EvalError {
    pub fn detail(&self) -> Option<String> {
        match self {
            EvalError::IncompatibleArrayDimensions { dims: None } => Some(
                "Arrays with differing dimensions are not compatible for concatenation.".into(),
            ),
            EvalError::IncompatibleArrayDimensions {
                dims: Some((a_dims, b_dims)),
            } => Some(format!(
                "Arrays of {} and {} dimensions are not compatible for concatenation.",
                a_dims, b_dims
            )),
            EvalError::InvalidIdentifier { detail, .. } => detail.as_deref().map(Into::into),
            EvalError::ArrayFillWrongArraySubscripts => {
                Some("Low bound array has different size than dimensions array.".into())
            }
            _ => None,
        }
    }

    pub fn hint(&self) -> Option<String> {
        match self {
            EvalError::InvalidBase64EndSequence => Some(
                "Input data is missing padding, is truncated, or is otherwise corrupted.".into(),
            ),
            EvalError::LikeEscapeTooLong => {
                Some("Escape string must be empty or one character.".into())
            }
            EvalError::MzTimestampOutOfRange(_) => Some(
                "Integer, numeric, and text casts to mz_timestamp must be in the form of whole \
                milliseconds since the Unix epoch. Values with fractional parts cannot be \
                converted to mz_timestamp."
                    .into(),
            ),
            _ => None,
        }
    }
}

impl std::error::Error for EvalError {}

impl From<ParseError> for EvalError {
    fn from(e: ParseError) -> EvalError {
        EvalError::Parse(e)
    }
}

impl From<ParseHexError> for EvalError {
    fn from(e: ParseHexError) -> EvalError {
        EvalError::ParseHex(e)
    }
}

impl From<InvalidArrayError> for EvalError {
    fn from(e: InvalidArrayError) -> EvalError {
        EvalError::InvalidArray(e)
    }
}

impl From<RegexCompilationError> for EvalError {
    fn from(e: RegexCompilationError) -> EvalError {
        EvalError::InvalidRegex(e.to_string().into())
    }
}

impl From<TypeFromOidError> for EvalError {
    fn from(e: TypeFromOidError) -> EvalError {
        EvalError::TypeFromOid(e.to_string().into())
    }
}

impl From<DateError> for EvalError {
    fn from(e: DateError) -> EvalError {
        match e {
            DateError::OutOfRange => EvalError::DateOutOfRange,
        }
    }
}

impl From<TimestampError> for EvalError {
    fn from(e: TimestampError) -> EvalError {
        match e {
            TimestampError::OutOfRange => EvalError::TimestampOutOfRange,
        }
    }
}

impl From<InvalidRangeError> for EvalError {
    fn from(e: InvalidRangeError) -> EvalError {
        EvalError::InvalidRange(e)
    }
}

impl RustType<ProtoEvalError> for EvalError {
    fn into_proto(&self) -> ProtoEvalError {
        use proto_eval_error::Kind::*;
        use proto_eval_error::*;
        let kind = match self {
            EvalError::CharacterNotValidForEncoding(v) => CharacterNotValidForEncoding(*v),
            EvalError::CharacterTooLargeForEncoding(v) => CharacterTooLargeForEncoding(*v),
            EvalError::DateBinOutOfRange(v) => DateBinOutOfRange(v.into_proto()),
            EvalError::DivisionByZero => DivisionByZero(()),
            EvalError::Unsupported {
                feature,
                discussion_no,
            } => Unsupported(ProtoUnsupported {
                feature: feature.into_proto(),
                discussion_no: discussion_no.into_proto(),
            }),
            EvalError::FloatOverflow => FloatOverflow(()),
            EvalError::FloatUnderflow => FloatUnderflow(()),
            EvalError::NumericFieldOverflow => NumericFieldOverflow(()),
            EvalError::Float32OutOfRange(val) => Float32OutOfRange(ProtoValueOutOfRange {
                value: val.to_string(),
            }),
            EvalError::Float64OutOfRange(val) => Float64OutOfRange(ProtoValueOutOfRange {
                value: val.to_string(),
            }),
            EvalError::Int16OutOfRange(val) => Int16OutOfRange(ProtoValueOutOfRange {
                value: val.to_string(),
            }),
            EvalError::Int32OutOfRange(val) => Int32OutOfRange(ProtoValueOutOfRange {
                value: val.to_string(),
            }),
            EvalError::Int64OutOfRange(val) => Int64OutOfRange(ProtoValueOutOfRange {
                value: val.to_string(),
            }),
            EvalError::UInt16OutOfRange(val) => Uint16OutOfRange(ProtoValueOutOfRange {
                value: val.to_string(),
            }),
            EvalError::UInt32OutOfRange(val) => Uint32OutOfRange(ProtoValueOutOfRange {
                value: val.to_string(),
            }),
            EvalError::UInt64OutOfRange(val) => Uint64OutOfRange(ProtoValueOutOfRange {
                value: val.to_string(),
            }),
            EvalError::MzTimestampOutOfRange(val) => MzTimestampOutOfRange(ProtoValueOutOfRange {
                value: val.to_string(),
            }),
            EvalError::MzTimestampStepOverflow => MzTimestampStepOverflow(()),
            EvalError::OidOutOfRange(val) => OidOutOfRange(ProtoValueOutOfRange {
                value: val.to_string(),
            }),
            EvalError::IntervalOutOfRange(val) => IntervalOutOfRange(ProtoValueOutOfRange {
                value: val.to_string(),
            }),
            EvalError::TimestampCannotBeNan => TimestampCannotBeNan(()),
            EvalError::TimestampOutOfRange => TimestampOutOfRange(()),
            EvalError::DateOutOfRange => DateOutOfRange(()),
            EvalError::CharOutOfRange => CharOutOfRange(()),
            EvalError::IndexOutOfRange {
                provided,
                valid_end,
            } => IndexOutOfRange(ProtoIndexOutOfRange {
                provided: *provided,
                valid_end: *valid_end,
            }),
            EvalError::InvalidBase64Equals => InvalidBase64Equals(()),
            EvalError::InvalidBase64Symbol(sym) => InvalidBase64Symbol(sym.into_proto()),
            EvalError::InvalidBase64EndSequence => InvalidBase64EndSequence(()),
            EvalError::InvalidTimezone(tz) => InvalidTimezone(tz.into_proto()),
            EvalError::InvalidTimezoneInterval => InvalidTimezoneInterval(()),
            EvalError::InvalidTimezoneConversion => InvalidTimezoneConversion(()),
            EvalError::InvalidLayer { max_layer, val } => InvalidLayer(ProtoInvalidLayer {
                max_layer: max_layer.into_proto(),
                val: *val,
            }),
            EvalError::InvalidArray(error) => InvalidArray(error.into_proto()),
            EvalError::InvalidEncodingName(v) => InvalidEncodingName(v.into_proto()),
            EvalError::InvalidHashAlgorithm(v) => InvalidHashAlgorithm(v.into_proto()),
            EvalError::InvalidByteSequence {
                byte_sequence,
                encoding_name,
            } => InvalidByteSequence(ProtoInvalidByteSequence {
                byte_sequence: byte_sequence.into_proto(),
                encoding_name: encoding_name.into_proto(),
            }),
            EvalError::InvalidJsonbCast { from, to } => InvalidJsonbCast(ProtoInvalidJsonbCast {
                from: from.into_proto(),
                to: to.into_proto(),
            }),
            EvalError::InvalidRegex(v) => InvalidRegex(v.into_proto()),
            EvalError::InvalidRegexFlag(v) => InvalidRegexFlag(v.into_proto()),
            EvalError::InvalidParameterValue(v) => InvalidParameterValue(v.into_proto()),
            EvalError::InvalidDatePart(part) => InvalidDatePart(part.into_proto()),
            EvalError::KeyCannotBeNull => KeyCannotBeNull(()),
            EvalError::NegSqrt => NegSqrt(()),
            EvalError::NegLimit => NegLimit(()),
            EvalError::NullCharacterNotPermitted => NullCharacterNotPermitted(()),
            EvalError::UnknownUnits(v) => UnknownUnits(v.into_proto()),
            EvalError::UnsupportedUnits(units, typ) => UnsupportedUnits(ProtoUnsupportedUnits {
                units: units.into_proto(),
                typ: typ.into_proto(),
            }),
            EvalError::UnterminatedLikeEscapeSequence => UnterminatedLikeEscapeSequence(()),
            EvalError::Parse(error) => Parse(error.into_proto()),
            EvalError::PrettyError(error) => PrettyError(error.into_proto()),
            EvalError::RedactError(error) => RedactError(error.into_proto()),
            EvalError::ParseHex(error) => ParseHex(error.into_proto()),
            EvalError::Internal(v) => Internal(v.into_proto()),
            EvalError::InfinityOutOfDomain(v) => InfinityOutOfDomain(v.into_proto()),
            EvalError::NegativeOutOfDomain(v) => NegativeOutOfDomain(v.into_proto()),
            EvalError::ZeroOutOfDomain(v) => ZeroOutOfDomain(v.into_proto()),
            EvalError::OutOfDomain(lower, upper, id) => OutOfDomain(ProtoOutOfDomain {
                lower: Some(lower.into_proto()),
                upper: Some(upper.into_proto()),
                id: id.into_proto(),
            }),
            EvalError::ComplexOutOfRange(v) => ComplexOutOfRange(v.into_proto()),
            EvalError::MultipleRowsFromSubquery => MultipleRowsFromSubquery(()),
            EvalError::NegativeRowsFromSubquery => NegativeRowsFromSubquery(()),
            EvalError::Undefined(v) => Undefined(v.into_proto()),
            EvalError::LikePatternTooLong => LikePatternTooLong(()),
            EvalError::LikeEscapeTooLong => LikeEscapeTooLong(()),
            EvalError::StringValueTooLong {
                target_type,
                length,
            } => StringValueTooLong(ProtoStringValueTooLong {
                target_type: target_type.into_proto(),
                length: length.into_proto(),
            }),
            EvalError::MultidimensionalArrayRemovalNotSupported => {
                MultidimensionalArrayRemovalNotSupported(())
            }
            EvalError::IncompatibleArrayDimensions { dims } => {
                IncompatibleArrayDimensions(ProtoIncompatibleArrayDimensions {
                    dims: dims.into_proto(),
                })
            }
            EvalError::TypeFromOid(v) => TypeFromOid(v.into_proto()),
            EvalError::InvalidRange(error) => InvalidRange(error.into_proto()),
            EvalError::InvalidRoleId(v) => InvalidRoleId(v.into_proto()),
            EvalError::InvalidPrivileges(v) => InvalidPrivileges(v.into_proto()),
            EvalError::InvalidCatalogJson(v) => InvalidCatalogJson(v.into_proto()),
            EvalError::LetRecLimitExceeded(v) => WmrRecursionLimitExceeded(v.into_proto()),
            EvalError::MultiDimensionalArraySearch => MultiDimensionalArraySearch(()),
            EvalError::MustNotBeNull(v) => MustNotBeNull(v.into_proto()),
            EvalError::InvalidIdentifier { ident, detail } => {
                InvalidIdentifier(ProtoInvalidIdentifier {
                    ident: ident.into_proto(),
                    detail: detail.into_proto(),
                })
            }
            EvalError::ArrayFillWrongArraySubscripts => ArrayFillWrongArraySubscripts(()),
            EvalError::MaxArraySizeExceeded(max_size) => {
                MaxArraySizeExceeded(u64::cast_from(*max_size))
            }
            EvalError::DateDiffOverflow { unit, a, b } => DateDiffOverflow(ProtoDateDiffOverflow {
                unit: unit.into_proto(),
                a: a.into_proto(),
                b: b.into_proto(),
            }),
            EvalError::IfNullError(s) => IfNullError(s.into_proto()),
            EvalError::LengthTooLarge => LengthTooLarge(()),
            EvalError::TempStorageBudgetExceeded => TempStorageBudgetExceeded(()),
            EvalError::AclArrayNullElement => AclArrayNullElement(()),
            EvalError::MzAclArrayNullElement => MzAclArrayNullElement(()),
            EvalError::InvalidIanaTimezoneId(s) => InvalidIanaTimezoneId(s.into_proto()),
        };
        ProtoEvalError { kind: Some(kind) }
    }

    fn from_proto(proto: ProtoEvalError) -> Result<Self, TryFromProtoError> {
        use proto_eval_error::Kind::*;
        match proto.kind {
            Some(kind) => match kind {
                CharacterNotValidForEncoding(v) => Ok(EvalError::CharacterNotValidForEncoding(v)),
                CharacterTooLargeForEncoding(v) => Ok(EvalError::CharacterTooLargeForEncoding(v)),
                DateBinOutOfRange(v) => Ok(EvalError::DateBinOutOfRange(v.into())),
                DivisionByZero(()) => Ok(EvalError::DivisionByZero),
                Unsupported(v) => Ok(EvalError::Unsupported {
                    feature: v.feature.into(),
                    discussion_no: v.discussion_no.into_rust()?,
                }),
                FloatOverflow(()) => Ok(EvalError::FloatOverflow),
                FloatUnderflow(()) => Ok(EvalError::FloatUnderflow),
                NumericFieldOverflow(()) => Ok(EvalError::NumericFieldOverflow),
                Float32OutOfRange(val) => Ok(EvalError::Float32OutOfRange(val.value.into())),
                Float64OutOfRange(val) => Ok(EvalError::Float64OutOfRange(val.value.into())),
                Int16OutOfRange(val) => Ok(EvalError::Int16OutOfRange(val.value.into())),
                Int32OutOfRange(val) => Ok(EvalError::Int32OutOfRange(val.value.into())),
                Int64OutOfRange(val) => Ok(EvalError::Int64OutOfRange(val.value.into())),
                Uint16OutOfRange(val) => Ok(EvalError::UInt16OutOfRange(val.value.into())),
                Uint32OutOfRange(val) => Ok(EvalError::UInt32OutOfRange(val.value.into())),
                Uint64OutOfRange(val) => Ok(EvalError::UInt64OutOfRange(val.value.into())),
                MzTimestampOutOfRange(val) => {
                    Ok(EvalError::MzTimestampOutOfRange(val.value.into()))
                }
                MzTimestampStepOverflow(()) => Ok(EvalError::MzTimestampStepOverflow),
                OidOutOfRange(val) => Ok(EvalError::OidOutOfRange(val.value.into())),
                IntervalOutOfRange(val) => Ok(EvalError::IntervalOutOfRange(val.value.into())),
                TimestampCannotBeNan(()) => Ok(EvalError::TimestampCannotBeNan),
                TimestampOutOfRange(()) => Ok(EvalError::TimestampOutOfRange),
                DateOutOfRange(()) => Ok(EvalError::DateOutOfRange),
                CharOutOfRange(()) => Ok(EvalError::CharOutOfRange),
                IndexOutOfRange(v) => Ok(EvalError::IndexOutOfRange {
                    provided: v.provided,
                    valid_end: v.valid_end,
                }),
                InvalidBase64Equals(()) => Ok(EvalError::InvalidBase64Equals),
                InvalidBase64Symbol(v) => char::from_proto(v).map(EvalError::InvalidBase64Symbol),
                InvalidBase64EndSequence(()) => Ok(EvalError::InvalidBase64EndSequence),
                InvalidTimezone(v) => Ok(EvalError::InvalidTimezone(v.into())),
                InvalidTimezoneInterval(()) => Ok(EvalError::InvalidTimezoneInterval),
                InvalidTimezoneConversion(()) => Ok(EvalError::InvalidTimezoneConversion),
                InvalidLayer(v) => Ok(EvalError::InvalidLayer {
                    max_layer: usize::from_proto(v.max_layer)?,
                    val: v.val,
                }),
                InvalidArray(error) => Ok(EvalError::InvalidArray(error.into_rust()?)),
                InvalidEncodingName(v) => Ok(EvalError::InvalidEncodingName(v.into())),
                InvalidHashAlgorithm(v) => Ok(EvalError::InvalidHashAlgorithm(v.into())),
                InvalidByteSequence(v) => Ok(EvalError::InvalidByteSequence {
                    byte_sequence: v.byte_sequence.into(),
                    encoding_name: v.encoding_name.into(),
                }),
                InvalidJsonbCast(v) => Ok(EvalError::InvalidJsonbCast {
                    from: v.from.into(),
                    to: v.to.into(),
                }),
                InvalidRegex(v) => Ok(EvalError::InvalidRegex(v.into())),
                InvalidRegexFlag(v) => Ok(EvalError::InvalidRegexFlag(char::from_proto(v)?)),
                InvalidParameterValue(v) => Ok(EvalError::InvalidParameterValue(v.into())),
                InvalidDatePart(part) => Ok(EvalError::InvalidDatePart(part.into())),
                KeyCannotBeNull(()) => Ok(EvalError::KeyCannotBeNull),
                NegSqrt(()) => Ok(EvalError::NegSqrt),
                NegLimit(()) => Ok(EvalError::NegLimit),
                NullCharacterNotPermitted(()) => Ok(EvalError::NullCharacterNotPermitted),
                UnknownUnits(v) => Ok(EvalError::UnknownUnits(v.into())),
                UnsupportedUnits(v) => {
                    Ok(EvalError::UnsupportedUnits(v.units.into(), v.typ.into()))
                }
                UnterminatedLikeEscapeSequence(()) => Ok(EvalError::UnterminatedLikeEscapeSequence),
                Parse(error) => Ok(EvalError::Parse(error.into_rust()?)),
                ParseHex(error) => Ok(EvalError::ParseHex(error.into_rust()?)),
                Internal(v) => Ok(EvalError::Internal(v.into())),
                InfinityOutOfDomain(v) => Ok(EvalError::InfinityOutOfDomain(v.into())),
                NegativeOutOfDomain(v) => Ok(EvalError::NegativeOutOfDomain(v.into())),
                ZeroOutOfDomain(v) => Ok(EvalError::ZeroOutOfDomain(v.into())),
                OutOfDomain(v) => Ok(EvalError::OutOfDomain(
                    v.lower.into_rust_if_some("ProtoDomainLimit::lower")?,
                    v.upper.into_rust_if_some("ProtoDomainLimit::upper")?,
                    v.id.into(),
                )),
                ComplexOutOfRange(v) => Ok(EvalError::ComplexOutOfRange(v.into())),
                MultipleRowsFromSubquery(()) => Ok(EvalError::MultipleRowsFromSubquery),
                NegativeRowsFromSubquery(()) => Ok(EvalError::NegativeRowsFromSubquery),
                Undefined(v) => Ok(EvalError::Undefined(v.into())),
                LikePatternTooLong(()) => Ok(EvalError::LikePatternTooLong),
                LikeEscapeTooLong(()) => Ok(EvalError::LikeEscapeTooLong),
                StringValueTooLong(v) => Ok(EvalError::StringValueTooLong {
                    target_type: v.target_type.into(),
                    length: usize::from_proto(v.length)?,
                }),
                MultidimensionalArrayRemovalNotSupported(()) => {
                    Ok(EvalError::MultidimensionalArrayRemovalNotSupported)
                }
                IncompatibleArrayDimensions(v) => Ok(EvalError::IncompatibleArrayDimensions {
                    dims: v.dims.into_rust()?,
                }),
                TypeFromOid(v) => Ok(EvalError::TypeFromOid(v.into())),
                InvalidRange(e) => Ok(EvalError::InvalidRange(e.into_rust()?)),
                InvalidRoleId(v) => Ok(EvalError::InvalidRoleId(v.into())),
                InvalidPrivileges(v) => Ok(EvalError::InvalidPrivileges(v.into())),
                InvalidCatalogJson(v) => Ok(EvalError::InvalidCatalogJson(v.into())),
                WmrRecursionLimitExceeded(v) => Ok(EvalError::LetRecLimitExceeded(v.into())),
                MultiDimensionalArraySearch(()) => Ok(EvalError::MultiDimensionalArraySearch),
                MustNotBeNull(v) => Ok(EvalError::MustNotBeNull(v.into())),
                InvalidIdentifier(v) => Ok(EvalError::InvalidIdentifier {
                    ident: v.ident.into(),
                    detail: v.detail.into_rust()?,
                }),
                ArrayFillWrongArraySubscripts(()) => Ok(EvalError::ArrayFillWrongArraySubscripts),
                MaxArraySizeExceeded(max_size) => {
                    Ok(EvalError::MaxArraySizeExceeded(usize::cast_from(max_size)))
                }
                DateDiffOverflow(v) => Ok(EvalError::DateDiffOverflow {
                    unit: v.unit.into(),
                    a: v.a.into(),
                    b: v.b.into(),
                }),
                IfNullError(v) => Ok(EvalError::IfNullError(v.into())),
                LengthTooLarge(()) => Ok(EvalError::LengthTooLarge),
                TempStorageBudgetExceeded(()) => Ok(EvalError::TempStorageBudgetExceeded),
                AclArrayNullElement(()) => Ok(EvalError::AclArrayNullElement),
                MzAclArrayNullElement(()) => Ok(EvalError::MzAclArrayNullElement),
                InvalidIanaTimezoneId(s) => Ok(EvalError::InvalidIanaTimezoneId(s.into())),
                PrettyError(s) => Ok(EvalError::PrettyError(s.into())),
                RedactError(s) => Ok(EvalError::RedactError(s.into())),
            },
            None => Err(TryFromProtoError::missing_field("ProtoEvalError::kind")),
        }
    }
}

impl RustType<ProtoDims> for (usize, usize) {
    fn into_proto(&self) -> ProtoDims {
        ProtoDims {
            f0: self.0.into_proto(),
            f1: self.1.into_proto(),
        }
    }

    fn from_proto(proto: ProtoDims) -> Result<Self, TryFromProtoError> {
        Ok((proto.f0.into_rust()?, proto.f1.into_rust()?))
    }
}

/// An [`EvalError`] that serializes as protobuf-encoded [`ProtoEvalError`]
/// bytes.
///
/// `EvalError`'s own serde impl mirrors the Rust enum, so every added error
/// variant or payload tweak would change a durable format. Use this wrapper
/// instead wherever an eval error is serialized into a durable, cross-version
/// format, such as the stable LIR plan format (the same role [`StableRow`]
/// plays for rows). `ProtoEvalError` already carries the needed backward
/// compatibility obligation: it is embedded in `ProtoDataflowError`, which
/// persist stores in the error side of `SourceData`, and the proto files are
/// covered by the buf breaking lint.
///
/// [`StableRow`]: mz_repr::StableRow
#[derive(
    Clone,
    Debug,
    Eq,
    PartialEq,
    Ord,
    PartialOrd,
    Hash,
    Serialize,
    Deserialize
)]
pub struct StableEvalError(#[serde(with = "stable_eval_error_proto")] pub EvalError);

/// Borrowing twin of [`StableEvalError`], to serialize without cloning. Both
/// serialize identically, under the `StableEvalError` container name.
#[derive(Debug, Serialize)]
#[serde(rename = "StableEvalError")]
pub struct StableEvalErrorRef<'a>(#[serde(with = "stable_eval_error_proto")] pub &'a EvalError);

impl From<EvalError> for StableEvalError {
    fn from(err: EvalError) -> Self {
        StableEvalError(err)
    }
}

impl std::ops::Deref for StableEvalError {
    type Target = EvalError;

    fn deref(&self) -> &EvalError {
        &self.0
    }
}

mod stable_eval_error_proto {
    use mz_proto::RustType;
    use prost::Message;
    use serde::de::Error;
    use serde::{Deserialize, Deserializer, Serializer};

    use crate::scalar::{EvalError, ProtoEvalError};

    pub fn serialize<S: Serializer, E: std::borrow::Borrow<EvalError>>(
        err: &E,
        serializer: S,
    ) -> Result<S::Ok, S::Error> {
        serializer.serialize_bytes(&err.borrow().into_proto().encode_to_vec())
    }

    pub fn deserialize<'de, D: Deserializer<'de>>(deserializer: D) -> Result<EvalError, D::Error> {
        let bytes = serde_bytes::ByteBuf::deserialize(deserializer)?;
        let proto = ProtoEvalError::decode(bytes.as_slice()).map_err(D::Error::custom)?;
        EvalError::from_proto(proto).map_err(D::Error::custom)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::scalar::func::variadic::Coalesce;

    // StableEvalError's wire format is proto bytes, so every EvalError
    // variant must roundtrip exactly through both a self-describing format
    // (JSON) and a compact binary one (bincode). This also exercises the
    // EvalError -> ProtoEvalError -> EvalError conversion itself, which has
    // no other roundtrip coverage.
    proptest! {
        #![proptest_config(ProptestConfig::with_cases(1000))]

        #[mz_ore::test]
        #[cfg_attr(miri, ignore)] // too slow
        fn stable_eval_error_serde_roundtrip(err in any::<EvalError>()) {
            let stable = StableEvalError(err);

            let json = serde_json::to_string(&stable).expect("serializes to JSON");
            let from_json: StableEvalError =
                serde_json::from_str(&json).expect("deserializes from JSON");
            prop_assert_eq!(&stable, &from_json);

            let bytes = bincode::serialize(&stable).expect("serializes to bincode");
            let from_bincode: StableEvalError =
                bincode::deserialize(&bytes).expect("deserializes from bincode");
            prop_assert_eq!(&stable, &from_bincode);
        }
    }

    /// An `OutOfDomain` with both limits unset is not constructible by any
    /// caller, but it decodes out of corrupted or forged `ProtoEvalError` bytes,
    /// and `DataflowErrorSer::Display` renders decoded errors on the peek path.
    /// Rendering it must not panic.
    #[mz_ore::test]
    fn test_unbounded_out_of_domain_renders() {
        let err = EvalError::OutOfDomain(DomainLimit::None, DomainLimit::None, "f".into());
        assert_eq!(
            err.to_string(),
            "function f is defined for numbers in an unspecified range"
        );
    }

    /// A budgeted arena must stop an amplifying expression that the per-call constant allows
    /// (SQL-431). `repeat(body, 20)` on a 1 MiB body is 20 MiB: far under
    /// `MAX_STRING_FUNC_RESULT_BYTES`, so nothing rejects it without a budget, and it is exactly
    /// the shape that made a webhook `CHECK` turn a bounded request into unbounded heap.
    #[mz_ore::test]
    #[cfg_attr(miri, ignore)] // multi-MB allocations; the small-size UB coverage is in `row.rs`
    fn test_repeat_respects_arena_budget() {
        use crate::scalar::func::RepeatString;

        let body = "a".repeat(1024 * 1024);
        let expr = MirScalarExpr::column(0).call_binary(
            MirScalarExpr::literal_ok(Datum::Int32(20), ReprScalarType::Int32),
            RepeatString,
        );
        let datums = [Datum::String(&body)];

        // Unbudgeted: allowed, and the arena really does hold the 20 MiB.
        let arena = RowArena::new();
        let datum = expr
            .eval(&datums, &arena)
            .expect("under the 100 MiB ceiling");
        assert_eq!(datum.unwrap_str().len(), 20 * 1024 * 1024);
        assert!(arena.allocated_bytes() >= 20 * 1024 * 1024);

        // Budgeted below the result: rejected, and the pre-check means the arena never grew, i.e.
        // the bytes were never allocated rather than allocated and then complained about.
        let arena = RowArena::with_budget(4 * 1024 * 1024);
        assert_eq!(
            expr.eval(&datums, &arena),
            Err(EvalError::LengthTooLarge),
            "an over-budget result must be refused"
        );
        assert_eq!(arena.allocated_bytes(), 0);

        // Budgeted above the result: unaffected.
        let arena = RowArena::with_budget(64 * 1024 * 1024);
        let datum = expr.eval(&datums, &arena).expect("within budget");
        assert_eq!(datum.unwrap_str().len(), 20 * 1024 * 1024);
    }

    /// The budget also has to bound both dimensions of an arena-building function: the packed
    /// result, and the transient it collects on its own stack first. The arena never sees the
    /// transient, so the post-call check can't catch it.
    #[mz_ore::test]
    #[cfg_attr(miri, ignore)] // multi-MB allocations; the small-size UB coverage is in `row.rs`
    fn test_arena_built_result_respects_budget() {
        use crate::scalar::func::variadic::{RegexpSplitToArray, StringToArray};

        let body = "a".repeat(256 * 1024);
        let expr = MirScalarExpr::call_variadic(
            StringToArray,
            vec![
                MirScalarExpr::column(0),
                MirScalarExpr::literal_ok(Datum::String("a"), ReprScalarType::String),
            ],
        );
        let datums = [Datum::String(&body)];

        let arena = RowArena::new();
        expr.eval(&datums, &arena).expect("no ceiling applies");
        let unbudgeted = arena.allocated_bytes();
        assert!(unbudgeted > 0);

        let arena = RowArena::with_budget(unbudgeted / 2);
        assert_eq!(
            expr.eval(&datums, &arena),
            Err(EvalError::TempStorageBudgetExceeded),
            "an over-budget arena-built result must be refused"
        );

        // A split collects every chunk into a `Vec<&str>` first, 16 bytes per fat pointer against
        // the 2 an empty chunk packs to. A budget the packed array fits under still refuses it.
        let intermediate = (body.len() + 1) * std::mem::size_of::<&str>();
        let budget = 4 * unbudgeted;
        assert!(
            unbudgeted < budget && budget < intermediate,
            "budget sits between"
        );
        let arena = RowArena::with_budget(budget);
        assert!(
            expr.eval(&datums, &arena).is_err(),
            "a split costing {intermediate} bytes to build must be refused by a {budget} byte budget"
        );
        assert_eq!(
            arena.allocated_bytes(),
            0,
            "refused before the transient was built"
        );

        // The regexp sibling builds the same transient and takes the same bound.
        let regexp_expr = MirScalarExpr::call_variadic(
            RegexpSplitToArray,
            vec![
                MirScalarExpr::column(0),
                MirScalarExpr::literal_ok(Datum::String("a"), ReprScalarType::String),
            ],
        );
        let arena = RowArena::with_budget(budget);
        assert!(
            regexp_expr.eval(&datums, &arena).is_err(),
            "a regexp split costing {intermediate} bytes to build must be refused too"
        );
        assert_eq!(arena.allocated_bytes(), 0);
    }

    /// `array_fill` sizes its result from a parameter rather than its input, so a budgeted arena
    /// has to refuse it in its own pre-check the way the string amplifiers do (SQL-431). Its result
    /// stays under the `array_fill` size ceiling, so without the budget-aware pre-check the only
    /// thing that would catch it is the evaluator's post-call check, after the spike has happened.
    #[mz_ore::test]
    #[cfg_attr(miri, ignore)] // multi-MB allocations; the small-size UB coverage is in `row.rs`
    fn test_array_fill_respects_arena_budget() {
        use crate::scalar::func::variadic::ArrayFill;
        use mz_repr::adt::array::ArrayDimension;

        // array_fill(1, ARRAY[fill_count]) builds a one-dimensional int array of `fill_count` ones.
        let fill_count: usize = 512 * 1024;
        let dims_storage = RowArena::new();
        let dims = dims_storage
            .try_make_datum(|packer| {
                packer.try_push_array(
                    &[ArrayDimension {
                        lower_bound: 1,
                        length: 1,
                    }],
                    [Datum::Int32(i32::try_from(fill_count).unwrap())],
                )
            })
            .unwrap();
        let expr = MirScalarExpr::call_variadic(
            ArrayFill {
                elem_type: mz_repr::SqlScalarType::Int32,
            },
            vec![MirScalarExpr::column(0), MirScalarExpr::column(1)],
        );
        let datums = [Datum::Int32(1), dims];

        // Unbudgeted: allowed, and the arena really holds the packed array.
        let arena = RowArena::new();
        expr.eval(&datums, &arena)
            .expect("under the array-size ceiling");
        assert!(arena.allocated_bytes() > 0);

        // Budgeted below what the call needs: refused by the pre-check, so nothing was allocated.
        // The intermediate `Vec<Datum>` alone is 512 Ki elements, well over this 1 MiB budget.
        let arena = RowArena::with_budget(1024 * 1024);
        assert_eq!(
            expr.eval(&datums, &arena),
            Err(EvalError::TempStorageBudgetExceeded),
            "an over-budget array_fill must be refused before it allocates"
        );
        assert_eq!(arena.allocated_bytes(), 0);

        // Budgeted well above both the packed result and the intermediate: unaffected.
        let arena = RowArena::with_budget(256 * 1024 * 1024);
        expr.eval(&datums, &arena).expect("within budget");
    }

    /// `array_remove` cannot grow its result, so no packed-size check would ever fire for it. The
    /// transient `Vec<Datum>` it filters into is a fresh input-scaled allocation on top of the input
    /// array's own bytes, and only a pre-check bounds that (SQL-431).
    #[mz_ore::test]
    #[cfg_attr(miri, ignore)] // multi-MB allocations; the small-size UB coverage is in `row.rs`
    fn test_array_remove_respects_arena_budget() {
        use crate::scalar::func::ArrayRemove;
        use mz_repr::adt::array::ArrayDimension;

        const ELEMS: usize = 256 * 1024;
        let input_storage = RowArena::new();
        let array = input_storage
            .try_make_datum(|packer| {
                packer.try_push_array(
                    &[ArrayDimension {
                        lower_bound: 1,
                        length: ELEMS,
                    }],
                    (0..ELEMS).map(|i| Datum::Int32(i32::try_from(i).unwrap())),
                )
            })
            .unwrap();
        let expr = MirScalarExpr::column(0).call_binary(MirScalarExpr::column(1), ArrayRemove);
        let datums = [array, Datum::Int32(0)];

        // Unbudgeted: allowed, and the arena holds the packed result.
        let arena = RowArena::new();
        expr.eval(&datums, &arena).expect("no ceiling applies");
        let unbudgeted = arena.allocated_bytes();
        assert!(unbudgeted > 0);

        // A `Datum` is several times what an `Int32` packs to, so a budget with room for the packed
        // result twice over still has to refuse the transient.
        let transient = ELEMS * std::mem::size_of::<Datum<'_>>();
        let budget = 2 * unbudgeted;
        assert!(budget < transient, "budget sits between");
        let arena = RowArena::with_budget(budget);
        assert_eq!(
            expr.eval(&datums, &arena),
            Err(EvalError::TempStorageBudgetExceeded),
            "an over-budget transient must be refused"
        );
        assert_eq!(arena.allocated_bytes(), 0);

        // Budgeted above both: unaffected.
        let arena = RowArena::with_budget(64 * 1024 * 1024);
        expr.eval(&datums, &arena).expect("within budget");
    }

    /// A budget has to bound what a single call allocates, not just what is observable between
    /// calls (SQL-431).
    ///
    /// The evaluator only polls the budget after `func.eval` has built its result and moved it into
    /// the arena, and erroring then does not give the bytes back. They stay resident until the arena
    /// drops, which for a webhook `CHECK` is the end of the request. So we assert on arena residency,
    /// not the returned `Result`. The overshoot is not a constant either. It scales with the body and
    /// with a multiplier the `CHECK` author picks at DDL time.
    #[mz_ore::test]
    #[cfg_attr(miri, ignore)] // multi-MB allocations; the small-size UB coverage is in `row.rs`
    fn test_single_call_respects_arena_budget() {
        use mz_ore::cast::CastLossy;

        use crate::scalar::func::variadic::{ArrayCreate, PadLeading, Translate};

        // Scaled down from the shipped 5 MiB body and 20 MiB budget. Every amplifier here is linear
        // in the body, so the ratios hold at any scale.
        const BODY_BYTES: usize = 1024 * 1024;
        const BUDGET: usize = 2 * 1024 * 1024;
        const WIDE: &str = "\u{1F4A5}"; // one character, four bytes

        let str_lit = |s| MirScalarExpr::literal_ok(Datum::String(s), ReprScalarType::String);
        // `ARRAY[body, ...]` is not a string function, so no ceiling of its own applies. The
        // multiplier is just how many times the author wrote `body`.
        let array_of = |n| {
            let elem_type = mz_repr::SqlScalarType::String;
            let refs = vec![MirScalarExpr::column(0); n];
            MirScalarExpr::call_variadic(ArrayCreate { elem_type }, refs)
        };
        let cases = [
            ("ARRAY[body x4]", array_of(4)),
            ("ARRAY[body x16]", array_of(16)),
            // `lpad`'s pre-check is budget-aware but compares `len`, a character count, against a
            // budget in bytes, so a 4-byte pad passes a check for exactly the budget then writes 4x.
            (
                "lpad(body, BUDGET, wide)",
                MirScalarExpr::call_variadic(
                    PadLeading,
                    vec![
                        MirScalarExpr::column(0),
                        MirScalarExpr::literal_ok(
                            Datum::Int32(i32::try_from(BUDGET).unwrap()),
                            ReprScalarType::Int32,
                        ),
                        str_lit(WIDE),
                    ],
                ),
            ),
            // `translate` has no pre-check at all, and widening each body byte is a 4x amplifier
            // that needs no length argument to drive it.
            (
                "translate(body, 'a', wide)",
                MirScalarExpr::call_variadic(
                    Translate,
                    vec![MirScalarExpr::column(0), str_lit("a"), str_lit(WIDE)],
                ),
            ),
        ];

        let body = "a".repeat(BODY_BYTES);
        let datums = [Datum::String(&body)];
        let mut over = Vec::new();
        for (name, expr) in cases {
            let arena = RowArena::with_budget(BUDGET);
            let _ = expr.eval(&datums, &arena); // refused or not, we only care what's left resident
            let held = arena.allocated_bytes();
            if held > BUDGET {
                let ratio = f64::cast_lossy(held) / f64::cast_lossy(BUDGET);
                over.push(format!("  {name}: held {held} bytes, {ratio:.1}x"));
            }
        }
        assert!(
            over.is_empty(),
            "a single call left a {BUDGET} byte arena holding more:\n{}",
            over.join("\n"),
        );
    }

    #[mz_ore::test]
    #[cfg_attr(miri, ignore)] // error: unsupported operation: can't call foreign function `rust_psm_stack_pointer` on OS `linux`
    fn test_reduce() {
        let relation_type: Vec<ReprColumnType> = vec![
            ReprScalarType::Int64.nullable(true),
            ReprScalarType::Int64.nullable(true),
            ReprScalarType::Int64.nullable(false),
        ]
        .into_iter()
        .collect();
        let col = MirScalarExpr::column;
        let int64_typ = ReprScalarType::Int64;
        let err = |e| MirScalarExpr::literal(Err(e), int64_typ.clone());
        let lit = |i| MirScalarExpr::literal_ok(Datum::Int64(i), int64_typ.clone());
        let null = || MirScalarExpr::literal_null(int64_typ.clone());

        struct TestCase {
            input: MirScalarExpr,
            output: MirScalarExpr,
        }

        let test_cases = vec![
            TestCase {
                input: MirScalarExpr::call_variadic(Coalesce, vec![lit(1)]),
                output: lit(1),
            },
            TestCase {
                input: MirScalarExpr::call_variadic(Coalesce, vec![lit(1), lit(2)]),
                output: lit(1),
            },
            TestCase {
                input: MirScalarExpr::call_variadic(Coalesce, vec![null(), lit(2), null()]),
                output: lit(2),
            },
            TestCase {
                input: MirScalarExpr::call_variadic(
                    Coalesce,
                    vec![null(), col(0), null(), col(1), lit(2), lit(3)],
                ),
                output: MirScalarExpr::call_variadic(Coalesce, vec![col(0), col(1), lit(2)]),
            },
            TestCase {
                input: MirScalarExpr::call_variadic(Coalesce, vec![col(0), col(2), col(1)]),
                output: MirScalarExpr::call_variadic(Coalesce, vec![col(0), col(2)]),
            },
            TestCase {
                input: MirScalarExpr::call_variadic(
                    Coalesce,
                    vec![lit(1), err(EvalError::DivisionByZero)],
                ),
                output: lit(1),
            },
            TestCase {
                input: MirScalarExpr::call_variadic(
                    Coalesce,
                    vec![
                        null(),
                        err(EvalError::DivisionByZero),
                        err(EvalError::NumericFieldOverflow),
                    ],
                ),
                output: err(EvalError::DivisionByZero),
            },
        ];

        for tc in test_cases {
            let mut actual = tc.input.clone();
            actual.reduce(&relation_type);
            assert!(
                actual == tc.output,
                "input: {}\nactual: {}\nexpected: {}",
                tc.input,
                actual,
                tc.output
            );
        }
    }

    /// Exercises the `unsafe` pointer stack in [`Visit::visit_mut_post`] with a
    /// closure that *replaces subtrees* (`*expr = ...`). Miri's aliasing model
    /// should shout if the "stack mirrors the call stack" becomes untrue.
    #[mz_ore::test]
    fn test_visit_mut_post_replace_subtrees() {
        let col = MirScalarExpr::column;
        let mut expr = col(0).if_then_else(col(1).if_then_else(col(2), col(3)), col(4));

        expr.visit_mut_post(&mut |expr: &mut MirScalarExpr| match expr {
            MirScalarExpr::Column(n, _) => *n += 1,
            MirScalarExpr::If { then, .. } => {
                let then = then.take();
                *expr = then;
            }
            _ => {}
        });

        // collapses to then-most branch
        assert_eq!(expr, col(3));
    }

    /// Exercises the `unsafe` pointer stack in [`Visit::visit_mut_pre_post`] with
    /// a `pre` that both *replaces the visited node wholesale* (`*expr = ...`)
    /// and *returns an explicit child set* borrowed from the freshly written
    /// value. Miri's aliasing model should shout if the "stack mirrors the call
    /// stack" becomes untrue.
    #[mz_ore::test]
    fn test_visit_mut_pre_post_explicit_children() {
        let col = MirScalarExpr::column;
        let mut expr = col(5)
            .if_then_else(col(6), col(7))
            .if_then_else(col(1).if_then_else(col(2), col(3)), col(4));

        // turns conditions into column 0 in pre
        // doesn't traverse conditions of ifs
        // adds 10 to all column refs in post (but not in conditions!)
        expr.visit_mut_pre_post(
            &mut |expr: &mut MirScalarExpr| -> Option<Vec<&mut MirScalarExpr>> {
                if let MirScalarExpr::If { .. } = expr {
                    let MirScalarExpr::If { then, els, .. } = expr else {
                        unreachable!()
                    };
                    let then = then.take();
                    let els = els.take();
                    *expr = MirScalarExpr::column(0).if_then_else(then, els);

                    let MirScalarExpr::If { then, els, .. } = expr else {
                        unreachable!()
                    };
                    Some(vec![then.as_mut(), els.as_mut()])
                } else {
                    // Leaves recurse with their default (empty) child set.
                    None
                }
            },
            &mut |expr: &mut MirScalarExpr| {
                if let MirScalarExpr::Column(n, _) = expr {
                    *n += 10;
                }
            },
        );

        // conditions become 0; everyone else += 10
        let expected = col(0).if_then_else(col(0).if_then_else(col(12), col(13)), col(14));
        assert_eq!(expr, expected);
    }
}