mz_expr/scalar/

func.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.
//
// Portions of this file are derived from the PostgreSQL project. The original
// source code is subject to the terms of the PostgreSQL license, a copy of
// which can be found in the LICENSE file at the root of this repository.

use std::borrow::Cow;
use std::cmp::Ordering;
use std::convert::{TryFrom, TryInto};
use std::str::FromStr;
use std::{iter, str};

use ::encoding::DecoderTrap;
use ::encoding::label::encoding_from_whatwg_label;
use aws_lc_rs::constant_time::verify_slices_are_equal;
use aws_lc_rs::digest;
use chrono::{DateTime, Duration, NaiveDate, NaiveDateTime, TimeZone, Timelike, Utc};
use chrono_tz::{OffsetComponents, OffsetName, Tz};
use dec::OrderedDecimal;
use itertools::Itertools;
use md5::{Digest, Md5};
use mz_expr_derive::sqlfunc;
use mz_ore::cast::{self, CastFrom};
use mz_ore::fmt::FormatBuffer;
use mz_ore::lex::LexBuf;
use mz_ore::option::OptionExt;
use mz_pgrepr::Type;
use mz_pgtz::timezone::{Timezone, TimezoneSpec};
use mz_repr::adt::array::{Array, ArrayDimension};
use mz_repr::adt::date::Date;
use mz_repr::adt::interval::{Interval, RoundBehavior};
use mz_repr::adt::jsonb::JsonbRef;
use mz_repr::adt::mz_acl_item::{AclMode, MzAclItem};
use mz_repr::adt::numeric::{self, Numeric};
use mz_repr::adt::range::Range;
use mz_repr::adt::regex::Regex;
use mz_repr::adt::timestamp::{CheckedTimestamp, TimestampLike};
use mz_repr::{
    ArrayRustType, Datum, DatumList, DatumMap, ExcludeNull, FromDatum, InputDatumType, Row,
    RowArena, SqlScalarType, strconv,
};
use mz_sql_parser::ast::display::{AstDisplay, FormatMode};
use mz_sql_pretty::{PrettyConfig, pretty_str};
use num::traits::CheckedNeg;

use crate::scalar::func::format::DateTimeFormat;
use crate::{EvalError, like_pattern};

#[macro_use]
mod macros;
mod binary;
mod encoding;
pub(crate) mod format;
pub(crate) mod impls;
mod unary;
mod unmaterializable;
pub mod variadic;

pub use binary::BinaryFunc;
pub use impls::*;
pub use unary::{EagerUnaryFunc, LazyUnaryFunc, UnaryFunc};
pub use unmaterializable::UnmaterializableFunc;
pub use variadic::VariadicFunc;

/// The canonical name of a scalar function.
///
/// For functions generated by `#[sqlfunc]` this is the name of the underlying
/// Rust function. Hand-written functions declare it explicitly. Test tooling
/// refers to exact function variants by these names, via
/// [`UnaryFunc::from_variant_name`] and its binary and variadic counterparts.
pub trait FuncName {
    const NAME: &'static str;
}

/// Declares the canonical [`FuncName`] of hand-written function structs.
/// Functions generated by `#[sqlfunc]` get theirs from that macro instead.
/// Structs generic over an expression type declare a leading `<E>`.
macro_rules! func_name {
    () => {};
    (<$param:ident> $ty:ty => $name:literal, $($rest:tt)*) => {
        impl<$param> FuncName for $ty {
            const NAME: &'static str = $name;
        }
        func_name!($($rest)*);
    };
    ($ty:ty => $name:literal, $($rest:tt)*) => {
        impl FuncName for $ty {
            const NAME: &'static str = $name;
        }
        func_name!($($rest)*);
    };
}

func_name! {
    AdjustNumericScale => "adjust_numeric_scale",
    AdjustTimestampPrecision => "adjust_timestamp_precision",
    AdjustTimestampTzPrecision => "adjust_timestamp_tz_precision",
    CaseLiteral => "case_literal",
    <E> CastArrayToArray<E> => "cast_array_to_array",
    <E> CastArrayToJsonb<E> => "cast_array_to_jsonb",
    CastArrayToString => "cast_array_to_string",
    CastDateToTimestamp => "cast_date_to_timestamp",
    CastDateToTimestampTz => "cast_date_to_timestamp_tz",
    CastFloat32ToNumeric => "cast_float32_to_numeric",
    CastFloat64ToNumeric => "cast_float64_to_numeric",
    CastInt16ToNumeric => "cast_int16_to_numeric",
    CastInt32ToNumeric => "cast_int32_to_numeric",
    CastInt64ToNumeric => "cast_int64_to_numeric",
    CastJsonbToNumeric => "cast_jsonb_to_numeric",
    <E> CastList1ToList2<E> => "cast_list1_to_list2",
    <E> CastListToJsonb<E> => "cast_list_to_jsonb",
    CastListToString => "cast_list_to_string",
    CastMapToString => "cast_map_to_string",
    CastRangeToString => "cast_range_to_string",
    <E> CastRecord1ToRecord2<E> => "cast_record1_to_record2",
    CastRecordToString => "cast_record_to_string",
    <E> CastStringToArray<E> => "cast_string_to_array",
    CastStringToChar => "cast_string_to_char",
    CastStringToInt2Vector => "cast_string_to_int2_vector",
    <E> CastStringToList<E> => "cast_string_to_list",
    <E> CastStringToMap<E> => "cast_string_to_map",
    CastStringToNumeric => "cast_string_to_numeric",
    <E> CastStringToRange<E> => "cast_string_to_range",
    CastStringToTimestamp => "cast_string_to_timestamp",
    CastStringToTimestampTz => "cast_string_to_timestamp_tz",
    CastStringToVarChar => "cast_string_to_var_char",
    CastTimestampToTimestampTz => "cast_timestamp_to_timestamp_tz",
    CastTimestampTzToTimestamp => "cast_timestamp_tz_to_timestamp",
    CastUint16ToNumeric => "cast_uint16_to_numeric",
    CastUint32ToNumeric => "cast_uint32_to_numeric",
    CastUint64ToNumeric => "cast_uint64_to_numeric",
    DatePartInterval => "date_part_interval",
    DatePartTime => "date_part_time",
    DatePartTimestamp => "date_part_timestamp",
    DatePartTimestampTz => "date_part_timestamp_tz",
    DateTruncTimestamp => "date_trunc_timestamp",
    DateTruncTimestampTz => "date_trunc_timestamp_tz",
    variadic::ErrorIfNull => "error_if_null",
    ExtractDate => "extract_date",
    ExtractInterval => "extract_interval",
    ExtractTime => "extract_time",
    ExtractTimestamp => "extract_timestamp",
    ExtractTimestampTz => "extract_timestamp_tz",
    IsLikeMatch => "is_like_match",
    IsRegexpMatch => "is_regexp_match",
    ListLengthMax => "list_length_max",
    MapBuildFromRecordList => "map_build_from_record_list",
    PadChar => "pad_char",
    variadic::RangeCreate => "range_create",
    RecordGet => "record_get",
    RegexpMatch => "regexp_match",
    RegexpReplace => "regexp_replace",
    RegexpSplitToArray => "regexp_split_to_array",
    TimezoneTime => "timezone_time",
    TimezoneTimestamp => "timezone_timestamp",
    TimezoneTimestampTz => "timezone_timestamp_tz",
    ToCharTimestamp => "to_char_timestamp",
    ToCharTimestampTz => "to_char_timestamp_tz",
    variadic::And => "and",
    variadic::Coalesce => "coalesce",
    variadic::Greatest => "greatest",
    variadic::Least => "least",
    variadic::Or => "or",
}

/// The maximum size of the result strings of certain string functions, such as `repeat` and `lpad`.
/// Chosen to be the smallest number to keep our tests passing without changing. 100MiB is probably
/// higher than what we want, but it's better than no limit.
///
/// Note: This number appears in our user-facing documentation in the function reference for every
/// function where it applies.
pub const MAX_STRING_FUNC_RESULT_BYTES: usize = 1024 * 1024 * 100;

/// The largest result a string function may build into `temp_storage`.
///
/// [`MAX_STRING_FUNC_RESULT_BYTES`] unless the arena carries a tighter budget, which is how an
/// expression evaluated in `environmentd` on behalf of a request (a webhook `CHECK`) is held to a
/// size proportionate to that request rather than to the constant, which is sized for a cluster.
///
/// A function that can predict its result size must consult this *before* building the result: the
/// arena's own budget is only observable after the bytes exist, which for an amplifying function is
/// exactly too late.
pub fn max_string_func_result_bytes(temp_storage: &RowArena) -> usize {
    std::cmp::min(
        MAX_STRING_FUNC_RESULT_BYTES,
        temp_storage.budget_remaining(),
    )
}

/// Refuses a collection `temp_storage`'s budget cannot afford, before it is packed.
///
/// The collection builders (`ARRAY[..]`, `LIST[..]`, `ROW(..)`, `MAP[..]`, `jsonb_build_*`) pack the
/// datums they are handed straight into `temp_storage`, so the result is as large as those datums
/// times however many times the expression names each one. Unlike the string functions bounded by
/// [`max_string_func_result_bytes`], `ARRAY[body, body, ..]` has no ceiling of its own.
///
/// Like that ceiling, this must be consulted *before* the result is built, since the arena's budget
/// is only observable once the bytes exist. [`mz_repr::datum_size`] is the size a datum occupies
/// once packed, so summing it bounds the result without allocating anything. Without a budget
/// `budget_remaining` is `usize::MAX` and nothing is refused.
pub fn check_datums_fit_budget<'a>(
    datums: impl IntoIterator<Item = Datum<'a>>,
    temp_storage: &RowArena,
) -> Result<(), EvalError> {
    let need: usize = datums
        .into_iter()
        .map(|d| mz_repr::datum_size(&d))
        .fold(0, usize::saturating_add);
    if need > temp_storage.budget_remaining() {
        return Err(EvalError::TempStorageBudgetExceeded);
    }
    Ok(())
}

/// Refuses an input-scaled transient the arena's budget cannot afford, before it is built.
///
/// A function that gathers `n_elems` of `elem_size` bytes into its own `Vec` before packing them
/// holds that allocation on the stack, where the arena never sees it. The evaluator's post-call
/// check counts arena bytes only, so a transient wider than the result it becomes can slip past a
/// budget the packed result fits under. A `Vec<&str>` of split chunks is one: a 16-byte fat pointer
/// per chunk against the ~2 bytes an empty chunk packs to. Pair this with
/// [`check_datums_fit_budget`], which bounds the packed result.
///
/// `n_elems` is a closure so an unbudgeted arena, which is every arena in a dataflow, never pays for
/// a count that can cost a pass over the input.
///
/// NOTE: the reformatters (`jsonb_pretty`, `pretty_sql`, `redact_sql`) are the known exception.
/// Sizing their output needs the same walk that produces it, so only the post-call check bounds
/// them.
pub fn check_build_fits_budget(
    n_elems: impl FnOnce() -> usize,
    elem_size: usize,
    temp_storage: &RowArena,
) -> Result<(), EvalError> {
    let budget_remaining = temp_storage.budget_remaining();
    // `usize::MAX` is the unbudgeted sentinel. Return before the count so an unbudgeted arena never
    // pays for it.
    if budget_remaining == usize::MAX {
        return Ok(());
    }
    if n_elems().saturating_mul(elem_size) > budget_remaining {
        return Err(EvalError::TempStorageBudgetExceeded);
    }
    Ok(())
}

pub fn jsonb_stringify<'a>(a: Datum<'a>, temp_storage: &'a RowArena) -> Option<&'a str> {
    match a {
        Datum::JsonNull => None,
        Datum::String(s) => Some(s),
        _ => {
            let s = cast_jsonb_to_string(JsonbRef::from_datum(a));
            Some(temp_storage.push_string(s))
        }
    }
}

#[sqlfunc(
    is_monotone = "(true, true)",
    is_infix_op = true,
    sqlname = "+",
    propagates_nulls = true
)]
fn add_int16(a: i16, b: i16) -> Result<i16, EvalError> {
    a.checked_add(b).ok_or(EvalError::NumericFieldOverflow)
}

#[sqlfunc(
    is_monotone = "(true, true)",
    is_infix_op = true,
    sqlname = "+",
    propagates_nulls = true
)]
fn add_int32(a: i32, b: i32) -> Result<i32, EvalError> {
    a.checked_add(b).ok_or(EvalError::NumericFieldOverflow)
}

#[sqlfunc(
    is_monotone = "(true, true)",
    is_infix_op = true,
    sqlname = "+",
    propagates_nulls = true
)]
fn add_int64(a: i64, b: i64) -> Result<i64, EvalError> {
    a.checked_add(b).ok_or(EvalError::NumericFieldOverflow)
}

#[sqlfunc(
    is_monotone = "(true, true)",
    is_infix_op = true,
    sqlname = "+",
    propagates_nulls = true
)]
fn add_uint16(a: u16, b: u16) -> Result<u16, EvalError> {
    a.checked_add(b)
        .ok_or_else(|| EvalError::UInt16OutOfRange(format!("{a} + {b}").into()))
}

#[sqlfunc(
    is_monotone = "(true, true)",
    is_infix_op = true,
    sqlname = "+",
    propagates_nulls = true
)]
fn add_uint32(a: u32, b: u32) -> Result<u32, EvalError> {
    a.checked_add(b)
        .ok_or_else(|| EvalError::UInt32OutOfRange(format!("{a} + {b}").into()))
}

#[sqlfunc(
    is_monotone = "(true, true)",
    is_infix_op = true,
    sqlname = "+",
    propagates_nulls = true
)]
fn add_uint64(a: u64, b: u64) -> Result<u64, EvalError> {
    a.checked_add(b)
        .ok_or_else(|| EvalError::UInt64OutOfRange(format!("{a} + {b}").into()))
}

#[sqlfunc(
    is_monotone = "(true, true)",
    is_infix_op = true,
    sqlname = "+",
    propagates_nulls = true
)]
fn add_float32(a: f32, b: f32) -> Result<f32, EvalError> {
    let sum = a + b;
    if sum.is_infinite() && !a.is_infinite() && !b.is_infinite() {
        Err(EvalError::FloatOverflow)
    } else {
        Ok(sum)
    }
}

#[sqlfunc(
    is_monotone = "(true, true)",
    is_infix_op = true,
    sqlname = "+",
    propagates_nulls = true
)]
fn add_float64(a: f64, b: f64) -> Result<f64, EvalError> {
    let sum = a + b;
    if sum.is_infinite() && !a.is_infinite() && !b.is_infinite() {
        Err(EvalError::FloatOverflow)
    } else {
        Ok(sum)
    }
}

// `Interval` is lex-ordered (months, days, micros), but adding an interval to a
// timestamp adds *calendar* months (with day-clamping) which does not respect
// that ordering: e.g. `i1 = {0 months, 31 days}` is lex-less than
// `i2 = {1 month, 0 days}`, but `2024-01-31 + i1 = 2024-03-02` is greater than
// `2024-01-31 + i2 = 2024-02-29`. Day-clamping plus preserved sub-day time also
// breaks monotonicity in the first argument near month boundaries.
#[sqlfunc(is_monotone = "(false, false)", is_infix_op = true, sqlname = "+")]
fn add_timestamp_interval(
    a: CheckedTimestamp<NaiveDateTime>,
    b: Interval,
) -> Result<CheckedTimestamp<NaiveDateTime>, EvalError> {
    add_timestamplike_interval(a, b)
}

#[sqlfunc(is_monotone = "(false, false)", is_infix_op = true, sqlname = "+")]
fn add_timestamp_tz_interval(
    a: CheckedTimestamp<DateTime<Utc>>,
    b: Interval,
) -> Result<CheckedTimestamp<DateTime<Utc>>, EvalError> {
    add_timestamplike_interval(a, b)
}

fn add_timestamplike_interval<T>(
    a: CheckedTimestamp<T>,
    b: Interval,
) -> Result<CheckedTimestamp<T>, EvalError>
where
    T: TimestampLike,
{
    let dt = a.date_time();
    let dt = add_timestamp_months(&dt, b.months)?;
    let dt = dt
        .checked_add_signed(b.duration_as_chrono())
        .ok_or(EvalError::TimestampOutOfRange)?;
    Ok(CheckedTimestamp::from_timestamplike(T::from_date_time(dt))?)
}

// See `add_timestamp_interval` for why this is not monotone.
#[sqlfunc(is_monotone = "(false, false)", is_infix_op = true, sqlname = "-")]
fn sub_timestamp_interval(
    a: CheckedTimestamp<NaiveDateTime>,
    b: Interval,
) -> Result<CheckedTimestamp<NaiveDateTime>, EvalError> {
    sub_timestamplike_interval(a, b)
}

#[sqlfunc(is_monotone = "(false, false)", is_infix_op = true, sqlname = "-")]
fn sub_timestamp_tz_interval(
    a: CheckedTimestamp<DateTime<Utc>>,
    b: Interval,
) -> Result<CheckedTimestamp<DateTime<Utc>>, EvalError> {
    sub_timestamplike_interval(a, b)
}

fn sub_timestamplike_interval<T>(
    a: CheckedTimestamp<T>,
    b: Interval,
) -> Result<CheckedTimestamp<T>, EvalError>
where
    T: TimestampLike,
{
    neg_interval_inner(b).and_then(|i| add_timestamplike_interval(a, i))
}

#[sqlfunc(is_monotone = "(true, true)", is_infix_op = true, sqlname = "+")]
fn add_date_time(
    date: Date,
    time: chrono::NaiveTime,
) -> Result<CheckedTimestamp<NaiveDateTime>, EvalError> {
    // A leap-second TIME (nanos >= 1e9) rolls over into the next minute,
    // matching what parsing the equivalent timestamp literal produces. The
    // leap representation must not enter a timestamp: it sorts before the
    // next second while epoch-style conversions count it at or past it,
    // breaking the monotonicity contracts filter pushdown relies on.
    let (extra_sec, nanos) = match time.nanosecond().checked_sub(1_000_000_000) {
        Some(nanos) => (1, nanos),
        None => (0, time.nanosecond()),
    };
    let dt = NaiveDate::from(date)
        .and_hms_nano_opt(time.hour(), time.minute(), time.second(), nanos)
        .unwrap()
        .checked_add_signed(chrono::Duration::try_seconds(extra_sec).unwrap())
        .ok_or(EvalError::TimestampOutOfRange)?;
    Ok(CheckedTimestamp::from_timestamplike(dt)?)
}

// Monotone in `date` (dates have no sub-day component, so day-clamping at month
// boundaries only causes results to collapse, never to reverse), but not in
// `interval`: e.g. `{0 months, 31 days}` is lex-less than `{1 month, 0 days}`,
// but adding the former to `2024-01-31` gives `2024-03-02` while the latter
// gives `2024-02-29`.
#[sqlfunc(is_monotone = "(true, false)", is_infix_op = true, sqlname = "+")]
fn add_date_interval(
    date: Date,
    interval: Interval,
) -> Result<CheckedTimestamp<NaiveDateTime>, EvalError> {
    let dt = NaiveDate::from(date).and_hms_opt(0, 0, 0).unwrap();
    let dt = add_timestamp_months(&dt, interval.months)?;
    let dt = dt
        .checked_add_signed(interval.duration_as_chrono())
        .ok_or(EvalError::TimestampOutOfRange)?;
    Ok(CheckedTimestamp::from_timestamplike(dt)?)
}

#[sqlfunc(
    // <time> + <interval> wraps!
    is_monotone = "(false, false)",
    is_infix_op = true,
    sqlname = "+",
    propagates_nulls = true
)]
fn add_time_interval(time: chrono::NaiveTime, interval: Interval) -> chrono::NaiveTime {
    let (t, _) = time.overflowing_add_signed(interval.duration_as_chrono());
    t
}

#[sqlfunc(
    is_monotone = "(true, false)",
    output_type = "Numeric",
    sqlname = "round",
    propagates_nulls = true
)]
fn round_numeric_binary(a: OrderedDecimal<Numeric>, mut b: i32) -> Result<Numeric, EvalError> {
    let mut a = a.0;
    let mut cx = numeric::cx_datum();
    let a_scale = numeric::get_scale(&a);
    if a.is_finite() && i64::from(b) > i64::from(a_scale) {
        // Rounding at or past `a`'s scale cannot change the value: right-pad
        // with zeroes via rescale. The rounding path below shifts left by `b`
        // first and overflows for large `b`.
        //
        // NOTE: Equal values can reach here with different scales, since
        // `Row` encoding folds trailing zeroes into the exponent. The result,
        // value or error, must not depend on which representation arrives: the
        // abstract interpreter reads its datums back out of a `Row`, and if it
        // calls infallible what the evaluator fails on, persist filter
        // pushdown discards parts it has to keep.
        //
        // `Infinity` and `NaN` report a scale of zero, but `rescale` on an
        // infinity yields `NaN` via invalid_operation, not the overflow
        // checked below, so the specials take the rounding path, which
        // propagates them unchanged, as PostgreSQL does.

        // Ensure rescale doesn't exceed max precision by putting a ceiling on
        // b equal to the maximum remaining scale the value can support.
        let max_remaining_scale = u32::from(numeric::NUMERIC_DATUM_MAX_PRECISION)
            - (numeric::get_precision(&a) - a_scale);
        b = match i32::try_from(max_remaining_scale) {
            Ok(max_remaining_scale) => std::cmp::min(b, max_remaining_scale),
            Err(_) => b,
        };
        cx.rescale(&mut a, &numeric::Numeric::from(-b));
    } else {
        // To avoid invalid operations, clamp b to be within 1 more than the
        // precision limit.
        const MAX_P_LIMIT: i32 = 1 + cast::u8_to_i32(numeric::NUMERIC_DATUM_MAX_PRECISION);
        b = std::cmp::min(MAX_P_LIMIT, b);
        b = std::cmp::max(-MAX_P_LIMIT, b);
        let mut b = numeric::Numeric::from(b);
        // Shift by 10^b; this put digit to round to in the one's place.
        cx.scaleb(&mut a, &b);
        cx.round(&mut a);
        // Negate exponent for shift back
        cx.neg(&mut b);
        cx.scaleb(&mut a, &b);
    }

    if cx.status().overflow() {
        Err(EvalError::FloatOverflow)
    } else if a.is_zero() {
        // simpler than handling cases where exponent has gotten set to some
        // value greater than the max precision, but all significant digits
        // were rounded away.
        Ok(numeric::Numeric::zero())
    } else {
        numeric::munge_numeric(&mut a).unwrap();
        Ok(a)
    }
}

#[sqlfunc(sqlname = "convert_from", propagates_nulls = true)]
fn convert_from<'a>(a: &'a [u8], b: &str) -> Result<&'a str, EvalError> {
    // Convert PostgreSQL-style encoding names[1] to WHATWG-style encoding names[2],
    // which the encoding library uses[3].
    // [1]: https://www.postgresql.org/docs/9.5/multibyte.html
    // [2]: https://encoding.spec.whatwg.org/
    // [3]: https://github.com/lifthrasiir/rust-encoding/blob/4e79c35ab6a351881a86dbff565c4db0085cc113/src/label.rs
    let encoding_name = b.to_lowercase().replace('_', "-").into_boxed_str();

    // Supporting other encodings is tracked by database-issues#797.
    if encoding_from_whatwg_label(&encoding_name).map(|e| e.name()) != Some("utf-8") {
        return Err(EvalError::InvalidEncodingName(encoding_name));
    }

    match str::from_utf8(a) {
        // Match PostgreSQL, which rejects NUL bytes because text values must
        // never contain them.
        Ok(from) if from.contains('\0') => Err(EvalError::InvalidByteSequence {
            byte_sequence: "0x00".into(),
            encoding_name,
        }),
        Ok(from) => Ok(from),
        Err(e) => Err(EvalError::InvalidByteSequence {
            byte_sequence: e.to_string().into(),
            encoding_name,
        }),
    }
}

#[sqlfunc]
fn encode(bytes: &[u8], format: &str) -> Result<String, EvalError> {
    let format = encoding::lookup_format(format)?;
    Ok(format.encode(bytes))
}

#[sqlfunc]
fn decode(string: &str, format: &str, temp_storage: &RowArena) -> Result<Vec<u8>, EvalError> {
    let format = encoding::lookup_format(format)?;
    let out = format.decode(string)?;
    if out.len() > max_string_func_result_bytes(temp_storage) {
        Err(EvalError::LengthTooLarge)
    } else {
        Ok(out)
    }
}

#[sqlfunc(sqlname = "length", propagates_nulls = true)]
fn encoded_bytes_char_length(a: &[u8], b: &str) -> Result<i32, EvalError> {
    // Convert PostgreSQL-style encoding names[1] to WHATWG-style encoding names[2],
    // which the encoding library uses[3].
    // [1]: https://www.postgresql.org/docs/9.5/multibyte.html
    // [2]: https://encoding.spec.whatwg.org/
    // [3]: https://github.com/lifthrasiir/rust-encoding/blob/4e79c35ab6a351881a86dbff565c4db0085cc113/src/label.rs
    let encoding_name = b.to_lowercase().replace('_', "-").into_boxed_str();

    let enc = match encoding_from_whatwg_label(&encoding_name) {
        Some(enc) => enc,
        None => return Err(EvalError::InvalidEncodingName(encoding_name)),
    };

    let decoded_string = match enc.decode(a, DecoderTrap::Strict) {
        Ok(s) => s,
        Err(e) => {
            return Err(EvalError::InvalidByteSequence {
                byte_sequence: e.into(),
                encoding_name,
            });
        }
    };

    let count = decoded_string.chars().count();
    i32::try_from(count).map_err(|_| EvalError::Int32OutOfRange(count.to_string().into()))
}

// TODO(benesch): remove potentially dangerous usage of `as`.
#[allow(clippy::as_conversions)]
pub fn add_timestamp_months<T: TimestampLike>(
    dt: &T,
    mut months: i32,
) -> Result<CheckedTimestamp<T>, EvalError> {
    if months == 0 {
        return Ok(CheckedTimestamp::from_timestamplike(dt.clone())?);
    }

    let (mut year, mut month, mut day) = (dt.year(), dt.month0() as i32, dt.day());
    let years = months / 12;
    year = year
        .checked_add(years)
        .ok_or(EvalError::TimestampOutOfRange)?;

    months %= 12;
    // positive modulus is easier to reason about
    if months < 0 {
        year -= 1;
        months += 12;
    }
    year += (month + months) / 12;
    month = (month + months) % 12;
    // account for dt.month0
    month += 1;

    // handle going from January 31st to February by saturation
    let mut new_d = chrono::NaiveDate::from_ymd_opt(year, month as u32, day);
    while new_d.is_none() {
        // If we have decremented day past 28 and are still receiving `None`,
        // then we have generally overflowed `NaiveDate`.
        if day < 28 {
            return Err(EvalError::TimestampOutOfRange);
        }
        day -= 1;
        new_d = chrono::NaiveDate::from_ymd_opt(year, month as u32, day);
    }
    let new_d = new_d.unwrap();

    // Neither postgres nor mysql support leap seconds, so this should be safe.
    //
    // Both my testing and https://dba.stackexchange.com/a/105829 support the
    // idea that we should ignore leap seconds
    let new_dt = new_d
        .and_hms_nano_opt(dt.hour(), dt.minute(), dt.second(), dt.nanosecond())
        .unwrap();
    let new_dt = T::from_date_time(new_dt);
    Ok(CheckedTimestamp::from_timestamplike(new_dt)?)
}

#[sqlfunc(
    is_monotone = "(true, true)",
    is_infix_op = true,
    sqlname = "+",
    propagates_nulls = true
)]
fn add_numeric(
    a: OrderedDecimal<Numeric>,
    b: OrderedDecimal<Numeric>,
) -> Result<Numeric, EvalError> {
    let mut cx = numeric::cx_datum();
    let mut a = a.0;
    cx.add(&mut a, &b.0);
    if cx.status().overflow() {
        Err(EvalError::FloatOverflow)
    } else {
        Ok(a)
    }
}

#[sqlfunc(
    is_monotone = "(true, true)",
    is_infix_op = true,
    sqlname = "+",
    propagates_nulls = true
)]
fn add_interval(a: Interval, b: Interval) -> Result<Interval, EvalError> {
    a.checked_add(&b)
        .ok_or_else(|| EvalError::IntervalOutOfRange(format!("{a} + {b}").into()))
}

#[sqlfunc(is_infix_op = true, sqlname = "&", propagates_nulls = true)]
fn bit_and_int16(a: i16, b: i16) -> i16 {
    a & b
}

#[sqlfunc(is_infix_op = true, sqlname = "&", propagates_nulls = true)]
fn bit_and_int32(a: i32, b: i32) -> i32 {
    a & b
}

#[sqlfunc(is_infix_op = true, sqlname = "&", propagates_nulls = true)]
fn bit_and_int64(a: i64, b: i64) -> i64 {
    a & b
}

#[sqlfunc(is_infix_op = true, sqlname = "&", propagates_nulls = true)]
fn bit_and_uint16(a: u16, b: u16) -> u16 {
    a & b
}

#[sqlfunc(is_infix_op = true, sqlname = "&", propagates_nulls = true)]
fn bit_and_uint32(a: u32, b: u32) -> u32 {
    a & b
}

#[sqlfunc(is_infix_op = true, sqlname = "&", propagates_nulls = true)]
fn bit_and_uint64(a: u64, b: u64) -> u64 {
    a & b
}

#[sqlfunc(is_infix_op = true, sqlname = "|", propagates_nulls = true)]
fn bit_or_int16(a: i16, b: i16) -> i16 {
    a | b
}

#[sqlfunc(is_infix_op = true, sqlname = "|", propagates_nulls = true)]
fn bit_or_int32(a: i32, b: i32) -> i32 {
    a | b
}

#[sqlfunc(is_infix_op = true, sqlname = "|", propagates_nulls = true)]
fn bit_or_int64(a: i64, b: i64) -> i64 {
    a | b
}

#[sqlfunc(is_infix_op = true, sqlname = "|", propagates_nulls = true)]
fn bit_or_uint16(a: u16, b: u16) -> u16 {
    a | b
}

#[sqlfunc(is_infix_op = true, sqlname = "|", propagates_nulls = true)]
fn bit_or_uint32(a: u32, b: u32) -> u32 {
    a | b
}

#[sqlfunc(is_infix_op = true, sqlname = "|", propagates_nulls = true)]
fn bit_or_uint64(a: u64, b: u64) -> u64 {
    a | b
}

#[sqlfunc(is_infix_op = true, sqlname = "#", propagates_nulls = true)]
fn bit_xor_int16(a: i16, b: i16) -> i16 {
    a ^ b
}

#[sqlfunc(is_infix_op = true, sqlname = "#", propagates_nulls = true)]
fn bit_xor_int32(a: i32, b: i32) -> i32 {
    a ^ b
}

#[sqlfunc(is_infix_op = true, sqlname = "#", propagates_nulls = true)]
fn bit_xor_int64(a: i64, b: i64) -> i64 {
    a ^ b
}

#[sqlfunc(is_infix_op = true, sqlname = "#", propagates_nulls = true)]
fn bit_xor_uint16(a: u16, b: u16) -> u16 {
    a ^ b
}

#[sqlfunc(is_infix_op = true, sqlname = "#", propagates_nulls = true)]
fn bit_xor_uint32(a: u32, b: u32) -> u32 {
    a ^ b
}

#[sqlfunc(is_infix_op = true, sqlname = "#", propagates_nulls = true)]
fn bit_xor_uint64(a: u64, b: u64) -> u64 {
    a ^ b
}

#[sqlfunc(is_infix_op = true, sqlname = "<<", propagates_nulls = true)]
// TODO(benesch): remove potentially dangerous usage of `as`.
#[allow(clippy::as_conversions)]
fn bit_shift_left_int16(a: i16, b: i32) -> i16 {
    // widen to i32 and then cast back to i16 in order emulate the C promotion rules used in by Postgres
    // when the rhs in the 16-31 range, e.g. (1 << 17 should evaluate to 0)
    // see https://github.com/postgres/postgres/blob/REL_14_STABLE/src/backend/utils/adt/int.c#L1460-L1476
    let lhs: i32 = a as i32;
    let rhs: u32 = b as u32;
    lhs.wrapping_shl(rhs) as i16
}

#[sqlfunc(is_infix_op = true, sqlname = "<<", propagates_nulls = true)]
// TODO(benesch): remove potentially dangerous usage of `as`.
#[allow(clippy::as_conversions)]
fn bit_shift_left_int32(lhs: i32, rhs: i32) -> i32 {
    let rhs = rhs as u32;
    lhs.wrapping_shl(rhs)
}

#[sqlfunc(is_infix_op = true, sqlname = "<<", propagates_nulls = true)]
// TODO(benesch): remove potentially dangerous usage of `as`.
#[allow(clippy::as_conversions)]
fn bit_shift_left_int64(lhs: i64, rhs: i32) -> i64 {
    let rhs = rhs as u32;
    lhs.wrapping_shl(rhs)
}

#[sqlfunc(is_infix_op = true, sqlname = "<<", propagates_nulls = true)]
// TODO(benesch): remove potentially dangerous usage of `as`.
#[allow(clippy::as_conversions)]
fn bit_shift_left_uint16(a: u16, b: u32) -> u16 {
    // widen to u32 and then cast back to u16 in order emulate the C promotion rules used in by Postgres
    // when the rhs in the 16-31 range, e.g. (1 << 17 should evaluate to 0)
    // see https://github.com/postgres/postgres/blob/REL_14_STABLE/src/backend/utils/adt/int.c#L1460-L1476
    let lhs: u32 = a as u32;
    let rhs: u32 = b;
    lhs.wrapping_shl(rhs) as u16
}

#[sqlfunc(is_infix_op = true, sqlname = "<<", propagates_nulls = true)]
fn bit_shift_left_uint32(a: u32, b: u32) -> u32 {
    let lhs = a;
    let rhs = b;
    lhs.wrapping_shl(rhs)
}

#[sqlfunc(
    output_type = "u64",
    is_infix_op = true,
    sqlname = "<<",
    propagates_nulls = true
)]
fn bit_shift_left_uint64(lhs: u64, rhs: u32) -> u64 {
    lhs.wrapping_shl(rhs)
}

#[sqlfunc(is_infix_op = true, sqlname = ">>", propagates_nulls = true)]
// TODO(benesch): remove potentially dangerous usage of `as`.
#[allow(clippy::as_conversions)]
fn bit_shift_right_int16(lhs: i16, rhs: i32) -> i16 {
    // widen to i32 and then cast back to i16 in order emulate the C promotion rules used in by Postgres
    // when the rhs in the 16-31 range, e.g. (-32767 >> 17 should evaluate to -1)
    // see https://github.com/postgres/postgres/blob/REL_14_STABLE/src/backend/utils/adt/int.c#L1460-L1476
    let lhs = lhs as i32;
    let rhs = rhs as u32;
    lhs.wrapping_shr(rhs) as i16
}

#[sqlfunc(is_infix_op = true, sqlname = ">>", propagates_nulls = true)]
// TODO(benesch): remove potentially dangerous usage of `as`.
#[allow(clippy::as_conversions)]
fn bit_shift_right_int32(lhs: i32, rhs: i32) -> i32 {
    lhs.wrapping_shr(rhs as u32)
}

#[sqlfunc(is_infix_op = true, sqlname = ">>", propagates_nulls = true)]
// TODO(benesch): remove potentially dangerous usage of `as`.
#[allow(clippy::as_conversions)]
fn bit_shift_right_int64(lhs: i64, rhs: i32) -> i64 {
    lhs.wrapping_shr(rhs as u32)
}

#[sqlfunc(is_infix_op = true, sqlname = ">>", propagates_nulls = true)]
// TODO(benesch): remove potentially dangerous usage of `as`.
#[allow(clippy::as_conversions)]
fn bit_shift_right_uint16(lhs: u16, rhs: u32) -> u16 {
    // widen to u32 and then cast back to u16 in order emulate the C promotion rules used in by Postgres
    // when the rhs in the 16-31 range, e.g. (-32767 >> 17 should evaluate to -1)
    // see https://github.com/postgres/postgres/blob/REL_14_STABLE/src/backend/utils/adt/int.c#L1460-L1476
    let lhs = lhs as u32;
    lhs.wrapping_shr(rhs) as u16
}

#[sqlfunc(is_infix_op = true, sqlname = ">>", propagates_nulls = true)]
fn bit_shift_right_uint32(lhs: u32, rhs: u32) -> u32 {
    lhs.wrapping_shr(rhs)
}

#[sqlfunc(is_infix_op = true, sqlname = ">>", propagates_nulls = true)]
fn bit_shift_right_uint64(lhs: u64, rhs: u32) -> u64 {
    lhs.wrapping_shr(rhs)
}

#[sqlfunc(
    is_monotone = "(true, true)",
    is_infix_op = true,
    sqlname = "-",
    propagates_nulls = true
)]
fn sub_int16(a: i16, b: i16) -> Result<i16, EvalError> {
    a.checked_sub(b).ok_or(EvalError::NumericFieldOverflow)
}

#[sqlfunc(
    is_monotone = "(true, true)",
    is_infix_op = true,
    sqlname = "-",
    propagates_nulls = true
)]
fn sub_int32(a: i32, b: i32) -> Result<i32, EvalError> {
    a.checked_sub(b).ok_or(EvalError::NumericFieldOverflow)
}

#[sqlfunc(
    is_monotone = "(true, true)",
    is_infix_op = true,
    sqlname = "-",
    propagates_nulls = true
)]
fn sub_int64(a: i64, b: i64) -> Result<i64, EvalError> {
    a.checked_sub(b).ok_or(EvalError::NumericFieldOverflow)
}

#[sqlfunc(
    is_monotone = "(true, true)",
    is_infix_op = true,
    sqlname = "-",
    propagates_nulls = true
)]
fn sub_uint16(a: u16, b: u16) -> Result<u16, EvalError> {
    a.checked_sub(b)
        .ok_or_else(|| EvalError::UInt16OutOfRange(format!("{a} - {b}").into()))
}

#[sqlfunc(
    is_monotone = "(true, true)",
    is_infix_op = true,
    sqlname = "-",
    propagates_nulls = true
)]
fn sub_uint32(a: u32, b: u32) -> Result<u32, EvalError> {
    a.checked_sub(b)
        .ok_or_else(|| EvalError::UInt32OutOfRange(format!("{a} - {b}").into()))
}

#[sqlfunc(
    is_monotone = "(true, true)",
    is_infix_op = true,
    sqlname = "-",
    propagates_nulls = true
)]
fn sub_uint64(a: u64, b: u64) -> Result<u64, EvalError> {
    a.checked_sub(b)
        .ok_or_else(|| EvalError::UInt64OutOfRange(format!("{a} - {b}").into()))
}

#[sqlfunc(
    is_monotone = "(true, true)",
    is_infix_op = true,
    sqlname = "-",
    propagates_nulls = true
)]
fn sub_float32(a: f32, b: f32) -> Result<f32, EvalError> {
    let difference = a - b;
    if difference.is_infinite() && !a.is_infinite() && !b.is_infinite() {
        Err(EvalError::FloatOverflow)
    } else {
        Ok(difference)
    }
}

#[sqlfunc(
    is_monotone = "(true, true)",
    is_infix_op = true,
    sqlname = "-",
    propagates_nulls = true
)]
fn sub_float64(a: f64, b: f64) -> Result<f64, EvalError> {
    let difference = a - b;
    if difference.is_infinite() && !a.is_infinite() && !b.is_infinite() {
        Err(EvalError::FloatOverflow)
    } else {
        Ok(difference)
    }
}

#[sqlfunc(
    is_monotone = "(true, true)",
    is_infix_op = true,
    sqlname = "-",
    propagates_nulls = true
)]
fn sub_numeric(
    a: OrderedDecimal<Numeric>,
    b: OrderedDecimal<Numeric>,
) -> Result<Numeric, EvalError> {
    let mut cx = numeric::cx_datum();
    let mut a = a.0;
    cx.sub(&mut a, &b.0);
    if cx.status().overflow() {
        Err(EvalError::FloatOverflow)
    } else {
        Ok(a)
    }
}

// `age(a, b)` is non-monotone in *both* arguments:
//
// * Lex order on `Interval` is `(months, days, micros)`, but the Postgres
//   `age` algorithm independently subtracts year/month/day/... fields and
//   then *borrows* across boundaries when a lower field goes negative. With
//   `b = 2024-02-15` fixed:
//     a = 2024-03-31  →  age = {1 month, 16 days}
//     a = 2024-04-01  →  age = {1 month, 15 days}
//     a = 2024-05-01  →  age = {2 months, 15 days}
//   As `a` increases past a month boundary, `months` jumps by 1 and `days`
//   drops, producing a lex-smaller interval than the previous step.
//
// * Holding `a` fixed and varying `b`, the result has a V-shape at `a == b`
//   (sign is flipped when `a < b`):
//     a = 2024-02-15, b = 2024-02-14  →  age = {0 months, 1 day}
//     a = 2024-02-15, b = 2024-02-15  →  age = {0 months, 0 days}
//     a = 2024-02-15, b = 2024-02-16  →  age = {0 months, 1 day}
#[sqlfunc(sqlname = "age")]
fn age_timestamp(
    a: CheckedTimestamp<chrono::NaiveDateTime>,
    b: CheckedTimestamp<chrono::NaiveDateTime>,
) -> Result<Interval, EvalError> {
    Ok(a.age(&b)?)
}

// See `age_timestamp` for why this is not monotone in either argument.
#[sqlfunc(sqlname = "age")]
fn age_timestamp_tz(
    a: CheckedTimestamp<chrono::DateTime<Utc>>,
    b: CheckedTimestamp<chrono::DateTime<Utc>>,
) -> Result<Interval, EvalError> {
    Ok(a.age(&b)?)
}

#[sqlfunc(is_monotone = "(true, true)", is_infix_op = true, sqlname = "-")]
fn sub_timestamp(
    a: CheckedTimestamp<NaiveDateTime>,
    b: CheckedTimestamp<NaiveDateTime>,
) -> Result<Interval, EvalError> {
    Interval::from_chrono_duration(a - b)
        .map_err(|e| EvalError::IntervalOutOfRange(e.to_string().into()))
}

#[sqlfunc(is_monotone = "(true, true)", is_infix_op = true, sqlname = "-")]
fn sub_timestamp_tz(
    a: CheckedTimestamp<chrono::DateTime<Utc>>,
    b: CheckedTimestamp<chrono::DateTime<Utc>>,
) -> Result<Interval, EvalError> {
    Interval::from_chrono_duration(a - b)
        .map_err(|e| EvalError::IntervalOutOfRange(e.to_string().into()))
}

#[sqlfunc(
    is_monotone = "(true, true)",
    is_infix_op = true,
    sqlname = "-",
    propagates_nulls = true
)]
fn sub_date(a: Date, b: Date) -> i32 {
    a - b
}

#[sqlfunc(is_monotone = "(true, true)", is_infix_op = true, sqlname = "-")]
fn sub_time(a: chrono::NaiveTime, b: chrono::NaiveTime) -> Result<Interval, EvalError> {
    Interval::from_chrono_duration(a - b)
        .map_err(|e| EvalError::IntervalOutOfRange(e.to_string().into()))
}

#[sqlfunc(
    is_monotone = "(true, true)",
    output_type = "Interval",
    is_infix_op = true,
    sqlname = "-",
    propagates_nulls = true
)]
fn sub_interval(a: Interval, b: Interval) -> Result<Interval, EvalError> {
    b.checked_neg()
        .and_then(|b| b.checked_add(&a))
        .ok_or_else(|| EvalError::IntervalOutOfRange(format!("{a} - {b}").into()))
}

// See `add_date_interval` for why this is not monotone in `interval`.
#[sqlfunc(
    is_monotone = "(true, false)",
    is_infix_op = true,
    sqlname = "-",
    propagates_nulls = true
)]
fn sub_date_interval(
    date: Date,
    interval: Interval,
) -> Result<CheckedTimestamp<NaiveDateTime>, EvalError> {
    let dt = NaiveDate::from(date).and_hms_opt(0, 0, 0).unwrap();
    let dt = interval
        .months
        .checked_neg()
        .ok_or_else(|| EvalError::IntervalOutOfRange(interval.months.to_string().into()))
        .and_then(|months| add_timestamp_months(&dt, months))?;
    let dt = dt
        .checked_sub_signed(interval.duration_as_chrono())
        .ok_or(EvalError::TimestampOutOfRange)?;
    Ok(dt.try_into()?)
}

#[sqlfunc(
    is_monotone = "(false, false)",
    is_infix_op = true,
    sqlname = "-",
    propagates_nulls = true
)]
fn sub_time_interval(time: chrono::NaiveTime, interval: Interval) -> chrono::NaiveTime {
    let (t, _) = time.overflowing_sub_signed(interval.duration_as_chrono());
    t
}

#[sqlfunc(
    is_monotone = "(true, true)",
    is_infix_op = true,
    sqlname = "*",
    propagates_nulls = true
)]
fn mul_int16(a: i16, b: i16) -> Result<i16, EvalError> {
    a.checked_mul(b).ok_or(EvalError::NumericFieldOverflow)
}

#[sqlfunc(
    is_monotone = "(true, true)",
    is_infix_op = true,
    sqlname = "*",
    propagates_nulls = true
)]
fn mul_int32(a: i32, b: i32) -> Result<i32, EvalError> {
    a.checked_mul(b).ok_or(EvalError::NumericFieldOverflow)
}

#[sqlfunc(
    is_monotone = "(true, true)",
    is_infix_op = true,
    sqlname = "*",
    propagates_nulls = true
)]
fn mul_int64(a: i64, b: i64) -> Result<i64, EvalError> {
    a.checked_mul(b).ok_or(EvalError::NumericFieldOverflow)
}

#[sqlfunc(
    is_monotone = "(true, true)",
    is_infix_op = true,
    sqlname = "*",
    propagates_nulls = true
)]
fn mul_uint16(a: u16, b: u16) -> Result<u16, EvalError> {
    a.checked_mul(b)
        .ok_or_else(|| EvalError::UInt16OutOfRange(format!("{a} * {b}").into()))
}

#[sqlfunc(
    is_monotone = "(true, true)",
    is_infix_op = true,
    sqlname = "*",
    propagates_nulls = true
)]
fn mul_uint32(a: u32, b: u32) -> Result<u32, EvalError> {
    a.checked_mul(b)
        .ok_or_else(|| EvalError::UInt32OutOfRange(format!("{a} * {b}").into()))
}

#[sqlfunc(
    is_monotone = "(true, true)",
    is_infix_op = true,
    sqlname = "*",
    propagates_nulls = true
)]
fn mul_uint64(a: u64, b: u64) -> Result<u64, EvalError> {
    a.checked_mul(b)
        .ok_or_else(|| EvalError::UInt64OutOfRange(format!("{a} * {b}").into()))
}

#[sqlfunc(
    is_monotone = (true, true),
    is_infinity_monotone = false,
    is_infix_op = true,
    sqlname = "*",
    propagates_nulls = true
)]
fn mul_float32(a: f32, b: f32) -> Result<f32, EvalError> {
    let product = a * b;
    if product.is_infinite() && !a.is_infinite() && !b.is_infinite() {
        Err(EvalError::FloatOverflow)
    } else if product == 0.0f32 && a != 0.0f32 && b != 0.0f32 {
        Err(EvalError::FloatUnderflow)
    } else {
        Ok(product)
    }
}

#[sqlfunc(
    is_monotone = "(true, true)",
    is_infinity_monotone = false,
    is_infix_op = true,
    sqlname = "*",
    propagates_nulls = true
)]
fn mul_float64(a: f64, b: f64) -> Result<f64, EvalError> {
    let product = a * b;
    if product.is_infinite() && !a.is_infinite() && !b.is_infinite() {
        Err(EvalError::FloatOverflow)
    } else if product == 0.0f64 && a != 0.0f64 && b != 0.0f64 {
        Err(EvalError::FloatUnderflow)
    } else {
        Ok(product)
    }
}

#[sqlfunc(
    is_monotone = "(true, true)",
    is_infinity_monotone = false,
    is_infix_op = true,
    sqlname = "*",
    propagates_nulls = true
)]
fn mul_numeric(mut a: Numeric, b: Numeric) -> Result<Numeric, EvalError> {
    let mut cx = numeric::cx_datum();
    cx.mul(&mut a, &b);
    let cx_status = cx.status();
    if cx_status.overflow() {
        Err(EvalError::FloatOverflow)
    } else if cx_status.subnormal() {
        Err(EvalError::FloatUnderflow)
    } else {
        numeric::munge_numeric(&mut a).unwrap();
        Ok(a)
    }
}

#[sqlfunc(
    is_monotone = "(false, false)",
    is_infix_op = true,
    sqlname = "*",
    propagates_nulls = true
)]
fn mul_interval(a: Interval, b: f64) -> Result<Interval, EvalError> {
    a.checked_mul(b)
        .ok_or_else(|| EvalError::IntervalOutOfRange(format!("{a} * {b}").into()))
}

#[sqlfunc(
    is_monotone = "(true, false)",
    is_infix_op = true,
    sqlname = "/",
    propagates_nulls = true
)]
fn div_int16(a: i16, b: i16) -> Result<i16, EvalError> {
    if b == 0 {
        Err(EvalError::DivisionByZero)
    } else {
        a.checked_div(b)
            .ok_or_else(|| EvalError::Int16OutOfRange(format!("{a} / {b}").into()))
    }
}

#[sqlfunc(
    is_monotone = "(true, false)",
    is_infix_op = true,
    sqlname = "/",
    propagates_nulls = true
)]
fn div_int32(a: i32, b: i32) -> Result<i32, EvalError> {
    if b == 0 {
        Err(EvalError::DivisionByZero)
    } else {
        a.checked_div(b)
            .ok_or_else(|| EvalError::Int32OutOfRange(format!("{a} / {b}").into()))
    }
}

#[sqlfunc(
    is_monotone = "(true, false)",
    is_infix_op = true,
    sqlname = "/",
    propagates_nulls = true
)]
fn div_int64(a: i64, b: i64) -> Result<i64, EvalError> {
    if b == 0 {
        Err(EvalError::DivisionByZero)
    } else {
        a.checked_div(b)
            .ok_or_else(|| EvalError::Int64OutOfRange(format!("{a} / {b}").into()))
    }
}

#[sqlfunc(
    is_monotone = "(true, false)",
    is_infix_op = true,
    sqlname = "/",
    propagates_nulls = true
)]
fn div_uint16(a: u16, b: u16) -> Result<u16, EvalError> {
    if b == 0 {
        Err(EvalError::DivisionByZero)
    } else {
        Ok(a / b)
    }
}

#[sqlfunc(
    is_monotone = "(true, false)",
    is_infix_op = true,
    sqlname = "/",
    propagates_nulls = true
)]
fn div_uint32(a: u32, b: u32) -> Result<u32, EvalError> {
    if b == 0 {
        Err(EvalError::DivisionByZero)
    } else {
        Ok(a / b)
    }
}

#[sqlfunc(
    is_monotone = "(true, false)",
    is_infix_op = true,
    sqlname = "/",
    propagates_nulls = true
)]
fn div_uint64(a: u64, b: u64) -> Result<u64, EvalError> {
    if b == 0 {
        Err(EvalError::DivisionByZero)
    } else {
        Ok(a / b)
    }
}

#[sqlfunc(
    is_monotone = "(true, false)",
    is_infinity_monotone = false,
    is_infix_op = true,
    sqlname = "/",
    propagates_nulls = true
)]
fn div_float32(a: f32, b: f32) -> Result<f32, EvalError> {
    if b == 0.0f32 && !a.is_nan() {
        Err(EvalError::DivisionByZero)
    } else {
        let quotient = a / b;
        if quotient.is_infinite() && !a.is_infinite() {
            Err(EvalError::FloatOverflow)
        } else if quotient == 0.0f32 && a != 0.0f32 && !b.is_infinite() {
            Err(EvalError::FloatUnderflow)
        } else {
            Ok(quotient)
        }
    }
}

#[sqlfunc(
    is_monotone = "(true, false)",
    is_infinity_monotone = false,
    is_infix_op = true,
    sqlname = "/",
    propagates_nulls = true
)]
fn div_float64(a: f64, b: f64) -> Result<f64, EvalError> {
    if b == 0.0f64 && !a.is_nan() {
        Err(EvalError::DivisionByZero)
    } else {
        let quotient = a / b;
        if quotient.is_infinite() && !a.is_infinite() {
            Err(EvalError::FloatOverflow)
        } else if quotient == 0.0f64 && a != 0.0f64 && !b.is_infinite() {
            Err(EvalError::FloatUnderflow)
        } else {
            Ok(quotient)
        }
    }
}

#[sqlfunc(
    is_monotone = "(true, false)",
    is_infinity_monotone = false,
    is_infix_op = true,
    sqlname = "/",
    propagates_nulls = true
)]
fn div_numeric(mut a: Numeric, b: Numeric) -> Result<Numeric, EvalError> {
    let mut cx = numeric::cx_datum();

    cx.div(&mut a, &b);
    let cx_status = cx.status();

    // checking the status for division by zero errors is insufficient because
    // the underlying library treats 0/0 as undefined and not division by zero.
    if b.is_zero() {
        Err(EvalError::DivisionByZero)
    } else if cx_status.overflow() {
        Err(EvalError::FloatOverflow)
    } else if cx_status.subnormal() {
        Err(EvalError::FloatUnderflow)
    } else {
        numeric::munge_numeric(&mut a).unwrap();
        Ok(a)
    }
}

#[sqlfunc(
    is_monotone = "(false, false)",
    is_infix_op = true,
    sqlname = "/",
    propagates_nulls = true
)]
fn div_interval(a: Interval, b: f64) -> Result<Interval, EvalError> {
    if b == 0.0 {
        Err(EvalError::DivisionByZero)
    } else {
        a.checked_div(b)
            .ok_or_else(|| EvalError::IntervalOutOfRange(format!("{a} / {b}").into()))
    }
}

#[sqlfunc(is_infix_op = true, sqlname = "%", propagates_nulls = true)]
fn mod_int16(a: i16, b: i16) -> Result<i16, EvalError> {
    if b == 0 {
        Err(EvalError::DivisionByZero)
    } else {
        Ok(a.checked_rem(b).unwrap_or(0))
    }
}

#[sqlfunc(is_infix_op = true, sqlname = "%", propagates_nulls = true)]
fn mod_int32(a: i32, b: i32) -> Result<i32, EvalError> {
    if b == 0 {
        Err(EvalError::DivisionByZero)
    } else {
        Ok(a.checked_rem(b).unwrap_or(0))
    }
}

#[sqlfunc(is_infix_op = true, sqlname = "%", propagates_nulls = true)]
fn mod_int64(a: i64, b: i64) -> Result<i64, EvalError> {
    if b == 0 {
        Err(EvalError::DivisionByZero)
    } else {
        Ok(a.checked_rem(b).unwrap_or(0))
    }
}

#[sqlfunc(is_infix_op = true, sqlname = "%", propagates_nulls = true)]
fn mod_uint16(a: u16, b: u16) -> Result<u16, EvalError> {
    if b == 0 {
        Err(EvalError::DivisionByZero)
    } else {
        Ok(a % b)
    }
}

#[sqlfunc(is_infix_op = true, sqlname = "%", propagates_nulls = true)]
fn mod_uint32(a: u32, b: u32) -> Result<u32, EvalError> {
    if b == 0 {
        Err(EvalError::DivisionByZero)
    } else {
        Ok(a % b)
    }
}

#[sqlfunc(is_infix_op = true, sqlname = "%", propagates_nulls = true)]
fn mod_uint64(a: u64, b: u64) -> Result<u64, EvalError> {
    if b == 0 {
        Err(EvalError::DivisionByZero)
    } else {
        Ok(a % b)
    }
}

#[sqlfunc(is_infix_op = true, sqlname = "%", propagates_nulls = true)]
fn mod_float32(a: f32, b: f32) -> Result<f32, EvalError> {
    if b == 0.0 {
        Err(EvalError::DivisionByZero)
    } else {
        Ok(a % b)
    }
}

#[sqlfunc(is_infix_op = true, sqlname = "%", propagates_nulls = true)]
fn mod_float64(a: f64, b: f64) -> Result<f64, EvalError> {
    if b == 0.0 {
        Err(EvalError::DivisionByZero)
    } else {
        Ok(a % b)
    }
}

#[sqlfunc(is_infix_op = true, sqlname = "%", propagates_nulls = true)]
fn mod_numeric(mut a: Numeric, b: Numeric) -> Result<Numeric, EvalError> {
    if b.is_zero() {
        return Err(EvalError::DivisionByZero);
    }
    let mut cx = numeric::cx_datum();
    // Postgres does _not_ use IEEE 754-style remainder
    cx.rem(&mut a, &b);
    numeric::munge_numeric(&mut a).unwrap();
    Ok(a)
}

fn neg_interval_inner(a: Interval) -> Result<Interval, EvalError> {
    a.checked_neg()
        .ok_or_else(|| EvalError::IntervalOutOfRange(a.to_string().into()))
}

fn log_guard_numeric(val: &Numeric, function_name: &str) -> Result<(), EvalError> {
    if val.is_negative() {
        return Err(EvalError::NegativeOutOfDomain(function_name.into()));
    }
    if val.is_zero() {
        return Err(EvalError::ZeroOutOfDomain(function_name.into()));
    }
    Ok(())
}

#[sqlfunc(sqlname = "log", propagates_nulls = true)]
fn log_base_numeric(mut a: Numeric, mut b: Numeric) -> Result<Numeric, EvalError> {
    log_guard_numeric(&a, "log")?;
    log_guard_numeric(&b, "log")?;
    let mut cx = numeric::cx_datum();
    cx.ln(&mut a);
    cx.ln(&mut b);
    cx.div(&mut b, &a);
    if a.is_zero() {
        Err(EvalError::DivisionByZero)
    } else {
        // This division can result in slightly wrong answers due to the
        // limitation of dividing irrational numbers. To correct that, see if
        // rounding off the value from its `numeric::NUMERIC_DATUM_MAX_PRECISION
        // - 1`th position results in an integral value.
        cx.set_precision(usize::from(numeric::NUMERIC_DATUM_MAX_PRECISION - 1))
            .expect("reducing precision below max always succeeds");
        let mut integral_check = b.clone();

        // `reduce` rounds to the context's final digit when the number of
        // digits in its argument exceeds its precision. We've contrived that to
        // happen by shrinking the context's precision by 1.
        cx.reduce(&mut integral_check);

        // Reduced integral values always have a non-negative exponent.
        let mut b = if integral_check.exponent() >= 0 {
            // We believe our result should have been an integral
            integral_check
        } else {
            b
        };

        numeric::munge_numeric(&mut b).unwrap();
        Ok(b)
    }
}

#[sqlfunc(propagates_nulls = true)]
fn power(a: f64, b: f64) -> Result<f64, EvalError> {
    if a == 0.0 && b.is_sign_negative() {
        return Err(EvalError::Undefined(
            "zero raised to a negative power".into(),
        ));
    }
    if a.is_sign_negative() && b.fract() != 0.0 {
        // Equivalent to PG error:
        // > a negative number raised to a non-integer power yields a complex result
        return Err(EvalError::ComplexOutOfRange("pow".into()));
    }
    let res = a.powf(b);
    if res.is_infinite() {
        return Err(EvalError::FloatOverflow);
    }
    if res == 0.0 && a != 0.0 {
        return Err(EvalError::FloatUnderflow);
    }
    Ok(res)
}

#[sqlfunc(propagates_nulls = true)]
fn uuid_generate_v5(a: uuid::Uuid, b: &str) -> uuid::Uuid {
    uuid::Uuid::new_v5(&a, b.as_bytes())
}

#[sqlfunc(output_type = "Numeric", propagates_nulls = true)]
fn power_numeric(mut a: Numeric, b: Numeric) -> Result<Numeric, EvalError> {
    if a.is_zero() {
        if b.is_zero() {
            return Ok(Numeric::from(1));
        }
        if b.is_negative() {
            return Err(EvalError::Undefined(
                "zero raised to a negative power".into(),
            ));
        }
    }
    if a.is_negative() && b.exponent() < 0 {
        // Equivalent to PG error:
        // > a negative number raised to a non-integer power yields a complex result
        return Err(EvalError::ComplexOutOfRange("pow".into()));
    }
    let mut cx = numeric::cx_datum();
    cx.pow(&mut a, &b);
    let cx_status = cx.status();
    if cx_status.overflow() || (cx_status.invalid_operation() && !b.is_negative()) {
        Err(EvalError::FloatOverflow)
    } else if cx_status.subnormal() || cx_status.invalid_operation() {
        Err(EvalError::FloatUnderflow)
    } else {
        numeric::munge_numeric(&mut a).unwrap();
        Ok(a)
    }
}

#[sqlfunc(propagates_nulls = true)]
fn get_bit(bytes: &[u8], index: i32) -> Result<i32, EvalError> {
    let err = EvalError::IndexOutOfRange {
        provided: index,
        valid_end: i32::try_from(bytes.len().saturating_mul(8)).unwrap_or(i32::MAX) - 1,
    };

    let index = usize::try_from(index).map_err(|_| err.clone())?;

    let byte_index = index / 8;
    let bit_index = index % 8;

    let i = bytes
        .get(byte_index)
        .map(|b| (*b >> bit_index) & 1)
        .ok_or(err)?;
    assert!(i == 0 || i == 1);
    Ok(i32::from(i))
}

#[sqlfunc(propagates_nulls = true)]
fn get_byte(bytes: &[u8], index: i32) -> Result<i32, EvalError> {
    let err = EvalError::IndexOutOfRange {
        provided: index,
        valid_end: i32::try_from(bytes.len()).unwrap_or(i32::MAX) - 1,
    };
    let i: &u8 = bytes
        .get(usize::try_from(index).map_err(|_| err.clone())?)
        .ok_or(err)?;
    Ok(i32::from(*i))
}

#[sqlfunc(sqlname = "constant_time_compare_bytes", propagates_nulls = true)]
pub fn constant_time_eq_bytes(a: &[u8], b: &[u8]) -> bool {
    verify_slices_are_equal(a, b).is_ok()
}

#[sqlfunc(sqlname = "constant_time_compare_strings", propagates_nulls = true)]
pub fn constant_time_eq_string(a: &str, b: &str) -> bool {
    verify_slices_are_equal(a.as_bytes(), b.as_bytes()).is_ok()
}

#[sqlfunc(is_infix_op = true, sqlname = "@>", propagates_nulls = true)]
fn range_contains_i32<'a>(a: Range<Datum<'a>>, b: i32) -> bool {
    a.contains_elem(&b)
}

#[sqlfunc(is_infix_op = true, sqlname = "@>", propagates_nulls = true)]
fn range_contains_i64<'a>(a: Range<Datum<'a>>, elem: i64) -> bool {
    a.contains_elem(&elem)
}

#[sqlfunc(is_infix_op = true, sqlname = "@>", propagates_nulls = true)]
fn range_contains_date<'a>(a: Range<Datum<'a>>, elem: Date) -> bool {
    a.contains_elem(&elem)
}

#[sqlfunc(is_infix_op = true, sqlname = "@>", propagates_nulls = true)]
fn range_contains_numeric<'a>(a: Range<Datum<'a>>, elem: OrderedDecimal<Numeric>) -> bool {
    a.contains_elem(&elem)
}

#[sqlfunc(is_infix_op = true, sqlname = "@>", propagates_nulls = true)]
fn range_contains_timestamp<'a>(
    a: Range<Datum<'a>>,
    elem: CheckedTimestamp<NaiveDateTime>,
) -> bool {
    a.contains_elem(&elem)
}

#[sqlfunc(is_infix_op = true, sqlname = "@>", propagates_nulls = true)]
fn range_contains_timestamp_tz<'a>(
    a: Range<Datum<'a>>,
    elem: CheckedTimestamp<DateTime<Utc>>,
) -> bool {
    a.contains_elem(&elem)
}

#[sqlfunc(is_infix_op = true, sqlname = "<@", propagates_nulls = true)]
fn range_contains_i32_rev<'a>(a: Range<Datum<'a>>, b: i32) -> bool {
    a.contains_elem(&b)
}

#[sqlfunc(is_infix_op = true, sqlname = "<@", propagates_nulls = true)]
fn range_contains_i64_rev<'a>(a: Range<Datum<'a>>, elem: i64) -> bool {
    a.contains_elem(&elem)
}

#[sqlfunc(is_infix_op = true, sqlname = "<@", propagates_nulls = true)]
fn range_contains_date_rev<'a>(a: Range<Datum<'a>>, elem: Date) -> bool {
    a.contains_elem(&elem)
}

#[sqlfunc(is_infix_op = true, sqlname = "<@", propagates_nulls = true)]
fn range_contains_numeric_rev<'a>(a: Range<Datum<'a>>, elem: OrderedDecimal<Numeric>) -> bool {
    a.contains_elem(&elem)
}

#[sqlfunc(is_infix_op = true, sqlname = "<@", propagates_nulls = true)]
fn range_contains_timestamp_rev<'a>(
    a: Range<Datum<'a>>,
    elem: CheckedTimestamp<NaiveDateTime>,
) -> bool {
    a.contains_elem(&elem)
}

#[sqlfunc(is_infix_op = true, sqlname = "<@", propagates_nulls = true)]
fn range_contains_timestamp_tz_rev<'a>(
    a: Range<Datum<'a>>,
    elem: CheckedTimestamp<DateTime<Utc>>,
) -> bool {
    a.contains_elem(&elem)
}

/// Macro to define binary function for various range operations.
/// Parameters:
/// 1. Unique binary function symbol.
/// 2. Range function symbol.
/// 3. SQL name for the function.
macro_rules! range_fn {
    ($fn:expr, $range_fn:expr, $sqlname:expr) => {
        paste::paste! {

            #[sqlfunc(
                output_type = "bool",
                is_infix_op = true,
                sqlname = $sqlname,
                propagates_nulls = true
            )]
            fn [< range_ $fn >]<'a>(a: Datum<'a>, b: Datum<'a>) -> Datum<'a>
            {
                if a.is_null() || b.is_null() { return Datum::Null }
                let l = a.unwrap_range();
                let r = b.unwrap_range();
                Datum::from(Range::<Datum<'a>>::$range_fn(&l, &r))
            }
        }
    };
}

// RangeContainsRange is either @> or <@ depending on the order of the arguments.
// It doesn't influence the result, but it does influence the display string.
range_fn!(contains_range, contains_range, "@>");
range_fn!(contains_range_rev, contains_range, "<@");
range_fn!(overlaps, overlaps, "&&");
range_fn!(after, after, ">>");
range_fn!(before, before, "<<");
range_fn!(overleft, overleft, "&<");
range_fn!(overright, overright, "&>");
range_fn!(adjacent, adjacent, "-|-");

#[sqlfunc(is_infix_op = true, sqlname = "+")]
fn range_union<T: Copy + Ord>(l: Range<T>, r: Range<T>) -> Result<Range<T>, EvalError> {
    Ok(l.union(&r)?)
}

#[sqlfunc(is_infix_op = true, sqlname = "*")]
fn range_intersection<T: Copy + Ord>(l: Range<T>, r: Range<T>) -> Range<T> {
    l.intersection(&r)
}

#[sqlfunc(
    output_type_expr = "input_types[0].scalar_type.without_modifiers().nullable(true)",
    is_infix_op = true,
    sqlname = "-",
    propagates_nulls = true,
    introduces_nulls = false
)]
fn range_difference<'a>(
    l: Range<Datum<'a>>,
    r: Range<Datum<'a>>,
) -> Result<Range<Datum<'a>>, EvalError> {
    Ok(l.difference(&r)?)
}

#[sqlfunc(is_infix_op = true, sqlname = "=", negate = "Some(NotEq.into())")]
fn eq<'a>(a: ExcludeNull<Datum<'a>>, b: ExcludeNull<Datum<'a>>) -> bool {
    // SQL equality demands that if either input is null, then the result should be null. However,
    // we don't need to handle this case here; it is handled when `BinaryFunc::eval` checks
    // `propagates_nulls`.
    a == b
}

#[sqlfunc(is_infix_op = true, sqlname = "!=", negate = "Some(Eq.into())")]
fn not_eq<'a>(a: ExcludeNull<Datum<'a>>, b: ExcludeNull<Datum<'a>>) -> bool {
    a != b
}

#[sqlfunc(
    is_monotone = "(true, true)",
    is_infix_op = true,
    sqlname = "<",
    negate = "Some(Gte.into())"
)]
fn lt<'a>(a: ExcludeNull<Datum<'a>>, b: ExcludeNull<Datum<'a>>) -> bool {
    a < b
}

#[sqlfunc(
    is_monotone = "(true, true)",
    is_infix_op = true,
    sqlname = "<=",
    negate = "Some(Gt.into())"
)]
fn lte<'a>(a: ExcludeNull<Datum<'a>>, b: ExcludeNull<Datum<'a>>) -> bool {
    a <= b
}

#[sqlfunc(
    is_monotone = "(true, true)",
    is_infix_op = true,
    sqlname = ">",
    negate = "Some(Lte.into())"
)]
fn gt<'a>(a: ExcludeNull<Datum<'a>>, b: ExcludeNull<Datum<'a>>) -> bool {
    a > b
}

#[sqlfunc(
    is_monotone = "(true, true)",
    is_infix_op = true,
    sqlname = ">=",
    negate = "Some(Lt.into())"
)]
fn gte<'a>(a: ExcludeNull<Datum<'a>>, b: ExcludeNull<Datum<'a>>) -> bool {
    a >= b
}

#[sqlfunc(sqlname = "tocharts", propagates_nulls = true)]
fn to_char_timestamp_format(ts: CheckedTimestamp<chrono::NaiveDateTime>, format: &str) -> String {
    let fmt = DateTimeFormat::compile(format);
    fmt.render(&*ts)
}

#[sqlfunc(sqlname = "tochartstz", propagates_nulls = true)]
fn to_char_timestamp_tz_format(
    ts: CheckedTimestamp<chrono::DateTime<Utc>>,
    format: &str,
) -> String {
    let fmt = DateTimeFormat::compile(format);
    fmt.render(&*ts)
}

#[sqlfunc(sqlname = "->", is_infix_op = true)]
fn jsonb_get_int64<'a>(a: JsonbRef<'a>, i: i64) -> Option<JsonbRef<'a>> {
    match a.into_datum() {
        Datum::List(list) => {
            let i = if i >= 0 {
                usize::cast_from(i.unsigned_abs())
            } else {
                // index backwards from the end
                let i = usize::cast_from(i.unsigned_abs());
                (list.iter().count()).wrapping_sub(i)
            };
            let v = list.iter().nth(i)?;
            // `v` should be valid jsonb because it came from a jsonb list, but we don't
            // panic on mismatch to avoid bringing down the whole system on corrupt data.
            // Instead, we'll return None.
            JsonbRef::try_from_result(Ok::<_, ()>(v)).ok()
        }
        Datum::Map(_) => None,
        _ => {
            // I have no idea why postgres does this, but we're stuck with it
            (i == 0 || i == -1).then_some(a)
        }
    }
}

#[sqlfunc(sqlname = "->>", is_infix_op = true)]
fn jsonb_get_int64_stringify<'a>(
    a: JsonbRef<'a>,
    i: i64,
    temp_storage: &'a RowArena,
) -> Option<&'a str> {
    let json = jsonb_get_int64(a, i)?;
    jsonb_stringify(json.into_datum(), temp_storage)
}

#[sqlfunc(sqlname = "->", is_infix_op = true)]
fn jsonb_get_string<'a>(a: JsonbRef<'a>, k: &str) -> Option<JsonbRef<'a>> {
    let dict = DatumMap::try_from_result(Ok::<_, ()>(a.into_datum())).ok()?;
    let v = dict.iter().find(|(k2, _v)| k == *k2).map(|(_k, v)| v)?;
    JsonbRef::try_from_result(Ok::<_, ()>(v)).ok()
}

#[sqlfunc(sqlname = "->>", is_infix_op = true)]
fn jsonb_get_string_stringify<'a>(
    a: JsonbRef<'a>,
    k: &str,
    temp_storage: &'a RowArena,
) -> Option<&'a str> {
    let v = jsonb_get_string(a, k)?;
    jsonb_stringify(v.into_datum(), temp_storage)
}

#[sqlfunc(sqlname = "#>", is_infix_op = true)]
fn jsonb_get_path<'a>(mut json: JsonbRef<'a>, b: Array<'a>) -> Option<JsonbRef<'a>> {
    let path = b.elements();
    for key in path.iter() {
        let key = match key {
            Datum::String(s) => s,
            Datum::Null => return None,
            _ => unreachable!("keys in jsonb_get_path known to be strings"),
        };
        let v = match json.into_datum() {
            Datum::Map(map) => map.iter().find(|(k, _)| key == *k).map(|(_k, v)| v),
            Datum::List(list) => {
                let i = strconv::parse_int64(key).ok()?;
                let i = if i >= 0 {
                    usize::cast_from(i.unsigned_abs())
                } else {
                    // index backwards from the end
                    let i = usize::cast_from(i.unsigned_abs());
                    (list.iter().count()).wrapping_sub(i)
                };
                list.iter().nth(i)
            }
            _ => return None,
        }?;
        json = JsonbRef::try_from_result(Ok::<_, ()>(v)).ok()?;
    }
    Some(json)
}

#[sqlfunc(sqlname = "#>>", is_infix_op = true)]
fn jsonb_get_path_stringify<'a>(
    a: JsonbRef<'a>,
    b: Array<'a>,
    temp_storage: &'a RowArena,
) -> Option<&'a str> {
    let json = jsonb_get_path(a, b)?;
    jsonb_stringify(json.into_datum(), temp_storage)
}

#[sqlfunc(is_infix_op = true, sqlname = "?")]
fn jsonb_contains_string<'a>(a: JsonbRef<'a>, k: &str) -> bool {
    // https://www.postgresql.org/docs/current/datatype-json.html#JSON-CONTAINMENT
    // When the left operand is SQL NULL (NULL::jsonb), JsonbRef::try_from_result rejects it,
    // so the binary evaluator never calls this function and returns NULL (see binary.rs).
    // So, this function only runs for non-null jsonb; a.into_datum() never sees Datum::Null.
    match a.into_datum() {
        Datum::List(list) => list.iter().any(|k2| Datum::from(k) == k2),
        Datum::Map(dict) => dict.iter().any(|(k2, _v)| k == k2),
        Datum::String(string) => string == k,
        _ => false,
    }
}

#[sqlfunc(is_infix_op = true, sqlname = "?", propagates_nulls = true)]
// Map keys are always text.
fn map_contains_key<'a>(map: DatumMap<'a>, k: &str) -> bool {
    map.iter().any(|(k2, _v)| k == k2)
}

#[sqlfunc(is_infix_op = true, sqlname = "?&")]
fn map_contains_all_keys<'a>(map: DatumMap<'a>, keys: Array<'a>) -> bool {
    keys.elements()
        .iter()
        .all(|key| !key.is_null() && map.iter().any(|(k, _v)| k == key.unwrap_str()))
}

#[sqlfunc(is_infix_op = true, sqlname = "?|", propagates_nulls = true)]
fn map_contains_any_keys<'a>(map: DatumMap<'a>, keys: Array<'a>) -> bool {
    keys.elements()
        .iter()
        .any(|key| !key.is_null() && map.iter().any(|(k, _v)| k == key.unwrap_str()))
}

#[sqlfunc(is_infix_op = true, sqlname = "@>", propagates_nulls = true)]
fn map_contains_map<'a>(map_a: DatumMap<'a>, b: DatumMap<'a>) -> bool {
    b.iter().all(|(b_key, b_val)| {
        map_a
            .iter()
            .any(|(a_key, a_val)| (a_key == b_key) && (a_val == b_val))
    })
}

#[sqlfunc(is_infix_op = true, sqlname = "->", propagates_nulls = true)]
fn map_get_value<'a, T: FromDatum<'a>>(a: DatumMap<'a, T>, target_key: &str) -> Option<T> {
    a.typed_iter()
        .find(|(key, _v)| target_key == *key)
        .map(|(_k, v)| v)
}

#[sqlfunc(is_infix_op = true, sqlname = "@>")]
fn list_contains_list<'a>(a: ExcludeNull<DatumList<'a>>, b: ExcludeNull<DatumList<'a>>) -> bool {
    // NULL is never equal to NULL. If NULL is an element of b, b cannot be contained in a, even if a contains NULL.
    if b.iter().contains(&Datum::Null) {
        false
    } else {
        b.iter()
            .all(|item_b| a.iter().any(|item_a| item_a == item_b))
    }
}

#[sqlfunc(is_infix_op = true, sqlname = "<@")]
fn list_contains_list_rev<'a>(
    a: ExcludeNull<DatumList<'a>>,
    b: ExcludeNull<DatumList<'a>>,
) -> bool {
    list_contains_list(b, a)
}

// TODO(jamii) nested loops are possibly not the fastest way to do this
#[sqlfunc(is_infix_op = true, sqlname = "@>")]
fn jsonb_contains_jsonb<'a>(a: JsonbRef<'a>, b: JsonbRef<'a>) -> bool {
    // https://www.postgresql.org/docs/current/datatype-json.html#JSON-CONTAINMENT
    fn contains(a: Datum, b: Datum, at_top_level: bool) -> bool {
        match (a, b) {
            (Datum::JsonNull, Datum::JsonNull) => true,
            (Datum::False, Datum::False) => true,
            (Datum::True, Datum::True) => true,
            (Datum::Numeric(a), Datum::Numeric(b)) => a == b,
            (Datum::String(a), Datum::String(b)) => a == b,
            (Datum::List(a), Datum::List(b)) => b
                .iter()
                .all(|b_elem| a.iter().any(|a_elem| contains(a_elem, b_elem, false))),
            (Datum::Map(a), Datum::Map(b)) => b.iter().all(|(b_key, b_val)| {
                a.iter()
                    .any(|(a_key, a_val)| (a_key == b_key) && contains(a_val, b_val, false))
            }),

            // fun special case
            (Datum::List(a), b) => {
                at_top_level && a.iter().any(|a_elem| contains(a_elem, b, false))
            }

            _ => false,
        }
    }
    contains(a.into_datum(), b.into_datum(), true)
}

#[sqlfunc(is_infix_op = true, sqlname = "||")]
fn jsonb_concat<'a>(
    a: JsonbRef<'a>,
    b: JsonbRef<'a>,
    temp_storage: &'a RowArena,
) -> Option<JsonbRef<'a>> {
    let res = match (a.into_datum(), b.into_datum()) {
        (Datum::Map(dict_a), Datum::Map(dict_b)) => {
            let mut pairs = dict_b.iter().chain(dict_a.iter()).collect::<Vec<_>>();
            // stable sort, so if keys collide dedup prefers dict_b
            pairs.sort_by(|(k1, _v1), (k2, _v2)| k1.cmp(k2));
            pairs.dedup_by(|(k1, _v1), (k2, _v2)| k1 == k2);
            temp_storage.make_datum(|packer| packer.push_dict(pairs))
        }
        (Datum::List(list_a), Datum::List(list_b)) => {
            let elems = list_a.iter().chain(list_b.iter());
            temp_storage.make_datum(|packer| packer.push_list(elems))
        }
        (Datum::List(list_a), b) => {
            let elems = list_a.iter().chain(Some(b));
            temp_storage.make_datum(|packer| packer.push_list(elems))
        }
        (a, Datum::List(list_b)) => {
            let elems = Some(a).into_iter().chain(list_b.iter());
            temp_storage.make_datum(|packer| packer.push_list(elems))
        }
        _ => return None,
    };
    Some(JsonbRef::from_datum(res))
}

#[sqlfunc(
    output_type_expr = "SqlScalarType::Jsonb.nullable(true)",
    is_infix_op = true,
    sqlname = "-",
    propagates_nulls = true,
    introduces_nulls = true
)]
fn jsonb_delete_int64<'a>(a: Datum<'a>, i: i64, temp_storage: &'a RowArena) -> Datum<'a> {
    match a {
        Datum::List(list) => {
            let i = if i >= 0 {
                usize::cast_from(i.unsigned_abs())
            } else {
                // index backwards from the end
                let i = usize::cast_from(i.unsigned_abs());
                (list.iter().count()).wrapping_sub(i)
            };
            let elems = list
                .iter()
                .enumerate()
                .filter(|(i2, _e)| i != *i2)
                .map(|(_, e)| e);
            temp_storage.make_datum(|packer| packer.push_list(elems))
        }
        _ => Datum::Null,
    }
}

#[sqlfunc(
    output_type_expr = "SqlScalarType::Jsonb.nullable(true)",
    is_infix_op = true,
    sqlname = "-",
    propagates_nulls = true,
    introduces_nulls = true
)]
fn jsonb_delete_string<'a>(a: Datum<'a>, k: &str, temp_storage: &'a RowArena) -> Datum<'a> {
    match a {
        Datum::List(list) => {
            let elems = list.iter().filter(|e| Datum::from(k) != *e);
            temp_storage.make_datum(|packer| packer.push_list(elems))
        }
        Datum::Map(dict) => {
            let pairs = dict.iter().filter(|(k2, _v)| k != *k2);
            temp_storage.make_datum(|packer| packer.push_dict(pairs))
        }
        _ => Datum::Null,
    }
}

#[sqlfunc(
    sqlname = "extractiv",
    propagates_nulls = true,
    introduces_nulls = false
)]
fn date_part_interval_numeric(units: &str, b: Interval) -> Result<Numeric, EvalError> {
    match units.parse() {
        Ok(units) => Ok(date_part_interval_inner::<Numeric>(units, b)?),
        Err(_) => Err(EvalError::UnknownUnits(units.into())),
    }
}

#[sqlfunc(
    sqlname = "date_partiv",
    propagates_nulls = true,
    introduces_nulls = false
)]
fn date_part_interval_f64(units: &str, b: Interval) -> Result<f64, EvalError> {
    match units.parse() {
        Ok(units) => Ok(date_part_interval_inner::<f64>(units, b)?),
        Err(_) => Err(EvalError::UnknownUnits(units.into())),
    }
}

#[sqlfunc(
    sqlname = "extractt",
    propagates_nulls = true,
    introduces_nulls = false
)]
fn date_part_time_numeric(units: &str, b: chrono::NaiveTime) -> Result<Numeric, EvalError> {
    match units.parse() {
        Ok(units) => Ok(date_part_time_inner::<Numeric>(units, b)?),
        Err(_) => Err(EvalError::UnknownUnits(units.into())),
    }
}

#[sqlfunc(
    sqlname = "date_partt",
    propagates_nulls = true,
    introduces_nulls = false
)]
fn date_part_time_f64(units: &str, b: chrono::NaiveTime) -> Result<f64, EvalError> {
    match units.parse() {
        Ok(units) => Ok(date_part_time_inner::<f64>(units, b)?),
        Err(_) => Err(EvalError::UnknownUnits(units.into())),
    }
}

#[sqlfunc(sqlname = "extractts", propagates_nulls = true)]
fn date_part_timestamp_timestamp_numeric(
    units: &str,
    ts: CheckedTimestamp<NaiveDateTime>,
) -> Result<Numeric, EvalError> {
    match units.parse() {
        Ok(units) => Ok(date_part_timestamp_inner::<_, Numeric>(units, &*ts)?),
        Err(_) => Err(EvalError::UnknownUnits(units.into())),
    }
}

#[sqlfunc(sqlname = "extracttstz", propagates_nulls = true)]
fn date_part_timestamp_timestamp_tz_numeric(
    units: &str,
    ts: CheckedTimestamp<DateTime<Utc>>,
) -> Result<Numeric, EvalError> {
    match units.parse() {
        Ok(units) => Ok(date_part_timestamp_inner::<_, Numeric>(units, &*ts)?),
        Err(_) => Err(EvalError::UnknownUnits(units.into())),
    }
}

#[sqlfunc(sqlname = "date_partts", propagates_nulls = true)]
fn date_part_timestamp_timestamp_f64(
    units: &str,
    ts: CheckedTimestamp<NaiveDateTime>,
) -> Result<f64, EvalError> {
    match units.parse() {
        Ok(units) => date_part_timestamp_inner(units, &*ts),
        Err(_) => Err(EvalError::UnknownUnits(units.into())),
    }
}

#[sqlfunc(sqlname = "date_parttstz", propagates_nulls = true)]
fn date_part_timestamp_timestamp_tz_f64(
    units: &str,
    ts: CheckedTimestamp<DateTime<Utc>>,
) -> Result<f64, EvalError> {
    match units.parse() {
        Ok(units) => date_part_timestamp_inner(units, &*ts),
        Err(_) => Err(EvalError::UnknownUnits(units.into())),
    }
}

#[sqlfunc(sqlname = "extractd", propagates_nulls = true)]
fn extract_date_units(units: &str, b: Date) -> Result<Numeric, EvalError> {
    match units.parse() {
        Ok(units) => Ok(extract_date_inner(units, b.into())?),
        Err(_) => Err(EvalError::UnknownUnits(units.into())),
    }
}

pub fn date_bin<T>(
    stride: Interval,
    source: CheckedTimestamp<T>,
    origin: CheckedTimestamp<T>,
) -> Result<CheckedTimestamp<T>, EvalError>
where
    T: TimestampLike,
{
    if stride.months != 0 {
        return Err(EvalError::DateBinOutOfRange(
            "timestamps cannot be binned into intervals containing months or years".into(),
        ));
    }

    let stride_ns = match stride.duration_as_chrono().num_nanoseconds() {
        Some(ns) if ns <= 0 => Err(EvalError::DateBinOutOfRange(
            "stride must be greater than zero".into(),
        )),
        Some(ns) => Ok(ns),
        None => Err(EvalError::DateBinOutOfRange(
            format!("stride cannot exceed {}/{} nanoseconds", i64::MAX, i64::MIN,).into(),
        )),
    }?;

    // Make sure the returned timestamp is at the start of the bin, even if the
    // origin is in the future. We do this here because `T` is not `Copy` and
    // gets moved by its subtraction operation.
    let sub_stride = origin > source;

    let tm_diff = (source - origin.clone()).num_nanoseconds().ok_or_else(|| {
        EvalError::DateBinOutOfRange(
            "source and origin must not differ more than 2^63 nanoseconds".into(),
        )
    })?;

    let remainder = tm_diff % stride_ns;
    let mut tm_delta = tm_diff - remainder;

    if sub_stride && remainder != 0 {
        tm_delta = tm_delta.checked_sub(stride_ns).ok_or_else(|| {
            EvalError::DateBinOutOfRange(
                "source and origin must not differ more than 2^63 nanoseconds".into(),
            )
        })?;
    }

    let res = origin
        .checked_add_signed(Duration::nanoseconds(tm_delta))
        .ok_or(EvalError::TimestampOutOfRange)?;
    Ok(CheckedTimestamp::from_timestamplike(res)?)
}

// Non-monotone in `stride`: the result is `origin + floor((source - origin) /
// stride) * stride`. For a fixed source like `2024-01-01 12:00:00`, a 1-day
// stride bins to `2024-01-01 00:00:00`, but a 2-day stride bins to
// `2023-12-31 00:00:00` — i.e. the lex-larger interval produces an earlier
// timestamp. Monotone in `source`.
#[sqlfunc(is_monotone = "(false, true)", sqlname = "bin_unix_epoch_timestamp")]
fn date_bin_timestamp(
    stride: Interval,
    source: CheckedTimestamp<NaiveDateTime>,
) -> Result<CheckedTimestamp<NaiveDateTime>, EvalError> {
    let origin =
        CheckedTimestamp::from_timestamplike(DateTime::from_timestamp(0, 0).unwrap().naive_utc())
            .expect("must fit");
    date_bin(stride, source, origin)
}

// See `date_bin_timestamp` for why this is not monotone in `stride`.
#[sqlfunc(is_monotone = "(false, true)", sqlname = "bin_unix_epoch_timestamptz")]
fn date_bin_timestamp_tz(
    stride: Interval,
    source: CheckedTimestamp<DateTime<Utc>>,
) -> Result<CheckedTimestamp<DateTime<Utc>>, EvalError> {
    let origin = CheckedTimestamp::from_timestamplike(DateTime::from_timestamp(0, 0).unwrap())
        .expect("must fit");
    date_bin(stride, source, origin)
}

#[sqlfunc(sqlname = "date_truncts", propagates_nulls = true)]
fn date_trunc_units_timestamp(
    units: &str,
    ts: CheckedTimestamp<NaiveDateTime>,
) -> Result<CheckedTimestamp<NaiveDateTime>, EvalError> {
    match units.parse() {
        Ok(units) => Ok(date_trunc_inner(units, &*ts)?.try_into()?),
        Err(_) => Err(EvalError::UnknownUnits(units.into())),
    }
}

#[sqlfunc(sqlname = "date_trunctstz", propagates_nulls = true)]
fn date_trunc_units_timestamp_tz(
    units: &str,
    ts: CheckedTimestamp<DateTime<Utc>>,
) -> Result<CheckedTimestamp<DateTime<Utc>>, EvalError> {
    match units.parse() {
        Ok(units) => Ok(date_trunc_inner(units, &*ts)?.try_into()?),
        Err(_) => Err(EvalError::UnknownUnits(units.into())),
    }
}

#[sqlfunc(sqlname = "date_trunciv", propagates_nulls = true)]
fn date_trunc_interval(units: &str, mut interval: Interval) -> Result<Interval, EvalError> {
    let dtf = units
        .parse()
        .map_err(|_| EvalError::UnknownUnits(units.into()))?;

    interval
        .truncate_low_fields(dtf, Some(0), RoundBehavior::Truncate)
        .expect(
            "truncate_low_fields should not fail with max_precision 0 and RoundBehavior::Truncate",
        );
    Ok(interval)
}

/// Parses a named timezone like `EST` or `America/New_York`, or a fixed-offset timezone like `-05:00`.
///
/// The interpretation of fixed offsets depend on whether the POSIX or ISO 8601 standard is being
/// used.
pub(crate) fn parse_timezone(tz: &str, spec: TimezoneSpec) -> Result<Timezone, EvalError> {
    Timezone::parse(tz, spec).map_err(|_| EvalError::InvalidTimezone(tz.into()))
}

/// Converts the time datum `b`, which is assumed to be in UTC, to the timezone that the interval datum `a` is assumed
/// to represent. The interval is not allowed to hold months, but there are no limits on the amount of seconds.
/// The interval acts like a `chrono::FixedOffset`, without the `-86,400 < x < 86,400` limitation.
#[sqlfunc(sqlname = "timezoneit")]
fn timezone_interval_time_binary(
    interval: Interval,
    time: chrono::NaiveTime,
) -> Result<chrono::NaiveTime, EvalError> {
    if interval.months != 0 {
        Err(EvalError::InvalidTimezoneInterval)
    } else {
        Ok(time.overflowing_add_signed(interval.duration_as_chrono()).0)
    }
}

/// Converts the timestamp datum `b`, which is assumed to be in the time of the timezone datum `a` to a timestamptz
/// in UTC. The interval is not allowed to hold months, but there are no limits on the amount of seconds.
/// The interval acts like a `chrono::FixedOffset`, without the `-86,400 < x < 86,400` limitation.
#[sqlfunc(sqlname = "timezoneits")]
fn timezone_interval_timestamp_binary(
    interval: Interval,
    ts: CheckedTimestamp<NaiveDateTime>,
) -> Result<CheckedTimestamp<DateTime<Utc>>, EvalError> {
    if interval.months != 0 {
        Err(EvalError::InvalidTimezoneInterval)
    } else {
        match ts.checked_sub_signed(interval.duration_as_chrono()) {
            Some(sub) => Ok(DateTime::from_naive_utc_and_offset(sub, Utc).try_into()?),
            None => Err(EvalError::TimestampOutOfRange),
        }
    }
}

/// Converts the UTC timestamptz datum `b`, to the local timestamp of the timezone datum `a`.
/// The interval is not allowed to hold months, but there are no limits on the amount of seconds.
/// The interval acts like a `chrono::FixedOffset`, without the `-86,400 < x < 86,400` limitation.
#[sqlfunc(sqlname = "timezoneitstz")]
fn timezone_interval_timestamp_tz_binary(
    interval: Interval,
    tstz: CheckedTimestamp<DateTime<Utc>>,
) -> Result<CheckedTimestamp<NaiveDateTime>, EvalError> {
    if interval.months != 0 {
        return Err(EvalError::InvalidTimezoneInterval);
    }
    match tstz
        .naive_utc()
        .checked_add_signed(interval.duration_as_chrono())
    {
        Some(dt) => Ok(dt.try_into()?),
        None => Err(EvalError::TimestampOutOfRange),
    }
}

#[sqlfunc(
    output_type_expr = r#"SqlScalarType::Record {
                fields: [
                    ("abbrev".into(), SqlScalarType::String.nullable(false)),
                    ("base_utc_offset".into(), SqlScalarType::Interval.nullable(false)),
                    ("dst_offset".into(), SqlScalarType::Interval.nullable(false)),
                ].into(),
                custom_id: None,
            }.nullable(true)"#,
    propagates_nulls = true,
    introduces_nulls = false
)]
fn timezone_offset<'a>(
    tz_str: &str,
    b: CheckedTimestamp<chrono::DateTime<Utc>>,
    temp_storage: &'a RowArena,
) -> Result<Datum<'a>, EvalError> {
    let tz = match Tz::from_str_insensitive(tz_str) {
        Ok(tz) => tz,
        Err(_) => return Err(EvalError::InvalidIanaTimezoneId(tz_str.into())),
    };
    let offset = tz.offset_from_utc_datetime(&b.naive_utc());
    // Zones without an alphabetic abbreviation get the numeric form rendered
    // from the offset, e.g. "+05". This matches PostgreSQL, whose tzdata files
    // have the same rendering applied by zic's %z expansion.
    let abbrev = match offset.abbreviation() {
        Some(abbrev) => abbrev.to_string(),
        None => {
            const SECONDS_PER_MINUTE: i64 = 60;
            const MINUTES_PER_HOUR: i64 = 60;
            let secs = (offset.base_utc_offset() + offset.dst_offset()).num_seconds();
            let sign = if secs < 0 { '-' } else { '+' };
            let (mins, s) = (
                secs.abs() / SECONDS_PER_MINUTE,
                secs.abs() % SECONDS_PER_MINUTE,
            );
            let (h, m) = (mins / MINUTES_PER_HOUR, mins % MINUTES_PER_HOUR);
            if s != 0 {
                // Unreachable for current tzdata: sub-minute offsets exist
                // only for pre-standardization history (e.g. Africa/Monrovia
                // until 1972), and tzdata gives all of them alphabetic names.
                // We render rather than trust that invariant forever.
                // chrono-tz's own Display impl instead asserts, and a scalar
                // function must not panic.
                format!("{sign}{h:02}{m:02}{s:02}")
            } else if m != 0 {
                // Fractional-hour zones, e.g. Asia/Kathmandu renders "+0545".
                format!("{sign}{h:02}{m:02}")
            } else {
                // Whole-hour zones, e.g. Asia/Almaty renders "+05".
                format!("{sign}{h:02}")
            }
        }
    };
    Ok(temp_storage.make_datum(|packer| {
        packer.push_list_with(|packer| {
            packer.push(Datum::from(abbrev.as_str()));
            packer.push(Datum::from(offset.base_utc_offset()));
            packer.push(Datum::from(offset.dst_offset()));
        });
    }))
}

/// Determines if an mz_aclitem contains one of the specified privileges. This will return true if
/// any of the listed privileges are contained in the mz_aclitem.
#[sqlfunc(
    sqlname = "mz_aclitem_contains_privilege",
    output_type = "bool",
    propagates_nulls = true
)]
fn mz_acl_item_contains_privilege(
    mz_acl_item: MzAclItem,
    privileges: &str,
) -> Result<bool, EvalError> {
    let acl_mode = AclMode::parse_multiple_privileges(privileges)
        .map_err(|e: anyhow::Error| EvalError::InvalidPrivileges(e.to_string().into()))?;
    let contains = !mz_acl_item.acl_mode.intersection(acl_mode).is_empty();
    Ok(contains)
}

#[sqlfunc]
// transliterated from postgres/src/backend/utils/adt/misc.c
fn parse_ident<'a>(ident: &'a str, strict: bool) -> Result<ArrayRustType<Cow<'a, str>>, EvalError> {
    fn is_ident_start(c: char) -> bool {
        matches!(c, 'A'..='Z' | 'a'..='z' | '_' | '\u{80}'..=char::MAX)
    }

    fn is_ident_cont(c: char) -> bool {
        matches!(c, '0'..='9' | '$') || is_ident_start(c)
    }

    let mut elems = vec![];
    let buf = &mut LexBuf::new(ident);

    let mut after_dot = false;

    buf.take_while(|ch| ch.is_ascii_whitespace());

    loop {
        let mut missing_ident = true;

        let c = buf.next();

        if c == Some('"') {
            let s = buf.take_while(|ch| !matches!(ch, '"'));

            if buf.next() != Some('"') {
                return Err(EvalError::InvalidIdentifier {
                    ident: ident.into(),
                    detail: Some("String has unclosed double quotes.".into()),
                });
            }
            elems.push(Cow::Borrowed(s));
            missing_ident = false;
        } else if c.map(is_ident_start).unwrap_or(false) {
            buf.prev();
            let s = buf.take_while(is_ident_cont);
            elems.push(Cow::Owned(s.to_ascii_lowercase()));
            missing_ident = false;
        }

        if missing_ident {
            if c == Some('.') {
                return Err(EvalError::InvalidIdentifier {
                    ident: ident.into(),
                    detail: Some("No valid identifier before \".\".".into()),
                });
            } else if after_dot {
                return Err(EvalError::InvalidIdentifier {
                    ident: ident.into(),
                    detail: Some("No valid identifier after \".\".".into()),
                });
            } else {
                return Err(EvalError::InvalidIdentifier {
                    ident: ident.into(),
                    detail: None,
                });
            }
        }

        buf.take_while(|ch| ch.is_ascii_whitespace());

        match buf.next() {
            Some('.') => {
                after_dot = true;

                buf.take_while(|ch| ch.is_ascii_whitespace());
            }
            Some(_) if strict => {
                return Err(EvalError::InvalidIdentifier {
                    ident: ident.into(),
                    detail: None,
                });
            }
            _ => break,
        }
    }

    Ok(elems.into())
}

fn regexp_split_to_array_re<'a>(
    text: &str,
    regexp: &Regex,
    temp_storage: &'a RowArena,
) -> Result<Datum<'a>, EvalError> {
    // Bound the transient `Vec<&str>` before the split builds it. The count follows the split's own
    // zero-length-match rule, so it refuses exactly the calls the split would build.
    check_build_fits_budget(
        || mz_regexp::regexp_split_to_array_count(text, regexp),
        std::mem::size_of::<&str>(),
        temp_storage,
    )?;
    let found = mz_regexp::regexp_split_to_array(text, regexp);
    // Splitting amplifies: each chunk is packed with its own tag and length, so a pattern that
    // splits per character costs several times the input.
    check_datums_fit_budget(found.iter().copied().map(Datum::String), temp_storage)?;
    let mut row = Row::default();
    let mut packer = row.packer();
    packer.try_push_array(
        &[ArrayDimension {
            lower_bound: 1,
            length: found.len(),
        }],
        found.into_iter().map(Datum::String),
    )?;
    Ok(temp_storage.push_unary_row(row))
}

// NOTE: no budget pre-check, see the exception on `check_build_fits_budget`.
#[sqlfunc(propagates_nulls = true)]
fn pretty_sql<'a>(sql: &str, width: i32, temp_storage: &'a RowArena) -> Result<&'a str, EvalError> {
    let width =
        usize::try_from(width).map_err(|_| EvalError::PrettyError("invalid width".into()))?;
    let pretty = pretty_str(
        sql,
        PrettyConfig {
            width,
            format_mode: FormatMode::Simple,
        },
    )
    .map_err(|e| EvalError::PrettyError(e.to_string().into()))?;
    let pretty = temp_storage.push_string(pretty);
    Ok(pretty)
}

// NOTE: no budget pre-check, see the exception on `check_build_fits_budget`.
#[sqlfunc]
fn redact_sql(sql: &str) -> Result<String, EvalError> {
    let stmts = mz_sql_parser::parser::parse_statements(sql)
        .map_err(|e| EvalError::RedactError(e.to_string().into()))?;
    match stmts.len() {
        1 => Ok(stmts[0].ast.to_ast_string_redacted()),
        n => Err(EvalError::RedactError(
            format!("expected a single statement, found {n}").into(),
        )),
    }
}

#[sqlfunc(propagates_nulls = true)]
fn starts_with(a: &str, b: &str) -> bool {
    a.starts_with(b)
}

#[sqlfunc(
    sqlname = "||",
    is_infix_op = true,
    propagates_nulls = true,
    // Text concatenation is monotonic in its second argument, because if I change the
    // second argument but don't change the first argument, then we won't find a difference
    // in that part of the concatenation result that came from the first argument, so we'll
    // find the difference that comes from changing the second argument.
    // (It's not monotonic in its first argument, because e.g.,
    // 'A' < 'AA' but 'AZ' > 'AAZ'.)
    is_monotone = (false, true),
)]
fn text_concat_binary(a: &str, b: &str, temp_storage: &RowArena) -> Result<String, EvalError> {
    if a.len() + b.len() > max_string_func_result_bytes(temp_storage) {
        return Err(EvalError::LengthTooLarge);
    }
    let mut buf = String::with_capacity(a.len() + b.len());
    buf.push_str(a);
    buf.push_str(b);
    Ok(buf)
}

#[sqlfunc(propagates_nulls = true, introduces_nulls = false)]
fn like_escape<'a>(
    pattern: &str,
    b: &str,
    temp_storage: &'a RowArena,
) -> Result<&'a str, EvalError> {
    let escape = like_pattern::EscapeBehavior::from_str(b)?;
    let normalized = like_pattern::normalize_pattern(pattern, escape)?;
    Ok(temp_storage.push_string(normalized))
}

#[sqlfunc(is_infix_op = true, sqlname = "like")]
fn is_like_match_case_sensitive(haystack: &str, pattern: &str) -> Result<bool, EvalError> {
    like_pattern::compile(pattern, false).map(|needle| needle.is_match(haystack))
}

#[sqlfunc(is_infix_op = true, sqlname = "ilike")]
fn is_like_match_case_insensitive(haystack: &str, pattern: &str) -> Result<bool, EvalError> {
    like_pattern::compile(pattern, true).map(|needle| needle.is_match(haystack))
}

#[sqlfunc(is_infix_op = true, sqlname = "~")]
fn is_regexp_match_case_sensitive(haystack: &str, needle: &str) -> Result<bool, EvalError> {
    let regex = build_regex(needle, "")?;
    Ok(regex.is_match(haystack))
}

#[sqlfunc(is_infix_op = true, sqlname = "~*")]
fn is_regexp_match_case_insensitive(haystack: &str, needle: &str) -> Result<bool, EvalError> {
    let regex = build_regex(needle, "i")?;
    Ok(regex.is_match(haystack))
}

fn regexp_match_static<'a>(
    haystack: Datum<'a>,
    temp_storage: &'a RowArena,
    needle: &regex::Regex,
) -> Result<Datum<'a>, EvalError> {
    let mut row = Row::default();
    let mut packer = row.packer();
    if needle.captures_len() > 1 {
        // The regex contains capture groups, so return an array containing the
        // matched text in each capture group, unless the entire match fails.
        // Individual capture groups may also be null if that group did not
        // participate in the match.
        match needle.captures(haystack.unwrap_str()) {
            None => packer.push(Datum::Null),
            Some(captures) => packer.try_push_array(
                &[ArrayDimension {
                    lower_bound: 1,
                    length: captures.len() - 1,
                }],
                // Skip the 0th capture group, which is the whole match.
                captures.iter().skip(1).map(|mtch| match mtch {
                    None => Datum::Null,
                    Some(mtch) => Datum::String(mtch.as_str()),
                }),
            )?,
        }
    } else {
        // The regex contains no capture groups, so return a one-element array
        // containing the match, or null if there is no match.
        match needle.find(haystack.unwrap_str()) {
            None => packer.push(Datum::Null),
            Some(mtch) => packer.try_push_array(
                &[ArrayDimension {
                    lower_bound: 1,
                    length: 1,
                }],
                iter::once(Datum::String(mtch.as_str())),
            )?,
        };
    };
    Ok(temp_storage.push_unary_row(row))
}

/// Sets `limit` based on the presence of 'g' in `flags` for use in `Regex::replacen`,
/// and removes 'g' from `flags` if present.
pub(crate) fn regexp_replace_parse_flags(flags: &str) -> (usize, Cow<'_, str>) {
    // 'g' means to replace all instead of the first. Use a Cow to avoid allocating in the fast
    // path. We could switch build_regex to take an iter which would also achieve that.
    let (limit, flags) = if flags.contains('g') {
        let flags = flags.replace('g', "");
        (0, Cow::Owned(flags))
    } else {
        (1, Cow::Borrowed(flags))
    };
    (limit, flags)
}

pub fn build_regex(needle: &str, flags: &str) -> Result<Regex, EvalError> {
    let mut case_insensitive = false;
    // Note: Postgres accepts it when both flags are present, taking the last one. We do the same.
    for f in flags.chars() {
        match f {
            'i' => {
                case_insensitive = true;
            }
            'c' => {
                case_insensitive = false;
            }
            _ => return Err(EvalError::InvalidRegexFlag(f)),
        }
    }
    Ok(Regex::new(needle, case_insensitive)?)
}

#[sqlfunc(sqlname = "repeat")]
fn repeat_string(string: &str, count: i32, temp_storage: &RowArena) -> Result<String, EvalError> {
    let len = usize::try_from(count).unwrap_or(0);
    if len.saturating_mul(string.len()) > max_string_func_result_bytes(temp_storage) {
        return Err(EvalError::LengthTooLarge);
    }
    Ok(string.repeat(len))
}

/// Constructs a new zero or one dimensional array out of an arbitrary number of
/// scalars.
///
/// If `datums` is empty, constructs a zero-dimensional array. Otherwise,
/// constructs a one dimensional array whose lower bound is one and whose length
/// is equal to `datums.len()`.
fn array_create_scalar<'a>(
    datums: &[Datum<'a>],
    temp_storage: &'a RowArena,
) -> Result<Datum<'a>, EvalError> {
    let mut dims = &[ArrayDimension {
        lower_bound: 1,
        length: datums.len(),
    }][..];
    if datums.is_empty() {
        // Per PostgreSQL, empty arrays are represented with zero dimensions,
        // not one dimension of zero length. We write this condition a little
        // strangely to satisfy the borrow checker while avoiding an allocation.
        dims = &[];
    }
    check_datums_fit_budget(datums.iter().copied(), temp_storage)?;
    let datum = temp_storage.try_make_datum(|packer| packer.try_push_array(dims, datums))?;
    Ok(datum)
}

fn stringify_datum<'a, B>(
    buf: &mut B,
    d: Datum<'a>,
    ty: &SqlScalarType,
) -> Result<strconv::Nestable, EvalError>
where
    B: FormatBuffer,
{
    use SqlScalarType::*;
    match &ty {
        AclItem => Ok(strconv::format_acl_item(buf, d.unwrap_acl_item())),
        Bool => Ok(strconv::format_bool(buf, d.unwrap_bool())),
        Int16 => Ok(strconv::format_int16(buf, d.unwrap_int16())),
        Int32 => Ok(strconv::format_int32(buf, d.unwrap_int32())),
        Int64 => Ok(strconv::format_int64(buf, d.unwrap_int64())),
        UInt16 => Ok(strconv::format_uint16(buf, d.unwrap_uint16())),
        UInt32 | Oid | RegClass | RegProc | RegType => {
            Ok(strconv::format_uint32(buf, d.unwrap_uint32()))
        }
        UInt64 => Ok(strconv::format_uint64(buf, d.unwrap_uint64())),
        Float32 => Ok(strconv::format_float32(buf, d.unwrap_float32())),
        Float64 => Ok(strconv::format_float64(buf, d.unwrap_float64())),
        Numeric { .. } => Ok(strconv::format_numeric(buf, &d.unwrap_numeric())),
        Date => Ok(strconv::format_date(buf, d.unwrap_date())),
        Time => Ok(strconv::format_time(buf, d.unwrap_time())),
        Timestamp { .. } => Ok(strconv::format_timestamp(buf, &d.unwrap_timestamp())),
        TimestampTz { .. } => Ok(strconv::format_timestamptz(buf, &d.unwrap_timestamptz())),
        Interval => Ok(strconv::format_interval(buf, d.unwrap_interval())),
        Bytes => Ok(strconv::format_bytes(buf, d.unwrap_bytes())),
        String | VarChar { .. } | PgLegacyName => Ok(strconv::format_string(buf, d.unwrap_str())),
        Char { length } => Ok(strconv::format_string(
            buf,
            &mz_repr::adt::char::format_str_pad(d.unwrap_str(), *length),
        )),
        PgLegacyChar => {
            format_pg_legacy_char(buf, d.unwrap_uint8())?;
            Ok(strconv::Nestable::MayNeedEscaping)
        }
        Jsonb => Ok(strconv::format_jsonb(buf, JsonbRef::from_datum(d))),
        Uuid => Ok(strconv::format_uuid(buf, d.unwrap_uuid())),
        Record { fields, .. } => {
            let mut fields = fields.iter();
            strconv::format_record(buf, d.unwrap_list(), |buf, d| {
                let (_name, ty) = fields.next().unwrap();
                if d.is_null() {
                    Ok(buf.write_null())
                } else {
                    stringify_datum(buf.nonnull_buffer(), d, &ty.scalar_type)
                }
            })
        }
        Array(elem_type) => strconv::format_array(
            buf,
            &d.unwrap_array().dims().into_iter().collect::<Vec<_>>(),
            d.unwrap_array().elements(),
            |buf, d| {
                if d.is_null() {
                    Ok(buf.write_null())
                } else {
                    stringify_datum(buf.nonnull_buffer(), d, elem_type)
                }
            },
        ),
        List { element_type, .. } => strconv::format_list(buf, d.unwrap_list(), |buf, d| {
            if d.is_null() {
                Ok(buf.write_null())
            } else {
                stringify_datum(buf.nonnull_buffer(), d, element_type)
            }
        }),
        Map { value_type, .. } => strconv::format_map(buf, &d.unwrap_map(), |buf, d| {
            if d.is_null() {
                Ok(buf.write_null())
            } else {
                stringify_datum(buf.nonnull_buffer(), d, value_type)
            }
        }),
        Int2Vector => strconv::format_legacy_vector(buf, d.unwrap_array().elements(), |buf, d| {
            stringify_datum(buf.nonnull_buffer(), d, &SqlScalarType::Int16)
        }),
        MzTimestamp { .. } => Ok(strconv::format_mz_timestamp(buf, d.unwrap_mz_timestamp())),
        Range { element_type } => strconv::format_range(buf, &d.unwrap_range(), |buf, d| match d {
            Some(d) => stringify_datum(buf.nonnull_buffer(), *d, element_type),
            None => Ok::<_, EvalError>(buf.write_null()),
        }),
        MzAclItem => Ok(strconv::format_mz_acl_item(buf, d.unwrap_mz_acl_item())),
    }
}

#[sqlfunc]
fn position(substring: &str, string: &str) -> Result<i32, EvalError> {
    let char_index = string.find(substring);

    if let Some(char_index) = char_index {
        // find the index in char space
        let string_prefix = &string[0..char_index];

        let num_prefix_chars = string_prefix.chars().count();
        let num_prefix_chars = i32::try_from(num_prefix_chars)
            .map_err(|_| EvalError::Int32OutOfRange(num_prefix_chars.to_string().into()))?;

        Ok(num_prefix_chars + 1)
    } else {
        Ok(0)
    }
}

#[sqlfunc]
fn strpos(string: &str, substring: &str) -> Result<i32, EvalError> {
    position(substring, string)
}

#[sqlfunc(
    propagates_nulls = true,
    // `left` is unfortunately not monotonic (at least for negative second arguments),
    // because 'aa' < 'z', but `left(_, -1)` makes 'a' > ''.
    is_monotone = (false, false)
)]
fn left<'a>(string: &'a str, b: i32) -> Result<&'a str, EvalError> {
    let n = i64::from(b);

    let mut byte_indices = string.char_indices().map(|(i, _)| i);

    let end_in_bytes = match n.cmp(&0) {
        Ordering::Equal => 0,
        Ordering::Greater => {
            let n = usize::try_from(n).map_err(|_| {
                EvalError::InvalidParameterValue(format!("invalid parameter n: {:?}", n).into())
            })?;
            // nth from the back
            byte_indices.nth(n).unwrap_or(string.len())
        }
        Ordering::Less => {
            let n = usize::try_from(n.abs() - 1).map_err(|_| {
                EvalError::InvalidParameterValue(format!("invalid parameter n: {:?}", n).into())
            })?;
            byte_indices.rev().nth(n).unwrap_or(0)
        }
    };

    Ok(&string[..end_in_bytes])
}

#[sqlfunc(propagates_nulls = true)]
fn right<'a>(string: &'a str, n: i32) -> Result<&'a str, EvalError> {
    let mut byte_indices = string.char_indices().map(|(i, _)| i);

    let start_in_bytes = if n == 0 {
        string.len()
    } else if n > 0 {
        let n = usize::try_from(n - 1).map_err(|_| {
            EvalError::InvalidParameterValue(format!("invalid parameter n: {:?}", n).into())
        })?;
        // nth from the back
        byte_indices.rev().nth(n).unwrap_or(0)
    } else if n == i32::MIN {
        // this seems strange but Postgres behaves like this
        0
    } else {
        let n = n.abs();
        let n = usize::try_from(n).map_err(|_| {
            EvalError::InvalidParameterValue(format!("invalid parameter n: {:?}", n).into())
        })?;
        byte_indices.nth(n).unwrap_or(string.len())
    };

    Ok(&string[start_in_bytes..])
}

#[sqlfunc(sqlname = "btrim", propagates_nulls = true)]
fn trim<'a>(a: &'a str, trim_chars: &str) -> &'a str {
    a.trim_matches(|c| trim_chars.contains(c))
}

#[sqlfunc(sqlname = "ltrim", propagates_nulls = true)]
fn trim_leading<'a>(a: &'a str, trim_chars: &str) -> &'a str {
    a.trim_start_matches(|c| trim_chars.contains(c))
}

#[sqlfunc(sqlname = "rtrim", propagates_nulls = true)]
fn trim_trailing<'a>(a: &'a str, trim_chars: &str) -> &'a str {
    a.trim_end_matches(|c| trim_chars.contains(c))
}

#[sqlfunc(
    sqlname = "array_length",
    propagates_nulls = true,
    introduces_nulls = true
)]
fn array_length<'a>(a: Array<'a>, b: i64) -> Result<Option<i32>, EvalError> {
    let i = match usize::try_from(b) {
        Ok(0) | Err(_) => return Ok(None),
        Ok(n) => n - 1,
    };
    Ok(match a.dims().into_iter().nth(i) {
        None => None,
        Some(dim) => Some(
            dim.length
                .try_into()
                .map_err(|_| EvalError::Int32OutOfRange(dim.length.to_string().into()))?,
        ),
    })
}

#[sqlfunc(is_infix_op = true)]
// TODO(benesch): remove potentially dangerous usage of `as`.
#[allow(clippy::as_conversions)]
fn array_lower<'a>(a: Array<'a>, i: i64) -> Result<Option<i32>, EvalError> {
    if i < 1 {
        return Ok(None);
    }
    a.dims()
        .into_iter()
        .nth(i as usize - 1)
        .map(|dim| {
            let (lower, _upper) = dim.dimension_bounds();
            lower
                .try_into()
                .map_err(|_| EvalError::Int32OutOfRange(lower.to_string().into()))
        })
        .transpose()
}

#[sqlfunc(
    output_type_expr = "input_types[0].scalar_type.without_modifiers().nullable(true)",
    sqlname = "array_remove",
    propagates_nulls = false,
    introduces_nulls = false
)]
fn array_remove<'a>(
    arr: Array<'a>,
    b: Datum<'a>,
    temp_storage: &'a RowArena,
) -> Result<Datum<'a>, EvalError> {
    // Zero-dimensional arrays are empty by definition
    if arr.dims().len() == 0 {
        return Ok(Datum::Array(arr));
    }

    // array_remove only supports one-dimensional arrays
    if arr.dims().len() > 1 {
        return Err(EvalError::MultidimensionalArrayRemovalNotSupported);
    }

    let mut dims = arr.dims().into_iter().collect::<Vec<_>>();
    // Removal can't grow the result, but the transient `Vec<Datum>` it filters into is a fresh
    // input-scaled allocation. One-dimensional by the check above, so dim 0's length is the count.
    check_build_fits_budget(
        || dims[0].length,
        std::mem::size_of::<Datum<'a>>(),
        temp_storage,
    )?;

    let elems: Vec<_> = arr.elements().iter().filter(|v| v != &b).collect();
    // This access is safe because `dims` is guaranteed to be non-empty
    dims[0] = ArrayDimension {
        lower_bound: 1,
        length: elems.len(),
    };

    Ok(temp_storage.try_make_datum(|packer| packer.try_push_array(&dims, elems))?)
}

#[sqlfunc(is_infix_op = true)]
// TODO(benesch): remove potentially dangerous usage of `as`.
#[allow(clippy::as_conversions)]
fn array_upper<'a>(a: Array<'a>, i: i64) -> Result<Option<i32>, EvalError> {
    if i < 1 {
        return Ok(None);
    }
    a.dims()
        .into_iter()
        .nth(i as usize - 1)
        .map(|dim| {
            let (_lower, upper) = dim.dimension_bounds();
            upper
                .try_into()
                .map_err(|_| EvalError::Int32OutOfRange(upper.to_string().into()))
        })
        .transpose()
}

#[sqlfunc(
    is_infix_op = true,
    sqlname = "array_contains",
    propagates_nulls = true,
    introduces_nulls = false
)]
fn array_contains<'a>(a: Datum<'a>, array: Array<'a>) -> bool {
    array.elements().iter().any(|e| e == a)
}

#[sqlfunc(is_infix_op = true, sqlname = "@>")]
fn array_contains_array<'a>(a: Array<'a>, b: Array<'a>) -> bool {
    let a = a.elements();
    let b = b.elements();

    // NULL is never equal to NULL. If NULL is an element of b, b cannot be contained in a, even if a contains NULL.
    if b.iter().contains(&Datum::Null) {
        false
    } else {
        b.iter()
            .all(|item_b| a.iter().any(|item_a| item_a == item_b))
    }
}

#[sqlfunc(is_infix_op = true, sqlname = "<@")]
fn array_contains_array_rev<'a>(a: Array<'a>, b: Array<'a>) -> bool {
    array_contains_array(b, a)
}

#[sqlfunc(
    output_type_expr = "input_types[0].scalar_type.without_modifiers().nullable(true)",
    is_infix_op = true,
    sqlname = "||",
    propagates_nulls = false,
    introduces_nulls = false
)]
fn array_array_concat<'a>(
    a: Option<Array<'a>>,
    b: Option<Array<'a>>,
    temp_storage: &'a RowArena,
) -> Result<Option<Array<'a>>, EvalError> {
    let Some(a_array) = a else {
        return Ok(b);
    };
    let Some(b_array) = b else {
        return Ok(a);
    };

    let a_dims: Vec<ArrayDimension> = a_array.dims().into_iter().collect();
    let b_dims: Vec<ArrayDimension> = b_array.dims().into_iter().collect();

    let a_ndims = a_dims.len();
    let b_ndims = b_dims.len();

    // Per PostgreSQL, if either of the input arrays is zero dimensional,
    // the output is the other array, no matter their dimensions.
    if a_ndims == 0 {
        return Ok(b);
    } else if b_ndims == 0 {
        return Ok(a);
    }

    // Postgres supports concatenating arrays of different dimensions,
    // as long as one of the arrays has the same type as an element of
    // the other array, i.e. `int[2][4] || int[4]` (or `int[4] || int[2][4]`)
    // works, because each element of `int[2][4]` is an `int[4]`.
    // This check is separate from the one below because Postgres gives a
    // specific error message if the number of dimensions differs by more
    // than one.
    // This cast is safe since MAX_ARRAY_DIMENSIONS is 6
    // Can be replaced by .abs_diff once it is stabilized
    // TODO(benesch): remove potentially dangerous usage of `as`.
    #[allow(clippy::as_conversions)]
    if (a_ndims as isize - b_ndims as isize).abs() > 1 {
        return Err(EvalError::IncompatibleArrayDimensions {
            dims: Some((a_ndims, b_ndims)),
        });
    }

    let mut dims;

    // After the checks above, we are certain that:
    // - neither array is zero dimensional nor empty
    // - both arrays have the same number of dimensions, or differ
    //   at most by one.
    match a_ndims.cmp(&b_ndims) {
        // If both arrays have the same number of dimensions, validate
        // that their inner dimensions are the same and concatenate the
        // arrays.
        Ordering::Equal => {
            if &a_dims[1..] != &b_dims[1..] {
                return Err(EvalError::IncompatibleArrayDimensions { dims: None });
            }
            dims = vec![ArrayDimension {
                lower_bound: a_dims[0].lower_bound,
                length: a_dims[0].length + b_dims[0].length,
            }];
            dims.extend(&a_dims[1..]);
        }
        // If `a` has less dimensions than `b`, this is an element-array
        // concatenation, which requires that `a` has the same dimensions
        // as an element of `b`.
        Ordering::Less => {
            if &a_dims[..] != &b_dims[1..] {
                return Err(EvalError::IncompatibleArrayDimensions { dims: None });
            }
            dims = vec![ArrayDimension {
                lower_bound: b_dims[0].lower_bound,
                // Since `a` is treated as an element of `b`, the length of
                // the first dimension of `b` is incremented by one, as `a` is
                // non-empty.
                length: b_dims[0].length + 1,
            }];
            dims.extend(a_dims);
        }
        // If `a` has more dimensions than `b`, this is an array-element
        // concatenation, which requires that `b` has the same dimensions
        // as an element of `a`.
        Ordering::Greater => {
            if &a_dims[1..] != &b_dims[..] {
                return Err(EvalError::IncompatibleArrayDimensions { dims: None });
            }
            dims = vec![ArrayDimension {
                lower_bound: a_dims[0].lower_bound,
                // Since `b` is treated as an element of `a`, the length of
                // the first dimension of `a` is incremented by one, as `b`
                // is non-empty.
                length: a_dims[0].length + 1,
            }];
            dims.extend(b_dims);
        }
    }

    let elems = a_array.elements().iter().chain(b_array.elements().iter());

    let datum = temp_storage.try_make_datum(|packer| packer.try_push_array(&dims, elems))?;
    Ok(Some(datum.unwrap_array()))
}

#[sqlfunc(
    is_infix_op = true,
    sqlname = "||",
    propagates_nulls = false,
    introduces_nulls = false
)]
fn list_list_concat<'a, T: FromDatum<'a>>(
    a: Option<DatumList<'a, T>>,
    b: Option<DatumList<'a, T>>,
    temp_storage: &'a RowArena,
) -> Option<DatumList<'a, T>> {
    let Some(a) = a else {
        return b;
    };
    let Some(b) = b else {
        return Some(a);
    };

    Some(temp_storage.make_datum_list(a.typed_iter().chain(b.typed_iter())))
}

#[sqlfunc(is_infix_op = true, sqlname = "||", propagates_nulls = false)]
fn list_element_concat<'a, T: FromDatum<'a>>(
    a: Option<DatumList<'a, T>>,
    b: T,
    temp_storage: &'a RowArena,
) -> DatumList<'a, T> {
    let a_elems = a.into_iter().flat_map(|a| a.typed_iter());
    temp_storage.make_datum_list(a_elems.chain(std::iter::once(b)))
}

// Note that the output type corresponds to the _second_ parameter's input type.
#[sqlfunc(is_infix_op = true, sqlname = "||", propagates_nulls = false)]
fn element_list_concat<'a, T: FromDatum<'a>>(
    a: T,
    b: Option<DatumList<'a, T>>,
    temp_storage: &'a RowArena,
) -> DatumList<'a, T> {
    let b_elems = b.into_iter().flat_map(|b| b.typed_iter());
    temp_storage.make_datum_list(std::iter::once(a).chain(b_elems))
}

#[sqlfunc(sqlname = "list_remove")]
fn list_remove<'a, T: FromDatum<'a>>(
    a: DatumList<'a, T>,
    b: T,
    temp_storage: &'a RowArena,
) -> DatumList<'a, T> {
    temp_storage.make_datum_list(a.typed_iter().filter(|elem| *elem != b))
}

#[sqlfunc(sqlname = "digest")]
fn digest_string(to_digest: &str, digest_fn: &str) -> Result<Vec<u8>, EvalError> {
    digest_inner(to_digest.as_bytes(), digest_fn)
}

#[sqlfunc(sqlname = "digest")]
fn digest_bytes(to_digest: &[u8], digest_fn: &str) -> Result<Vec<u8>, EvalError> {
    digest_inner(to_digest, digest_fn)
}

fn digest_inner(bytes: &[u8], digest_fn: &str) -> Result<Vec<u8>, EvalError> {
    match digest_fn {
        "md5" => Ok(Md5::digest(bytes).to_vec()),
        "sha1" => Ok(digest::digest(&digest::SHA1_FOR_LEGACY_USE_ONLY, bytes)
            .as_ref()
            .to_vec()),
        "sha224" => Ok(digest::digest(&digest::SHA224, bytes).as_ref().to_vec()),
        "sha256" => Ok(digest::digest(&digest::SHA256, bytes).as_ref().to_vec()),
        "sha384" => Ok(digest::digest(&digest::SHA384, bytes).as_ref().to_vec()),
        "sha512" => Ok(digest::digest(&digest::SHA512, bytes).as_ref().to_vec()),
        other => Err(EvalError::InvalidHashAlgorithm(other.into())),
    }
}

#[sqlfunc]
fn mz_render_typmod(oid: u32, typmod: i32) -> String {
    match Type::from_oid_and_typmod(oid, typmod) {
        Ok(typ) => typ.constraint().display_or("").to_string(),
        // Match dubious PostgreSQL behavior of outputting the unmodified
        // `typmod` when positive if the type OID/typmod is invalid.
        Err(_) if typmod >= 0 => format!("({typmod})"),
        Err(_) => "".into(),
    }
}

#[cfg(test)]
mod test {
    use chrono::prelude::*;
    use mz_repr::PropDatum;
    use proptest::prelude::*;

    use super::*;
    use crate::{Eval, MirScalarExpr};

    #[mz_ore::test]
    fn variant_names_unique() {
        // `from_variant_name` resolves the first variant with a matching
        // canonical name, so a duplicate name (from a typo in `func_name!` or
        // colliding function names) would silently shadow a variant.
        fn assert_unique(enum_name: &str, names: impl Iterator<Item = &'static str>) {
            let mut seen = std::collections::BTreeSet::new();
            for name in names {
                assert!(seen.insert(name), "duplicate {enum_name} name: {name}");
            }
        }
        assert_unique("UnaryFunc", UnaryFunc::variant_names());
        assert_unique("BinaryFunc", BinaryFunc::variant_names());
        assert_unique("VariadicFunc", VariadicFunc::variant_names());
    }

    #[mz_ore::test]
    fn add_interval_months() {
        let dt = ym(2000, 1);

        assert_eq!(add_timestamp_months(&*dt, 0).unwrap(), dt);
        assert_eq!(add_timestamp_months(&*dt, 1).unwrap(), ym(2000, 2));
        assert_eq!(add_timestamp_months(&*dt, 12).unwrap(), ym(2001, 1));
        assert_eq!(add_timestamp_months(&*dt, 13).unwrap(), ym(2001, 2));
        assert_eq!(add_timestamp_months(&*dt, 24).unwrap(), ym(2002, 1));
        assert_eq!(add_timestamp_months(&*dt, 30).unwrap(), ym(2002, 7));

        // and negatives
        assert_eq!(add_timestamp_months(&*dt, -1).unwrap(), ym(1999, 12));
        assert_eq!(add_timestamp_months(&*dt, -12).unwrap(), ym(1999, 1));
        assert_eq!(add_timestamp_months(&*dt, -13).unwrap(), ym(1998, 12));
        assert_eq!(add_timestamp_months(&*dt, -24).unwrap(), ym(1998, 1));
        assert_eq!(add_timestamp_months(&*dt, -30).unwrap(), ym(1997, 7));

        // and going over a year boundary by less than a year
        let dt = ym(1999, 12);
        assert_eq!(add_timestamp_months(&*dt, 1).unwrap(), ym(2000, 1));
        let end_of_month_dt = NaiveDate::from_ymd_opt(1999, 12, 31)
            .unwrap()
            .and_hms_opt(9, 9, 9)
            .unwrap();
        assert_eq!(
            // leap year
            add_timestamp_months(&end_of_month_dt, 2).unwrap(),
            NaiveDate::from_ymd_opt(2000, 2, 29)
                .unwrap()
                .and_hms_opt(9, 9, 9)
                .unwrap()
                .try_into()
                .unwrap(),
        );
        assert_eq!(
            // not leap year
            add_timestamp_months(&end_of_month_dt, 14).unwrap(),
            NaiveDate::from_ymd_opt(2001, 2, 28)
                .unwrap()
                .and_hms_opt(9, 9, 9)
                .unwrap()
                .try_into()
                .unwrap(),
        );
    }

    fn ym(year: i32, month: u32) -> CheckedTimestamp<NaiveDateTime> {
        NaiveDate::from_ymd_opt(year, month, 1)
            .unwrap()
            .and_hms_opt(9, 9, 9)
            .unwrap()
            .try_into()
            .unwrap()
    }

    #[mz_ore::test]
    fn array_lower_upper_respect_lower_bound() {
        use mz_repr::adt::array::ArrayDimension;
        use mz_repr::{Datum, RowArena};

        let arena = RowArena::new();

        // Builds a one-dimensional array with the given lower bound and length,
        // then returns (array_lower(_, 1), array_upper(_, 1)).
        let bounds = |lower_bound: isize, length: usize| {
            let dims = [ArrayDimension {
                lower_bound,
                length,
            }];
            let elems = vec![Datum::Int32(0); length];
            let datum = arena.make_datum(|packer| packer.try_push_array(&dims, elems).unwrap());
            let arr = match datum {
                Datum::Array(arr) => arr,
                other => panic!("expected array, got {other:?}"),
            };
            (array_lower(arr, 1).unwrap(), array_upper(arr, 1).unwrap())
        };

        // Default lower bound of 1: array_fill(0, ARRAY[3]).
        assert_eq!(bounds(1, 3), (Some(1), Some(3)));
        // Lower bound of 5: array_fill(0, ARRAY[3], ARRAY[5]) => [5:7].
        assert_eq!(bounds(5, 3), (Some(5), Some(7)));
        // Negative lower bound: array_fill(0, ARRAY[3], ARRAY[-3]) => [-3:-1].
        assert_eq!(bounds(-3, 3), (Some(-3), Some(-1)));

        // Out-of-range dimensions return None rather than the bound.
        let dims = [ArrayDimension {
            lower_bound: 5,
            length: 3,
        }];
        let elems = vec![Datum::Int32(0); 3];
        let datum = arena.make_datum(|packer| packer.try_push_array(&dims, elems).unwrap());
        let arr = match datum {
            Datum::Array(arr) => arr,
            other => panic!("expected array, got {other:?}"),
        };
        assert_eq!(array_lower(arr, 0).unwrap(), None);
        assert_eq!(array_upper(arr, 0).unwrap(), None);
        assert_eq!(array_lower(arr, 2).unwrap(), None);
        assert_eq!(array_upper(arr, 2).unwrap(), None);
    }

    #[mz_ore::test]
    #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `decNumberFromInt32` on OS `linux`
    fn test_is_monotone() {
        use proptest::prelude::*;

        /// Asserts that the function is either monotonically increasing or decreasing over
        /// the given sets of arguments.
        fn assert_monotone<'a, const N: usize>(
            expr: &MirScalarExpr,
            arena: &'a RowArena,
            datums: &[[Datum<'a>; N]],
        ) {
            // TODO: assertions for nulls, errors
            let Ok(results) = datums
                .iter()
                .map(|args| expr.eval(args.as_slice(), arena))
                .collect::<Result<Vec<_>, _>>()
            else {
                return;
            };

            let forward = results.iter().tuple_windows().all(|(a, b)| a <= b);
            let reverse = results.iter().tuple_windows().all(|(a, b)| a >= b);
            assert!(
                forward || reverse,
                "expected {expr} to be monotone, but passing {datums:?} returned {results:?}"
            );
        }

        fn proptest_binary<'a>(
            func: BinaryFunc,
            arena: &'a RowArena,
            left: impl Strategy<Value = PropDatum>,
            right: impl Strategy<Value = PropDatum>,
        ) {
            let (left_monotone, right_monotone) = func.is_monotone();
            let expr = MirScalarExpr::CallBinary {
                func,
                expr1: Box::new(MirScalarExpr::column(0)),
                expr2: Box::new(MirScalarExpr::column(1)),
            };
            proptest!(|(
                mut left in proptest::array::uniform3(left),
                mut right in proptest::array::uniform3(right),
            )| {
                left.sort();
                right.sort();
                if left_monotone {
                    for r in &right {
                        let args: Vec<[_; 2]> = left
                            .iter()
                            .map(|l| [Datum::from(l), Datum::from(r)])
                            .collect();
                        assert_monotone(&expr, arena, &args);
                    }
                }
                if right_monotone {
                    for l in &left {
                        let args: Vec<[_; 2]> = right
                            .iter()
                            .map(|r| [Datum::from(l), Datum::from(r)])
                            .collect();
                        assert_monotone(&expr, arena, &args);
                    }
                }
            });
        }

        let interesting_strs: Vec<_> = SqlScalarType::String.interesting_datums().collect();
        let str_datums = proptest::strategy::Union::new([
            proptest::string::string_regex("[A-Z]{0,10}")
                .expect("valid regex")
                .prop_map(|s| PropDatum::String(s.to_string()))
                .boxed(),
            (0..interesting_strs.len())
                .prop_map(move |i| {
                    let Datum::String(val) = interesting_strs[i] else {
                        unreachable!("interesting strings has non-strings")
                    };
                    PropDatum::String(val.to_string())
                })
                .boxed(),
        ]);

        let interesting_i32s: Vec<Datum<'static>> =
            SqlScalarType::Int32.interesting_datums().collect();
        let i32_datums = proptest::strategy::Union::new([
            any::<i32>().prop_map(PropDatum::Int32).boxed(),
            (0..interesting_i32s.len())
                .prop_map(move |i| {
                    let Datum::Int32(val) = interesting_i32s[i] else {
                        unreachable!("interesting int32 has non-i32s")
                    };
                    PropDatum::Int32(val)
                })
                .boxed(),
            (-10i32..10).prop_map(PropDatum::Int32).boxed(),
        ]);

        let arena = RowArena::new();

        // It would be interesting to test all funcs here, but we currently need to hardcode
        // the generators for the argument types, which makes this tedious. Choose an interesting
        // subset for now.
        proptest_binary(
            BinaryFunc::AddInt32(AddInt32),
            &arena,
            &i32_datums,
            &i32_datums,
        );
        proptest_binary(SubInt32.into(), &arena, &i32_datums, &i32_datums);
        proptest_binary(MulInt32.into(), &arena, &i32_datums, &i32_datums);
        proptest_binary(DivInt32.into(), &arena, &i32_datums, &i32_datums);
        proptest_binary(TextConcatBinary.into(), &arena, &str_datums, &str_datums);
        proptest_binary(Left.into(), &arena, &str_datums, &i32_datums);
    }
}