mz_expr/

interpret.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::BTreeMap;
use std::fmt::Debug;

use mz_repr::{Datum, ReprColumnType, ReprRelationType, ReprScalarType, Row, RowArena};

use crate::scalar::func::variadic::And;
use crate::{
    BinaryFunc, Eval, EvalError, MapFilterProject, MfpPlan, MirScalarExpr, UnaryFunc,
    UnmaterializableFunc, VariadicFunc, func,
};
/// Whether a datum is a floating-point or numeric `NaN`.
///
/// `NaN` sorts as the maximum of the numeric/float [Datum] ordering, but it is
/// a fixed point of most functions we treat as monotone (e.g. negation maps
/// `NaN` to `NaN` while flipping the sign of every other value). A range whose
/// bounds include `NaN` therefore breaks the monotonicity assumption that
/// [ResultSpec::flat_map] relies on, so the range-narrowing shortcut must be
/// skipped in that case.
fn datum_is_nan(datum: Datum) -> bool {
    match datum {
        Datum::Float32(f) => f.is_nan(),
        Datum::Float64(f) => f.is_nan(),
        Datum::Numeric(n) => n.0.is_nan(),
        _ => false,
    }
}

/// Whether a datum is a floating-point or numeric infinity.
fn datum_is_infinite(datum: Datum) -> bool {
    match datum {
        Datum::Float32(f) => f.is_infinite(),
        Datum::Float64(f) => f.is_infinite(),
        Datum::Numeric(n) => n.0.is_infinite(),
        _ => false,
    }
}

/// An inclusive range of non-null datum values.
#[derive(Clone, Eq, PartialEq, Debug)]
enum Values<'a> {
    /// This range contains no values.
    Empty,
    /// An inclusive range. Invariant: the first element is always <= the second.
    // TODO: a variant for small sets of data would avoid losing precision here.
    Within(Datum<'a>, Datum<'a>),
    /// Constraints on structured fields, useful for recursive structures like maps.
    /// Fields that are not present in the map default to Values::All.
    // TODO: consider using this variant, or similar, for Datum::List.
    Nested(BTreeMap<Datum<'a>, ResultSpec<'a>>),
    /// This range might contain any value. Since we're overapproximating, this is often used
    /// as a "safe fallback" when we can't determine the right boundaries for a range.
    All,
}

impl<'a> Values<'a> {
    fn just(a: Datum<'a>) -> Values<'a> {
        match a {
            Datum::Map(datum_map) => Values::Nested(
                datum_map
                    .iter()
                    .map(|(key, val)| (key.into(), ResultSpec::value(val)))
                    .collect(),
            ),
            other => Self::Within(other, other),
        }
    }

    fn union(self, other: Values<'a>) -> Values<'a> {
        match (self, other) {
            (Values::Empty, r) => r,
            (r, Values::Empty) => r,
            (Values::Within(a0, a1), Values::Within(b0, b1)) => {
                Values::Within(a0.min(b0), a1.max(b1))
            }
            (Values::Nested(a), Values::Nested(mut b)) => {
                // `Nested(map)` treats keys missing from `map` as fully unconstrained, so a
                // key present in only one side of the union must be treated as `anything`
                // on the other side. Because `x ∪ anything = anything`, such keys drop
                // out of the merged map (the Nested default is already "anything").
                let mut merged = BTreeMap::new();
                for (key, a_spec) in a {
                    if let Some(b_spec) = b.remove(&key) {
                        let unioned = a_spec.union(b_spec);
                        if unioned != ResultSpec::anything() {
                            merged.insert(key, unioned);
                        }
                    }
                }
                if merged.is_empty() {
                    Values::All
                } else {
                    Values::Nested(merged)
                }
            }
            _ => Values::All,
        }
    }

    fn intersect(self, other: Values<'a>) -> Values<'a> {
        match (self, other) {
            (Values::Empty, _) => Values::Empty,
            (_, Values::Empty) => Values::Empty,
            (Values::Within(a0, a1), Values::Within(b0, b1)) => {
                let min = a0.max(b0);
                let max = a1.min(b1);
                if min <= max {
                    Values::Within(min, max)
                } else {
                    Values::Empty
                }
            }
            (Values::Nested(mut a), Values::Nested(b)) => {
                for (datum, other_spec) in b {
                    let spec = a.entry(datum).or_insert_with(ResultSpec::anything);
                    *spec = spec.clone().intersect(other_spec);
                }
                Values::Nested(a)
            }
            (Values::All, v) => v,
            (v, Values::All) => v,
            // A `Within` range and a `Nested` (map) spec can genuinely overlap:
            // `Datum` order places `Map` between `List` and `Numeric`, so a
            // range straddling those tags contains map values. We can't compute
            // the precise intersection, so return the (more structured) `Nested`
            // side as a sound over-approximation. Returning `Empty` here would
            // drop real values and let pushdown wrongly discard a part.
            (nested @ Values::Nested(_), Values::Within(_, _))
            | (Values::Within(_, _), nested @ Values::Nested(_)) => nested,
        }
    }

    fn may_contain(&self, value: Datum<'a>) -> bool {
        match self {
            Values::Empty => false,
            Values::Within(min, max) => *min <= value && value <= *max,
            Values::All => true,
            Values::Nested(field_map) => match value {
                Datum::Map(datum_map) => {
                    datum_map
                        .iter()
                        .all(|(key, val)| match field_map.get(&key.into()) {
                            None => true,
                            Some(nested) => nested.may_contain(val),
                        })
                }
                _ => false,
            },
        }
    }

    /// Returns the sole datum in this value set, if it is known to be a single
    /// value. Returns `None` otherwise (for empty sets, ranges with distinct
    /// endpoints, structured constraints, and the unconstrained set).
    ///
    /// Prefer this over pattern-matching on [Values::Within] directly when you
    /// only need the "single known value" case: it's robust against future
    /// variants of [Values] (e.g. a small-set representation) automatically
    /// degrading to "not a single value" rather than silently mis-matching.
    fn as_single(&self) -> Option<Datum<'a>> {
        match self {
            Values::Within(a, b) if a == b => Some(*a),
            _ => None,
        }
    }
}

/// An approximation of the set of values an expression might have, including whether or not it
/// might be null or an error value. This is generally an _overapproximation_, in the sense that
/// [ResultSpec::may_contain] may return true even if the argument is not included in the set.
/// (However, it should never return false when the value _is_ included!)
#[derive(Debug, Clone, Eq, PartialEq)]
pub struct ResultSpec<'a> {
    /// True if the expression may evaluate to [Datum::Null].
    nullable: bool,
    /// True if the expression may evaluate to an error.
    fallible: bool,
    /// The range of possible (non-null) values that the expression may evaluate to.
    values: Values<'a>,
}

impl<'a> ResultSpec<'a> {
    /// No results match this spec. (For example, an empty table.)
    pub fn nothing() -> Self {
        ResultSpec {
            nullable: false,
            fallible: false,
            values: Values::Empty,
        }
    }

    /// Every result matches this spec.
    pub fn anything() -> Self {
        ResultSpec {
            nullable: true,
            fallible: true,
            values: Values::All,
        }
    }

    /// Every result matches this spec.
    pub fn any_infallible() -> Self {
        ResultSpec {
            nullable: true,
            fallible: false,
            values: Values::All,
        }
    }

    /// A spec that only matches null.
    pub fn null() -> Self {
        ResultSpec {
            nullable: true,
            ..Self::nothing()
        }
    }

    /// A spec that only matches error values.
    pub fn fails() -> Self {
        ResultSpec {
            fallible: true,
            ..Self::nothing()
        }
    }

    /// A spec that matches all values of a given type.
    pub fn has_type(col: &ReprColumnType, fallible: bool) -> ResultSpec<'a> {
        let values = match &col.scalar_type {
            ReprScalarType::Bool => Values::Within(Datum::False, Datum::True),
            // TODO: add bounds for other bounded types, like integers
            _ => Values::All,
        };
        ResultSpec {
            nullable: col.nullable,
            fallible,
            values,
        }
    }

    /// A spec that only matches the given value.
    pub fn value(value: Datum<'a>) -> ResultSpec<'a> {
        match value {
            Datum::Null => Self::null(),
            nonnull => ResultSpec {
                values: Values::just(nonnull),
                ..Self::nothing()
            },
        }
    }

    /// A spec that matches values between the given (non-null) min and max.
    pub fn value_between(min: Datum<'a>, max: Datum<'a>) -> ResultSpec<'a> {
        assert!(!min.is_null());
        assert!(!max.is_null());
        if min <= max {
            ResultSpec {
                values: Values::Within(min, max),
                ..ResultSpec::nothing()
            }
        } else {
            ResultSpec::nothing()
        }
    }

    /// A spec that matches any non-null value.
    pub fn value_all() -> ResultSpec<'a> {
        ResultSpec {
            values: Values::All,
            ..ResultSpec::nothing()
        }
    }

    /// A spec that matches Datum::Maps of the given type.
    pub fn map_spec(map: BTreeMap<Datum<'a>, ResultSpec<'a>>) -> ResultSpec<'a> {
        ResultSpec {
            values: Values::Nested(map),
            ..ResultSpec::nothing()
        }
    }

    /// Given two specs, returns a new spec that matches anything that either original spec would match.
    pub fn union(self, other: ResultSpec<'a>) -> ResultSpec<'a> {
        ResultSpec {
            nullable: self.nullable || other.nullable,
            fallible: self.fallible || other.fallible,
            values: self.values.union(other.values),
        }
    }

    /// Given two specs, returns a new spec that only matches things that both original specs would match.
    pub fn intersect(self, other: ResultSpec<'a>) -> ResultSpec<'a> {
        ResultSpec {
            nullable: self.nullable && other.nullable,
            fallible: self.fallible && other.fallible,
            values: self.values.intersect(other.values),
        }
    }

    /// Check if a particular value matches the spec.
    pub fn may_contain(&self, value: Datum<'a>) -> bool {
        if value == Datum::Null {
            return self.nullable;
        }

        self.values.may_contain(value)
    }

    /// Check if an error value matches the spec.
    pub fn may_fail(&self) -> bool {
        self.fallible
    }

    /// Whether this spec is pinned to a single concrete (non-null) value.
    ///
    /// When it is, evaluating a function on the input is exact rather than an
    /// endpoint-sampled approximation of a range, so the interpreter can trust
    /// the sampled fallibility. When it isn't, endpoint sampling cannot prove a
    /// `could_error` function infallible over the range (see the fallibility
    /// handling in [`ColumnSpecs::unary`] and friends).
    fn is_single_value(&self) -> bool {
        self.values.as_single().is_some()
    }

    /// Whether the value range might include a floating-point or numeric
    /// infinity. Infinities sort at the extremes of the order, so a `Within`
    /// range includes one only as an endpoint.
    fn may_be_infinite(&self) -> bool {
        match &self.values {
            Values::Within(min, max) => datum_is_infinite(*min) || datum_is_infinite(*max),
            Values::All => true,
            Values::Empty | Values::Nested(_) => false,
        }
    }

    /// This method "maps" a function across the `ResultSpec`.
    ///
    /// As mentioned above, `ResultSpec` represents an approximate set of results.
    /// If we actually stored each result in the set, `flat_map` could be implemented by passing
    /// each result to the function one-by-one and unioning the resulting sets. This is possible
    /// when our values set is empty or contains a single datum, but when it contains a range,
    /// we can't enumerate all possible values of the set. We handle this by:
    /// - tracking whether the function is monotone, in which case we can map the range by just
    ///   mapping the endpoints;
    /// - using a safe default when we can't infer a tighter bound on the set, eg. [Self::anything].
    fn flat_map(
        &self,
        is_monotone: bool,
        mut result_map: impl FnMut(Result<Datum<'a>, EvalError>) -> ResultSpec<'a>,
    ) -> ResultSpec<'a> {
        let null_spec = if self.nullable {
            result_map(Ok(Datum::Null))
        } else {
            ResultSpec::nothing()
        };

        let error_spec = if self.fallible {
            // Since we only care about whether / not an error is possible, and not the specific
            // error, create an arbitrary error here.
            // NOTE! This assumes that functions do not discriminate on the type of the error.
            let map_err = result_map(Err(EvalError::Internal("".into())));
            let raise_err = ResultSpec::fails();
            // SQL has a very loose notion of evaluation order: https://www.postgresql.org/docs/current/sql-expressions.html#SYNTAX-EXPRESS-EVAL
            // Here, we account for the possibility that the expression is evaluated strictly,
            // raising the error, or that it's evaluated lazily by the result_map function
            // (which may return a non-error result even when given an error as input).
            raise_err.union(map_err)
        } else {
            ResultSpec::nothing()
        };

        let values_spec = match self.values {
            Values::Empty => ResultSpec::nothing(),
            // If this range contains a single datum, just call the function.
            Values::Within(min, max) if min == max => result_map(Ok(min)),
            // If this is a range of booleans, we know all the values... just try them.
            Values::Within(Datum::False, Datum::True) => {
                result_map(Ok(Datum::False)).union(result_map(Ok(Datum::True)))
            }
            // Otherwise, if our function is monotonic, we can try mapping the input
            // range to an output range. A range whose bounds include `NaN` is
            // excluded: `NaN` is ordered as the maximum but is a fixed point of
            // most monotone functions, so evaluating the endpoints does not
            // bound the interior. Such ranges fall through to the
            // overapproximation below.
            Values::Within(min, max) if is_monotone && !datum_is_nan(min) && !datum_is_nan(max) => {
                let min_result = result_map(Ok(min));
                let max_result = result_map(Ok(max));
                // Value, null, and error are orthogonal channels. Monotonicity
                // lets us bound the *values* by the endpoints, but only when both
                // endpoints actually produced a value; null and error can't be
                // bounded from value endpoints, so we just union whatever the
                // endpoints reported on those channels. (A function that errors
                // on an interior value while the endpoints don't is handled by
                // the fallibility guard in `unary`/`binary`/`variadic`, not
                // here.)
                match (min_result, max_result) {
                    // Both endpoints produced a value: bound the interior values
                    // by their union, and carry any null/error from the endpoints.
                    (
                        ResultSpec {
                            nullable: n1,
                            fallible: f1,
                            values: a_values @ Values::Within(..),
                        },
                        ResultSpec {
                            nullable: n2,
                            fallible: f2,
                            values: b_values @ Values::Within(..),
                        },
                    ) => ResultSpec {
                        nullable: n1 || n2,
                        fallible: f1 || f2,
                        values: a_values.union(b_values),
                    },
                    // If both endpoints map purely to null, assume the whole
                    // range maps to null. (Both endpoints *erroring* is NOT
                    // enough: an interior input can still produce a value, e.g.
                    // a cast that rejects both bounds but accepts a value
                    // between them.)
                    (
                        ResultSpec {
                            nullable: true,
                            fallible: false,
                            values: Values::Empty,
                        },
                        ResultSpec {
                            nullable: true,
                            fallible: false,
                            values: Values::Empty,
                        },
                    ) => ResultSpec::null(),
                    // Otherwise we can't bound the interior values.
                    _ => ResultSpec::anything(),
                }
            }
            // TODO: we could return a narrower result for eg. `Values::Nested` with all-`Within` fields.
            Values::Within(_, _) | Values::Nested(_) | Values::All => ResultSpec::anything(),
        };

        null_spec.union(error_spec).union(values_spec)
    }
}

/// [Abstract interpretation](https://en.wikipedia.org/wiki/Abstract_interpretation) for
/// [MirScalarExpr].
///
/// [MirScalarExpr::eval] implements a "concrete interpreter" for the expression type: given an
/// expression and specific column values as input, it returns a specific value for the output.
/// This could be reimplemented using this trait... but it's most useful for "abstract"
/// interpretations of the expression, where we generalize about sets of possible inputs and outputs.
/// See [Trace] and [ColumnSpecs] for how this can be useful in practice.
pub trait Interpreter {
    type Summary: Clone + Debug + Sized;

    /// A column of the input row.
    fn column(&self, id: usize) -> Self::Summary;

    /// A literal value.
    /// (Stored as a row, because we can't own a Datum.)
    fn literal(&self, result: &Result<Row, EvalError>, col_type: &ReprColumnType) -> Self::Summary;
    /// A call to an unmaterializable function.
    ///
    /// These functions cannot be evaluated by `MirScalarExpr::eval`. They must
    /// be transformed away by a higher layer.
    fn unmaterializable(&self, func: &UnmaterializableFunc) -> Self::Summary;

    /// A function call that takes one expression as an argument.
    fn unary(&self, func: &UnaryFunc, expr: Self::Summary) -> Self::Summary;

    /// A function call that takes two expressions as arguments.
    fn binary(&self, func: &BinaryFunc, left: Self::Summary, right: Self::Summary)
    -> Self::Summary;

    /// A function call that takes an arbitrary number of arguments.
    fn variadic(&self, func: &VariadicFunc, exprs: Vec<Self::Summary>) -> Self::Summary;

    /// Conditionally evaluated expressions.
    fn cond(&self, cond: Self::Summary, then: Self::Summary, els: Self::Summary) -> Self::Summary;

    /// Evaluate an entire expression, by delegating to the fine-grained methods on [Interpreter].
    fn expr(&self, expr: &MirScalarExpr) -> Self::Summary {
        match expr {
            MirScalarExpr::Column(id, _name) => self.column(*id),
            MirScalarExpr::Literal(value, col_type) => self.literal(value, col_type),
            MirScalarExpr::CallUnmaterializable(func) => self.unmaterializable(func),
            MirScalarExpr::CallUnary { func, expr } => {
                let expr_range = self.expr(expr);
                self.unary(func, expr_range)
            }
            MirScalarExpr::CallBinary { func, expr1, expr2 } => {
                let expr1_range = self.expr(expr1);
                let expr2_range = self.expr(expr2);
                self.binary(func, expr1_range, expr2_range)
            }
            MirScalarExpr::CallVariadic { func, exprs } => {
                let exprs: Vec<_> = exprs.into_iter().map(|e| self.expr(e)).collect();
                self.variadic(func, exprs)
            }
            MirScalarExpr::If { cond, then, els } => {
                let cond_range = self.expr(cond);
                let then_range = self.expr(then);
                let els_range = self.expr(els);
                self.cond(cond_range, then_range, els_range)
            }
        }
    }

    /// Specifically, this evaluates the map and filters stages of an MFP: summarize each of the
    /// map expressions, then `and` together all of the filters.
    fn mfp_filter(&self, mfp: &MapFilterProject) -> Self::Summary {
        let mfp_eval = MfpEval::new(self, mfp.input_arity, &mfp.expressions);
        // NB: self should not be used after this point!
        let predicates = mfp
            .predicates
            .iter()
            .map(|(_, e)| mfp_eval.expr(e))
            .collect();
        mfp_eval.variadic(&And.into(), predicates)
    }

    /// Similar to [Self::mfp_filter], but includes the additional temporal filters that have been
    /// broken out.
    fn mfp_plan_filter(&self, plan: &MfpPlan) -> Self::Summary {
        let mfp_eval = MfpEval::new(self, plan.mfp.input_arity, &plan.mfp.expressions);
        // NB: self should not be used after this point!
        let mut results: Vec<_> = plan
            .mfp
            .predicates
            .iter()
            .map(|(_, e)| mfp_eval.expr(e))
            .collect();
        let mz_now = mfp_eval.unmaterializable(&UnmaterializableFunc::MzNow);
        for bound in &plan.lower_bounds {
            let bound_range = mfp_eval.expr(bound);
            let result = mfp_eval.binary(&BinaryFunc::Lte(func::Lte), bound_range, mz_now.clone());
            results.push(result);
        }
        for bound in &plan.upper_bounds {
            let bound_range = mfp_eval.expr(bound);
            let result = mfp_eval.binary(&BinaryFunc::Gte(func::Gte), bound_range, mz_now.clone());
            results.push(result);
        }
        self.variadic(&And.into(), results)
    }
}

/// Wrap another interpreter, but tack a few extra columns on at the end. An internal implementation
/// detail of `eval_mfp` and `eval_mfp_plan`.
pub(crate) struct MfpEval<'a, E: Interpreter + ?Sized> {
    evaluator: &'a E,
    input_arity: usize,
    expressions: Vec<E::Summary>,
}

impl<'a, E: Interpreter + ?Sized> MfpEval<'a, E> {
    pub(crate) fn new(evaluator: &'a E, input_arity: usize, expressions: &[MirScalarExpr]) -> Self {
        let mut mfp_eval = MfpEval {
            evaluator,
            input_arity,
            expressions: vec![],
        };
        for expr in expressions {
            let result = mfp_eval.expr(expr);
            mfp_eval.expressions.push(result);
        }
        mfp_eval
    }
}

impl<'a, E: Interpreter + ?Sized> Interpreter for MfpEval<'a, E> {
    type Summary = E::Summary;

    fn column(&self, id: usize) -> Self::Summary {
        if id < self.input_arity {
            self.evaluator.column(id)
        } else {
            self.expressions[id - self.input_arity].clone()
        }
    }

    fn literal(&self, result: &Result<Row, EvalError>, col_type: &ReprColumnType) -> Self::Summary {
        self.evaluator.literal(result, col_type)
    }

    fn unmaterializable(&self, func: &UnmaterializableFunc) -> Self::Summary {
        self.evaluator.unmaterializable(func)
    }

    fn unary(&self, func: &UnaryFunc, expr: Self::Summary) -> Self::Summary {
        self.evaluator.unary(func, expr)
    }

    fn binary(
        &self,
        func: &BinaryFunc,
        left: Self::Summary,
        right: Self::Summary,
    ) -> Self::Summary {
        self.evaluator.binary(func, left, right)
    }

    fn variadic(&self, func: &VariadicFunc, exprs: Vec<Self::Summary>) -> Self::Summary {
        self.evaluator.variadic(func, exprs)
    }

    fn cond(&self, cond: Self::Summary, then: Self::Summary, els: Self::Summary) -> Self::Summary {
        self.evaluator.cond(cond, then, els)
    }
}

/// A unary function we've added special-case handling for; including:
/// - A three-argument function, taking and returning [ResultSpec]s. This
///   overrides the default function-handling logic entirely.
/// - Metadata on whether / not this function is pushdownable. See [Trace].
struct SpecialUnary {
    map_fn: for<'a, 'b> fn(&'b ColumnSpecs<'a>, ResultSpec<'a>) -> ResultSpec<'a>,
    pushdownable: bool,
}

impl SpecialUnary {
    /// Returns the special-case handling for a particular function, if it exists.
    fn for_func(func: &UnaryFunc) -> Option<SpecialUnary> {
        /// Eager in the same sense as `func.rs` uses the term; this assumes that
        /// nulls and errors propagate up, and we only need to define the behaviour
        /// on values.
        fn eagerly<'b>(
            spec: ResultSpec<'b>,
            value_fn: impl FnOnce(Values<'b>) -> ResultSpec<'b>,
        ) -> ResultSpec<'b> {
            let result = match spec.values {
                Values::Empty => ResultSpec::nothing(),
                other => value_fn(other),
            };
            ResultSpec {
                fallible: spec.fallible || result.fallible,
                nullable: spec.nullable || result.nullable,
                values: result.values,
            }
        }
        match func {
            UnaryFunc::TryParseMonotonicIso8601Timestamp(_) => Some(SpecialUnary {
                map_fn: |specs, range| {
                    let expr = MirScalarExpr::CallUnary {
                        func: UnaryFunc::TryParseMonotonicIso8601Timestamp(
                            crate::func::TryParseMonotonicIso8601Timestamp,
                        ),
                        expr: Box::new(MirScalarExpr::column(0)),
                    };
                    let eval = |d| specs.eval_result(expr.eval(&[d], specs.arena));

                    eagerly(range, |values| {
                        match values {
                            Values::Within(a, b) if a == b => eval(a),
                            Values::Within(a, b) => {
                                let spec = eval(a).union(eval(b));
                                let values_spec = if spec.nullable {
                                    // At least one of the endpoints of the range wasn't a valid
                                    // timestamp. We can't compute a precise range in this case.
                                    // If we used the general is_monotone handling, that code would
                                    // incorrectly assume the whole range mapped to null if each
                                    // endpoint did.
                                    ResultSpec::value_all()
                                } else {
                                    spec
                                };
                                // A range of strings will always contain strings that don't parse
                                // as timestamps - so unlike the general case, we'll assume null
                                // is present in every range of output values.
                                values_spec.union(ResultSpec::null())
                            }
                            // Otherwise, assume the worst: this function may return either a valid
                            // value or null.
                            _ => ResultSpec::any_infallible(),
                        }
                    })
                },
                pushdownable: true,
            }),
            _ => None,
        }
    }
}

/// The abstract-domain counterpart of a [BinaryFunc]: a binary function
/// we've added special-case handling for; including:
/// - Either a complete override of [ResultSpec] computation, or a way to
///   compute monotonicity dynamically from the input specs.
/// - Metadata on whether / not this function is pushdownable. See [Trace].
///
/// Note: today a function can have *either* a handler override *or* a
/// dynamic-monotonicity verdict, but not both. If a future function wants
/// both, promote [AbstractFuncHandler] from an enum to a struct with two
/// optional fields.
struct AbstractFunc {
    handler: AbstractFuncHandler,
    /// `(left, right)`: per-argument pushdownability hint consumed by
    /// [Trace]. `true` for an argument means the function preserves enough
    /// structure that, with sufficient information about that argument's
    /// range, the output spec can be predicted — i.e. the predicate is a
    /// pushdown candidate when that argument is constant or a tight range.
    pushdownable: (bool, bool),
}

/// How an [AbstractFunc] computes the output [ResultSpec].
enum AbstractFuncHandler {
    /// Completely override the spec computation; the default flat-map machinery
    /// is bypassed.
    Override(for<'a> fn(ResultSpec<'a>, ResultSpec<'a>) -> ResultSpec<'a>),
    /// Use the default flat-map machinery, but with a monotonicity verdict that
    /// depends on the input specs. This lets us claim monotonicity for cases
    /// the static `LazyBinaryFunc::is_monotone` annotation can't safely claim:
    /// for instance, `t + INTERVAL '1' day` is monotone in `t`, but `t + i`
    /// generally isn't (the calendar-month / day-clamping arithmetic in
    /// `add_timestamp_interval` is non-monotone when `i.months != 0`).
    DynamicMonotone(fn(&ResultSpec<'_>, &ResultSpec<'_>) -> (bool, bool)),
}

impl AbstractFunc {
    /// Returns the special-case handling for a particular function, if it exists.
    fn for_func(func: &BinaryFunc) -> Option<AbstractFunc> {
        /// Eager in the same sense as `func.rs` uses the term; this assumes that
        /// nulls and errors propagate up, and we only need to define the behaviour
        /// on values.
        fn eagerly<'b>(
            left: ResultSpec<'b>,
            right: ResultSpec<'b>,
            value_fn: impl FnOnce(Values<'b>, Values<'b>) -> ResultSpec<'b>,
        ) -> ResultSpec<'b> {
            let result = match (left.values, right.values) {
                (Values::Empty, _) | (_, Values::Empty) => ResultSpec::nothing(),
                (l, r) => value_fn(l, r),
            };
            ResultSpec {
                fallible: left.fallible || right.fallible || result.fallible,
                nullable: left.nullable || right.nullable || result.nullable,
                values: result.values,
            }
        }

        fn jsonb_get_string<'b>(
            left: ResultSpec<'b>,
            right: ResultSpec<'b>,
            stringify: bool,
        ) -> ResultSpec<'b> {
            eagerly(left, right, |left, right| {
                let nested_spec = match (left, right) {
                    (Values::Nested(mut map_spec), Values::Within(key, key2)) if key == key2 => {
                        map_spec.remove(&key)
                    }
                    _ => None,
                };

                if let Some(field_spec) = nested_spec {
                    if stringify {
                        // We only preserve value-range information when stringification
                        // is a noop. (Common in real queries.)
                        let values = match field_spec.values {
                            Values::Empty => Values::Empty,
                            Values::Within(min @ Datum::String(_), max @ Datum::String(_)) => {
                                Values::Within(min, max)
                            }
                            Values::Within(_, _) | Values::Nested(_) | Values::All => Values::All,
                        };
                        ResultSpec {
                            values,
                            ..field_spec
                        }
                    } else {
                        field_spec
                    }
                } else {
                    // The implementation of `jsonb_get_string` always returns
                    // `Ok(...)`. Morally, everything has a string
                    // representation, and the worst you can get is a NULL,
                    // which maps to a NULL.
                    ResultSpec::any_infallible()
                }
            })
        }

        fn eq<'b>(left: ResultSpec<'b>, right: ResultSpec<'b>) -> ResultSpec<'b> {
            eagerly(left, right, |left, right| {
                // `eq` might return true if there's any overlap between the range of its two arguments...
                let maybe_true = match left.clone().intersect(right.clone()) {
                    Values::Empty => ResultSpec::nothing(),
                    _ => ResultSpec::value(Datum::True),
                };

                // ...and may return false if the union contains at least two distinct values.
                // Note that the `Empty` case is handled by `eagerly` above.
                let maybe_false = match left.union(right) {
                    Values::Within(a, b) if a == b => ResultSpec::nothing(),
                    _ => ResultSpec::value(Datum::False),
                };

                maybe_true.union(maybe_false)
            })
        }

        /// `add_timestamp_interval` and friends do calendar-month arithmetic
        /// with day-clamping, which is non-monotone in either argument when
        /// `interval.months != 0`. But when `interval.months == 0` the
        /// operation reduces to adding a fixed number of microseconds, which
        /// *is* monotone in both arguments. The static `is_monotone`
        /// annotation has to pick the conservative answer; this dynamic check
        /// recovers filter pushdown for the common case of literal
        /// `INTERVAL '<N>' day`-style predicates.
        fn timestamp_plus_interval_monotone(
            _left: &ResultSpec<'_>,
            right: &ResultSpec<'_>,
        ) -> (bool, bool) {
            let months_zero = matches!(
                right.values.as_single(),
                Some(Datum::Interval(i)) if i.months == 0,
            );
            (months_zero, months_zero)
        }

        match func {
            BinaryFunc::JsonbGetString(_) => Some(AbstractFunc {
                handler: AbstractFuncHandler::Override(|l, r| jsonb_get_string(l, r, false)),
                pushdownable: (true, false),
            }),
            BinaryFunc::JsonbGetStringStringify(_) => Some(AbstractFunc {
                handler: AbstractFuncHandler::Override(|l, r| jsonb_get_string(l, r, true)),
                pushdownable: (true, false),
            }),
            BinaryFunc::Eq(_) => Some(AbstractFunc {
                handler: AbstractFuncHandler::Override(eq),
                pushdownable: (true, true),
            }),
            BinaryFunc::AddTimestampInterval(_)
            | BinaryFunc::AddTimestampTzInterval(_)
            | BinaryFunc::SubTimestampInterval(_)
            | BinaryFunc::SubTimestampTzInterval(_) => Some(AbstractFunc {
                handler: AbstractFuncHandler::DynamicMonotone(timestamp_plus_interval_monotone),
                // For [Trace]: we *might* be pushdownable in the first argument
                // (we are when the interval is a literal with no months). The
                // interval argument is reported as non-pushdownable so that
                // `t_col +/- col_interval` doesn't get routed through pushdown
                // for no benefit; if both sides are constants the predicate
                // collapses anyway.
                pushdownable: (true, false),
            }),
            _ => None,
        }
    }
}

#[derive(Clone, Debug)]
pub struct ColumnSpec<'a> {
    pub col_type: ReprColumnType,
    pub range: ResultSpec<'a>,
}

/// An interpreter that:
/// - stores both the type and the range of possible values for every column and
///   unmaterializable function. (See the `push_` methods.)
/// - given an expression (or MFP, etc.), returns the range of possible results that evaluating that
///   expression might have. (See the `eval_` methods.)
#[derive(Clone, Debug)]
pub struct ColumnSpecs<'a> {
    pub relation: &'a ReprRelationType,
    pub columns: Vec<ResultSpec<'a>>,
    pub unmaterializables: BTreeMap<UnmaterializableFunc, ResultSpec<'a>>,
    pub arena: &'a RowArena,
}

impl<'a> ColumnSpecs<'a> {
    // Interpreting a variadic function can lead to exponential blowup: there are up to 4 possibly-
    // interesting values for each argument (error, null, range bounds...) and in the worst case
    // we may need to test every combination. We mitigate that here in two ways:
    // - Adding a linear-time optimization for associative functions like AND, OR, COALESCE.
    // - Limiting the number of arguments we'll pass on to eval. If this limit is crossed, we'll
    //   return our default / safe overapproximation instead.
    const MAX_EVAL_ARGS: usize = 6;

    /// Create a new, empty set of column specs. (Initially, the only assumption we make about the
    /// data in the column is that it matches the type.)
    pub fn new(relation: &'a ReprRelationType, arena: &'a RowArena) -> Self {
        let columns = relation
            .column_types
            .iter()
            .map(|ct| ResultSpec::has_type(ct, false))
            .collect();
        ColumnSpecs {
            relation,
            columns,
            unmaterializables: Default::default(),
            arena,
        }
    }

    /// Restrict the set of possible values in a given column. (By intersecting it with the existing
    /// spec.)
    pub fn push_column(&mut self, id: usize, update: ResultSpec<'a>) {
        let range = self.columns.get_mut(id).expect("valid column id");
        *range = range.clone().intersect(update);
    }

    /// Restrict the set of possible values a given unmaterializable func might return. (By
    /// intersecting it with the existing spec.)
    pub fn push_unmaterializable(&mut self, func: UnmaterializableFunc, update: ResultSpec<'a>) {
        let range = self
            .unmaterializables
            .entry(func.clone())
            .or_insert_with(|| ResultSpec::has_type(&func.output_type(), true));
        *range = range.clone().intersect(update);
    }

    fn eval_result<'b, E>(&self, result: Result<Datum<'b>, E>) -> ResultSpec<'a> {
        match result {
            Ok(Datum::Null) => ResultSpec {
                nullable: true,
                ..ResultSpec::nothing()
            },
            Ok(d) => ResultSpec {
                values: Values::just(self.arena.make_datum(|packer| packer.push(d))),
                ..ResultSpec::nothing()
            },
            Err(_) => ResultSpec {
                fallible: true,
                ..ResultSpec::nothing()
            },
        }
    }

    fn set_literal(expr: &mut MirScalarExpr, update: Result<Datum, EvalError>) {
        match expr {
            MirScalarExpr::Literal(literal, col_type) => match update {
                Err(error) => *literal = Err(error),
                Ok(datum) => {
                    assert!(
                        datum.is_instance_of(col_type),
                        "{datum:?} must be an instance of {col_type:?}"
                    );
                    match literal {
                        // Reuse the allocation if we can
                        Ok(row) => row.packer().push(datum),
                        literal => *literal = Ok(Row::pack_slice(&[datum])),
                    }
                }
            },
            _ => panic!("not a literal"),
        }
    }

    fn set_argument(expr: &mut MirScalarExpr, arg: usize, value: Result<Datum, EvalError>) {
        match (expr, arg) {
            (MirScalarExpr::CallUnary { expr, .. }, 0) => Self::set_literal(expr, value),
            (MirScalarExpr::CallBinary { expr1, .. }, 0) => Self::set_literal(expr1, value),
            (MirScalarExpr::CallBinary { expr2, .. }, 1) => Self::set_literal(expr2, value),
            (MirScalarExpr::CallVariadic { exprs, .. }, n) if n < exprs.len() => {
                Self::set_literal(&mut exprs[n], value)
            }
            _ => panic!("illegal argument for expression"),
        }
    }

    /// A literal with the given type and a trivial default value. Callers should ensure that
    /// [Self::set_literal] is called on the resulting expression to give it a meaningful value
    /// before evaluating.
    fn placeholder(col_type: ReprColumnType) -> MirScalarExpr {
        MirScalarExpr::Literal(Err(EvalError::Internal("".into())), col_type)
    }
}

impl<'a> Interpreter for ColumnSpecs<'a> {
    type Summary = ColumnSpec<'a>;

    fn column(&self, id: usize) -> Self::Summary {
        let col_type = self.relation.column_types[id].clone();
        let range = self.columns[id].clone();
        ColumnSpec { col_type, range }
    }

    fn literal(&self, result: &Result<Row, EvalError>, col_type: &ReprColumnType) -> Self::Summary {
        let col_type = col_type.clone();
        let range = self.eval_result(result.as_ref().map(|row| {
            self.arena
                .make_datum(|packer| packer.push(row.unpack_first()))
        }));
        ColumnSpec { col_type, range }
    }

    fn unmaterializable(&self, func: &UnmaterializableFunc) -> Self::Summary {
        let col_type = func.output_type();
        let range = self
            .unmaterializables
            .get(func)
            .cloned()
            .unwrap_or_else(|| ResultSpec::has_type(&func.output_type(), true));
        ColumnSpec { col_type, range }
    }

    fn unary(&self, func: &UnaryFunc, summary: Self::Summary) -> Self::Summary {
        let fallible = func.could_error() || summary.range.fallible;
        // Endpoint sampling proves a monotone function's output value range, but
        // it cannot prove the function never errors on an interior value of a
        // multi-valued range: monotonicity says nothing about where errors
        // occur (e.g. `round(_, scale)` overflows on inputs whose exponent hits
        // a bad branch, `numeric::mz_timestamp` rejects fractional inputs). So
        // we do not let it conclude a could-error function is infallible over
        // such a range.
        let input_multivalued = !summary.range.is_single_value();
        let mapped_spec = if let Some(special) = SpecialUnary::for_func(func) {
            (special.map_fn)(self, summary.range)
        } else {
            let is_monotone = func.is_monotone();
            let mut expr = MirScalarExpr::CallUnary {
                func: func.clone(),
                expr: Box::new(Self::placeholder(summary.col_type.clone())),
            };
            summary.range.flat_map(is_monotone, |datum| {
                Self::set_argument(&mut expr, 0, datum);
                self.eval_result(expr.eval(&[], self.arena))
            })
        };

        let col_type = func.output_type(summary.col_type);

        let mut range = mapped_spec.intersect(ResultSpec::has_type(&col_type, fallible));
        // `intersect` only ANDs the fallible flag, so `has_type` above narrows
        // the value and null domain but cannot surface an error that endpoint
        // sampling missed. Force it for a could-error function over a range.
        if fallible && input_multivalued {
            range.fallible = true;
        }
        ColumnSpec { col_type, range }
    }

    fn binary(
        &self,
        func: &BinaryFunc,
        left: Self::Summary,
        right: Self::Summary,
    ) -> Self::Summary {
        let fallible = func.could_error() || left.range.fallible || right.range.fallible;
        // See the note in `unary`: endpoint sampling cannot rule out an interior
        // error, so a could-error function is fallible over a multi-valued range.
        let inputs_multivalued = !left.range.is_single_value() || !right.range.is_single_value();
        let operand_may_be_infinite = left.range.may_be_infinite() || right.range.may_be_infinite();

        let special = AbstractFunc::for_func(func);
        let (left_monotonic, right_monotonic) = match &special {
            Some(AbstractFunc {
                handler: AbstractFuncHandler::DynamicMonotone(monotone_fn),
                ..
            }) => monotone_fn(&left.range, &right.range),
            _ => func.is_monotone(),
        };

        let mapped_spec = match special {
            Some(AbstractFunc {
                handler: AbstractFuncHandler::Override(f),
                ..
            }) => f(left.range, right.range),
            _ => {
                let mut expr = MirScalarExpr::CallBinary {
                    func: func.clone(),
                    expr1: Box::new(Self::placeholder(left.col_type.clone())),
                    expr2: Box::new(Self::placeholder(right.col_type.clone())),
                };
                left.range.flat_map(left_monotonic, |left_result| {
                    Self::set_argument(&mut expr, 0, left_result);
                    right.range.flat_map(right_monotonic, |right_result| {
                        Self::set_argument(&mut expr, 1, right_result);
                        self.eval_result(expr.eval(&[], self.arena))
                    })
                })
            }
        };

        let col_type = func.output_type(&[left.col_type, right.col_type]);

        let mut range = mapped_spec.intersect(ResultSpec::has_type(&col_type, fallible));
        // `intersect` only ANDs the fallible flag, so force the interior
        // fallibility it cannot add (see the note in `unary`).
        if fallible && inputs_multivalued {
            range.fallible = true;
        }
        // Some functions (multiplication, division) are not corner-sampleable
        // when an operand may be infinite: their indeterminate forms (`∞ * 0`,
        // `∞ / ∞`) evaluate to a value the endpoints don't bound, and that value
        // can be reached only from the interior (e.g. `[-∞, +∞] * 0` maps both
        // endpoints to `NaN` while its finite interior maps to `0`, and
        // `finite / ∞ = 0` is stepped over when both endpoints are `∞ / ∞ =
        // NaN`). Fall back to the full value domain for them.
        if operand_may_be_infinite && !func.is_infinity_monotone() {
            range.values = Values::All;
        }
        ColumnSpec { col_type, range }
    }

    fn variadic(&self, func: &VariadicFunc, args: Vec<Self::Summary>) -> Self::Summary {
        let fallible = func.could_error() || args.iter().any(|s| s.range.fallible);
        let inputs_multivalued = args.iter().any(|s| !s.range.is_single_value());
        if func.is_associative() && args.len() > 2 {
            // To avoid a combinatorial explosion, evaluate large variadic calls as a series of
            // smaller ones, since associativity guarantees we'll get compatible results.
            return args
                .into_iter()
                .reduce(|a, b| self.variadic(func, vec![a, b]))
                .expect("reducing over a non-empty argument list");
        }

        let mapped_spec = if args.len() >= Self::MAX_EVAL_ARGS {
            ResultSpec::anything()
        } else {
            fn eval_loop<'a>(
                is_monotonic: bool,
                expr: &mut MirScalarExpr,
                args: &[ColumnSpec<'a>],
                index: usize,
                datum_map: &mut impl FnMut(&MirScalarExpr) -> ResultSpec<'a>,
            ) -> ResultSpec<'a> {
                if index >= args.len() {
                    datum_map(expr)
                } else {
                    args[index].range.flat_map(is_monotonic, |datum| {
                        ColumnSpecs::set_argument(expr, index, datum);
                        eval_loop(is_monotonic, expr, args, index + 1, datum_map)
                    })
                }
            }

            let mut fn_expr = MirScalarExpr::CallVariadic {
                func: func.clone(),
                exprs: args
                    .iter()
                    .map(|spec| Self::placeholder(spec.col_type.clone()))
                    .collect(),
            };
            eval_loop(func.is_monotone(), &mut fn_expr, &args, 0, &mut |expr| {
                self.eval_result(expr.eval(&[], self.arena))
            })
        };

        let col_types = args.into_iter().map(|spec| spec.col_type).collect();
        let col_type = func.output_type(col_types);

        let mut range = mapped_spec.intersect(ResultSpec::has_type(&col_type, fallible));
        // `intersect` only ANDs the fallible flag, so force the interior
        // fallibility it cannot add (see the note in `unary`).
        if fallible && inputs_multivalued {
            range.fallible = true;
        }

        ColumnSpec { col_type, range }
    }

    fn cond(&self, cond: Self::Summary, then: Self::Summary, els: Self::Summary) -> Self::Summary {
        let col_type = then
            .col_type
            .union(&els.col_type)
            .expect("failed type union for cond during abstract interpretation");

        let range = cond
            .range
            .flat_map(true, |datum| match datum {
                Ok(Datum::True) => then.range.clone(),
                // A false OR null condition takes the `els` branch, matching
                // `MirScalarExpr::eval` (`Datum::False | Datum::Null => els`).
                // Mapping null to `fails()` here would drop `els` from the value
                // channel and let pushdown wrongly rule out the else result.
                Ok(Datum::False) | Ok(Datum::Null) => els.range.clone(),
                _ => ResultSpec::fails(),
            })
            .intersect(ResultSpec::has_type(&col_type, true));

        ColumnSpec { col_type, range }
    }

    /// Override the default implementations of [Self::mfp_filter] and
    /// [Self::mfp_plan_filter] so that the fallibility of MFP expressions
    /// surfaces in the result, even when the expression's result column isn't
    /// referenced by a predicate or temporal bound.
    ///
    /// The runtime MFP evaluator runs every expression once all the preceding
    /// predicates pass (see [`crate::SafeMfpPlan::evaluate_inner`]), so an
    /// expression that errors on the actual data will turn the whole row into
    /// an `Err` — even if no predicate or bound mentions that expression. The
    /// default `mfp_filter` / `mfp_plan_filter` only AND together the
    /// predicates and bounds, so the AND result misses the expression's
    /// `fallible` flag and persist filter pushdown can wrongly discard a part
    /// that actually produces error rows. See database-issues#9656.
    fn mfp_filter(&self, mfp: &MapFilterProject) -> Self::Summary {
        let mfp_eval = MfpEval::new(self, mfp.input_arity, &mfp.expressions);
        let predicates = mfp
            .predicates
            .iter()
            .map(|(_, e)| mfp_eval.expr(e))
            .collect();
        let mut result = self.variadic(&And.into(), predicates);
        if mfp_eval.expressions.iter().any(|s| s.range.fallible) {
            result.range.fallible = true;
        }
        result
    }

    fn mfp_plan_filter(&self, plan: &MfpPlan) -> Self::Summary {
        let mfp_eval = MfpEval::new(self, plan.mfp.input_arity, &plan.mfp.expressions);
        let mut results: Vec<_> = plan
            .mfp
            .predicates
            .iter()
            .map(|(_, e)| mfp_eval.expr(e))
            .collect();
        let mz_now = mfp_eval.unmaterializable(&UnmaterializableFunc::MzNow);
        for bound in &plan.lower_bounds {
            let bound_range = mfp_eval.expr(bound);
            let result = mfp_eval.binary(&BinaryFunc::Lte(func::Lte), bound_range, mz_now.clone());
            results.push(result);
        }
        for bound in &plan.upper_bounds {
            let bound_range = mfp_eval.expr(bound);
            let result = mfp_eval.binary(&BinaryFunc::Gte(func::Gte), bound_range, mz_now.clone());
            results.push(result);
        }
        let mut result = self.variadic(&And.into(), results);
        if mfp_eval.expressions.iter().any(|s| s.range.fallible) {
            result.range.fallible = true;
        }
        result
    }
}

/// An interpreter that returns whether or not a particular expression is "pushdownable".
/// Broadly speaking, an expression is pushdownable if the result of evaluating the expression
/// depends on the range of possible column values in a way that `ColumnSpecs` is able to reason about.
///
/// In practice, we internally need to distinguish between expressions that are trivially predicable
/// (because they're constant) and expressions that depend on the column ranges themselves.
/// See the [TraceSummary] variants for those distinctions, and [TraceSummary::pushdownable] for
/// the overall assessment.
#[derive(Debug)]
pub struct Trace;

/// A summary type for the [Trace] interpreter.
///
/// The ordering of this type is meaningful: the "smaller" the summary, the more information we have
/// about the possible values of the expression. This means we can eg. use `max` in the
/// interpreter below to find the summary for a function-call expression based on the summaries
/// of its arguments.
#[derive(Copy, Clone, Debug, PartialOrd, PartialEq, Ord, Eq)]
pub enum TraceSummary {
    /// The expression is constant: we can evaluate it without any runtime information.
    /// This corresponds to a `ResultSpec` of a single value.
    Constant,
    /// The expression depends on runtime information, but in "predictable" way... ie. if we know
    /// the range of possible values for all columns and unmaterializable functions, we can
    /// predict the possible values of the output.
    /// This corresponds to a `ResultSpec` of a perhaps range of values.
    Dynamic,
    /// The expression depends on runtime information in an unpredictable way.
    /// This corresponds to a `ResultSpec::value_all()` or something similarly vague.
    Unknown,
}

impl TraceSummary {
    /// We say that a function is "pushdownable" for a particular
    /// argument if `ColumnSpecs` can determine the spec of the function's output given the input spec for
    /// that argument. (In practice, this is true when either the function is monotone in that argument
    /// or it's been special-cased in the interpreter.)
    fn apply_fn(self, pushdownable: bool) -> Self {
        match self {
            TraceSummary::Constant => TraceSummary::Constant,
            TraceSummary::Dynamic => match pushdownable {
                true => TraceSummary::Dynamic,
                false => TraceSummary::Unknown,
            },
            TraceSummary::Unknown => TraceSummary::Unknown,
        }
    }

    /// We say that an expression is "pushdownable" if it's either constant or dynamic.
    pub fn pushdownable(self) -> bool {
        match self {
            TraceSummary::Constant | TraceSummary::Dynamic => true,
            TraceSummary::Unknown => false,
        }
    }
}

impl Interpreter for Trace {
    type Summary = TraceSummary;

    fn column(&self, _id: usize) -> Self::Summary {
        TraceSummary::Dynamic
    }

    fn literal(
        &self,
        _result: &Result<Row, EvalError>,
        _col_type: &ReprColumnType,
    ) -> Self::Summary {
        TraceSummary::Constant
    }

    fn unmaterializable(&self, _func: &UnmaterializableFunc) -> Self::Summary {
        TraceSummary::Dynamic
    }

    fn unary(&self, func: &UnaryFunc, expr: Self::Summary) -> Self::Summary {
        let pushdownable = match SpecialUnary::for_func(func) {
            None => func.is_monotone(),
            Some(special) => special.pushdownable,
        };
        expr.apply_fn(pushdownable)
    }

    fn binary(
        &self,
        func: &BinaryFunc,
        left: Self::Summary,
        right: Self::Summary,
    ) -> Self::Summary {
        let (left_pushdownable, right_pushdownable) = match AbstractFunc::for_func(func) {
            None => func.is_monotone(),
            Some(special) => special.pushdownable,
        };
        left.apply_fn(left_pushdownable)
            .max(right.apply_fn(right_pushdownable))
    }

    fn variadic(&self, func: &VariadicFunc, exprs: Vec<Self::Summary>) -> Self::Summary {
        if !func.is_associative() && exprs.len() >= ColumnSpecs::MAX_EVAL_ARGS {
            // We can't efficiently evaluate functions with very large argument lists;
            // see the comment on ColumnSpecs::MAX_EVAL_ARGS for details.
            return TraceSummary::Unknown;
        }

        let pushdownable_fn = func.is_monotone();
        exprs
            .into_iter()
            .map(|pushdownable_arg| pushdownable_arg.apply_fn(pushdownable_fn))
            .max()
            .unwrap_or(TraceSummary::Constant)
    }

    fn cond(&self, cond: Self::Summary, then: Self::Summary, els: Self::Summary) -> Self::Summary {
        // We don't actually need to be able to predict the condition precisely to predict the output,
        // since we can union the ranges of the two branches for a conservative estimate.
        let cond = cond.min(TraceSummary::Dynamic);
        cond.max(then).max(els)
    }
}

#[cfg(test)]
mod tests {
    use itertools::Itertools;
    use mz_repr::adt::datetime::DateTimeUnits;
    use mz_repr::{Datum, PropDatum, RowArena, SqlScalarType};
    use proptest::prelude::*;
    use proptest::sample::{Index, select};

    use crate::func::*;
    use crate::scalar::func::variadic::Concat;
    use crate::{BinaryFunc, MirScalarExpr, UnaryFunc};

    use super::*;

    #[derive(Debug)]
    struct ExpressionData {
        relation_type: ReprRelationType,
        specs: Vec<ResultSpec<'static>>,
        rows: Vec<Row>,
        expr: MirScalarExpr,
    }

    // Currently there's no good way to check whether a particular function accepts a particular
    // type as argument, which means we need to list everything out explicitly here. Restrict our interest
    // to a reasonable number of functions, to keep things tractable
    // TODO: replace this with function-level info once it's available.
    const NUM_TYPE: ReprScalarType = ReprScalarType::Numeric;
    static SCALAR_TYPES: &[ReprScalarType] = &[
        ReprScalarType::Bool,
        ReprScalarType::Jsonb,
        NUM_TYPE,
        ReprScalarType::Int32,
        ReprScalarType::Float32,
        ReprScalarType::Float64,
        ReprScalarType::Date,
        ReprScalarType::Timestamp,
        ReprScalarType::MzTimestamp,
        ReprScalarType::Interval,
        ReprScalarType::String,
    ];

    const INTERESTING_UNARY_FUNCS: &[UnaryFunc] = {
        &[
            UnaryFunc::CastNumericToMzTimestamp(CastNumericToMzTimestamp),
            UnaryFunc::CastTimestampToMzTimestamp(CastTimestampToMzTimestamp),
            UnaryFunc::NegNumeric(NegNumeric),
            UnaryFunc::NegFloat64(NegFloat64),
            UnaryFunc::CastJsonbToNumeric(CastJsonbToNumeric(None)),
            UnaryFunc::CastJsonbToBool(CastJsonbToBool),
            UnaryFunc::CastJsonbToString(CastJsonbToString),
            UnaryFunc::DateTruncTimestamp(DateTruncTimestamp(DateTimeUnits::Epoch)),
            UnaryFunc::ExtractTimestamp(ExtractTimestamp(DateTimeUnits::Epoch)),
            UnaryFunc::ExtractDate(ExtractDate(DateTimeUnits::Epoch)),
            UnaryFunc::Not(Not),
            UnaryFunc::IsNull(IsNull),
            UnaryFunc::IsFalse(IsFalse),
            UnaryFunc::TryParseMonotonicIso8601Timestamp(TryParseMonotonicIso8601Timestamp),
        ]
    };

    fn unary_typecheck(func: &UnaryFunc, arg: &ReprColumnType) -> bool {
        use UnaryFunc::*;
        match func {
            CastNumericToMzTimestamp(_) | NegNumeric(_) => arg.scalar_type == NUM_TYPE,
            NegFloat64(_) => arg.scalar_type == ReprScalarType::Float64,
            CastTimestampToMzTimestamp(_) => arg.scalar_type == ReprScalarType::Timestamp,
            CastJsonbToNumeric(_) | CastJsonbToBool(_) | CastJsonbToString(_) => {
                arg.scalar_type == ReprScalarType::Jsonb
            }
            ExtractTimestamp(_) | DateTruncTimestamp(_) => {
                arg.scalar_type == ReprScalarType::Timestamp
            }
            ExtractDate(_) => arg.scalar_type == ReprScalarType::Date,
            Not(_) => arg.scalar_type == ReprScalarType::Bool,
            IsNull(_) => true,
            TryParseMonotonicIso8601Timestamp(_) => arg.scalar_type == ReprScalarType::String,
            _ => false,
        }
    }

    fn interesting_binary_funcs() -> Vec<BinaryFunc> {
        vec![
            AddTimestampInterval.into(),
            AddNumeric.into(),
            SubNumeric.into(),
            MulNumeric.into(),
            DivNumeric.into(),
            AddFloat64.into(),
            SubFloat64.into(),
            MulFloat64.into(),
            DivFloat64.into(),
            MulFloat32.into(),
            DivFloat32.into(),
            RoundNumericBinary.into(),
            Eq.into(),
            Lt.into(),
            Gt.into(),
            Lte.into(),
            Gte.into(),
            DateTruncUnitsTimestamp.into(),
            JsonbGetString.into(),
            JsonbGetStringStringify.into(),
        ]
    }

    fn binary_typecheck(func: &BinaryFunc, arg0: &ReprColumnType, arg1: &ReprColumnType) -> bool {
        use BinaryFunc::*;
        match func {
            AddTimestampInterval(_) => {
                arg0.scalar_type == ReprScalarType::Timestamp
                    && arg1.scalar_type == ReprScalarType::Interval
            }
            AddNumeric(_) | SubNumeric(_) | MulNumeric(_) | DivNumeric(_) => {
                arg0.scalar_type == NUM_TYPE && arg1.scalar_type == NUM_TYPE
            }
            AddFloat64(_) | SubFloat64(_) | MulFloat64(_) | DivFloat64(_) => {
                arg0.scalar_type == ReprScalarType::Float64
                    && arg1.scalar_type == ReprScalarType::Float64
            }
            MulFloat32(_) | DivFloat32(_) => {
                arg0.scalar_type == ReprScalarType::Float32
                    && arg1.scalar_type == ReprScalarType::Float32
            }
            RoundNumeric(_) => {
                arg0.scalar_type == NUM_TYPE && arg1.scalar_type == ReprScalarType::Int32
            }
            Eq(_) | Lt(_) | Gt(_) | Lte(_) | Gte(_) => arg0.scalar_type == arg1.scalar_type,
            DateTruncTimestamp(_) => {
                arg0.scalar_type == ReprScalarType::String
                    && arg1.scalar_type == ReprScalarType::Timestamp
            }
            JsonbGetString(_) | JsonbGetStringStringify(_) => {
                arg0.scalar_type == ReprScalarType::Jsonb
                    && arg1.scalar_type == ReprScalarType::String
            }
            _ => false,
        }
    }

    const INTERESTING_VARIADIC_FUNCS: &[VariadicFunc] = {
        use crate::scalar::func::variadic as v;
        use VariadicFunc::*;
        &[
            Coalesce(v::Coalesce),
            Greatest(v::Greatest),
            Least(v::Least),
            And(v::And),
            Or(v::Or),
            Concat(v::Concat),
            ConcatWs(v::ConcatWs),
        ]
    };

    fn variadic_typecheck(func: &VariadicFunc, args: &[ReprColumnType]) -> bool {
        use VariadicFunc::*;
        fn all_eq<'a>(
            iter: impl IntoIterator<Item = &'a ReprColumnType>,
            other: &ReprScalarType,
        ) -> bool {
            iter.into_iter().all(|t| t.scalar_type == *other)
        }
        match func {
            Coalesce(_) | Greatest(_) | Least(_) => match args {
                [] => true,
                [first, rest @ ..] => all_eq(rest, &first.scalar_type),
            },
            And(_) | Or(_) => all_eq(args, &ReprScalarType::Bool),
            Concat(_) => all_eq(args, &ReprScalarType::String),
            ConcatWs(_) => args.len() > 1 && all_eq(args, &ReprScalarType::String),
            _ => false,
        }
    }

    fn gen_datums_for_type(typ: &ReprColumnType) -> BoxedStrategy<Datum<'static>> {
        let mut values: Vec<Datum<'static>> = SqlScalarType::from_repr(&typ.scalar_type)
            .interesting_datums()
            .collect();
        if typ.nullable {
            values.push(Datum::Null)
        }
        select(values).boxed()
    }

    fn gen_column() -> impl Strategy<Value = (ReprColumnType, Datum<'static>, ResultSpec<'static>)>
    {
        let col_type = (select(SCALAR_TYPES), any::<bool>())
            .prop_map(|(t, b)| t.nullable(b))
            .prop_filter("need at least one value", |c| {
                SqlScalarType::from_repr(&c.scalar_type)
                    .interesting_datums()
                    .count()
                    > 0
            });

        let result_spec = select(vec![
            ResultSpec::nothing(),
            ResultSpec::null(),
            ResultSpec::anything(),
            ResultSpec::value_all(),
        ]);

        (col_type, result_spec).prop_flat_map(|(col, result_spec)| {
            gen_datums_for_type(&col).prop_map(move |datum| {
                let result_spec = result_spec.clone().union(ResultSpec::value(datum));
                (col.clone(), datum, result_spec)
            })
        })
    }

    fn gen_expr_for_relation(
        relation: &ReprRelationType,
    ) -> BoxedStrategy<(MirScalarExpr, ReprColumnType)> {
        let column_gen = {
            let column_types = relation.column_types.clone();
            any::<Index>()
                .prop_map(move |idx| {
                    let id = idx.index(column_types.len());
                    (MirScalarExpr::column(id), column_types[id].clone())
                })
                .boxed()
        };

        let literal_gen = (select(SCALAR_TYPES), any::<bool>())
            .prop_map(|(s, b)| s.nullable(b))
            .prop_flat_map(|ct| {
                let error_gen = any::<EvalError>().prop_map(Err).boxed();
                let value_gen = gen_datums_for_type(&ct)
                    .prop_map(move |datum| Ok(Row::pack_slice(&[datum])))
                    .boxed();
                error_gen.prop_union(value_gen).prop_map(move |result| {
                    (MirScalarExpr::Literal(result, ct.clone()), ct.clone())
                })
            })
            .boxed();

        column_gen
            .prop_union(literal_gen)
            .prop_recursive(4, 64, 8, |self_gen| {
                let unary_gen = (select(INTERESTING_UNARY_FUNCS), self_gen.clone())
                    .prop_filter_map("unary func", |(func, (expr_in, type_in))| {
                        if !unary_typecheck(&func, &type_in) {
                            return None;
                        }
                        let type_out = func.output_type(type_in);
                        let expr_out = MirScalarExpr::CallUnary {
                            func,
                            expr: Box::new(expr_in),
                        };
                        Some((expr_out, type_out))
                    })
                    .boxed();
                let binary_gen = (
                    select(interesting_binary_funcs()),
                    self_gen.clone(),
                    self_gen.clone(),
                )
                    .prop_filter_map(
                        "binary func",
                        |(func, (expr_left, type_left), (expr_right, type_right))| {
                            if !binary_typecheck(&func, &type_left, &type_right) {
                                return None;
                            }
                            let type_out = func.output_type(&[type_left, type_right]);
                            let expr_out = MirScalarExpr::CallBinary {
                                func,
                                expr1: Box::new(expr_left),
                                expr2: Box::new(expr_right),
                            };
                            Some((expr_out, type_out))
                        },
                    )
                    .boxed();
                let variadic_gen = (
                    select(INTERESTING_VARIADIC_FUNCS),
                    prop::collection::vec(self_gen.clone(), 1..4),
                )
                    .prop_filter_map("variadic func", |(func, exprs)| {
                        let (exprs_in, type_in): (_, Vec<_>) = exprs.into_iter().unzip();
                        if !variadic_typecheck(&func, &type_in) {
                            return None;
                        }
                        let type_out = func.output_type(type_in);
                        let expr_out = MirScalarExpr::CallVariadic {
                            func,
                            exprs: exprs_in,
                        };
                        Some((expr_out, type_out))
                    })
                    .boxed();
                // Generate `If` nodes without the heavy rejection that filtering
                // three independent subexprs for a bool condition and matching
                // branch types would incur. The condition is a boolean literal
                // (so it can be `True`, `False`, or `Null` — exercising `cond`'s
                // value channel, including the null-takes-`els` path), and `els`
                // is a literal of `then`'s type so the branches always unify.
                let if_gen = {
                    let bool_type = ReprScalarType::Bool.nullable(true);
                    let cond_gen = gen_datums_for_type(&bool_type).prop_map(move |datum| {
                        MirScalarExpr::Literal(Ok(Row::pack_slice(&[datum])), bool_type.clone())
                    });
                    (cond_gen, self_gen.clone())
                        .prop_flat_map(|(cond_expr, (then_expr, then_type))| {
                            let out_type = then_type.clone();
                            gen_datums_for_type(&then_type).prop_map(move |datum| {
                                let els_expr = MirScalarExpr::Literal(
                                    Ok(Row::pack_slice(&[datum])),
                                    out_type.clone(),
                                );
                                let expr_out = MirScalarExpr::If {
                                    cond: Box::new(cond_expr.clone()),
                                    then: Box::new(then_expr.clone()),
                                    els: Box::new(els_expr),
                                };
                                (expr_out, out_type.clone())
                            })
                        })
                        .boxed()
                };

                unary_gen
                    .prop_union(binary_gen)
                    .boxed()
                    .prop_union(variadic_gen)
                    .boxed()
                    .prop_union(if_gen)
            })
            .boxed()
    }

    fn gen_expr_data() -> impl Strategy<Value = ExpressionData> {
        let columns = prop::collection::vec(gen_column(), 1..10);
        columns.prop_flat_map(|data| {
            let (columns, datums, specs): (Vec<_>, Vec<_>, Vec<_>) = data.into_iter().multiunzip();
            let relation = ReprRelationType::new(columns);
            let row = Row::pack_slice(&datums);
            gen_expr_for_relation(&relation).prop_map(move |(expr, _)| ExpressionData {
                relation_type: relation.clone(),
                specs: specs.clone(),
                rows: vec![row.clone()],
                expr,
            })
        })
    }

    #[mz_ore::test]
    #[cfg_attr(miri, ignore)] // error: unsupported operation: can't call foreign function `decContextDefault` on OS `linux`
    fn test_trivial_spec_matches() {
        fn check(datum: PropDatum) -> Result<(), TestCaseError> {
            let datum: Datum = (&datum).into();
            let spec = if datum.is_null() {
                ResultSpec::null()
            } else {
                ResultSpec::value(datum)
            };
            assert!(spec.may_contain(datum));
            Ok(())
        }

        proptest!(|(datum in mz_repr::arb_datum(true))| {
            check(datum)?;
        });

        assert!(ResultSpec::fails().may_fail());
    }

    #[mz_ore::test]
    #[cfg_attr(miri, ignore)] // error: unsupported operation: can't call foreign function `decContextDefault` on OS `linux`
    fn test_equivalence() {
        fn check(data: ExpressionData) -> Result<(), TestCaseError> {
            let ExpressionData {
                relation_type,
                specs,
                rows,
                expr,
            } = data;

            // We want to ensure that the spec we get when evaluating an expression using
            // `ColumnSpecs` always contains the _actual_ value of that column when evaluated with
            // eval. (This is an important correctness property of abstract interpretation.)
            let arena = RowArena::new();
            let mut interpreter = ColumnSpecs::new(&relation_type, &arena);
            for (id, spec) in specs.into_iter().enumerate() {
                interpreter.push_column(id, spec);
            }

            let spec = interpreter.expr(&expr);

            for row in &rows {
                let datums: Vec<_> = row.iter().collect();
                let eval_result = expr.eval(&datums, &arena);
                match eval_result {
                    Ok(value) => {
                        assert!(spec.range.may_contain(value))
                    }
                    Err(_) => {
                        assert!(spec.range.may_fail());
                    }
                }
            }

            Ok(())
        }

        proptest!(|(data in gen_expr_data())| {
            check(data)?;
        });
    }

    /// A column whose spec is a genuine `value_between(lo, hi)` range with
    /// `lo < hi`, paired with a concrete row value `mid` drawn from strictly
    /// inside `[lo, hi]`. This is the shape persist filter pushdown actually
    /// feeds the interpreter: min/max stats become a `Values::Within` range,
    /// and the interpreter narrows it through each function (relying on
    /// monotonicity) without ever seeing the interior values. `gen_column`
    /// only ever produces single-value or `anything` specs, so it never
    /// exercises the range-narrowing path.
    fn gen_range_column()
    -> impl Strategy<Value = (ReprColumnType, Datum<'static>, ResultSpec<'static>)> {
        select(SCALAR_TYPES)
            .prop_map(|t| t.nullable(false))
            .prop_filter("need at least two distinct values for a range", |c| {
                let mut datums: Vec<Datum> = SqlScalarType::from_repr(&c.scalar_type)
                    .interesting_datums()
                    .filter(|d| !d.is_null())
                    .collect();
                datums.sort();
                datums.dedup();
                datums.len() >= 2
            })
            .prop_flat_map(|col| {
                let mut datums: Vec<Datum<'static>> = SqlScalarType::from_repr(&col.scalar_type)
                    .interesting_datums()
                    .filter(|d| !d.is_null())
                    .collect();
                datums.sort();
                datums.dedup();
                (
                    Just(col),
                    Just(datums),
                    any::<Index>(),
                    any::<Index>(),
                    any::<Index>(),
                )
                    .prop_map(|(col, datums, a, b, c)| {
                        let n = datums.len();
                        let mut idxs = [a.index(n), b.index(n), c.index(n)];
                        idxs.sort();
                        let lo = datums[idxs[0]];
                        let mid = datums[idxs[1]];
                        let hi = datums[idxs[2]];
                        let spec = ResultSpec::value_between(lo, hi);
                        (col, mid, spec)
                    })
            })
    }

    fn gen_range_expr_data() -> impl Strategy<Value = ExpressionData> {
        let columns = prop::collection::vec(gen_range_column(), 1..10);
        columns.prop_flat_map(|data| {
            let (columns, datums, specs): (Vec<_>, Vec<_>, Vec<_>) = data.into_iter().multiunzip();
            let relation = ReprRelationType::new(columns);
            let row = Row::pack_slice(&datums);
            gen_expr_for_relation(&relation).prop_map(move |(expr, _)| ExpressionData {
                relation_type: relation.clone(),
                specs: specs.clone(),
                rows: vec![row.clone()],
                expr,
            })
        })
    }

    /// Regression test for database-issues#9656 (PER-50).
    ///
    /// Like [`test_equivalence`], but the column specs are genuine
    /// `value_between(lo, hi)` ranges rather than single values. This is the
    /// input shape persist filter pushdown produces from min/max stats, and it
    /// is the one that drives the interpreter's monotonicity-based range
    /// narrowing. If a function narrows a range to a spec that does not contain
    /// the value the concrete evaluator produces for an interior input, the
    /// interpreter can wrongly rule out a matching row and pushdown discards a
    /// part it should have kept, triggering the audit panic.
    #[mz_ore::test]
    #[cfg_attr(miri, ignore)] // error: unsupported operation: can't call foreign function `decContextDefault` on OS `linux`
    fn test_equivalence_ranges() {
        fn check(data: ExpressionData) -> Result<(), TestCaseError> {
            let ExpressionData {
                relation_type,
                specs,
                rows,
                expr,
            } = data;

            let arena = RowArena::new();
            let mut interpreter = ColumnSpecs::new(&relation_type, &arena);
            for (id, spec) in specs.into_iter().enumerate() {
                interpreter.push_column(id, spec);
            }

            let spec = interpreter.expr(&expr);

            for row in &rows {
                let datums: Vec<_> = row.iter().collect();
                let eval_result = expr.eval(&datums, &arena);
                match eval_result {
                    Ok(value) => {
                        prop_assert!(
                            spec.range.may_contain(value),
                            "interpreter ruled out a value the evaluator produced \
                             for an interior input: expr={expr:?} row={row:?} \
                             value={value:?} spec={:?}",
                            spec.range,
                        );
                    }
                    Err(_) => {
                        prop_assert!(
                            spec.range.may_fail(),
                            "interpreter ruled out an error the evaluator produced \
                             for an interior input: expr={expr:?} row={row:?}",
                        );
                    }
                }
            }

            Ok(())
        }

        // The expression generator rejects many function/type combinations
        // (see the `prop_filter_map`s in `gen_expr_for_relation`), so the
        // per-run local-reject budget has to be raised well above proptest's
        // default to let enough cases through.
        let config = ProptestConfig {
            cases: 2048,
            max_local_rejects: 1 << 20,
            ..ProptestConfig::default()
        };
        proptest!(config, |(data in gen_range_expr_data())| {
            check(data)?;
        });
    }

    /// The abstract-domain lattice laws `ColumnSpecs` relies on: `union` must
    /// over-approximate (contain everything either operand contains), and
    /// `intersect` must contain every value BOTH operands contain. An
    /// intersect-law violation is a false negative — a value the interpreter
    /// silently drops. The interpreter forms straddling `Within` ranges
    /// internally (e.g. `Values::union` of differently-typed endpoints in `cond`
    /// branches or `eq`), so this is not purely hypothetical. See
    /// database-issues#9656.
    #[mz_ore::test]
    #[cfg_attr(miri, ignore)] // error: unsupported operation: can't call foreign function `decContextDefault` on OS `linux`
    fn test_result_spec_lattice_laws() {
        // A recipe for a `ResultSpec`, carrying owned `PropDatum`s so the spec
        // (which borrows `Datum`s) can be materialized inside the test.
        #[derive(Debug, Clone)]
        enum Recipe {
            Nothing,
            Null,
            Fails,
            Anything,
            ValueAll,
            Value(PropDatum),
            Between(PropDatum, PropDatum),
            Map(Vec<(PropDatum, Recipe)>),
            Union(Box<Recipe>, Box<Recipe>),
        }

        fn materialize(recipe: &Recipe) -> ResultSpec<'_> {
            match recipe {
                Recipe::Nothing => ResultSpec::nothing(),
                Recipe::Null => ResultSpec::null(),
                Recipe::Fails => ResultSpec::fails(),
                Recipe::Anything => ResultSpec::anything(),
                Recipe::ValueAll => ResultSpec::value_all(),
                Recipe::Value(pd) => ResultSpec::value(pd.into()),
                Recipe::Between(a, b) => {
                    let (a, b): (Datum, Datum) = (a.into(), b.into());
                    if a.is_null() || b.is_null() {
                        ResultSpec::nothing()
                    } else if a <= b {
                        ResultSpec::value_between(a, b)
                    } else {
                        ResultSpec::value_between(b, a)
                    }
                }
                Recipe::Map(entries) => {
                    let mut map = BTreeMap::new();
                    for (key, val) in entries {
                        let key: Datum = key.into();
                        if !key.is_null() {
                            map.insert(key, materialize(val));
                        }
                    }
                    ResultSpec::map_spec(map)
                }
                Recipe::Union(a, b) => materialize(a).union(materialize(b)),
            }
        }

        fn recipe_strategy() -> impl Strategy<Value = Recipe> {
            let leaf = proptest::strategy::Union::new(vec![
                Just(Recipe::Nothing).boxed(),
                Just(Recipe::Null).boxed(),
                Just(Recipe::Fails).boxed(),
                Just(Recipe::Anything).boxed(),
                Just(Recipe::ValueAll).boxed(),
                mz_repr::arb_datum(false).prop_map(Recipe::Value).boxed(),
                (mz_repr::arb_datum(false), mz_repr::arb_datum(false))
                    .prop_map(|(a, b)| Recipe::Between(a, b))
                    .boxed(),
            ]);
            leaf.prop_recursive(3, 24, 4, |inner| {
                proptest::strategy::Union::new(vec![
                    prop::collection::vec((mz_repr::arb_datum(false), inner.clone()), 0..3)
                        .prop_map(Recipe::Map)
                        .boxed(),
                    (inner.clone(), inner.clone())
                        .prop_map(|(a, b)| Recipe::Union(Box::new(a), Box::new(b)))
                        .boxed(),
                ])
            })
        }

        fn check(a: Recipe, b: Recipe, v: PropDatum) -> Result<(), TestCaseError> {
            let a_spec = materialize(&a);
            let b_spec = materialize(&b);
            let v: Datum = (&v).into();

            let in_a = a_spec.may_contain(v);
            let in_b = b_spec.may_contain(v);

            if in_a || in_b {
                prop_assert!(
                    a_spec.clone().union(b_spec.clone()).may_contain(v),
                    "union dropped a value: a={a:?} b={b:?} v={v:?}",
                );
            }
            if in_a && in_b {
                prop_assert!(
                    a_spec.intersect(b_spec).may_contain(v),
                    "intersect dropped a common value: a={a:?} b={b:?} v={v:?}",
                );
            }
            Ok(())
        }

        proptest!(
            ProptestConfig::with_cases(4096),
            |(a in recipe_strategy(), b in recipe_strategy(), v in mz_repr::arb_datum(true))| {
                check(a, b, v)?;
            }
        );
    }

    /// Deterministic regression test for database-issues#9656 (PER-50), the
    /// minimal case [`test_equivalence_ranges`] shrinks to.
    ///
    /// A numeric column whose stats range spans `[-Infinity, NaN]` (NaN is the
    /// maximum of the numeric Datum order) feeds `-column`. `NegNumeric` is
    /// declared monotone, so the interpreter would narrow the output to
    /// `[neg(-Infinity), neg(NaN)] = [Infinity, NaN]` and wrongly rule out the
    /// value `1` that the evaluator produces for the interior input `-1`. That
    /// false negative is exactly what makes persist filter pushdown discard a
    /// part it should keep, tripping the audit panic.
    #[mz_ore::test]
    #[cfg_attr(miri, ignore)] // error: unsupported operation: can't call foreign function `decContextDefault` on OS `linux`
    fn test_neg_numeric_nan_range() {
        use mz_repr::adt::numeric::Numeric;

        let neg = MirScalarExpr::CallUnary {
            func: UnaryFunc::NegNumeric(NegNumeric),
            expr: Box::new(MirScalarExpr::column(0)),
        };

        let relation = ReprRelationType::new(vec![ReprScalarType::Numeric.nullable(false)]);
        let arena = RowArena::new();
        let mut interpreter = ColumnSpecs::new(&relation, &arena);
        interpreter.push_column(
            0,
            ResultSpec::value_between(
                Datum::from(Numeric::from(f64::NEG_INFINITY)),
                Datum::from(Numeric::from(f64::NAN)),
            ),
        );

        let spec = interpreter.expr(&neg);

        // `-1` is an interior value of the input range, and `-(-1) = 1`.
        let actual = neg
            .eval(&[Datum::from(Numeric::from(-1.0f64))], &arena)
            .expect("eval succeeds");
        assert!(
            spec.range.may_contain(actual),
            "interpreter must not rule out {actual:?}, which the evaluator \
             produces for an interior input; got spec {:?}",
            spec.range,
        );
    }

    /// Deterministic regression test for database-issues#9656 (PER-50), the
    /// fallibility variant [`test_equivalence_ranges`] also surfaces.
    ///
    /// `cast_numeric_to_mz_timestamp` is declared monotone but errors on
    /// fractional inputs, which are dense in the interior of any range. The
    /// range `[0, 2]` has integer endpoints that both cast cleanly, so the
    /// interpreter's endpoint sampling never observes an error. It must instead
    /// surface the function's own `could_error`, otherwise persist filter
    /// pushdown discards a part whose interior rows (e.g. `1.5`) produce error
    /// rows.
    #[mz_ore::test]
    #[cfg_attr(miri, ignore)] // error: unsupported operation: can't call foreign function `decContextDefault` on OS `linux`
    fn test_fallible_monotone_interior_error() {
        use mz_repr::adt::numeric::Numeric;

        let cast = MirScalarExpr::CallUnary {
            func: UnaryFunc::CastNumericToMzTimestamp(CastNumericToMzTimestamp),
            expr: Box::new(MirScalarExpr::column(0)),
        };

        let relation = ReprRelationType::new(vec![ReprScalarType::Numeric.nullable(false)]);
        let arena = RowArena::new();
        let mut interpreter = ColumnSpecs::new(&relation, &arena);
        interpreter.push_column(
            0,
            ResultSpec::value_between(
                Datum::from(Numeric::from(0.0f64)),
                Datum::from(Numeric::from(2.0f64)),
            ),
        );

        let spec = interpreter.expr(&cast);

        // `1.5` is an interior value of `[0, 2]`, and casting a fractional
        // numeric to mz_timestamp errors.
        let interior = Datum::from(Numeric::from(1.5f64));
        assert!(
            cast.eval(&[interior], &arena).is_err(),
            "precondition: a fractional numeric fails to cast to mz_timestamp",
        );
        assert!(
            spec.range.may_fail(),
            "interpreter must surface that a monotone-but-fallible function may \
             error on an interior value it never sampled; got spec {:?}",
            spec.range,
        );
    }

    /// Regression test for database-issues#9656.
    ///
    /// The interpreter must surface the fallibility of MFP expressions that
    /// aren't referenced by any predicate or temporal bound. The runtime MFP
    /// evaluator runs every expression once predicates pass, so an expression
    /// that errors on the actual data makes the whole row an `Err` — and
    /// `filter_result` must keep the part to emit that error.
    #[mz_ore::test]
    #[cfg_attr(miri, ignore)]
    fn test_mfp_unreferenced_fallible_expression() {
        use crate::scalar::func::CastStringToUuid;

        // MFP: one expression that always errors on the input range, and one
        // predicate that always passes. The expression's result column is
        // *not* referenced by the predicate, so the default interpreter
        // implementation would AND together just `True` and miss the
        // fallibility.
        let mfp = MapFilterProject {
            expressions: vec![MirScalarExpr::CallUnary {
                func: UnaryFunc::CastStringToUuid(CastStringToUuid),
                expr: Box::new(MirScalarExpr::column(0)),
            }],
            predicates: vec![(
                1,
                MirScalarExpr::literal_ok(Datum::True, ReprScalarType::Bool),
            )],
            projection: vec![0, 1],
            input_arity: 1,
        };

        let relation = ReprRelationType::new(vec![ReprScalarType::String.nullable(false)]);
        let arena = RowArena::new();
        let mut interpreter = ColumnSpecs::new(&relation, &arena);
        // "not-a-uuid" is in the stats range and definitely doesn't parse as a UUID.
        interpreter.push_column(
            0,
            ResultSpec::value_between(Datum::String("not-a-uuid"), Datum::String("not-a-uuid")),
        );
        let spec = interpreter.mfp_filter(&mfp);
        assert!(
            spec.range.may_fail(),
            "an MFP expression that errors on the stats range must propagate \
             fallibility, otherwise persist filter pushdown can wrongly discard \
             a part that produces error rows",
        );
    }

    /// Proptest companion to [`test_mfp_unreferenced_fallible_expression`]:
    /// directly verifies the fallibility claim of [`ColumnSpecs::mfp_filter`]
    /// against the runtime MFP semantics. For a random expression placed in
    /// `MapFilterProject::expressions` (i.e. as an unreferenced Map step), if
    /// evaluating the expression on a row drawn from the stats range produces
    /// an error at runtime, then the interpreter's summary must report
    /// `may_fail()`. Without the `expressions.any(|s| s.range.fallible)` patch
    /// in `mfp_filter`, the AND over an empty predicate list collapses to
    /// `True` and the runtime error is wrongly ruled out.
    #[mz_ore::test]
    #[cfg_attr(miri, ignore)]
    fn test_mfp_filter_fallibility_equivalence() {
        fn check(data: ExpressionData) -> Result<(), TestCaseError> {
            let ExpressionData {
                relation_type,
                specs,
                rows,
                expr,
            } = data;

            let input_arity = relation_type.column_types.len();
            let mfp = MapFilterProject {
                expressions: vec![expr.clone()],
                predicates: vec![],
                projection: (0..input_arity).collect(),
                input_arity,
            };

            let arena = RowArena::new();
            let mut interpreter = ColumnSpecs::new(&relation_type, &arena);
            for (id, spec) in specs.into_iter().enumerate() {
                interpreter.push_column(id, spec);
            }
            let summary = interpreter.mfp_filter(&mfp);

            for row in &rows {
                let datums: Vec<_> = row.iter().collect();
                if expr.eval(&datums, &arena).is_err() {
                    prop_assert!(
                        summary.range.may_fail(),
                        "mfp_filter must surface the fallibility of an \
                         unreferenced MFP expression: row {:?} errored at \
                         runtime but the interpreter ruled out errors",
                        row,
                    );
                }
            }
            Ok(())
        }

        proptest!(|(data in gen_expr_data())| {
            check(data)?;
        });
    }

    #[mz_ore::test]
    fn test_mfp() {
        // Regression test for https://github.com/MaterializeInc/database-issues/issues/5736
        use MirScalarExpr::*;

        let mfp = MapFilterProject {
            expressions: vec![],
            predicates: vec![
                // Always fails on the known input range
                (
                    1,
                    CallUnary {
                        func: UnaryFunc::IsNull(IsNull),
                        expr: Box::new(CallBinary {
                            func: MulInt32.into(),
                            expr1: Box::new(MirScalarExpr::column(0)),
                            expr2: Box::new(MirScalarExpr::column(0)),
                        }),
                    },
                ),
                // Always returns false on the known input range
                (
                    1,
                    CallBinary {
                        func: Eq.into(),
                        expr1: Box::new(MirScalarExpr::column(0)),
                        expr2: Box::new(MirScalarExpr::literal_ok(
                            Datum::Int32(1727694505),
                            ReprScalarType::Int32,
                        )),
                    },
                ),
            ],
            projection: vec![],
            input_arity: 1,
        };

        let relation = ReprRelationType::new(vec![ReprScalarType::Int32.nullable(true)]);
        let arena = RowArena::new();
        let mut interpreter = ColumnSpecs::new(&relation, &arena);
        interpreter.push_column(0, ResultSpec::value(Datum::Int32(-1294725158)));
        let spec = interpreter.mfp_filter(&mfp);
        assert!(spec.range.may_fail());
    }

    #[mz_ore::test]
    fn test_concat() {
        let expr = MirScalarExpr::call_variadic(
            Concat,
            vec![
                MirScalarExpr::column(0),
                MirScalarExpr::literal_ok(Datum::String("a"), ReprScalarType::String),
                MirScalarExpr::literal_ok(Datum::String("b"), ReprScalarType::String),
            ],
        );

        let relation = ReprRelationType::new(vec![ReprScalarType::String.nullable(false)]);
        let arena = RowArena::new();
        let interpreter = ColumnSpecs::new(&relation, &arena);
        let spec = interpreter.expr(&expr);
        assert!(spec.range.may_contain(Datum::String("blab")));
    }

    #[mz_ore::test]
    fn test_eval_range() {
        // Example inspired by the tumbling windows temporal filter in the docs.
        //
        // NOTE: `may_fail()` is `true` in both cases below. `DivInt64`,
        // `MulInt64`, and `CastInt64ToMzTimestamp` can all error, and the
        // interpreter no longer infers infallibility from endpoint sampling for
        // an erroring function over a multi-valued range (it cannot prove the
        // function doesn't error on an interior value). For this data none of
        // them actually errors, so this is a conservative over-approximation
        // that keeps the part rather than pruning it. The value channel is still
        // narrowed precisely (`may_contain` below is exact), so a follow-up that
        // makes the fallibility flag argument-aware could recover pruning here.
        let period_ms = MirScalarExpr::literal_ok(Datum::Int64(10), ReprScalarType::Int64);
        let expr = MirScalarExpr::CallBinary {
            func: Gte.into(),
            expr1: Box::new(MirScalarExpr::CallUnmaterializable(
                UnmaterializableFunc::MzNow,
            )),
            expr2: Box::new(MirScalarExpr::CallUnary {
                func: UnaryFunc::CastInt64ToMzTimestamp(CastInt64ToMzTimestamp),
                expr: Box::new(MirScalarExpr::CallBinary {
                    func: MulInt64.into(),
                    expr1: Box::new(period_ms.clone()),
                    expr2: Box::new(MirScalarExpr::CallBinary {
                        func: DivInt64.into(),
                        expr1: Box::new(MirScalarExpr::column(0)),
                        expr2: Box::new(period_ms),
                    }),
                }),
            }),
        };
        let relation = ReprRelationType::new(vec![ReprScalarType::Int64.nullable(false)]);

        {
            // Non-overlapping windows
            let arena = RowArena::new();
            let mut interpreter = ColumnSpecs::new(&relation, &arena);
            interpreter.push_unmaterializable(
                UnmaterializableFunc::MzNow,
                ResultSpec::value_between(
                    Datum::MzTimestamp(10.into()),
                    Datum::MzTimestamp(20.into()),
                ),
            );
            interpreter.push_column(0, ResultSpec::value_between(30i64.into(), 40i64.into()));

            let range_out = interpreter.expr(&expr).range;
            assert!(range_out.may_contain(Datum::False));
            assert!(!range_out.may_contain(Datum::True));
            assert!(!range_out.may_contain(Datum::Null));
            assert!(range_out.may_fail());
        }

        {
            // Overlapping windows
            let arena = RowArena::new();
            let mut interpreter = ColumnSpecs::new(&relation, &arena);
            interpreter.push_unmaterializable(
                UnmaterializableFunc::MzNow,
                ResultSpec::value_between(
                    Datum::MzTimestamp(10.into()),
                    Datum::MzTimestamp(35.into()),
                ),
            );
            interpreter.push_column(0, ResultSpec::value_between(30i64.into(), 40i64.into()));

            let range_out = interpreter.expr(&expr).range;
            assert!(range_out.may_contain(Datum::False));
            assert!(range_out.may_contain(Datum::True));
            assert!(!range_out.may_contain(Datum::Null));
            assert!(range_out.may_fail());
        }
    }

    #[mz_ore::test]
    #[cfg_attr(miri, ignore)] // error: unsupported operation: can't call foreign function `decNumberFromInt32` on OS `linux`
    fn test_jsonb() {
        let arena = RowArena::new();

        let expr = MirScalarExpr::column(0)
            .call_binary(
                MirScalarExpr::literal_ok(Datum::from("ts"), ReprScalarType::String),
                JsonbGetString,
            )
            .call_unary(CastJsonbToNumeric(None));

        let relation = ReprRelationType::new(vec![ReprScalarType::Jsonb.nullable(true)]);
        let mut interpreter = ColumnSpecs::new(&relation, &arena);
        interpreter.push_column(
            0,
            ResultSpec::map_spec(
                [(
                    "ts".into(),
                    ResultSpec::value_between(
                        Datum::Numeric(100.into()),
                        Datum::Numeric(300.into()),
                    ),
                )]
                .into_iter()
                .collect(),
            ),
        );

        let range_out = interpreter.expr(&expr).range;
        assert!(!range_out.may_contain(Datum::Numeric(0.into())));
        assert!(range_out.may_contain(Datum::Numeric(200.into())));
        assert!(!range_out.may_contain(Datum::Numeric(400.into())));
    }

    #[mz_ore::test]
    fn test_nested_union_partial_overlap() {
        // `Nested(map)` constrains a key only when the key is present in `map`; absent
        // keys mean "anything". So the union of two Nested specs must drop any key
        // that's missing from one side, because `x ∪ anything = anything`. Only keys
        // present in *both* sides survive (with their per-key specs unioned).
        let a = ResultSpec::map_spec(
            [
                ("x".into(), ResultSpec::value(Datum::String("a"))),
                ("y".into(), ResultSpec::value(Datum::String("b"))),
                ("c".into(), ResultSpec::value(Datum::String("c"))),
            ]
            .into_iter()
            .collect(),
        );
        let b = ResultSpec::map_spec(
            [
                ("x".into(), ResultSpec::value(Datum::String("a2"))),
                ("y".into(), ResultSpec::value(Datum::String("b2"))),
                ("z".into(), ResultSpec::value(Datum::String("z"))),
            ]
            .into_iter()
            .collect(),
        );

        let unioned = a.union(b);

        // Push the unioned spec through `->> <key>`: keys only in one side must
        // admit NULL (the other side is unconstrained, so the field could be absent
        // there); shared keys must include both observed values.
        let arena = RowArena::new();
        let relation = ReprRelationType::new(vec![ReprScalarType::Jsonb.nullable(false)]);

        // Key only in `a`: the union must admit NULL.
        {
            let mut interpreter = ColumnSpecs::new(&relation, &arena);
            interpreter.push_column(0, unioned.clone());
            let expr = MirScalarExpr::column(0).call_binary(
                MirScalarExpr::literal_ok(Datum::from("c"), ReprScalarType::String),
                JsonbGetStringStringify,
            );
            assert!(interpreter.expr(&expr).range.may_contain(Datum::Null));
        }

        // Key only in `b`: symmetric.
        {
            let mut interpreter = ColumnSpecs::new(&relation, &arena);
            interpreter.push_column(0, unioned.clone());
            let expr = MirScalarExpr::column(0).call_binary(
                MirScalarExpr::literal_ok(Datum::from("z"), ReprScalarType::String),
                JsonbGetStringStringify,
            );
            assert!(interpreter.expr(&expr).range.may_contain(Datum::Null));
        }

        // Key in both: result must include both observed values.
        {
            let mut interpreter = ColumnSpecs::new(&relation, &arena);
            interpreter.push_column(0, unioned);
            let expr = MirScalarExpr::column(0).call_binary(
                MirScalarExpr::literal_ok(Datum::from("x"), ReprScalarType::String),
                JsonbGetStringStringify,
            );
            let x_range = interpreter.expr(&expr).range;
            assert!(x_range.may_contain(Datum::String("a")));
            assert!(x_range.may_contain(Datum::String("a2")));
        }
    }

    #[mz_ore::test]
    #[cfg_attr(miri, ignore)] // unsupported foreign call in numeric decoding
    fn test_case_over_jsonb_columns() {
        // Regression test for PER-6: when CASE picks between two JSON columns whose
        // observed keys are disjoint, filter pushdown must not prune parts where
        // accessing a key only present in one branch might yield NULL in the other.
        let arena = RowArena::new();

        // `(CASE WHEN col0 THEN col1 ELSE col2 END) ->> 'y' IS NULL`
        let expr = MirScalarExpr::If {
            cond: Box::new(MirScalarExpr::column(0)),
            then: Box::new(MirScalarExpr::column(1)),
            els: Box::new(MirScalarExpr::column(2)),
        }
        .call_binary(
            MirScalarExpr::literal_ok(Datum::from("y"), ReprScalarType::String),
            JsonbGetStringStringify,
        )
        .call_unary(UnaryFunc::IsNull(IsNull));

        let relation = ReprRelationType::new(vec![
            ReprScalarType::Bool.nullable(false),
            ReprScalarType::Jsonb.nullable(false),
            ReprScalarType::Jsonb.nullable(false),
        ]);
        let mut interpreter = ColumnSpecs::new(&relation, &arena);
        interpreter.push_column(0, ResultSpec::value_between(Datum::False, Datum::True));
        interpreter.push_column(
            1,
            ResultSpec::map_spec(
                [("x".into(), ResultSpec::value(Datum::String("a")))]
                    .into_iter()
                    .collect(),
            ),
        );
        interpreter.push_column(
            2,
            ResultSpec::map_spec(
                [("y".into(), ResultSpec::value(Datum::String("b")))]
                    .into_iter()
                    .collect(),
            ),
        );

        let range_out = interpreter.expr(&expr).range;
        // When the CASE selects column 1, "y" is absent and `->> 'y'` yields NULL, so
        // `IS NULL` is True. The filter must not prune a part that could match.
        assert!(range_out.may_contain(Datum::True));
    }

    #[mz_ore::test]
    fn test_like() {
        let arena = RowArena::new();

        let expr = MirScalarExpr::CallUnary {
            func: UnaryFunc::IsLikeMatch(IsLikeMatch(
                crate::like_pattern::compile("%whatever%", true).unwrap(),
            )),
            expr: Box::new(MirScalarExpr::column(0)),
        };

        let relation = ReprRelationType::new(vec![ReprScalarType::String.nullable(true)]);
        let mut interpreter = ColumnSpecs::new(&relation, &arena);
        interpreter.push_column(
            0,
            ResultSpec::value_between(Datum::String("aardvark"), Datum::String("zebra")),
        );

        let range_out = interpreter.expr(&expr).range;
        assert!(
            !range_out.fallible,
            "like function should not error on non-error input"
        );
        assert!(range_out.may_contain(Datum::True));
        assert!(range_out.may_contain(Datum::False));
        assert!(range_out.may_contain(Datum::Null));
    }

    #[mz_ore::test]
    fn test_try_parse_monotonic_iso8601_timestamp() {
        use chrono::NaiveDateTime;

        let arena = RowArena::new();

        let expr = MirScalarExpr::CallUnary {
            func: UnaryFunc::TryParseMonotonicIso8601Timestamp(TryParseMonotonicIso8601Timestamp),
            expr: Box::new(MirScalarExpr::column(0)),
        };

        let relation = ReprRelationType::new(vec![ReprScalarType::String.nullable(true)]);
        // Test the case where we have full timestamps as bounds.
        let mut interpreter = ColumnSpecs::new(&relation, &arena);
        interpreter.push_column(
            0,
            ResultSpec::value_between(
                Datum::String("2024-01-11T00:00:00.000Z"),
                Datum::String("2024-01-11T20:00:00.000Z"),
            ),
        );

        let timestamp = |ts| {
            Datum::Timestamp(
                NaiveDateTime::parse_from_str(ts, "%Y-%m-%dT%H:%M:%S")
                    .unwrap()
                    .try_into()
                    .unwrap(),
            )
        };

        let range_out = interpreter.expr(&expr).range;
        assert!(!range_out.fallible);
        assert!(range_out.nullable);
        assert!(!range_out.may_contain(timestamp("2024-01-10T10:00:00")));
        assert!(range_out.may_contain(timestamp("2024-01-11T10:00:00")));
        assert!(!range_out.may_contain(timestamp("2024-01-12T10:00:00")));

        // Test the case where we have truncated / useless bounds.
        let mut interpreter = ColumnSpecs::new(&relation, &arena);
        interpreter.push_column(
            0,
            ResultSpec::value_between(Datum::String("2024-01-1"), Datum::String("2024-01-2")),
        );

        let range_out = interpreter.expr(&expr).range;
        assert!(!range_out.fallible);
        assert!(range_out.nullable);
        assert!(range_out.may_contain(timestamp("2024-01-10T10:00:00")));
        assert!(range_out.may_contain(timestamp("2024-01-11T10:00:00")));
        assert!(range_out.may_contain(timestamp("2024-01-12T10:00:00")));

        // Test the case where only one bound is truncated
        let mut interpreter = ColumnSpecs::new(&relation, &arena);
        interpreter.push_column(
            0,
            ResultSpec::value_between(
                Datum::String("2024-01-1"),
                Datum::String("2024-01-12T10:00:00"),
            )
            .union(ResultSpec::null()),
        );

        let range_out = interpreter.expr(&expr).range;
        assert!(!range_out.fallible);
        assert!(range_out.nullable);
        assert!(range_out.may_contain(timestamp("2024-01-10T10:00:00")));
        assert!(range_out.may_contain(timestamp("2024-01-11T10:00:00")));
        assert!(range_out.may_contain(timestamp("2024-01-12T10:00:00")));

        // Test the case where the upper and lower bound are identical
        let mut interpreter = ColumnSpecs::new(&relation, &arena);
        interpreter.push_column(
            0,
            ResultSpec::value_between(
                Datum::String("2024-01-11T10:00:00.000Z"),
                Datum::String("2024-01-11T10:00:00.000Z"),
            ),
        );

        let range_out = interpreter.expr(&expr).range;
        assert!(!range_out.fallible);
        assert!(!range_out.nullable);
        assert!(!range_out.may_contain(timestamp("2024-01-10T10:00:00")));
        assert!(range_out.may_contain(timestamp("2024-01-11T10:00:00")));
        assert!(!range_out.may_contain(timestamp("2024-01-12T10:00:00")));
    }

    #[mz_ore::test]
    fn test_inequality() {
        let arena = RowArena::new();

        let expr = MirScalarExpr::column(0).call_binary(
            MirScalarExpr::CallUnmaterializable(UnmaterializableFunc::MzNow),
            Gte,
        );

        let relation = ReprRelationType::new(vec![ReprScalarType::MzTimestamp.nullable(true)]);
        let mut interpreter = ColumnSpecs::new(&relation, &arena);
        interpreter.push_column(
            0,
            ResultSpec::value_between(
                Datum::MzTimestamp(1704736444949u64.into()),
                Datum::MzTimestamp(1704736444949u64.into()),
            )
            .union(ResultSpec::null()),
        );
        interpreter.push_unmaterializable(
            UnmaterializableFunc::MzNow,
            ResultSpec::value_between(
                Datum::MzTimestamp(1704738791000u64.into()),
                Datum::MzTimestamp(18446744073709551615u64.into()),
            ),
        );

        let range_out = interpreter.expr(&expr).range;
        assert!(
            !range_out.fallible,
            "<= function should not error on non-error input"
        );
        assert!(!range_out.may_contain(Datum::True));
        assert!(range_out.may_contain(Datum::False));
        assert!(range_out.may_contain(Datum::Null));
    }

    /// Regression test for database-issues#9656.
    ///
    /// Adding an `Interval` to a `Timestamp` is non-monotone in the interval
    /// argument: the lex order of intervals (months, days, micros) does not
    /// respect calendar-month arithmetic with day-clamping. The interpreter
    /// must therefore not assume monotonicity, otherwise persist filter
    /// pushdown can incorrectly conclude that a part has no matching rows.
    #[mz_ore::test]
    #[cfg_attr(miri, ignore)]
    fn test_add_timestamp_interval_non_monotone() {
        use chrono::NaiveDateTime;
        use mz_repr::adt::interval::Interval;
        use mz_repr::adt::timestamp::CheckedTimestamp;
        use mz_repr::{Datum, Row};

        let arena = RowArena::new();

        // The setup: a timestamp literal `t = 2024-01-31 00:00:00`, and an
        // interval column whose stats-range spans
        // `[{0 months, 31 days, 0 us}, {1 month, 0 days, 0 us}]`. In lex order,
        // the 31-day interval is the lower bound and the 1-month interval is
        // the upper bound. The function values at the endpoints are:
        //   t + {0,31,0} = 2024-03-02
        //   t + {1, 0,0} = 2024-02-29
        // But an *interior* interval like {0, 60, 0} maps to 2024-03-31, which
        // lies far outside `[Feb 29, Mar 2]`. Under the (incorrect) monotone
        // assumption, the interpreter would conclude the output is in that
        // narrow window, and rule out predicates like `>= 2024-03-15`.
        let ts_lit = |s: &str| {
            let mut row = Row::default();
            row.packer().push(Datum::Timestamp(
                CheckedTimestamp::from_timestamplike(
                    NaiveDateTime::parse_from_str(s, "%Y-%m-%dT%H:%M:%S").unwrap(),
                )
                .unwrap(),
            ));
            MirScalarExpr::Literal(Ok(row), ReprScalarType::Timestamp.nullable(false))
        };
        let interval = |months: i32, days: i32, micros: i64| {
            Datum::Interval(Interval {
                months,
                days,
                micros,
            })
        };

        // Expression: `(timestamp_lit + interval_col) >= 2024-03-15`.
        let expr = ts_lit("2024-01-31T00:00:00")
            .call_binary(MirScalarExpr::column(0), AddTimestampInterval)
            .call_binary(ts_lit("2024-03-15T00:00:00"), Gte);

        let relation = ReprRelationType::new(vec![ReprScalarType::Interval.nullable(false)]);
        let mut interpreter = ColumnSpecs::new(&relation, &arena);
        interpreter.push_column(
            0,
            ResultSpec::value_between(interval(0, 31, 0), interval(1, 0, 0)),
        );

        let range_out = interpreter.expr(&expr).range;
        // The actual data may include e.g. `{0, 60, 0}` → 2024-03-31, which
        // satisfies `>= 2024-03-15`. The interpreter must admit `True` so that
        // filter pushdown does not skip the part. Under the buggy
        // `(true, true)` annotation, the output range would be
        // `[Feb 29, Mar 2]`, all of which is `< Mar 15`, and the interpreter
        // would (wrongly) admit only `False`.
        assert!(
            range_out.may_contain(Datum::True),
            "interpreter incorrectly ruled out matching rows; \
             add_timestamp_interval is not monotone in the interval argument",
        );
    }

    /// Companion test to `test_add_timestamp_interval_non_monotone`: when the
    /// interval argument is a literal with `months == 0`, the function reduces
    /// to a pure linear shift in microseconds and *is* monotone in the
    /// timestamp. The dynamic-monotonicity handler in `AbstractFunc` should
    /// recover the tight output range in that case, so that filter pushdown
    /// can still narrow predicates like `t - INTERVAL '1' day < literal`.
    #[mz_ore::test]
    #[cfg_attr(miri, ignore)]
    fn test_timestamp_plus_interval_dynamic_monotone() {
        use chrono::NaiveDateTime;
        use mz_repr::adt::interval::Interval;
        use mz_repr::adt::timestamp::CheckedTimestamp;
        use mz_repr::{Datum, Row};

        let arena = RowArena::new();

        let ts = |s: &str| {
            Datum::Timestamp(
                CheckedTimestamp::from_timestamplike(
                    NaiveDateTime::parse_from_str(s, "%Y-%m-%dT%H:%M:%S").unwrap(),
                )
                .unwrap(),
            )
        };
        let interval_lit = |months: i32, days: i32, micros: i64| {
            let mut row = Row::default();
            row.packer().push(Datum::Interval(Interval {
                months,
                days,
                micros,
            }));
            MirScalarExpr::Literal(Ok(row), ReprScalarType::Interval.nullable(false))
        };

        let relation = ReprRelationType::new(vec![ReprScalarType::Timestamp.nullable(false)]);

        // (a) `t_col - INTERVAL '1' day < 2024-01-15`, with `t_col` ranging
        // over `[2024-01-15, 2024-01-20]`. With the days-only interval, the
        // subtraction is monotone, so endpoints alone determine the output:
        // [2024-01-14, 2024-01-19]. Only `2024-01-14` satisfies `< 2024-01-15`,
        // so both True and False are reachable.
        {
            let expr = MirScalarExpr::column(0)
                .call_binary(interval_lit(0, 1, 0), SubTimestampInterval)
                .call_binary(
                    MirScalarExpr::Literal(
                        Ok({
                            let mut r = Row::default();
                            r.packer().push(ts("2024-01-15T00:00:00"));
                            r
                        }),
                        ReprScalarType::Timestamp.nullable(false),
                    ),
                    Lt,
                );
            let mut interpreter = ColumnSpecs::new(&relation, &arena);
            interpreter.push_column(
                0,
                ResultSpec::value_between(ts("2024-01-15T00:00:00"), ts("2024-01-20T00:00:00")),
            );
            let range_out = interpreter.expr(&expr).range;
            assert!(
                range_out.may_contain(Datum::True),
                "day-only interval should preserve tight bounds",
            );
            assert!(
                range_out.may_contain(Datum::False),
                "day-only interval should preserve tight bounds",
            );
        }

        // (b) Same predicate, but with `t_col` strictly *after* the literal:
        // `[2024-01-17, 2024-01-20]`. Output of `t - 1 day`:
        // `[2024-01-16, 2024-01-19]`, none of which is `< 2024-01-15`. The
        // interpreter must rule out `True`.
        {
            let expr = MirScalarExpr::column(0)
                .call_binary(interval_lit(0, 1, 0), SubTimestampInterval)
                .call_binary(
                    MirScalarExpr::Literal(
                        Ok({
                            let mut r = Row::default();
                            r.packer().push(ts("2024-01-15T00:00:00"));
                            r
                        }),
                        ReprScalarType::Timestamp.nullable(false),
                    ),
                    Lt,
                );
            let mut interpreter = ColumnSpecs::new(&relation, &arena);
            interpreter.push_column(
                0,
                ResultSpec::value_between(ts("2024-01-17T00:00:00"), ts("2024-01-20T00:00:00")),
            );
            let range_out = interpreter.expr(&expr).range;
            assert!(
                !range_out.may_contain(Datum::True),
                "day-only interval should narrow out impossible matches",
            );
        }

        // (c) With a *month*-bearing literal interval, the operation is no
        // longer monotone (day-clamping), so the dynamic-monotonicity handler
        // must fall back to `anything()` — the interpreter cannot rule out
        // either outcome even when the column range is narrow.
        {
            let expr = MirScalarExpr::column(0)
                .call_binary(interval_lit(1, 0, 0), SubTimestampInterval)
                .call_binary(
                    MirScalarExpr::Literal(
                        Ok({
                            let mut r = Row::default();
                            r.packer().push(ts("2024-01-15T00:00:00"));
                            r
                        }),
                        ReprScalarType::Timestamp.nullable(false),
                    ),
                    Lt,
                );
            let mut interpreter = ColumnSpecs::new(&relation, &arena);
            interpreter.push_column(
                0,
                ResultSpec::value_between(ts("2024-01-17T00:00:00"), ts("2024-01-20T00:00:00")),
            );
            let range_out = interpreter.expr(&expr).range;
            assert!(
                range_out.may_contain(Datum::True),
                "month-bearing interval must conservatively admit True",
            );
            assert!(
                range_out.may_contain(Datum::False),
                "month-bearing interval must conservatively admit False",
            );
        }
    }

    /// Proptest companion to [`test_timestamp_plus_interval_dynamic_monotone`]:
    /// the dynamic-monotonicity handler in [`AbstractFunc`] claims that
    /// `add_timestamp_interval(t, i)` is monotone in `t` whenever `i.months == 0`
    /// (the only case it actually claims monotonicity for at runtime: the
    /// matches above require the right argument to be a single value with
    /// `months == 0`). This proptest verifies that claim directly against the
    /// function impl by sampling random timestamps and zero-month intervals
    /// and checking that input ordering is preserved in the output.
    #[mz_ore::test]
    #[cfg_attr(miri, ignore)]
    fn proptest_timestamp_plus_interval_monotone_when_months_zero() {
        use mz_repr::adt::interval::Interval;
        use mz_repr::{Datum, RowArena, SqlScalarType, arb_datum_for_scalar};
        use proptest::prelude::*;

        let timestamp_strat = || arb_datum_for_scalar(SqlScalarType::Timestamp { precision: None });
        // Lex order on `Interval` does *not* match total-microseconds order when
        // both days and micros vary independently (e.g. `{0, 0, 86_400_000_001}`
        // is lex-less than `{0, 1, 0}` but evaluates to a strictly larger
        // timestamp), so we only claim monotonicity for *fixed* zero-month
        // intervals — which is exactly what the DynamicMonotone handler does.
        // The proptest accordingly varies `t` with `i` held constant.
        let zero_month_interval_strat =
            (any::<i32>(), any::<i64>()).prop_map(|(days, micros)| Interval {
                months: 0,
                days,
                micros,
            });

        let expr = MirScalarExpr::CallBinary {
            func: AddTimestampInterval.into(),
            expr1: Box::new(MirScalarExpr::column(0)),
            expr2: Box::new(MirScalarExpr::column(1)),
        };
        let arena = RowArena::new();

        proptest!(|(
            t1 in timestamp_strat(),
            t2 in timestamp_strat(),
            i in zero_month_interval_strat,
        )| {
            let t1 = match t1 { PropDatum::Timestamp(t) => t, _ => unreachable!() };
            let t2 = match t2 { PropDatum::Timestamp(t) => t, _ => unreachable!() };
            let i = Datum::Interval(i);
            let r1 = expr.eval(&[Datum::Timestamp(t1), i], &arena);
            let r2 = expr.eval(&[Datum::Timestamp(t2), i], &arena);
            // Only compare when both calls succeed; the monotonicity claim
            // applies only within the success domain.
            if let (Ok(Datum::Timestamp(r1)), Ok(Datum::Timestamp(r2))) = (r1, r2) {
                prop_assert_eq!(t1.cmp(&t2), r1.cmp(&r2));
            }
        });
    }

    /// Regression test for `date_bin_timestamp`, which is non-monotone in the
    /// `stride` argument: a larger stride can bin a source timestamp to an
    /// *earlier* result than a smaller stride, because the bin alignment to
    /// the unix epoch depends on the stride magnitude rather than on lex order.
    #[mz_ore::test]
    #[cfg_attr(miri, ignore)]
    fn test_date_bin_timestamp_non_monotone() {
        use chrono::NaiveDateTime;
        use mz_repr::adt::interval::Interval;
        use mz_repr::adt::timestamp::CheckedTimestamp;
        use mz_repr::{Datum, Row};

        let arena = RowArena::new();

        let ts_lit = |s: &str| {
            let mut row = Row::default();
            row.packer().push(Datum::Timestamp(
                CheckedTimestamp::from_timestamplike(
                    NaiveDateTime::parse_from_str(s, "%Y-%m-%dT%H:%M:%S").unwrap(),
                )
                .unwrap(),
            ));
            MirScalarExpr::Literal(Ok(row), ReprScalarType::Timestamp.nullable(false))
        };
        let interval = |months: i32, days: i32, micros: i64| {
            Datum::Interval(Interval {
                months,
                days,
                micros,
            })
        };

        // Expression: `date_bin(stride_col, 2024-01-01 12:00:00) > 2024-01-01 06:00:00`.
        // stride_col ranges over `[1 day, 2 days]`.
        //
        // Endpoint evaluations:
        //   1 day stride → bins to 2024-01-01 00:00:00
        //   2 day stride → bins to 2023-12-31 00:00:00
        //
        // Interior strides produce results *outside* that endpoint box. For
        // example, a 1.5-day stride (i.e. `{0 months, 1 day, 12 h micros}`,
        // which sorts between the two endpoints in lex order) bins
        // 2024-01-01 12:00:00 to exactly 2024-01-01 12:00:00 — well above the
        // endpoint maximum of 2024-01-01 00:00:00. With the buggy
        // `(true, true)` annotation, the interpreter narrows the output to
        // `[Dec 31 00:00, Jan 1 00:00]`, both of which are `<= Jan 1 06:00`,
        // so the predicate is wrongly proved `False`. With the non-monotone
        // fix the output is `anything()`, so `True` is correctly admitted.
        let expr = MirScalarExpr::column(0)
            .call_binary(ts_lit("2024-01-01T12:00:00"), DateBinTimestamp)
            .call_binary(ts_lit("2024-01-01T06:00:00"), Gt);

        let relation = ReprRelationType::new(vec![ReprScalarType::Interval.nullable(false)]);
        let mut interpreter = ColumnSpecs::new(&relation, &arena);
        interpreter.push_column(
            0,
            ResultSpec::value_between(interval(0, 1, 0), interval(0, 2, 0)),
        );

        let range_out = interpreter.expr(&expr).range;
        assert!(
            range_out.may_contain(Datum::True),
            "date_bin is not monotone in the stride argument; \
             interior strides can produce outputs outside the endpoint-bounded \
             box, so the interpreter must admit True for `>`-style predicates",
        );
    }

    #[mz_ore::test]
    fn test_trace() {
        use super::Trace;

        let expr = MirScalarExpr::column(0).call_binary(
            MirScalarExpr::column(1)
                .call_binary(MirScalarExpr::column(3).call_unary(NegInt64), AddInt64),
            Gte,
        );
        let summary = Trace.expr(&expr);
        assert!(summary.pushdownable());
    }
}