mz_sql/plan/

typeconv.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.

//! Maintains a catalog of valid casts between [`mz_repr::SqlScalarType`]s, as well as
//! other cast-related functions.

use std::cell::RefCell;
use std::collections::BTreeMap;
use std::sync::LazyLock;

use dynfmt::{Format, SimpleCurlyFormat};
use itertools::Itertools;
use mz_expr::func;
use mz_expr::func::variadic::{JsonbBuildObject, RecordCreate};
use mz_expr::func::{CastArrayToJsonb, CastListToJsonb};
use mz_ore::soft_panic_or_log;
use mz_repr::{
    ColumnName, Datum, Row, SqlColumnType, SqlRelationType, SqlScalarBaseType, SqlScalarType,
};
use mz_sql_parser::ast::CastFailureMode;

use crate::catalog::TypeCategory;
use crate::plan::error::PlanError;
use crate::plan::hir::{
    AbstractColumnType, CoercibleScalarExpr, CoercibleScalarType, HirScalarExpr, NameMetadata,
    UnaryFunc,
};
use crate::plan::query::{ExprContext, QueryContext};
use crate::plan::scope::Scope;

/// Like func::sql_impl_func, but for casts.
fn sql_impl_cast(expr: &'static str) -> CastTemplate {
    let invoke = crate::func::sql_impl(expr);
    CastTemplate::strict_only(move |ecx, _ccx, from_type, _to_type| {
        // Oddly, this needs to be able to gracefully fail so we can detect unmet dependencies.
        let mut out = invoke(ecx, vec![from_type.clone()]).ok()?;
        Some(move |e| {
            out.splice_parameters(&[e], 0);
            out
        })
    })
}

fn sql_impl_cast_per_context(casts: &[(CastContext, &'static str)]) -> CastTemplate {
    let casts: BTreeMap<CastContext, _> = casts
        .iter()
        .map(|(ccx, expr)| (ccx.clone(), crate::func::sql_impl(expr)))
        .collect();
    CastTemplate::strict_only(move |ecx, ccx, from_type, _to_type| {
        let invoke = &casts[&ccx];
        let r = invoke(ecx, vec![from_type.clone()]);
        let mut out = r.ok()?;
        Some(move |e| {
            out.splice_parameters(&[e], 0);
            out
        })
    })
}

/// A cast is a function that takes a `ScalarExpr` to another `ScalarExpr`.
type Cast = Box<dyn FnOnce(HirScalarExpr) -> HirScalarExpr>;

/// A cast template is a function that produces a `Cast` given a concrete input
/// and output type and a failure mode. A template can return `None` to
/// indicate that it is incapable of producing a cast for the specified types,
/// or for the specified failure mode.
///
/// Cast templates are used to share code for similar casts, where the input or
/// output type is of one "category" of type. For example, a single cast
/// template handles converting from strings to any list type. Without cast
/// templates, we'd have to enumerate every possible list -> list conversion,
/// which is impractical.
struct CastTemplate(
    Box<
        dyn Fn(
                &ExprContext,
                CastContext,
                CastFailureMode,
                &SqlScalarType,
                &SqlScalarType,
            ) -> Option<Cast>
            + Send
            + Sync,
    >,
);

impl CastTemplate {
    /// Builds a template from a builder of the erroring cast.
    ///
    /// Under [`CastFailureMode::NullFallback`] the builder is run once on a
    /// placeholder of the input type, and its result must be that placeholder
    /// wrapped in zero or more `CallUnary` nodes with nothing else in between.
    /// The template then emits those stages with each one that could error
    /// wrapped in `TryCast`, so a failure at any stage yields NULL, while a
    /// stage that cannot error is emitted as is and keeps its nullability.
    ///
    /// A builder whose result has any other shape, for example one that
    /// compares or branches on its input, makes `TRY_CAST` report that the
    /// cast cannot fall back to NULL, even when the strict cast cannot error.
    /// `test_try_cast_wraps_every_stage_of_every_cast` in `mz_adapter` plans
    /// every entry of the cast table in both modes and fails on such a
    /// builder. A template whose builder has another shape on purpose uses
    /// [`CastTemplate::strict_only`].
    fn new<T, C>(t: T) -> CastTemplate
    where
        T: Fn(&ExprContext, CastContext, &SqlScalarType, &SqlScalarType) -> Option<C>
            + Send
            + Sync
            + 'static,
        C: FnOnce(HirScalarExpr) -> HirScalarExpr + 'static,
    {
        CastTemplate(Box::new(move |ecx, ccx, mode, from_ty, to_ty| {
            let cast = t(ecx, ccx, from_ty, to_ty)?;
            match mode {
                CastFailureMode::Error => Some(Box::new(cast)),
                CastFailureMode::NullFallback => {
                    let stages = extract_unary_stages_for_try_cast(from_ty, cast)?;
                    Some(Box::new(move |mut expr: HirScalarExpr| {
                        for stage in stages {
                            expr = expr.call_unary(UnaryFunc::try_cast_if_fallible(stage));
                        }
                        expr
                    }))
                }
            }
        }))
    }

    /// Builds a template that supports only [`CastFailureMode::Error`].
    ///
    /// For casts implemented in SQL: their failures come from `mz_error_if_null`
    /// calls and from subquery cardinality, neither of which a scalar-level
    /// wrapper can turn into NULL.
    fn strict_only<T, C>(t: T) -> CastTemplate
    where
        T: Fn(&ExprContext, CastContext, &SqlScalarType, &SqlScalarType) -> Option<C>
            + Send
            + Sync
            + 'static,
        C: FnOnce(HirScalarExpr) -> HirScalarExpr + 'static,
    {
        CastTemplate(Box::new(move |ecx, ccx, mode, from_ty, to_ty| match mode {
            CastFailureMode::Error => Some(Box::new(t(ecx, ccx, from_ty, to_ty)?)),
            CastFailureMode::NullFallback => None,
        }))
    }
}

/// The unary functions a strict cast from `from` applies to its input,
/// innermost first, or `None` if the cast has any other shape. See
/// [`CastTemplate::new`] for the shape a cast must have.
///
/// Every stage after the first must propagate NULLs, so that the NULL a
/// failed stage produces flows through the rest of the chain unchanged:
/// `f(g(e))` matches `try[f](try[g](e))` only when `f` returns NULL on a NULL
/// input. A violation is reported as a planner bug rather than refused
/// silently, since every cast function propagates NULLs and a new one that
/// does not is a bug in that function.
fn extract_unary_stages_for_try_cast(
    from: &SqlScalarType,
    cast: impl FnOnce(HirScalarExpr) -> HirScalarExpr,
) -> Option<Vec<UnaryFunc>> {
    // Templates never inspect their input, so a dummy of the input type serves
    // to discover the stages they apply. Built directly because
    // `HirScalarExpr::literal` asserts that the datum inhabits the type, which
    // `Datum::Dummy` never does.
    let sentinel = HirScalarExpr::Literal(
        Row::pack([Datum::Dummy]),
        from.clone().nullable(false),
        NameMetadata::default(),
    );
    let mut expr = cast(sentinel.clone());
    let mut stages = vec![];
    loop {
        match expr {
            HirScalarExpr::CallUnary {
                func, expr: inner, ..
            } => {
                stages.push(func);
                expr = *inner;
            }
            other if other == sentinel => break,
            _ => return None,
        }
    }
    stages.reverse();
    if let Some(stage) = stages
        .iter()
        .skip(1)
        .find(|stage| !stage.propagates_nulls())
    {
        soft_panic_or_log!("cast stage {stage} does not propagate NULLs; TRY_CAST cannot wrap it");
        return None;
    }
    Some(stages)
}

impl From<UnaryFunc> for CastTemplate {
    fn from(u: UnaryFunc) -> CastTemplate {
        CastTemplate::new(move |_ecx, _ccx, _from, _to| {
            let u = u.clone();
            Some(move |expr: HirScalarExpr| expr.call_unary(u))
        })
    }
}

impl<const N: usize> From<[UnaryFunc; N]> for CastTemplate {
    fn from(funcs: [UnaryFunc; N]) -> CastTemplate {
        CastTemplate::new(move |_ecx, _ccx, _from, _to| {
            let funcs = funcs.clone();
            Some(move |mut expr: HirScalarExpr| {
                for func in funcs {
                    expr = expr.call_unary(func.clone());
                }
                expr
            })
        })
    }
}

/// STRING to REG*
///
/// A reg* type represents a specific type of object by oid.
///
/// Casting from a string to a reg*:
/// - Accepts `-`, the absent reference, as OID 0 in all cases, matching
///   PostgreSQL's `parseDashOrOid`.
/// - Accepts a string that looks like an OID and converts the value to the
///   specified reg* type. This is available in all cases except explicitly
///   casting text values to regclass (e.g. `SELECT '2'::text::regclass`)
/// - Resolves non-OID-appearing strings to objects. If this string resolves to
///   more than one OID (e.g. functions), it errors.
///
/// The below code provides a template to accomplish this for various reg*
/// types. Arguments in order are:
/// - 0: type catalog name this is casting to
/// - 1: the category of this reg for the error message
/// - 2: Whether or not to permit passing through numeric values as OIDs
const STRING_REG_CAST_TEMPLATE: &str = "
(SELECT
CASE
    WHEN $1 IS NULL THEN NULL
-- Handle the absent reference, which the reg* text output spells `-`. PostgreSQL's
-- `parseDashOrOid` accepts it for every reg* type, including text to regclass.
    WHEN $1 = '-' THEN 0::pg_catalog.oid::pg_catalog.{0}
-- Handle OID-like input, if available via {2}
    WHEN {2} AND pg_catalog.substring($1, 1, 1) BETWEEN '0' AND '9' THEN
        $1::pg_catalog.oid::pg_catalog.{0}
    ELSE (
    -- String case; look up that the item exists
        SELECT o.oid
        FROM mz_unsafe.mz_error_if_null(
            (
                -- We need to ensure a distinct here in the case of e.g. functions,
                -- where multiple items share a GlobalId.
                SELECT DISTINCT id AS name_id
                FROM mz_internal.mz_resolve_object_name('{0}', $1)
            ),
            -- TODO: Support the correct error code for does not exist (42883).
            '{1} \"' || $1 || '\" does not exist'
        ) AS i (name_id),
        -- Lateral lets us error separately from DNE case
        LATERAL (
            SELECT
                CASE
            -- Handle too many OIDs
                WHEN mz_catalog.list_length(mz_catalog.list_agg(oid)) > 1 THEN
                    mz_unsafe.mz_error_if_null(
                        NULL::pg_catalog.{0},
                        'more than one {1} named \"' || $1 || '\"'
                    )
            -- Resolve object name's OID if we know there is only one
                ELSE
                    CAST(mz_catalog.list_agg(oid)[1] AS pg_catalog.{0})
                END
            FROM mz_catalog.mz_objects
            WHERE id = name_id
            GROUP BY id
        ) AS o (oid)
    )
END)";

static STRING_TO_REGCLASS_EXPLICIT: LazyLock<String> = LazyLock::new(|| {
    SimpleCurlyFormat
        .format(STRING_REG_CAST_TEMPLATE, ["regclass", "relation", "false"])
        .unwrap()
        .to_string()
});

static STRING_TO_REGCLASS_COERCED: LazyLock<String> = LazyLock::new(|| {
    SimpleCurlyFormat
        .format(STRING_REG_CAST_TEMPLATE, ["regclass", "relation", "true"])
        .unwrap()
        .to_string()
});

static STRING_TO_REGPROC: LazyLock<String> = LazyLock::new(|| {
    SimpleCurlyFormat
        .format(STRING_REG_CAST_TEMPLATE, ["regproc", "function", "true"])
        .unwrap()
        .to_string()
});

static STRING_TO_REGTYPE: LazyLock<String> = LazyLock::new(|| {
    SimpleCurlyFormat
        .format(STRING_REG_CAST_TEMPLATE, ["regtype", "type", "true"])
        .unwrap()
        .to_string()
});

// OID 0 renders as `-` to match PostgreSQL's `regprocout`, `regclassout` and
// `regtypeout`, which all spell the absent reference that way. A nonzero OID
// that names nothing still renders as its digits, also as PostgreSQL does.
const REG_STRING_CAST_TEMPLATE: &str = "(
SELECT
    CASE
      WHEN CAST($1 AS pg_catalog.oid) = 0::pg_catalog.oid THEN '-'
      ELSE
        COALESCE(mz_internal.mz_global_id_to_name(o.id), CAST($1 AS pg_catalog.oid)::pg_catalog.text)
    END
    AS text
FROM
  (
        SELECT
          (
            SELECT DISTINCT id
            FROM
              mz_catalog.mz_objects AS o
                JOIN
                  mz_internal.mz_object_oid_alias AS a
                  ON o.type = a.object_type
            WHERE
              oid = CAST($1 AS pg_catalog.oid)
                AND
              a.oid_alias = '{0}'
          )
      )
      AS o
)";

static REGCLASS_TO_STRING: LazyLock<String> = LazyLock::new(|| {
    SimpleCurlyFormat
        .format(REG_STRING_CAST_TEMPLATE, ["regclass"])
        .unwrap()
        .to_string()
});

static REGPROC_TO_STRING: LazyLock<String> = LazyLock::new(|| {
    SimpleCurlyFormat
        .format(REG_STRING_CAST_TEMPLATE, ["regproc"])
        .unwrap()
        .to_string()
});

static REGTYPE_TO_STRING: LazyLock<String> = LazyLock::new(|| {
    SimpleCurlyFormat
        .format(REG_STRING_CAST_TEMPLATE, ["regtype"])
        .unwrap()
        .to_string()
});

/// Describes the context of a cast.
///
/// n.b. this type derived `Ord, PartialOrd` and the ordering of these values
/// has semantics meaning; casts are only permitted when the caller's cast
/// context is geq the ccx we store, which is the minimum required context.
#[derive(Debug, Clone, Copy, Eq, PartialEq, Ord, PartialOrd)]
pub enum CastContext {
    /// Implicit casts are "no-brainer" casts that apply automatically in
    /// expressions. They are typically lossless, such as `SqlScalarType::Int32` to
    /// `SqlScalarType::Int64`.
    Implicit,
    /// Assignment casts are "reasonable" casts that make sense to apply
    /// automatically in `INSERT` statements, but are surprising enough that
    /// they don't apply implicitly in expressions.
    Assignment,
    /// Explicit casts are casts that are possible but may be surprising, like
    /// casting `SqlScalarType::Json` to `SqlScalarType::Int32`, and therefore they do
    /// not happen unless explicitly requested by the user with a cast operator.
    Explicit,
    /// Coerced casts permit different behavior when a type is coerced from a
    /// string literal vs. a value of type `pg_catalog::text`.
    ///
    /// The only call site that should pass this value in to this module is
    /// string coercion.
    Coerced,
}

/// The implementation of a cast.
struct CastImpl {
    template: CastTemplate,
    context: CastContext,
}

macro_rules! casts(
    {
        $(
            $from_to:expr => $cast_context:ident: $cast_template:expr
        ),+
    } => {{
        let mut m = BTreeMap::new();
        $(
            m.insert($from_to, CastImpl {
                template: $cast_template.into(),
                context: CastContext::$cast_context,
            });
        )+
        m
    }};
);

static VALID_CASTS: LazyLock<BTreeMap<(SqlScalarBaseType, SqlScalarBaseType), CastImpl>> =
    LazyLock::new(|| {
        use SqlScalarBaseType::*;
        use UnaryFunc::*;

        casts! {
            // BOOL
            (Bool, Int32) => Explicit: CastBoolToInt32(func::CastBoolToInt32),
            (Bool, Int64) => Explicit: CastBoolToInt64(func::CastBoolToInt64),
            (Bool, String) => Assignment: CastBoolToString(func::CastBoolToString),

            //INT16
            (Int16, Int32) => Implicit: CastInt16ToInt32(func::CastInt16ToInt32),
            (Int16, Int64) => Implicit: CastInt16ToInt64(func::CastInt16ToInt64),
            (Int16, UInt16) => Assignment: CastInt16ToUint16(func::CastInt16ToUint16),
            (Int16, UInt32) => Assignment: CastInt16ToUint32(func::CastInt16ToUint32),
            (Int16, UInt64) => Assignment: CastInt16ToUint64(func::CastInt16ToUint64),
            (Int16, Float32) => Implicit: CastInt16ToFloat32(func::CastInt16ToFloat32),
            (Int16, Float64) => Implicit: CastInt16ToFloat64(func::CastInt16ToFloat64),
            (Int16, Numeric) => Implicit: CastTemplate::new(|_ecx, _ccx, _from_type, to_type| {
                let s = to_type.unwrap_numeric_max_scale();
                let f = CastInt16ToNumeric(func::CastInt16ToNumeric(s));
                Some(move |e: HirScalarExpr| e.call_unary(f))
            }),
            (Int16, Oid) => Implicit: [
                CastInt16ToInt32(func::CastInt16ToInt32),
                CastInt32ToOid(func::CastInt32ToOid),
            ],
            (Int16, RegClass) => Implicit: [
                CastInt16ToInt32(func::CastInt16ToInt32),
                CastInt32ToOid(func::CastInt32ToOid),
                CastOidToRegClass(func::CastOidToRegClass),
            ],
            (Int16, RegProc) => Implicit: [
                CastInt16ToInt32(func::CastInt16ToInt32),
                CastInt32ToOid(func::CastInt32ToOid),
                CastOidToRegProc(func::CastOidToRegProc),
            ],
            (Int16, RegType) => Implicit: [
                CastInt16ToInt32(func::CastInt16ToInt32),
                CastInt32ToOid(func::CastInt32ToOid),
                CastOidToRegType(func::CastOidToRegType),
            ],
            (Int16, String) => Assignment: CastInt16ToString(func::CastInt16ToString),

            //INT32
            (Int32, Bool) => Explicit: CastInt32ToBool(func::CastInt32ToBool),
            (Int32, Oid) => Implicit: CastInt32ToOid(func::CastInt32ToOid),
            (Int32, RegClass) => Implicit: [
                CastInt32ToOid(func::CastInt32ToOid),
                CastOidToRegClass(func::CastOidToRegClass),
            ],
            (Int32, RegProc) => Implicit: [
                CastInt32ToOid(func::CastInt32ToOid),
                CastOidToRegProc(func::CastOidToRegProc),
            ],
            (Int32, RegType) => Implicit: [
                CastInt32ToOid(func::CastInt32ToOid),
                CastOidToRegType(func::CastOidToRegType),
            ],
            (Int32, PgLegacyChar) => Explicit:
                CastInt32ToPgLegacyChar(func::CastInt32ToPgLegacyChar),
            (Int32, Int16) => Assignment: CastInt32ToInt16(func::CastInt32ToInt16),
            (Int32, Int64) => Implicit: CastInt32ToInt64(func::CastInt32ToInt64),
            (Int32, UInt16) => Assignment: CastInt32ToUint16(func::CastInt32ToUint16),
            (Int32, UInt32) => Assignment: CastInt32ToUint32(func::CastInt32ToUint32),
            (Int32, UInt64) => Assignment: CastInt32ToUint64(func::CastInt32ToUint64),
            (Int32, Float32) => Implicit: CastInt32ToFloat32(func::CastInt32ToFloat32),
            (Int32, Float64) => Implicit: CastInt32ToFloat64(func::CastInt32ToFloat64),
            (Int32, Numeric) => Implicit: CastTemplate::new(|_ecx, _ccx, _from_type, to_type| {
                let s = to_type.unwrap_numeric_max_scale();
                let f = CastInt32ToNumeric(func::CastInt32ToNumeric(s));
                Some(move |e: HirScalarExpr| e.call_unary(f))
            }),
            (Int32, String) => Assignment: CastInt32ToString(func::CastInt32ToString),

            // INT64
            (Int64, Bool) => Explicit: CastInt64ToBool(func::CastInt64ToBool),
            (Int64, Int16) => Assignment: CastInt64ToInt16(func::CastInt64ToInt16),
            (Int64, Int32) => Assignment: CastInt64ToInt32(func::CastInt64ToInt32),
            (Int64, UInt16) => Assignment: CastInt64ToUint16(func::CastInt64ToUint16),
            (Int64, UInt32) => Assignment: CastInt64ToUint32(func::CastInt64ToUint32),
            (Int64, UInt64) => Assignment: CastInt64ToUint64(func::CastInt64ToUint64),
            (Int64, Numeric) => Implicit: CastTemplate::new(|_ecx, _ccx, _from_type, to_type| {
                let s = to_type.unwrap_numeric_max_scale();
                let f = CastInt64ToNumeric(func::CastInt64ToNumeric(s));
                Some(move |e: HirScalarExpr| e.call_unary(f))
            }),
            (Int64, Float32) => Implicit: CastInt64ToFloat32(func::CastInt64ToFloat32),
            (Int64, Float64) => Implicit: CastInt64ToFloat64(func::CastInt64ToFloat64),
            (Int64, Oid) => Implicit: CastInt64ToOid(func::CastInt64ToOid),
            (Int64, RegClass) => Implicit: [
                CastInt64ToOid(func::CastInt64ToOid),
                CastOidToRegClass(func::CastOidToRegClass),
            ],
            (Int64, RegProc) => Implicit: [
                CastInt64ToOid(func::CastInt64ToOid),
                CastOidToRegProc(func::CastOidToRegProc),
            ],
            (Int64, RegType) => Implicit: [
                CastInt64ToOid(func::CastInt64ToOid),
                CastOidToRegType(func::CastOidToRegType),
            ],
            (Int64, String) => Assignment: CastInt64ToString(func::CastInt64ToString),

            // UINT16
            (UInt16, UInt32) => Implicit: CastUint16ToUint32(func::CastUint16ToUint32),
            (UInt16, UInt64) => Implicit: CastUint16ToUint64(func::CastUint16ToUint64),
            (UInt16, Int16) => Assignment: CastUint16ToInt16(func::CastUint16ToInt16),
            (UInt16, Int32) => Implicit: CastUint16ToInt32(func::CastUint16ToInt32),
            (UInt16, Int64) => Implicit: CastUint16ToInt64(func::CastUint16ToInt64),
            (UInt16, Numeric) => Implicit: CastTemplate::new(|_ecx, _ccx, _from_type, to_type| {
                let s = to_type.unwrap_numeric_max_scale();
                let f = CastUint16ToNumeric(func::CastUint16ToNumeric(s));
                Some(move |e: HirScalarExpr| e.call_unary(f))
            }),
            (UInt16, Float32) => Implicit: CastUint16ToFloat32(func::CastUint16ToFloat32),
            (UInt16, Float64) => Implicit: CastUint16ToFloat64(func::CastUint16ToFloat64),
            (UInt16, String) => Assignment: CastUint16ToString(func::CastUint16ToString),

            // UINT32
            (UInt32, UInt16) => Assignment: CastUint32ToUint16(func::CastUint32ToUint16),
            (UInt32, UInt64) => Implicit: CastUint32ToUint64(func::CastUint32ToUint64),
            (UInt32, Int16) => Assignment: CastUint32ToInt16(func::CastUint32ToInt16),
            (UInt32, Int32) => Assignment: CastUint32ToInt32(func::CastUint32ToInt32),
            (UInt32, Int64) => Implicit: CastUint32ToInt64(func::CastUint32ToInt64),
            (UInt32, Numeric) => Implicit: CastTemplate::new(|_ecx, _ccx, _from_type, to_type| {
                let s = to_type.unwrap_numeric_max_scale();
                let f = CastUint32ToNumeric(func::CastUint32ToNumeric(s));
                Some(move |e: HirScalarExpr| e.call_unary(f))
            }),
            (UInt32, Float32) => Implicit: CastUint32ToFloat32(func::CastUint32ToFloat32),
            (UInt32, Float64) => Implicit: CastUint32ToFloat64(func::CastUint32ToFloat64),
            (UInt32, String) => Assignment: CastUint32ToString(func::CastUint32ToString),

            // UINT64
            (UInt64, UInt16) => Assignment: CastUint64ToUint16(func::CastUint64ToUint16),
            (UInt64, UInt32) => Assignment: CastUint64ToUint32(func::CastUint64ToUint32),
            (UInt64, Int16) => Assignment: CastUint64ToInt16(func::CastUint64ToInt16),
            (UInt64, Int32) => Assignment: CastUint64ToInt32(func::CastUint64ToInt32),
            (UInt64, Int64) => Assignment: CastUint64ToInt64(func::CastUint64ToInt64),
            (UInt64, Numeric) => Implicit: CastTemplate::new(|_ecx, _ccx, _from_type, to_type| {
                let s = to_type.unwrap_numeric_max_scale();
                let f = CastUint64ToNumeric(func::CastUint64ToNumeric(s));
                Some(move |e: HirScalarExpr| e.call_unary(f))
            }),
            (UInt64, Float32) => Implicit: CastUint64ToFloat32(func::CastUint64ToFloat32),
            (UInt64, Float64) => Implicit: CastUint64ToFloat64(func::CastUint64ToFloat64),
            (UInt64, String) => Assignment: CastUint64ToString(func::CastUint64ToString),

            // MZ_TIMESTAMP
            (MzTimestamp, String) => Assignment:
                CastMzTimestampToString(func::CastMzTimestampToString),
            (MzTimestamp, Timestamp) => Assignment:
                CastMzTimestampToTimestamp(func::CastMzTimestampToTimestamp),
            (MzTimestamp, TimestampTz) => Assignment:
                CastMzTimestampToTimestampTz(
                    func::CastMzTimestampToTimestampTz,
                ),
            (String, MzTimestamp) => Assignment:
                CastStringToMzTimestamp(func::CastStringToMzTimestamp),
            (UInt64, MzTimestamp) => Implicit:
                CastUint64ToMzTimestamp(func::CastUint64ToMzTimestamp),
            (UInt32, MzTimestamp) => Implicit:
                CastUint32ToMzTimestamp(func::CastUint32ToMzTimestamp),
            (Int64, MzTimestamp) => Implicit:
                CastInt64ToMzTimestamp(func::CastInt64ToMzTimestamp),
            (Int32, MzTimestamp) => Implicit:
                CastInt32ToMzTimestamp(func::CastInt32ToMzTimestamp),
            (Numeric, MzTimestamp) => Implicit:
                CastNumericToMzTimestamp(func::CastNumericToMzTimestamp),
            (Timestamp, MzTimestamp) => Implicit:
                CastTimestampToMzTimestamp(func::CastTimestampToMzTimestamp),
            (TimestampTz, MzTimestamp) => Implicit:
                CastTimestampTzToMzTimestamp(
                    func::CastTimestampTzToMzTimestamp,
                ),
            (Date, MzTimestamp) => Implicit:
                CastDateToMzTimestamp(func::CastDateToMzTimestamp),

            // OID
            (Oid, Int32) => Assignment: CastOidToInt32(func::CastOidToInt32),
            (Oid, Int64) => Assignment: CastOidToInt64(func::CastOidToInt64),
            (Oid, String) => Explicit: CastOidToString(func::CastOidToString),
            (Oid, RegClass) => Implicit: CastOidToRegClass(func::CastOidToRegClass),
            (Oid, RegProc) => Implicit: CastOidToRegProc(func::CastOidToRegProc),
            (Oid, RegType) => Implicit: CastOidToRegType(func::CastOidToRegType),

            // REGCLASS
            (RegClass, Oid) => Implicit: CastRegClassToOid(func::CastRegClassToOid),
            (RegClass, String) => Explicit: sql_impl_cast(&REGCLASS_TO_STRING),

            // REGPROC
            (RegProc, Oid) => Implicit: CastRegProcToOid(func::CastRegProcToOid),
            (RegProc, String) => Explicit: sql_impl_cast(&REGPROC_TO_STRING),

            // REGTYPE
            (RegType, Oid) => Implicit: CastRegTypeToOid(func::CastRegTypeToOid),
            (RegType, String) => Explicit: sql_impl_cast(&REGTYPE_TO_STRING),

            // FLOAT32
            (Float32, Int16) => Assignment: CastFloat32ToInt16(func::CastFloat32ToInt16),
            (Float32, Int32) => Assignment: CastFloat32ToInt32(func::CastFloat32ToInt32),
            (Float32, Int64) => Assignment: CastFloat32ToInt64(func::CastFloat32ToInt64),
            (Float32, UInt16) => Assignment: CastFloat32ToUint16(func::CastFloat32ToUint16),
            (Float32, UInt32) => Assignment: CastFloat32ToUint32(func::CastFloat32ToUint32),
            (Float32, UInt64) => Assignment: CastFloat32ToUint64(func::CastFloat32ToUint64),
            (Float32, Float64) => Implicit: CastFloat32ToFloat64(func::CastFloat32ToFloat64),
            (Float32, Numeric) => Assignment: CastTemplate::new(|_ecx, _ccx, _from_type, to_type| {
                let s = to_type.unwrap_numeric_max_scale();
                let f = CastFloat32ToNumeric(func::CastFloat32ToNumeric(s));
                Some(move |e: HirScalarExpr| e.call_unary(f))
            }),
            (Float32, String) => Assignment: CastFloat32ToString(func::CastFloat32ToString),

            // FLOAT64
            (Float64, Int16) => Assignment: CastFloat64ToInt16(func::CastFloat64ToInt16),
            (Float64, Int32) => Assignment: CastFloat64ToInt32(func::CastFloat64ToInt32),
            (Float64, Int64) => Assignment: CastFloat64ToInt64(func::CastFloat64ToInt64),
            (Float64, UInt16) => Assignment: CastFloat64ToUint16(func::CastFloat64ToUint16),
            (Float64, UInt32) => Assignment: CastFloat64ToUint32(func::CastFloat64ToUint32),
            (Float64, UInt64) => Assignment: CastFloat64ToUint64(func::CastFloat64ToUint64),
            (Float64, Float32) => Assignment: CastFloat64ToFloat32(func::CastFloat64ToFloat32),
            (Float64, Numeric) => Assignment: CastTemplate::new(|_ecx, _ccx, _from_type, to_type| {
                let s = to_type.unwrap_numeric_max_scale();
                let f = CastFloat64ToNumeric(func::CastFloat64ToNumeric(s));
                Some(move |e: HirScalarExpr| e.call_unary(f))
            }),
            (Float64, String) => Assignment: CastFloat64ToString(func::CastFloat64ToString),

            // DATE
            (Date, Timestamp) => Implicit: CastTemplate::new(|_ecx, _ccx, _from_type, to_type| {
                let p = to_type.unwrap_timestamp_precision();
                let f = CastDateToTimestamp(func::CastDateToTimestamp(p));
                Some(move |e: HirScalarExpr| e.call_unary(f))
            }),
            (Date, TimestampTz) => Implicit: CastTemplate::new(|_ecx, _ccx, _from_type, to_type| {
                let p = to_type.unwrap_timestamp_precision();
                let f = CastDateToTimestampTz(func::CastDateToTimestampTz(p));
                Some(move |e: HirScalarExpr| e.call_unary(f))
            }),
            (Date, String) => Assignment: CastDateToString(func::CastDateToString),

            // TIME
            (Time, Interval) => Implicit: CastTimeToInterval(func::CastTimeToInterval),
            (Time, String) => Assignment: CastTimeToString(func::CastTimeToString),

            // TIMESTAMP
            (Timestamp, Date) => Assignment: CastTimestampToDate(func::CastTimestampToDate),
            (Timestamp, TimestampTz) => Implicit: CastTemplate::new(
                |_ecx, _ccx, from_type, to_type|
            {
                let from = from_type.unwrap_timestamp_precision();
                let to = to_type.unwrap_timestamp_precision();
                let f = CastTimestampToTimestampTz(
                    func::CastTimestampToTimestampTz { from, to },
                );
                Some(move |e: HirScalarExpr| e.call_unary(f))
            }),
            (Timestamp, Timestamp) => Assignment: CastTemplate::new(
                |_ecx, _ccx, from_type, to_type|
            {
                let from = from_type.unwrap_timestamp_precision();
                let to = to_type.unwrap_timestamp_precision();
                let f = AdjustTimestampPrecision(
                    func::AdjustTimestampPrecision { from, to },
                );
                Some(move |e: HirScalarExpr| e.call_unary(f))
            }),
            (Timestamp, Time) => Assignment: CastTimestampToTime(func::CastTimestampToTime),
            (Timestamp, String) => Assignment: CastTimestampToString(func::CastTimestampToString),

            // TIMESTAMPTZ
            (TimestampTz, Date) => Assignment: CastTimestampTzToDate(func::CastTimestampTzToDate),
            (TimestampTz, Timestamp) => Assignment: CastTemplate::new(
                |_ecx, _ccx, from_type, to_type|
            {
                let from = from_type.unwrap_timestamp_precision();
                let to = to_type.unwrap_timestamp_precision();
                let f = CastTimestampTzToTimestamp(
                    func::CastTimestampTzToTimestamp { from, to },
                );
                Some(move |e: HirScalarExpr| e.call_unary(f))
            }),
            (TimestampTz, TimestampTz) => Assignment: CastTemplate::new(
                |_ecx, _ccx, from_type, to_type|
            {
                let from = from_type.unwrap_timestamp_precision();
                let to = to_type.unwrap_timestamp_precision();
                let f = AdjustTimestampTzPrecision(
                    func::AdjustTimestampTzPrecision { from, to },
                );
                Some(move |e: HirScalarExpr| e.call_unary(f))
            }),
            (TimestampTz, Time) => Assignment:
                CastTimestampTzToTime(func::CastTimestampTzToTime),
            (TimestampTz, String) => Assignment:
                CastTimestampTzToString(func::CastTimestampTzToString),

            // INTERVAL
            (Interval, Time) => Assignment: CastIntervalToTime(func::CastIntervalToTime),
            (Interval, String) => Assignment: CastIntervalToString(func::CastIntervalToString),

            // BYTES
            (Bytes, String) => Assignment: CastBytesToString(func::CastBytesToString),

            // STRING
            (String, Bool) => Explicit: CastStringToBool(func::CastStringToBool),
            (String, Int16) => Explicit: CastStringToInt16(func::CastStringToInt16),
            (String, Int32) => Explicit: CastStringToInt32(func::CastStringToInt32),
            (String, Int64) => Explicit: CastStringToInt64(func::CastStringToInt64),
            (String, UInt16) => Explicit: CastStringToUint16(func::CastStringToUint16),
            (String, UInt32) => Explicit: CastStringToUint32(func::CastStringToUint32),
            (String, UInt64) => Explicit: CastStringToUint64(func::CastStringToUint64),
            (String, Oid) => Explicit: CastStringToOid(func::CastStringToOid),

            // STRING to REG*
            // A reg* type represents a specific type of object by oid.
            // Converting from string to reg* does a lookup of the object name
            // in the corresponding mz_catalog table and expects exactly one object to match it.
            // You can also specify (in postgres) a string that's a valid
            // int4 and it'll happily cast it (without verifying that the int4 matches
            // an object oid).
            // TODO: Support the correct error code for does not exist (42883).
            (String, RegClass) => Explicit: sql_impl_cast_per_context(
                &[
                    (CastContext::Explicit, &STRING_TO_REGCLASS_EXPLICIT),
                    (CastContext::Coerced, &STRING_TO_REGCLASS_COERCED)
                ]
            ),
            (String, RegProc) => Explicit: sql_impl_cast(&STRING_TO_REGPROC),
            (String, RegType) => Explicit: sql_impl_cast(&STRING_TO_REGTYPE),

            (String, Float32) => Explicit: CastStringToFloat32(func::CastStringToFloat32),
            (String, Float64) => Explicit: CastStringToFloat64(func::CastStringToFloat64),
            (String, Numeric) => Explicit: CastTemplate::new(|_ecx, _ccx, _from_type, to_type| {
                let s = to_type.unwrap_numeric_max_scale();
                let f = CastStringToNumeric(func::CastStringToNumeric(s));
                Some(move |e: HirScalarExpr| e.call_unary(f))
            }),
            (String, Date) => Explicit: CastStringToDate(func::CastStringToDate),
            (String, Time) => Explicit: CastStringToTime(func::CastStringToTime),
            (String, Timestamp) => Explicit: CastTemplate::new(|_ecx, _ccx, _from_type, to_type| {
                let p = to_type.unwrap_timestamp_precision();
                let f = CastStringToTimestamp(func::CastStringToTimestamp(p));
                Some(move |e: HirScalarExpr| e.call_unary(f))
            }),
            (String, TimestampTz) => Explicit: CastTemplate::new(|_ecx, _ccx, _from_type, to_type| {
                let p = to_type.unwrap_timestamp_precision();
                let f = CastStringToTimestampTz(
                    func::CastStringToTimestampTz(p),
                );
                Some(move |e: HirScalarExpr| e.call_unary(f))
            }),
            (String, Interval) => Explicit: CastStringToInterval(func::CastStringToInterval),
            (String, Bytes) => Explicit: CastStringToBytes(func::CastStringToBytes),
            (String, Jsonb) => Explicit: CastStringToJsonb(func::CastStringToJsonb),
            (String, Uuid) => Explicit: CastStringToUuid(func::CastStringToUuid),
            (String, Array) => Explicit: CastTemplate::new(|ecx, ccx, from_type, to_type| {
                let return_ty = to_type.clone();
                let to_el_type = to_type.unwrap_array_element_type();
                let cast_expr = plan_hypothetical_cast(ecx, ccx, from_type, to_el_type)?;
                Some(|e: HirScalarExpr| {
                    e.call_unary(UnaryFunc::CastStringToArray(
                        func::CastStringToArray {
                            return_ty,
                            cast_expr: Box::new(cast_expr),
                        },
                    ))
                })
            }),
            (String, List) => Explicit: CastTemplate::new(|ecx, ccx, from_type, to_type| {
                let return_ty = to_type.clone();
                let to_el_type = to_type.unwrap_list_element_type();
                let cast_expr = plan_hypothetical_cast(ecx, ccx, from_type, to_el_type)?;
                Some(|e: HirScalarExpr| {
                    e.call_unary(UnaryFunc::CastStringToList(
                        func::CastStringToList {
                            return_ty,
                            cast_expr: Box::new(cast_expr),
                        },
                    ))
                })
            }),
            (String, Map) => Explicit: CastTemplate::new(|ecx, ccx, from_type, to_type| {
                let return_ty = to_type.clone();
                let to_val_type = to_type.unwrap_map_value_type();
                let cast_expr = plan_hypothetical_cast(ecx, ccx, from_type, to_val_type)?;
                Some(|e: HirScalarExpr| {
                    e.call_unary(UnaryFunc::CastStringToMap(
                        func::CastStringToMap {
                            return_ty,
                            cast_expr: Box::new(cast_expr),
                        },
                    ))
                })
            }),
            (String, Range) => Explicit: CastTemplate::new(|ecx, ccx, from_type, to_type| {
                let return_ty = to_type.clone();
                let to_el_type = to_type.unwrap_range_element_type();
                let cast_expr = plan_hypothetical_cast(ecx, ccx, from_type, to_el_type)?;
                Some(|e: HirScalarExpr| {
                    e.call_unary(UnaryFunc::CastStringToRange(
                        func::CastStringToRange {
                            return_ty,
                            cast_expr: Box::new(cast_expr),
                        },
                    ))
                })
            }),
            (String, Int2Vector) => Explicit: CastStringToInt2Vector(func::CastStringToInt2Vector),
            (String, Char) => Implicit: CastTemplate::new(|_ecx, ccx, _from_type, to_type| {
                let length = to_type.unwrap_char_length();
                let fail_on_len = ccx != CastContext::Explicit;
                let f = CastStringToChar(func::CastStringToChar {
                    length, fail_on_len,
                });
                Some(move |e: HirScalarExpr| e.call_unary(f))
            }),
            (String, VarChar) => Implicit: CastTemplate::new(|_ecx, ccx, _from_type, to_type| {
                let length = to_type.unwrap_varchar_max_length();
                let fail_on_len = ccx != CastContext::Explicit;
                let f = CastStringToVarChar(func::CastStringToVarChar {
                    length, fail_on_len,
                });
                Some(move |e: HirScalarExpr| e.call_unary(f))
            }),
            (String, PgLegacyChar) => Assignment:
                CastStringToPgLegacyChar(func::CastStringToPgLegacyChar),
            // CHAR
            (Char, String) => Implicit: CastCharToString(func::CastCharToString),
            (Char, Char) => Implicit: CastTemplate::new(|_ecx, ccx, _from_type, to_type| {
                let length = to_type.unwrap_char_length();
                let fail_on_len = ccx != CastContext::Explicit;
                let f = CastStringToChar(func::CastStringToChar {
                    length, fail_on_len,
                });
                Some(move |e: HirScalarExpr| e.call_unary(f))
            }),
            (Char, VarChar) => Implicit: CastTemplate::new(|_ecx, ccx, _from_type, to_type| {
                let length = to_type.unwrap_varchar_max_length();
                let fail_on_len = ccx != CastContext::Explicit;
                let f = CastStringToVarChar(func::CastStringToVarChar {
                    length, fail_on_len,
                });
                Some(move |e: HirScalarExpr| e.call_unary(f))
            }),
            (Char, PgLegacyChar) => Assignment:
                CastStringToPgLegacyChar(func::CastStringToPgLegacyChar),

            // VARCHAR
            (VarChar, String) => Implicit: CastVarCharToString(func::CastVarCharToString),
            (VarChar, Char) => Implicit: CastTemplate::new(|_ecx, ccx, _from_type, to_type| {
                let length = to_type.unwrap_char_length();
                let fail_on_len = ccx != CastContext::Explicit;
                let f = CastStringToChar(func::CastStringToChar {
                    length, fail_on_len,
                });
                Some(move |e: HirScalarExpr| e.call_unary(f))
            }),
            (VarChar, VarChar) => Implicit: CastTemplate::new(|_ecx, ccx, _from_type, to_type| {
                let length = to_type.unwrap_varchar_max_length();
                let fail_on_len = ccx != CastContext::Explicit;
                let f = CastStringToVarChar(func::CastStringToVarChar {
                    length, fail_on_len,
                });
                Some(move |e: HirScalarExpr| e.call_unary(f))
            }),
            (VarChar, PgLegacyChar) => Assignment:
                CastStringToPgLegacyChar(func::CastStringToPgLegacyChar),

            // PG LEGACY CHAR
            (PgLegacyChar, String) => Implicit:
                CastPgLegacyCharToString(func::CastPgLegacyCharToString),
            (PgLegacyChar, Char) => Assignment:
                CastPgLegacyCharToChar(func::CastPgLegacyCharToChar),
            (PgLegacyChar, VarChar) => Assignment:
                CastPgLegacyCharToVarChar(func::CastPgLegacyCharToVarChar),
            (PgLegacyChar, Int32) => Explicit:
                CastPgLegacyCharToInt32(func::CastPgLegacyCharToInt32),

            // PG LEGACY NAME
            // Under the hood VarChars and Name's are just Strings, so we
            // can re-use existing methods
            // on Strings and VarChars instead of defining new ones.
            (PgLegacyName, String) => Implicit: CastVarCharToString(func::CastVarCharToString),
            (PgLegacyName, Char) => Assignment: CastTemplate::new(|_ecx, ccx, _from_type, to_type| {
                let length = to_type.unwrap_char_length();
                let fail_on_len = ccx != CastContext::Explicit;
                let f = CastStringToChar(func::CastStringToChar {
                    length, fail_on_len,
                });
                Some(move |e: HirScalarExpr| e.call_unary(f))
            }),
            (PgLegacyName, VarChar) => Assignment: CastTemplate::new(
                |_ecx, ccx, _from_type, to_type|
            {
                let length = to_type.unwrap_varchar_max_length();
                let fail_on_len = ccx != CastContext::Explicit;
                let f = CastStringToVarChar(func::CastStringToVarChar {
                    length, fail_on_len,
                });
                Some(move |e: HirScalarExpr| e.call_unary(f))
            }),
            (String, PgLegacyName) => Implicit:
                CastStringToPgLegacyName(func::CastStringToPgLegacyName),
            (Char, PgLegacyName) => Implicit:
                CastStringToPgLegacyName(func::CastStringToPgLegacyName),
            (VarChar, PgLegacyName) => Implicit:
                CastStringToPgLegacyName(func::CastStringToPgLegacyName),

            // RECORD
            (Record, String) => Assignment: CastTemplate::new(|_ecx, _ccx, from_type, _to_type| {
                let ty = from_type.clone();
                Some(|e: HirScalarExpr| {
                    e.call_unary(CastRecordToString(
                        func::CastRecordToString { ty },
                    ))
                })
            }),
            (Record, Record) => Implicit: CastTemplate::new(|ecx, ccx, from_type, to_type| {
                let from_fields = from_type.unwrap_record_element_type();
                let to_fields = to_type.unwrap_record_element_type();
                if from_fields.len() != to_fields.len() {
                    return None;
                }

                if let (
                    l @ SqlScalarType::Record {
                        custom_id: Some(..), ..
                    },
                    r,
                ) = (from_type, to_type)
                {
                    // Changing `from`'s custom_id requires at least
                    // Assignment context
                    if ccx == CastContext::Implicit && l != r {
                        return None;
                    }
                }

                let cast_exprs = from_fields
                    .iter()
                    .zip_eq(to_fields)
                    .map(|(f, t)| plan_hypothetical_cast(ecx, ccx, f, t))
                    .collect::<Option<Box<_>>>()?;
                let to = to_type.clone();
                Some(|e: HirScalarExpr| {
                    e.call_unary(CastRecord1ToRecord2(
                        func::CastRecord1ToRecord2 {
                            return_ty: to,
                            cast_exprs,
                        },
                    ))
                })
            }),

            // ARRAY
            (Array, String) => Assignment: CastTemplate::new(|_ecx, _ccx, from_type, _to_type| {
                let ty = from_type.clone();
                Some(|e: HirScalarExpr| {
                    e.call_unary(CastArrayToString(
                        func::CastArrayToString { ty },
                    ))
                })
            }),
            (Array, List) => Explicit: CastArrayToListOneDim(func::CastArrayToListOneDim),
            (Array, Array) => Explicit: CastTemplate::new(|ecx, ccx, from_type, to_type| {
                let inner_from_type = from_type.unwrap_array_element_type();
                let inner_to_type = to_type.unwrap_array_element_type();
                let cast_expr = plan_hypothetical_cast(
                    ecx, ccx, inner_from_type, inner_to_type,
                )?;
                let return_ty = to_type.clone();

                Some(move |e: HirScalarExpr| {
                    e.call_unary(CastArrayToArray(func::CastArrayToArray {
                        return_ty,
                        cast_expr: Box::new(cast_expr),
                    }))
                })
            }),

            // INT2VECTOR
            (Int2Vector, Array) => Implicit: CastTemplate::new(|ecx, ccx, _from_type, to_type| {
                let inner_to_type = to_type.unwrap_array_element_type();
                // Int2Vector elements are Int16, so if the target element type
                // differs we need to also cast the array elements after the
                // Int2Vector -> Array(Int16) conversion. (Postgres seems to do the same.)
                let element_cast = if inner_to_type != &SqlScalarType::Int16 {
                    let cast_expr = plan_hypothetical_cast(
                        ecx, ccx, &SqlScalarType::Int16, inner_to_type
                    )?;
                    Some((to_type.clone(), cast_expr))
                } else {
                    None
                };
                Some(move |e: HirScalarExpr| {
                    let arr = e.call_unary(
                        UnaryFunc::CastInt2VectorToArray(func::CastInt2VectorToArray)
                    );
                    match element_cast {
                        Some((return_ty, cast_expr)) => {
                            arr.call_unary(CastArrayToArray(
                                func::CastArrayToArray { return_ty, cast_expr: Box::new(cast_expr) }
                            ))
                        }
                        None => arr,
                    }
                })
            }),
            (Int2Vector, String) => Explicit: CastInt2VectorToString(func::CastInt2VectorToString),

            // LIST
            (List, String) => Assignment: CastTemplate::new(|_ecx, _ccx, from_type, _to_type| {
                let ty = from_type.clone();
                Some(|e: HirScalarExpr| {
                    e.call_unary(CastListToString(
                        func::CastListToString { ty },
                    ))
                })
            }),
            (List, List) => Implicit: CastTemplate::new(|ecx, ccx, from_type, to_type| {

                if let (
                    l @ SqlScalarType::List {
                        custom_id: Some(..), ..
                    },
                    r,
                ) = (from_type, to_type)
                {
                    // Changing `from`'s custom_id requires at least
                    // Assignment context
                    if ccx == CastContext::Implicit && !l.base_eq(r) {
                        return None;
                    }
                }

                let return_ty = to_type.clone();
                let from_el_type = from_type.unwrap_list_element_type();
                let to_el_type = to_type.unwrap_list_element_type();
                let cast_expr = plan_hypothetical_cast(
                    ecx, ccx, from_el_type, to_el_type,
                )?;
                Some(|e: HirScalarExpr| {
                    e.call_unary(UnaryFunc::CastList1ToList2(
                        func::CastList1ToList2 {
                            return_ty,
                            cast_expr: Box::new(cast_expr),
                        },
                    ))
                })
            }),

            // MAP
            (Map, String) => Assignment: CastTemplate::new(|_ecx, _ccx, from_type, _to_type| {
                let ty = from_type.clone();
                Some(|e: HirScalarExpr| e.call_unary(CastMapToString(func::CastMapToString { ty })))
            }),

            // JSONB
            (Jsonb, Bool) => Explicit: CastJsonbToBool(func::CastJsonbToBool),
            (Jsonb, Int16) => Explicit: CastJsonbToInt16(func::CastJsonbToInt16),
            (Jsonb, Int32) => Explicit: CastJsonbToInt32(func::CastJsonbToInt32),
            (Jsonb, Int64) => Explicit: CastJsonbToInt64(func::CastJsonbToInt64),
            (Jsonb, Float32) => Explicit: CastJsonbToFloat32(func::CastJsonbToFloat32),
            (Jsonb, Float64) => Explicit: CastJsonbToFloat64(func::CastJsonbToFloat64),
            (Jsonb, Numeric) => Explicit: CastTemplate::new(|_ecx, _ccx, _from_type, to_type| {
                let s = to_type.unwrap_numeric_max_scale();
                let f = CastJsonbToNumeric(func::CastJsonbToNumeric(s));
                Some(move |e: HirScalarExpr| e.call_unary(f))
            }),
            (Jsonb, String) => Assignment: CastJsonbToString(func::CastJsonbToString),

            // UUID
            (Uuid, String) => Assignment: CastUuidToString(func::CastUuidToString),

            // Numeric
            (Numeric, Numeric) => Assignment: CastTemplate::new(|_ecx, _ccx, _from_type, to_type| {
                let scale = to_type.unwrap_numeric_max_scale();
                Some(move |e: HirScalarExpr| match scale {
                    None => e,
                    Some(scale) => e.call_unary(
                        UnaryFunc::AdjustNumericScale(
                            func::AdjustNumericScale(scale),
                        ),
                    ),
                })
            }),
            (Numeric, Float32) => Implicit: CastNumericToFloat32(func::CastNumericToFloat32),
            (Numeric, Float64) => Implicit: CastNumericToFloat64(func::CastNumericToFloat64),
            (Numeric, Int16) => Assignment: CastNumericToInt16(func::CastNumericToInt16),
            (Numeric, Int32) => Assignment: CastNumericToInt32(func::CastNumericToInt32),
            (Numeric, Int64) => Assignment: CastNumericToInt64(func::CastNumericToInt64),
            (Numeric, UInt16) => Assignment: CastNumericToUint16(func::CastNumericToUint16),
            (Numeric, UInt32) => Assignment: CastNumericToUint32(func::CastNumericToUint32),
            (Numeric, UInt64) => Assignment: CastNumericToUint64(func::CastNumericToUint64),
            (Numeric, String) => Assignment: CastNumericToString(func::CastNumericToString),

            // Range
            (Range, String) => Assignment: CastTemplate::new(|_ecx, _ccx, from_type, _to_type| {
                let ty = from_type.clone();
                Some(|e: HirScalarExpr| {
                    e.call_unary(CastRangeToString(
                        func::CastRangeToString { ty },
                    ))
                })
            }),

            // MzAclItem
            (MzAclItem, String) => Explicit: sql_impl_cast("(
                SELECT
                    (CASE
                        WHEN grantee_role_id = 'p' THEN ''
                        ELSE COALESCE(grantee_role.name, grantee_role_id)
                    END)
                    || '='
                    || mz_internal.mz_aclitem_privileges($1)
                    || '/'
                    || COALESCE(grantor_role.name, grantor_role_id)
                FROM
                    (SELECT mz_internal.mz_aclitem_grantee($1) AS grantee_role_id),
                    (SELECT mz_internal.mz_aclitem_grantor($1) AS grantor_role_id)
                LEFT JOIN mz_catalog.mz_roles AS grantee_role ON grantee_role_id = grantee_role.id
                LEFT JOIN mz_catalog.mz_roles AS grantor_role ON grantor_role_id = grantor_role.id
            )"),
            (MzAclItem, AclItem) => Explicit: sql_impl_cast("(
                SELECT makeaclitem(
                    (CASE mz_internal.mz_aclitem_grantee($1)
                        WHEN 'p' THEN 0
                        ELSE (SELECT oid FROM mz_catalog.mz_roles
                            WHERE id = mz_internal.mz_aclitem_grantee($1))
                    END),
                    (SELECT oid FROM mz_catalog.mz_roles
                        WHERE id = mz_internal.mz_aclitem_grantor($1)),
                    (SELECT array_to_string(
                        mz_internal.mz_format_privileges(
                            mz_internal.mz_aclitem_privileges($1)
                        ), ',')),
                    -- GRANT OPTION isn't implemented so we hardcode false.
                    false
                )
            )"),

            // AclItem
            (AclItem, String) => Explicit: sql_impl_cast("(
                SELECT
                    (CASE grantee_oid
                        WHEN 0 THEN ''
                        ELSE COALESCE(grantee_role.name, grantee_oid::text)
                    END)
                    || '='
                    || mz_internal.aclitem_privileges($1)
                    || '/'
                    || COALESCE(grantor_role.name, grantor_oid::text)
                FROM
                    (SELECT mz_internal.aclitem_grantee($1) AS grantee_oid),
                    (SELECT mz_internal.aclitem_grantor($1) AS grantor_oid)
                LEFT JOIN mz_catalog.mz_roles AS grantee_role ON grantee_oid = grantee_role.oid
                LEFT JOIN mz_catalog.mz_roles AS grantor_role ON grantor_oid = grantor_role.oid
            )"),
            (AclItem, MzAclItem) => Explicit: sql_impl_cast("(
                SELECT mz_internal.make_mz_aclitem(
                    (CASE mz_internal.aclitem_grantee($1)
                        WHEN 0 THEN 'p'
                        ELSE (SELECT id FROM mz_catalog.mz_roles
                            WHERE oid = mz_internal.aclitem_grantee($1))
                    END),
                    (SELECT id FROM mz_catalog.mz_roles
                        WHERE oid = mz_internal.aclitem_grantor($1)),
                    (SELECT array_to_string(
                        mz_internal.mz_format_privileges(
                            mz_internal.aclitem_privileges($1)
                        ), ','))
                )
            )")
        }
    });

/// Error returned when a cast is not available or not allowed.
#[derive(Debug)]
pub enum CastError {
    /// No cast exists between the given types (or not in this cast context).
    InvalidCast {
        ccx: CastContext,
        from: String,
        to: String,
    },
    /// Cast would apply but is disallowed (e.g. range over an unsupported element type).
    UnsupportedRangeElementType { element_type_name: String },
    /// The cast exists but has no null-fallback form, so `TRY_CAST` cannot use it.
    NullFallbackUnsupported { from: String, to: String },
}

impl CastError {
    /// Convert to a [`PlanError`] for use in planning; requires the expression name for context.
    pub fn into_plan_error(self, name: String) -> PlanError {
        match self {
            CastError::InvalidCast { ccx, from, to } => PlanError::InvalidCast {
                name,
                ccx,
                from,
                to,
            },
            CastError::UnsupportedRangeElementType { element_type_name } => {
                PlanError::UnsupportedRangeElementType { element_type_name }
            }
            CastError::NullFallbackUnsupported { from, to } => {
                PlanError::TryCastUnsupported { from, to }
            }
        }
    }
}

/// Get casts directly between two [`SqlScalarType`]s, with control over the
/// allowed [`CastContext`].
///
/// Returns `Err` when the cast is not supported: either no cast exists
/// ([`CastError::InvalidCast`]) or the cast is disallowed
/// ([`CastError::UnsupportedRangeElementType`], e.g. range over float/uint).
fn get_cast(
    ecx: &ExprContext,
    ccx: CastContext,
    mode: CastFailureMode,
    from: &SqlScalarType,
    to: &SqlScalarType,
) -> Result<Cast, CastError> {
    use CastContext::*;

    // The identity cast cannot fail, so it is the same in every failure mode.
    if from == to || (ccx == Implicit && from.base_eq(to)) {
        return Ok(Box::new(|expr| expr));
    }

    // Reject casts to range types with unsupported element types at plan time.
    if let SqlScalarType::Range { element_type } = to {
        validate_range_element_type(ecx, element_type)?;
    }

    let imp = match VALID_CASTS.get(&(from.into(), to.into())) {
        Some(imp) => imp,
        None => {
            return Err(CastError::InvalidCast {
                ccx,
                from: ecx.humanize_sql_scalar_type(from, false),
                to: ecx.humanize_sql_scalar_type(to, false),
            });
        }
    };
    let template = if ccx >= imp.context {
        Some(&imp.template)
    } else {
        None
    };
    let build = |mode| template.and_then(|template| (template.0)(ecx, ccx, mode, from, to));
    match build(mode) {
        Some(cast) => Ok(cast),
        // A template declines a failure mode by returning `None`, the same way
        // it declines a type pair. Tell the two apart so that the error names
        // the actual problem.
        None if mode == CastFailureMode::NullFallback
            && build(CastFailureMode::Error).is_some() =>
        {
            Err(CastError::NullFallbackUnsupported {
                from: ecx.humanize_sql_scalar_type(from, false),
                to: ecx.humanize_sql_scalar_type(to, false),
            })
        }
        None => Err(CastError::InvalidCast {
            ccx,
            from: ecx.humanize_sql_scalar_type(from, false),
            to: ecx.humanize_sql_scalar_type(to, false),
        }),
    }
}

/// The base type pairs that have an entry in the cast table, for tests that
/// want to prove they exercised every one.
pub fn valid_cast_base_type_pairs() -> impl Iterator<Item = (SqlScalarBaseType, SqlScalarBaseType)>
{
    VALID_CASTS.keys().copied()
}

/// Converts an expression to `SqlScalarType::String`.
///
/// Most types have an explicit cast to string, but a few do not: `mz_aclitem`
/// has no text cast at all, and the `reg*` casts are implemented as subqueries
/// that `plan_cast` rejects in contexts that disallow subqueries (e.g. a
/// `RETURNING` clause or a `to_jsonb` element cast). We surface those as a
/// planning error rather than panicking.
pub fn to_string(ecx: &ExprContext, expr: HirScalarExpr) -> Result<HirScalarExpr, PlanError> {
    plan_cast(ecx, CastContext::Explicit, expr, &SqlScalarType::String)
}

/// Converts an expression to `SqlScalarType::Jsonb`.
///
/// The rules are as follows:
///   * `SqlScalarType::Boolean`s become JSON booleans.
///   * All numeric types are converted to `Float64`s, then become JSON numbers.
///   * Records are converted to a JSON object where the record's field names
///     are the keys of the object, and the record's fields are recursively
///     converted to JSON by `to_jsonb`.
///   * Other types are converted to strings by their usual cast function and
///     become JSON strings. That cast can fail (see `to_string`), in which case
///     we propagate the planning error instead of panicking.
pub fn to_jsonb(ecx: &ExprContext, expr: HirScalarExpr) -> Result<HirScalarExpr, PlanError> {
    use SqlScalarType::*;

    Ok(match ecx.scalar_type(&expr) {
        Bool | Jsonb | Numeric { .. } => {
            expr.call_unary(UnaryFunc::CastJsonbableToJsonb(func::CastJsonbableToJsonb))
        }
        Int16 | Int32 | Int64 | UInt16 | UInt32 | UInt64 | Float32 | Float64 => plan_cast(
            ecx,
            CastContext::Explicit,
            expr,
            &Numeric { max_scale: None },
        )
        .expect("cast known to exist")
        .call_unary(UnaryFunc::CastJsonbableToJsonb(func::CastJsonbableToJsonb)),
        Record { fields, .. } => {
            let mut exprs = vec![];
            for (i, (name, _ty)) in fields.iter().enumerate() {
                exprs.push(HirScalarExpr::literal(
                    Datum::String(name),
                    SqlScalarType::String,
                ));
                exprs.push(to_jsonb(
                    ecx,
                    expr.clone()
                        .call_unary(UnaryFunc::RecordGet(func::RecordGet(i))),
                )?);
            }
            HirScalarExpr::call_variadic(JsonbBuildObject, exprs)
        }
        ref ty @ List {
            ref element_type, ..
        }
        | ref ty @ Array(ref element_type) => {
            // Construct a new expression context with one column whose type
            // is the container's element type.
            let qcx = QueryContext::root(ecx.qcx.scx, ecx.qcx.lifetime);
            let ecx = ExprContext {
                qcx: &qcx,
                name: "to_jsonb",
                scope: &Scope::empty(),
                relation_type: &SqlRelationType::new(vec![element_type.clone().nullable(true)]),
                allow_aggregates: false,
                allow_subqueries: false,
                allow_parameters: false,
                allow_windows: false,
            };

            // Create an element-casting expression by calling `to_jsonb` on
            // an expression that references the first column in a row.
            let cast_element = to_jsonb(&ecx, HirScalarExpr::column(0))?;
            let cast_element = cast_element
                .lower_uncorrelated(ecx.catalog().system_vars())
                .expect("to_jsonb does not produce correlated expressions on uncorrelated input");

            // The `Cast{Array|List}ToJsonb` functions take the element-casting
            // expression as an argument and evaluate the expression against
            // each element of the container at runtime.
            let func = match ty {
                List { .. } => UnaryFunc::CastListToJsonb(CastListToJsonb {
                    cast_element: Box::new(cast_element),
                }),
                Array { .. } => UnaryFunc::CastArrayToJsonb(CastArrayToJsonb {
                    cast_element: Box::new(cast_element),
                }),
                _ => unreachable!("validated above"),
            };

            expr.call_unary(func)
        }
        Date
        | Time
        | Timestamp { .. }
        | TimestampTz { .. }
        | Interval
        | PgLegacyChar
        | PgLegacyName
        | Bytes
        | String
        | Char { .. }
        | VarChar { .. }
        | Uuid
        | Oid
        | Map { .. }
        | RegProc
        | RegType
        | RegClass
        | Int2Vector
        | MzTimestamp
        | Range { .. }
        | MzAclItem
        | AclItem => to_string(ecx, expr)?
            .call_unary(UnaryFunc::CastJsonbableToJsonb(func::CastJsonbableToJsonb)),
    })
}

/// Guesses the most-common type among a set of [`SqlScalarType`]s that all members
/// can be cast to. Returns `None` if a common type cannot be deduced.
///
/// Note that this function implements the type-determination components of
/// Postgres' ["`UNION`, `CASE`, and Related Constructs"][union-type-conv] type
/// conversion.
///
/// [union-type-conv]: https://www.postgresql.org/docs/12/typeconv-union-case.html
pub fn guess_best_common_type(
    ecx: &ExprContext,
    types: &[CoercibleScalarType],
) -> Result<SqlScalarType, PlanError> {
    // This function is a translation of `select_common_type` in PostgreSQL with
    // the addition of our near match logic, which supports Materialize
    // non-linear type promotions.
    // https://github.com/postgres/postgres/blob/d1b307eef/
    //   src/backend/parser/parse_coerce.c#L1288-L1308

    // If every type is a literal record with the same number of fields, the
    // best common type is a record with that number of fields. We recursively
    // guess the best type for each field.
    if let Some(CoercibleScalarType::Record(field_tys)) = types.first() {
        if types
            .iter()
            .all(|t| matches!(t, CoercibleScalarType::Record(fts) if field_tys.len() == fts.len()))
        {
            let mut fields = vec![];
            for i in 0..field_tys.len() {
                let name = ColumnName::from(format!("f{}", fields.len() + 1));
                let mut guesses = vec![];
                let mut nullable = false;
                for ty in types {
                    let field_ty = match ty {
                        CoercibleScalarType::Record(fts) => fts[i].clone(),
                        _ => unreachable!(),
                    };
                    if field_ty.nullable() {
                        nullable = true;
                    }
                    guesses.push(field_ty.scalar_type());
                }
                let guess = guess_best_common_type(ecx, &guesses)?;
                fields.push((name, guess.nullable(nullable)));
            }
            return Ok(SqlScalarType::Record {
                fields: fields.into(),
                custom_id: None,
            });
        }
    }

    // Remove unknown types, and collect them.
    let mut types: Vec<_> = types.into_iter().filter_map(|v| v.as_coerced()).collect();

    // In the case of mixed ints and uints, replace uints with their near match
    let contains_int = types.iter().any(|t| {
        matches!(
            t,
            SqlScalarType::Int16 | SqlScalarType::Int32 | SqlScalarType::Int64
        )
    });

    for t in types.iter_mut() {
        if contains_int
            && matches!(
                t,
                SqlScalarType::UInt16 | SqlScalarType::UInt32 | SqlScalarType::UInt64
            )
        {
            *t = t.near_match().expect("unsigned ints have near matches")
        }
    }

    let mut types = types.iter();

    let mut candidate = match types.next() {
        // If no known types, fall back to `String`.
        None => return Ok(SqlScalarType::String),
        // Start by guessing the first type.
        Some(t) => t,
    };

    let preferred_type = TypeCategory::from_type(candidate).preferred_type();

    for typ in types {
        if TypeCategory::from_type(candidate) != TypeCategory::from_type(typ) {
            // The next type is in a different category; give up.
            sql_bail!(
                "{} types {} and {} cannot be matched",
                ecx.name,
                ecx.humanize_sql_scalar_type(candidate, false),
                ecx.humanize_sql_scalar_type(typ, false),
            );
        };

        // If this type is the preferred type, make it the candidate.
        if preferred_type.as_ref() != Some(candidate)
            && can_cast(ecx, CastContext::Implicit, candidate, typ)
            && !can_cast(ecx, CastContext::Implicit, typ, candidate)
        {
            // The current candidate is not the preferred type for its category
            // and the next type is implicitly convertible to the current
            // candidate, but not vice-versa, so take the next type as the new
            // candidate.
            candidate = typ;
        }
    }
    Ok(candidate.without_modifiers())
}

pub fn plan_coerce<'a>(
    ecx: &'a ExprContext,
    e: CoercibleScalarExpr,
    coerce_to: &SqlScalarType,
) -> Result<HirScalarExpr, PlanError> {
    plan_coerce_with_failure_mode(ecx, e, coerce_to, CastFailureMode::Error)
}

/// Like [`plan_coerce`], but the casts that coerce a literal to `coerce_to`
/// use the given failure mode. `TRY_CAST('abc' AS int4)` reaches its cast
/// through this coercion, so the mode has to apply here too.
pub fn plan_coerce_with_failure_mode<'a>(
    ecx: &'a ExprContext,
    e: CoercibleScalarExpr,
    coerce_to: &SqlScalarType,
    mode: CastFailureMode,
) -> Result<HirScalarExpr, PlanError> {
    use CoercibleScalarExpr::*;

    Ok(match e {
        Coerced(e) => e,

        LiteralNull => HirScalarExpr::literal_null(coerce_to.clone()),

        LiteralString(s) => {
            let lit = HirScalarExpr::literal(Datum::String(&s), SqlScalarType::String);
            // Per PostgreSQL, string literal explicitly casts to the base type.
            // The caller is responsible for applying any desired modifiers
            // (with either implicit or explicit semantics) via a separate call
            // to `plan_cast`.
            let coerce_to_base = &coerce_to.without_modifiers();
            plan_cast_with_failure_mode(ecx, CastContext::Coerced, mode, lit, coerce_to_base)?
        }

        LiteralRecord(exprs) => {
            let arity = exprs.len();
            // Under `NullFallback` the target field types are not pushed into
            // the literal. Fields are typed on their own, so every field
            // conversion happens inside the record cast that follows, where
            // the `TryCast` wrapper covers it, and a bad field yields NULL for
            // the whole record rather than a record with a NULL field. This
            // mirrors how `plan_cast` treats collection constructors.
            let coercions = match (coerce_to, mode) {
                (SqlScalarType::Record { fields, .. }, CastFailureMode::Error)
                    if fields.len() == arity =>
                {
                    fields
                        .iter()
                        .map(|(_name, ty)| &ty.scalar_type)
                        .cloned()
                        .collect()
                }
                _ => vec![SqlScalarType::String; exprs.len()],
            };
            let mut out = vec![];
            for (e, coerce_to) in exprs.into_iter().zip_eq(coercions) {
                out.push(plan_coerce_with_failure_mode(ecx, e, &coerce_to, mode)?);
            }
            HirScalarExpr::call_variadic(
                RecordCreate {
                    field_names: (0..arity)
                        .map(|i| ColumnName::from(format!("f{}", i + 1)))
                        .collect(),
                },
                out,
            )
        }

        Parameter(n) => {
            let prev = ecx.param_types().borrow_mut().insert(n, coerce_to.clone());
            if let Some(prev) = prev {
                if prev != *coerce_to {
                    sql_bail!(
                        "there are contradicting constraints for \
                         the type of parameter ${}: \
                         should be both {} and {}",
                        n,
                        ecx.humanize_sql_scalar_type(&prev, false),
                        ecx.humanize_sql_scalar_type(coerce_to, false),
                    );
                }
            }
            HirScalarExpr::parameter(n)
        }
    })
}

/// Returns an error if the given type is not a supported range element type
/// (PostgreSQL built-in range types: int32, int64, date, numeric, timestamp,
/// timestamptz).
fn validate_range_element_type(
    ecx: &ExprContext,
    element_type: &SqlScalarType,
) -> Result<(), CastError> {
    let allowed = matches!(
        element_type,
        SqlScalarType::Int32
            | SqlScalarType::Int64
            | SqlScalarType::Date
            | SqlScalarType::Numeric { .. }
            | SqlScalarType::Timestamp { .. }
            | SqlScalarType::TimestampTz { .. }
    );
    if allowed {
        Ok(())
    } else {
        Err(CastError::UnsupportedRangeElementType {
            element_type_name: ecx.humanize_sql_scalar_type(element_type, false),
        })
    }
}

/// Similar to `plan_cast`, but for situations where you only know the type of
/// the input expression (`from`) and not the expression itself. The returned
/// expression refers to the first column of some imaginary row, where the first
/// column is assumed to have type `from`.
///
/// If casting from `from` to `to` is not possible, returns `None`.
pub fn plan_hypothetical_cast(
    ecx: &ExprContext,
    ccx: CastContext,
    from: &SqlScalarType,
    to: &SqlScalarType,
) -> Option<mz_expr::MirScalarExpr> {
    plan_hypothetical_cast_with_failure_mode(ecx, ccx, CastFailureMode::Error, from, to)
}

/// Similar to `plan_hypothetical_cast`, but with a `CastFailureMode` to specify behavior.
pub fn plan_hypothetical_cast_with_failure_mode(
    ecx: &ExprContext,
    ccx: CastContext,
    mode: CastFailureMode,
    from: &SqlScalarType,
    to: &SqlScalarType,
) -> Option<mz_expr::MirScalarExpr> {
    // Reconstruct an expression context where the expression is evaluated on
    // the "first column" of some imaginary row.
    let mut scx = ecx.qcx.scx.clone();
    scx.param_types = RefCell::new(BTreeMap::new());
    let qcx = QueryContext::root(&scx, ecx.qcx.lifetime);
    let relation_type = SqlRelationType {
        column_types: vec![SqlColumnType {
            nullable: true,
            scalar_type: from.clone(),
        }],
        keys: vec![vec![0]],
    };
    let ecx = ExprContext {
        qcx: &qcx,
        name: "plan_hypothetical_cast",
        scope: &Scope::empty(),
        relation_type: &relation_type,
        allow_aggregates: false,
        allow_subqueries: true,
        allow_parameters: true,
        allow_windows: false,
    };

    let col_expr = HirScalarExpr::column(0);

    // Determine the `ScalarExpr` required to cast our column to the target
    // component type.
    //
    // NOTE: the templates plan element casts through `plan_hypothetical_cast`,
    // which is always strict, whatever the enclosing cast's failure mode. Under
    // `TRY_CAST` the enclosing function is what gets wrapped, so a bad element
    // makes the whole value NULL rather than producing a value with NULL holes
    // in it.
    plan_cast_with_failure_mode(&ecx, ccx, mode, col_expr, to)
        .ok()?
        // TODO(jkosh44) Support casts that have correlated implementations.
        .lower_uncorrelated(ecx.catalog().system_vars())
        .ok()
}

/// Plans a cast between [`SqlScalarType`]s, specifying which types of casts are
/// permitted using [`CastContext`].
///
/// # Errors
///
/// If a cast between the `ScalarExpr`'s base type and the specified type is:
/// - Not possible, e.g. `Bytes` to `Interval`
/// - Not permitted, e.g. implicitly casting from `Float64` to `Float32`.
/// - Not implemented yet
pub fn plan_cast(
    ecx: &ExprContext,
    ccx: CastContext,
    expr: HirScalarExpr,
    to: &SqlScalarType,
) -> Result<HirScalarExpr, PlanError> {
    plan_cast_with_failure_mode(ecx, ccx, CastFailureMode::Error, expr, to)
}

/// Like [`plan_cast`], but with the given failure mode. Under
/// [`CastFailureMode::NullFallback`] every stage of the cast, including the
/// intermediate string cast that string-like types route through, is wrapped so
/// that a failure anywhere yields NULL.
pub fn plan_cast_with_failure_mode(
    ecx: &ExprContext,
    ccx: CastContext,
    mode: CastFailureMode,
    expr: HirScalarExpr,
    to: &SqlScalarType,
) -> Result<HirScalarExpr, PlanError> {
    let from = ecx.scalar_type(&expr);

    // Close over `ccx`, `mode`, `from`, and `to` to simplify error messages in
    // the face of intermediate expressions.
    let cast_inner = |from, to, expr| {
        get_cast(ecx, ccx, mode, from, to)
            .map(|cast| cast(expr))
            .map_err(|e| e.into_plan_error(ecx.name.into()))
    };

    // String-like types get special handling to match PostgreSQL.
    // See: https://github.com/postgres/postgres/blob/6b04abdfc/
    //   src/backend/parser/parse_coerce.c#L3205-L3223
    let from_category = TypeCategory::from_type(&from);
    let to_category = TypeCategory::from_type(to);
    if from_category == TypeCategory::String && to_category != TypeCategory::String {
        // Converting from stringlike to something non-stringlike. Handle as if
        // `from` were a `SqlScalarType::String`.
        //
        // `"char"` is string-like but stores a byte (`Datum::UInt8`), so the
        // string-source cast functions cannot consume it directly. Use its
        // registered cast when one exists (`int4`, by byte value) and
        // otherwise render it to text first.
        let expr = match from {
            SqlScalarType::PgLegacyChar => match get_cast(ecx, ccx, mode, &from, to) {
                Ok(cast) => return Ok(cast(expr)),
                Err(_) => cast_inner(&from, &SqlScalarType::String, expr)?,
            },
            _ => expr,
        };
        cast_inner(&SqlScalarType::String, to, expr)
    } else if from_category != TypeCategory::String && to_category == TypeCategory::String {
        // Converting from non-stringlike to something stringlike. Convert to a
        // `SqlScalarType::String` and then to the desired type.
        let expr = cast_inner(&from, &SqlScalarType::String, expr)?;
        cast_inner(&SqlScalarType::String, to, expr)
    } else {
        // Standard cast.
        cast_inner(&from, to, expr)
    }
}

/// Reports whether it is possible to perform a cast from the specified types.
pub fn can_cast(
    ecx: &ExprContext,
    ccx: CastContext,
    cast_from: &SqlScalarType,
    cast_to: &SqlScalarType,
) -> bool {
    get_cast(ecx, ccx, CastFailureMode::Error, cast_from, cast_to).is_ok()
}

#[cfg(test)]
mod tests {
    use mz_expr::func::{CastInt64ToInt32, CastStringToInt64, Eq as EqFunc, IsNull};
    use mz_repr::SqlScalarType;

    use super::extract_unary_stages_for_try_cast;
    use crate::plan::hir::HirScalarExpr;

    #[mz_ore::test]
    fn cast_stages_of_chain_are_innermost_first() {
        let stages =
            extract_unary_stages_for_try_cast(&SqlScalarType::String, |e: HirScalarExpr| {
                e.call_unary(CastStringToInt64.into())
                    .call_unary(CastInt64ToInt32.into())
            });
        assert_eq!(
            stages,
            Some(vec![CastStringToInt64.into(), CastInt64ToInt32.into()])
        );
    }

    #[mz_ore::test]
    fn cast_stages_of_identity_is_empty() {
        assert_eq!(
            extract_unary_stages_for_try_cast(&SqlScalarType::Int32, |e: HirScalarExpr| e),
            Some(vec![])
        );
    }

    #[mz_ore::test]
    fn cast_stages_rejects_non_unary_shapes() {
        let compares = |e: HirScalarExpr| {
            let zero = HirScalarExpr::literal_null(SqlScalarType::Int32);
            e.call_binary(zero, EqFunc)
        };
        assert_eq!(
            extract_unary_stages_for_try_cast(&SqlScalarType::Int32, compares),
            None
        );
        let ignores_input = |_: HirScalarExpr| HirScalarExpr::literal_null(SqlScalarType::Int32);
        assert_eq!(
            extract_unary_stages_for_try_cast(&SqlScalarType::Int32, ignores_input),
            None
        );
    }

    /// A stage after the first that does not propagate NULLs would turn the
    /// NULL of an earlier failure into a value, so wrapping must refuse it.
    #[mz_ore::test]
    fn cast_stages_rejects_non_null_propagating_later_stage() {
        let chain = |e: HirScalarExpr| {
            e.call_unary(CastInt64ToInt32.into())
                .call_unary(IsNull.into())
        };
        let outcome = mz_ore::panic::catch_unwind(|| {
            extract_unary_stages_for_try_cast(&SqlScalarType::Int64, chain)
        });
        if mz_ore::assert::soft_assertions_enabled() {
            assert!(outcome.is_err(), "soft assertion should have fired");
        } else {
            assert_eq!(outcome.expect("no panic"), None);
        }
    }
}