spg_engine/

conversions.rs

//! Type conversions — Value/literal <-> text/bytes/special-format. The
//! coercion entry point (`coerce_value`) plus every parser/formatter it
//! leans on: bytea, text/2-D arrays, hstore, ranges, money, time, year,
//! and literal->Value. Split out of `lib.rs` (v7.32 engine
//! modularisation); a self-contained cluster (its members call each
//! other), depending only on spg_storage/spg_sql, `eval`, and `numeric`.

use alloc::string::ToString;
use alloc::vec::Vec;

use spg_sql::ast::{ColumnTypeName, Expr, Literal, UnOp, VecEncoding as SqlVecEncoding};
use spg_storage::{ColumnSchema, DataType, StorageError, Value, VecEncoding};

use crate::EngineError;
use crate::eval::{self, EvalContext, EvalError};
use crate::numeric::{
    numeric_from_float, numeric_from_integer, numeric_rescale, numeric_round_to_integer,
    parse_numeric_text,
};

/// v7.10.4 — decode a BYTEA literal. Accepts:
///   * `\xDEADBEEF` (case-insensitive hex; whitespace stripped)
///   * `Hello\000world` (backslash escape form; `\\` for literal backslash)
///   * Anything else → raw UTF-8 bytes of the input (PG accepts this too).
/// v7.39 (round 325, V57) — errors are PG's own, verbatim:
/// `invalid hexadecimal digit: "Z"` (naming the offending character) and
/// `invalid hexadecimal data: odd number of digits`. They used to be SPG
/// phrasings wrapped in `cannot parse "…" as BYTEA: `.
/// PostgreSQL's pseudo-types, canonically spelled.
///
/// v7.38.19 — `SELECT typname FROM pg_type WHERE typtype = 'p'` on
/// PostgreSQL 18.4, minus `any` and `_record`. `any` is a reserved word,
/// so PG answers a syntax error rather than a type error and matching
/// its list here would replace one with the other; `_record` is the
/// array spelling and is not written by hand.
///
/// A pseudo-type has no storage. PG refuses a column declared with one,
/// and it refuses it as an INVALID TABLE DEFINITION (42P16) rather than
/// an undefined type (42704) -- the name exists, it just cannot hold a
/// value. SPG answered `type "cstring" does not exist`, which is the
/// wrong class and the wrong claim.
pub(crate) fn pseudo_type(name: &str) -> Option<&'static str> {
    const NAMES: &[&str] = &[
        "anyarray",
        "anycompatible",
        "anycompatiblearray",
        "anycompatiblemultirange",
        "anycompatiblenonarray",
        "anycompatiblerange",
        "anyelement",
        "anyenum",
        "anymultirange",
        "anynonarray",
        "anyrange",
        "cstring",
        "event_trigger",
        "fdw_handler",
        "index_am_handler",
        "internal",
        "language_handler",
        "pg_ddl_command",
        "record",
        "table_am_handler",
        "trigger",
        "tsm_handler",
        "unknown",
        "void",
    ];
    NAMES.iter().find(|n| n.eq_ignore_ascii_case(name)).copied()
}

pub(crate) fn decode_bytea_literal(s: &str) -> Result<alloc::vec::Vec<u8>, alloc::string::String> {
    let s = s.trim();
    if let Some(hex) = s.strip_prefix("\\x").or_else(|| s.strip_prefix("\\X")) {
        // Hex form. Each pair of hex digits → one byte.
        let cleaned: alloc::string::String = hex.chars().filter(|c| !c.is_whitespace()).collect();
        if cleaned.len() % 2 != 0 {
            return Err(alloc::string::String::from(
                "invalid hexadecimal data: odd number of digits",
            ));
        }
        let mut out = alloc::vec::Vec::with_capacity(cleaned.len() / 2);
        let cleaned_bytes = cleaned.as_bytes();
        for i in (0..cleaned_bytes.len()).step_by(2) {
            let hi = hex_nibble(cleaned_bytes[i]).map_err(|()| bad_hex_digit(cleaned_bytes[i]))?;
            let lo = hex_nibble(cleaned_bytes[i + 1])
                .map_err(|()| bad_hex_digit(cleaned_bytes[i + 1]))?;
            out.push((hi << 4) | lo);
        }
        return Ok(out);
    }
    // Escape form or raw. Walk char-by-char; `\\` and `\NNN` octal
    // sequences decode; anything else is a literal byte.
    let bytes = s.as_bytes();
    let mut out = alloc::vec::Vec::with_capacity(bytes.len());
    let mut i = 0;
    while i < bytes.len() {
        let b = bytes[i];
        if b == b'\\' && i + 1 < bytes.len() {
            let n = bytes[i + 1];
            if n == b'\\' {
                out.push(b'\\');
                i += 2;
                continue;
            }
            if n.is_ascii_digit()
                && i + 3 < bytes.len()
                && bytes[i + 2].is_ascii_digit()
                && bytes[i + 3].is_ascii_digit()
            {
                let oct = |x: u8| (x - b'0') as u32;
                let v = oct(n) * 64 + oct(bytes[i + 2]) * 8 + oct(bytes[i + 3]);
                if v <= 0xFF {
                    out.push(v as u8);
                    i += 4;
                    continue;
                }
            }
        }
        out.push(b);
        i += 1;
    }
    Ok(out)
}

pub(crate) fn hex_nibble(b: u8) -> Result<u8, ()> {
    match b {
        b'0'..=b'9' => Ok(b - b'0'),
        b'a'..=b'f' => Ok(b - b'a' + 10),
        b'A'..=b'F' => Ok(b - b'A' + 10),
        _ => Err(()),
    }
}

/// PG names the character it choked on.
fn bad_hex_digit(b: u8) -> alloc::string::String {
    alloc::format!("invalid hexadecimal digit: \"{}\"", b as char)
}

/// v7.37.5 γ — uniform array-of-scalar shape detector. Returns
/// `Some(kind)` only when every non-NULL element fits the same
/// new-array element type; `None` falls back to the legacy
/// `array_literal_widen` Int/BigInt/Text path.
#[derive(Clone, Copy)]
enum UniformArrayKind {
    Bool,
    Float,
    Numeric,
    Date,
    Timestamp,
    Uuid,
    Bytes,
    Interval,
    Money,
}

impl UniformArrayKind {
    fn build(self, items: alloc::vec::Vec<Value<'static>>) -> Value<'static> {
        match self {
            Self::Bool => Value::BoolArray(
                items
                    .into_iter()
                    .map(|v| match v {
                        Value::Null => None,
                        Value::Bool(b) => Some(b),
                        _ => unreachable!("uniform Bool"),
                    })
                    .collect(),
            ),
            Self::Float => Value::FloatArray(
                items
                    .into_iter()
                    .map(|v| match v {
                        Value::Null => None,
                        Value::Float(x) => Some(x),
                        _ => unreachable!("uniform Float"),
                    })
                    .collect(),
            ),
            Self::Numeric => Value::NumericArray(
                items
                    .into_iter()
                    .map(|v| match v {
                        Value::Null => None,
                        Value::Numeric { scaled, scale, .. } => Some((scaled, scale)),
                        _ => unreachable!("uniform Numeric"),
                    })
                    .collect(),
            ),
            Self::Date => Value::DateArray(
                items
                    .into_iter()
                    .map(|v| match v {
                        Value::Null => None,
                        Value::Date(d) => Some(d),
                        _ => unreachable!("uniform Date"),
                    })
                    .collect(),
            ),
            Self::Timestamp => Value::TimestampArray(
                items
                    .into_iter()
                    .map(|v| match v {
                        Value::Null => None,
                        Value::Timestamp(t) => Some(t),
                        _ => unreachable!("uniform Timestamp"),
                    })
                    .collect(),
            ),
            Self::Uuid => Value::UuidArray(
                items
                    .into_iter()
                    .map(|v| match v {
                        Value::Null => None,
                        Value::Uuid(b) => Some(b),
                        _ => unreachable!("uniform Uuid"),
                    })
                    .collect(),
            ),
            Self::Bytes => Value::BytesArray(
                items
                    .into_iter()
                    .map(|v| match v {
                        Value::Null => None,
                        Value::Bytes(b) => Some(b.into_owned()),
                        _ => unreachable!("uniform Bytes"),
                    })
                    .collect(),
            ),
            Self::Interval => Value::IntervalArray(
                items
                    .into_iter()
                    .map(|v| match v {
                        Value::Null => None,
                        Value::Interval {
                            months,
                            days,
                            micros,
                            kind,
                        } => Some(spg_storage::IntervalSpan {
                            months,
                            days,
                            micros,
                            kind,
                        }),
                        _ => unreachable!("uniform Interval"),
                    })
                    .collect(),
            ),
            Self::Money => Value::MoneyArray(
                items
                    .into_iter()
                    .map(|v| match v {
                        Value::Null => None,
                        Value::Money(c) => Some(c),
                        _ => unreachable!("uniform Money"),
                    })
                    .collect(),
            ),
        }
    }
}

fn widen_uniform_typed(items: &[Value<'static>]) -> Option<UniformArrayKind> {
    let mut kind: Option<UniformArrayKind> = None;
    let mut saw_non_null = false;
    for v in items {
        let this = match v {
            Value::Null => continue,
            Value::Bool(_) => UniformArrayKind::Bool,
            Value::Float(_) => UniformArrayKind::Float,
            Value::Numeric { .. } => UniformArrayKind::Numeric,
            Value::Date(_) => UniformArrayKind::Date,
            Value::Timestamp(_) => UniformArrayKind::Timestamp,
            Value::Uuid(_) => UniformArrayKind::Uuid,
            Value::Bytes(_) => UniformArrayKind::Bytes,
            Value::Interval { .. } => UniformArrayKind::Interval,
            Value::Money(_) => UniformArrayKind::Money,
            // Int / BigInt / Text / Json — defer to the legacy
            // Int/Text widen below so the existing IntArray /
            // BigIntArray / TextArray behaviour is unchanged.
            _ => return None,
        };
        match kind {
            None => kind = Some(this),
            Some(prev) if discriminant_eq(prev, this) => {}
            Some(_) => return None,
        }
        saw_non_null = true;
    }
    if saw_non_null { kind } else { None }
}

fn discriminant_eq(a: UniformArrayKind, b: UniformArrayKind) -> bool {
    matches!(
        (a, b),
        (UniformArrayKind::Bool, UniformArrayKind::Bool)
            | (UniformArrayKind::Float, UniformArrayKind::Float)
            | (UniformArrayKind::Numeric, UniformArrayKind::Numeric)
            | (UniformArrayKind::Date, UniformArrayKind::Date)
            | (UniformArrayKind::Timestamp, UniformArrayKind::Timestamp)
            | (UniformArrayKind::Uuid, UniformArrayKind::Uuid)
            | (UniformArrayKind::Bytes, UniformArrayKind::Bytes)
            | (UniformArrayKind::Interval, UniformArrayKind::Interval)
            | (UniformArrayKind::Money, UniformArrayKind::Money)
    )
}

/// v7.10.11 — decode a PG TEXT[] external array form
/// (`{a,b,NULL}` with optional double-quoted elements). The
/// engine takes a leading/trailing `{`/`}` and splits at commas.
/// Quoted elements (`"hello, world"`) preserve embedded commas;
/// `\\` and `\"` decode to literal backslash / quote. Plain
/// unquoted `NULL` (case-insensitive) maps to `None`.
/// v7.11.13 — pick the array type for `ARRAY[lit, …]` from the
/// element values. Single-element-type rules:
///   - all NULL / all Text → TextArray
///   - all Int (or Int+NULL) → IntArray
///   - any BigInt without Text → BigIntArray (widening)
///   - any Text → TextArray (fallback; non-string elements
///     render as text)
pub(crate) fn array_literal_widen(items: alloc::vec::Vec<Value<'static>>) -> Value<'static> {
    // v7.37.5 γ — first, detect a uniform new-array-type. If every
    // non-NULL element shares one of the array-of-scalar element
    // shapes (Bool / Float / Numeric / Date / Timestamp / Uuid /
    // Bytes / Interval), build the matching typed array directly
    // so INSERT to a typed column doesn't have to go through the
    // TextArray fallback + coerce chain.
    // v7.39 (read01 round 75) — rows that are themselves arrays make a 2-D array.
    // This path (the INSERT literal one) did not know 2-D at all, so an
    // `ARRAY[ARRAY[…]]` in a VALUES list silently collapsed to text[] — the same
    // per-variant hole, in the builder next door.
    if let Some(m) = crate::eval::values::build_2d_from_rows(&items) {
        return m;
    }
    if let Some(arr) = widen_uniform_typed(&items) {
        return arr.build(items);
    }
    let mut has_text = false;
    let mut has_bigint = false;
    let mut has_int = false;
    for v in &items {
        match v {
            Value::Null => {}
            Value::Text(_) | Value::Json(_) => has_text = true,
            Value::BigInt(_) => has_bigint = true,
            Value::Int(_) | Value::SmallInt(_) => has_int = true,
            _ => has_text = true,
        }
    }
    if has_text || (!has_bigint && !has_int) {
        let out: alloc::vec::Vec<Option<alloc::string::String>> = items
            .into_iter()
            .map(|v| match v {
                Value::Null => None,
                Value::Text(s) | Value::Json(s) => Some(s.into_owned()),
                other => Some(alloc::format!("{other:?}")),
            })
            .collect();
        return Value::TextArray(out);
    }
    if has_bigint {
        let out: alloc::vec::Vec<Option<i64>> = items
            .into_iter()
            .map(|v| match v {
                Value::Null => None,
                Value::Int(n) => Some(i64::from(n)),
                Value::SmallInt(n) => Some(i64::from(n)),
                Value::BigInt(n) => Some(n),
                _ => unreachable!("widen: unexpected non-integer in BigInt path"),
            })
            .collect();
        return Value::BigIntArray(out);
    }
    let out: alloc::vec::Vec<Option<i32>> = items
        .into_iter()
        .map(|v| match v {
            Value::Null => None,
            Value::Int(n) => Some(n),
            Value::SmallInt(n) => Some(i32::from(n)),
            _ => unreachable!("widen: unexpected non-i32-compatible in Int path"),
        })
        .collect();
    Value::IntArray(out)
}

/// v7.39 (round 325, V57) — PG's message for a literal that will not
/// become an array, DETAIL and all. Measured on PG 18.4 (INSERT into a
/// typed column):
///
/// | literal | DETAIL |
/// |---|---|
/// | `abc` | `Array value must start with "{" or dimension information.` |
/// | `{1,2` | `Unexpected end of input.` |
/// | `{1,2}}` · `{1,2}x` | `Junk after closing right brace.` |
/// | `{1,}` | `Unexpected "}" character.` |
///
/// The `" DETAIL: "` separator is the one the wire splits into the
/// ErrorResponse `D` field. An element that fails to convert is NOT this
/// error: PG reports the ELEMENT type's own input-syntax error, which is
/// what the per-element coercion below already produces.
#[must_use]
pub(crate) fn malformed_array_literal(text: &str) -> alloc::string::String {
    let t = text.trim();
    let detail = if !t.starts_with('{') {
        "Array value must start with \"{\" or dimension information."
    } else {
        // The array ends at the FIRST unquoted `}` — the same rule the
        // decoder applies, so `{1,2}}` is junk after the brace rather
        // than an unterminated literal.
        match first_unquoted_close_brace(&t[1..]) {
            None => "Unexpected end of input.",
            Some(close) => {
                let inner = &t[1..1 + close];
                if !t[1 + close + 1..].trim().is_empty() {
                    "Junk after closing right brace."
                } else if inner.trim_end().ends_with(',') {
                    "Unexpected \"}\" character."
                } else {
                    "Unexpected end of input."
                }
            }
        }
    };
    alloc::format!("malformed array literal: \"{text}\" DETAIL: {detail}")
}

/// Byte offset of the first `}` outside quotes, if any.
fn first_unquoted_close_brace(body: &str) -> Option<usize> {
    let bs = body.as_bytes();
    let mut in_quote = false;
    let mut k = 0;
    while k < bs.len() {
        match bs[k] {
            b'\\' if in_quote => k += 1,
            b'"' => in_quote = !in_quote,
            b'}' if !in_quote => return Some(k),
            _ => {}
        }
        k += 1;
    }
    None
}

pub(crate) fn decode_text_array_literal(
    s: &str,
) -> Result<alloc::vec::Vec<Option<alloc::string::String>>, &'static str> {
    let trimmed = s.trim();
    // v7.39 (round 325, V57) — the array ends at the FIRST unquoted `}`,
    // and anything after it is junk. Peeling one brace off each end let
    // `{1,2}}` through as the elements `1` and `2}`, so the failure was
    // reported as a bad INTEGER rather than PG's "Junk after closing right
    // brace." — a wrong diagnosis, not just wrong words.
    let body = trimmed
        .strip_prefix('{')
        .ok_or("TEXT[] literal must be enclosed in '{...}'")?;
    let close =
        first_unquoted_close_brace(body).ok_or("TEXT[] literal must be enclosed in '{...}'")?;
    if !body[close + 1..].trim().is_empty() {
        return Err("junk after closing right brace");
    }
    let inner = &body[..close];
    let mut out: alloc::vec::Vec<Option<alloc::string::String>> = alloc::vec::Vec::new();
    if inner.trim().is_empty() {
        return Ok(out);
    }
    let bytes = inner.as_bytes();
    let mut i = 0;
    while i <= bytes.len() {
        // Skip leading whitespace.
        while i < bytes.len() && (bytes[i] == b' ' || bytes[i] == b'\t') {
            i += 1;
        }
        // Quoted element.
        if i < bytes.len() && bytes[i] == b'"' {
            i += 1; // open quote
            let mut buf = alloc::string::String::new();
            while i < bytes.len() && bytes[i] != b'"' {
                if bytes[i] == b'\\' && i + 1 < bytes.len() {
                    buf.push(bytes[i + 1] as char);
                    i += 2;
                } else {
                    buf.push(bytes[i] as char);
                    i += 1;
                }
            }
            if i >= bytes.len() {
                return Err("unterminated quoted element");
            }
            i += 1; // close quote
            out.push(Some(buf));
        } else {
            // Unquoted element — read until next comma or end.
            let start = i;
            while i < bytes.len() && bytes[i] != b',' {
                i += 1;
            }
            let raw = inner[start..i].trim();
            // v7.39 (round 325, V57) — PG rejects an empty UNQUOTED
            // element (`{1,}` is `Unexpected "}" character.`); it used to
            // become an empty string, which then failed as a bad element
            // of whatever the array's type was.
            if raw.is_empty() {
                return Err("empty array element");
            }
            if raw.eq_ignore_ascii_case("NULL") {
                out.push(None);
            } else {
                out.push(Some(alloc::string::ToString::to_string(raw)));
            }
        }
        // Skip whitespace, expect comma or end.
        while i < bytes.len() && (bytes[i] == b' ' || bytes[i] == b'\t') {
            i += 1;
        }
        if i >= bytes.len() {
            break;
        }
        if bytes[i] != b',' {
            return Err("expected ',' between TEXT[] elements");
        }
        i += 1;
    }
    Ok(out)
}

/// v7.10.11 — encode a TEXT[] back into the PG external array
/// form. NULL elements become the literal `NULL`; elements
/// containing commas, quotes, backslashes, or braces are
/// double-quoted with `\\` / `\"` escapes.
pub(crate) fn encode_text_array(items: &[Option<alloc::string::String>]) -> alloc::string::String {
    let mut out = alloc::string::String::with_capacity(2 + items.len() * 8);
    out.push('{');
    for (i, item) in items.iter().enumerate() {
        if i > 0 {
            out.push(',');
        }
        match item {
            None => out.push_str("NULL"),
            Some(s) => {
                let needs_quote = s.is_empty()
                    || s.eq_ignore_ascii_case("NULL")
                    || s.chars()
                        .any(|c| matches!(c, ',' | '{' | '}' | '"' | '\\' | ' ' | '\t'));
                if needs_quote {
                    out.push('"');
                    for c in s.chars() {
                        if c == '"' || c == '\\' {
                            out.push('\\');
                        }
                        out.push(c);
                    }
                    out.push('"');
                } else {
                    out.push_str(s);
                }
            }
        }
    }
    out.push('}');
    out
}

/// v7.10.4 — encode BYTEA bytes in PG hex output format
/// (`\x` prefix, lowercase hex pairs). Used by Text-side
/// round-trip + the wire layer's text-mode encoder.
pub(crate) fn encode_bytea_hex(b: &[u8]) -> alloc::string::String {
    let mut out = alloc::string::String::with_capacity(2 + 2 * b.len());
    out.push_str("\\x");
    for byte in b {
        let hi = byte >> 4;
        let lo = byte & 0x0F;
        out.push(hex_digit(hi));
        out.push(hex_digit(lo));
    }
    out
}

pub(crate) const fn hex_digit(n: u8) -> char {
    match n {
        0..=9 => (b'0' + n) as char,
        10..=15 => (b'a' + n - 10) as char,
        _ => '?',
    }
}

/// v7.17.0 Phase 3.P0-39 — parse a PG `hstore` text literal into
/// a flat key→value map. Empty string → empty map. Duplicate
/// keys keep the FIRST occurrence (PG18-measured, round 780; the old
/// note claimed last-write-wins).
///
/// Accepted shapes (minimal subset):
///   * `'a=>1, b=>2'`            — bareword keys/values
///   * `'"a"=>"1", "b"=>"2"'`    — quoted keys/values
///   * `'a=>NULL'`               — case-insensitive NULL token
///     surfaces as `None` (no quotes around NULL)
///
/// Returns None on parse failure → caller surfaces as hard error.
pub(crate) fn parse_hstore_str(
    s: &str,
) -> Option<Vec<(alloc::string::String, Option<alloc::string::String>)>> {
    let bytes = s.as_bytes();
    let mut i = 0;
    let mut out: Vec<(alloc::string::String, Option<alloc::string::String>)> = Vec::new();
    let skip_ws = |bytes: &[u8], i: &mut usize| {
        while *i < bytes.len() && matches!(bytes[*i], b' ' | b'\t' | b'\n' | b'\r') {
            *i += 1;
        }
    };
    let parse_token = |bytes: &[u8], i: &mut usize| -> Option<alloc::string::String> {
        if *i >= bytes.len() {
            return None;
        }
        if bytes[*i] == b'"' {
            *i += 1;
            let mut out = alloc::string::String::new();
            while *i < bytes.len() {
                match bytes[*i] {
                    b'"' => {
                        *i += 1;
                        return Some(out);
                    }
                    b'\\' if *i + 1 < bytes.len() => {
                        out.push(bytes[*i + 1] as char);
                        *i += 2;
                    }
                    c => {
                        out.push(c as char);
                        *i += 1;
                    }
                }
            }
            None
        } else {
            let start = *i;
            while *i < bytes.len()
                && !matches!(bytes[*i], b' ' | b'\t' | b'\n' | b'\r' | b',' | b'=')
            {
                *i += 1;
            }
            if *i == start {
                return None;
            }
            Some(alloc::str::from_utf8(&bytes[start..*i]).ok()?.to_string())
        }
    };
    skip_ws(bytes, &mut i);
    while i < bytes.len() {
        let key = parse_token(bytes, &mut i)?;
        skip_ws(bytes, &mut i);
        if i + 1 >= bytes.len() || bytes[i] != b'=' || bytes[i + 1] != b'>' {
            return None;
        }
        i += 2;
        skip_ws(bytes, &mut i);
        // Check for unquoted NULL token (case-insensitive).
        let val_token = if i + 4 <= bytes.len()
            && bytes[i..i + 4].eq_ignore_ascii_case(b"NULL")
            && (i + 4 == bytes.len() || matches!(bytes[i + 4], b' ' | b'\t' | b',' | b'\n' | b'\r'))
        {
            i += 4;
            None
        } else {
            Some(parse_token(bytes, &mut i)?)
        };
        // v7.39 (round 780, F31-D1) — PG's hstore_in keeps the FIRST
        // occurrence of a duplicate key (measured: 'a=>1, a=>2' is
        // "a"=>"1"); the old arm replaced it and the comment claimed
        // last-write-wins matched PG.
        if out.iter().any(|(k, _)| k == &key) {
            // keep the first
        } else {
            out.push((key, val_token));
        }
        skip_ws(bytes, &mut i);
        if i >= bytes.len() {
            break;
        }
        if bytes[i] == b',' {
            i += 1;
            skip_ws(bytes, &mut i);
            continue;
        }
        return None;
    }
    Some(out)
}

/// v7.17.0 Phase 3.P0-39 — render a hstore as canonical PG text
/// form `"k"=>"v"` (keys and non-NULL values always quoted;
/// NULL token is bare).
pub(crate) fn format_hstore_str(
    pairs: &[(alloc::string::String, Option<alloc::string::String>)],
) -> alloc::string::String {
    let mut out = alloc::string::String::new();
    for (i, (k, v)) in pairs.iter().enumerate() {
        if i > 0 {
            out.push_str(", ");
        }
        out.push('"');
        out.push_str(k);
        out.push_str("\"=>");
        match v {
            None => out.push_str("NULL"),
            Some(val) => {
                out.push('"');
                out.push_str(val);
                out.push('"');
            }
        }
    }
    out
}

/// v7.17.0 Phase 3.P0-39 — pub re-export so pgwire + sqllogictest
/// share the single hstore renderer.
pub fn format_hstore_text(
    pairs: &[(alloc::string::String, Option<alloc::string::String>)],
) -> alloc::string::String {
    format_hstore_str(pairs)
}

// ─── v7.17.0 Phase 3.P0-40 — 2D array parse + display ─────────

/// Split a PG external 2D-array literal `'{{a,b},{c,d}}'` into
/// per-row token lists. Returns Err on shape mismatch.
pub(crate) fn split_2d_literal(s: &str) -> Result<Vec<Vec<alloc::string::String>>, &'static str> {
    let s = s.trim();
    let outer = s
        .strip_prefix('{')
        .and_then(|x| x.strip_suffix('}'))
        .ok_or("missing outer '{...}' braces")?;
    let trimmed = outer.trim();
    if trimmed.is_empty() {
        return Ok(Vec::new());
    }
    let mut rows: Vec<Vec<alloc::string::String>> = Vec::new();
    let mut i = 0;
    let bytes = trimmed.as_bytes();
    while i < bytes.len() {
        while i < bytes.len() && matches!(bytes[i], b' ' | b'\t' | b'\n' | b'\r' | b',') {
            i += 1;
        }
        if i >= bytes.len() {
            break;
        }
        if bytes[i] != b'{' {
            return Err("expected '{' opening a row");
        }
        i += 1;
        let row_start = i;
        let mut depth = 1;
        while i < bytes.len() && depth > 0 {
            match bytes[i] {
                b'{' => depth += 1,
                b'}' => depth -= 1,
                _ => {}
            }
            if depth > 0 {
                i += 1;
            }
        }
        if depth != 0 {
            return Err("unbalanced '{...}' in row");
        }
        let row_text = &trimmed[row_start..i];
        i += 1;
        let cells: Vec<alloc::string::String> = if row_text.trim().is_empty() {
            Vec::new()
        } else {
            row_text.split(',').map(|t| t.trim().to_string()).collect()
        };
        rows.push(cells);
    }
    if let Some(first) = rows.first() {
        let cols = first.len();
        for r in &rows {
            if r.len() != cols {
                return Err("ragged 2D array (rows have different column counts)");
            }
        }
    }
    Ok(rows)
}

pub(crate) fn parse_int_2d_literal(s: &str) -> Result<Vec<Vec<Option<i32>>>, &'static str> {
    let raw = split_2d_literal(s)?;
    raw.into_iter()
        .map(|row| {
            row.into_iter()
                .map(|cell| {
                    if cell.eq_ignore_ascii_case("NULL") {
                        Ok(None)
                    } else {
                        cell.parse::<i32>()
                            .map(Some)
                            .map_err(|_| "invalid int element")
                    }
                })
                .collect()
        })
        .collect()
}

pub(crate) fn parse_bigint_2d_literal(s: &str) -> Result<Vec<Vec<Option<i64>>>, &'static str> {
    let raw = split_2d_literal(s)?;
    raw.into_iter()
        .map(|row| {
            row.into_iter()
                .map(|cell| {
                    if cell.eq_ignore_ascii_case("NULL") {
                        Ok(None)
                    } else {
                        cell.parse::<i64>()
                            .map(Some)
                            .map_err(|_| "invalid bigint element")
                    }
                })
                .collect()
        })
        .collect()
}

pub(crate) fn parse_text_2d_literal(
    s: &str,
) -> Result<Vec<Vec<Option<alloc::string::String>>>, &'static str> {
    let raw = split_2d_literal(s)?;
    Ok(raw
        .into_iter()
        .map(|row| {
            row.into_iter()
                .map(|cell| {
                    if cell.eq_ignore_ascii_case("NULL") {
                        None
                    } else {
                        Some(cell.trim_matches('"').to_string())
                    }
                })
                .collect()
        })
        .collect())
}

pub(crate) fn format_int_2d_text(rows: &[Vec<Option<i32>>]) -> alloc::string::String {
    let mut out = alloc::string::String::from("{");
    for (i, row) in rows.iter().enumerate() {
        if i > 0 {
            out.push(',');
        }
        out.push('{');
        for (j, cell) in row.iter().enumerate() {
            if j > 0 {
                out.push(',');
            }
            match cell {
                None => out.push_str("NULL"),
                Some(n) => out.push_str(&alloc::format!("{n}")),
            }
        }
        out.push('}');
    }
    out.push('}');
    out
}

pub(crate) fn format_bigint_2d_text(rows: &[Vec<Option<i64>>]) -> alloc::string::String {
    let mut out = alloc::string::String::from("{");
    for (i, row) in rows.iter().enumerate() {
        if i > 0 {
            out.push(',');
        }
        out.push('{');
        for (j, cell) in row.iter().enumerate() {
            if j > 0 {
                out.push(',');
            }
            match cell {
                None => out.push_str("NULL"),
                Some(n) => out.push_str(&alloc::format!("{n}")),
            }
        }
        out.push('}');
    }
    out.push('}');
    out
}

pub(crate) fn format_text_2d_text(
    rows: &[Vec<Option<alloc::string::String>>],
) -> alloc::string::String {
    let mut out = alloc::string::String::from("{");
    for (i, row) in rows.iter().enumerate() {
        if i > 0 {
            out.push(',');
        }
        out.push('{');
        for (j, cell) in row.iter().enumerate() {
            if j > 0 {
                out.push(',');
            }
            match cell {
                None => out.push_str("NULL"),
                Some(s) => out.push_str(s),
            }
        }
        out.push('}');
    }
    out.push('}');
    out
}

/// v7.17.0 Phase 3.P0-40 — pub re-exports so pgwire + sqllogictest
/// share the single 2D-array renderer.
pub fn format_int_2d_text_pub(rows: &[Vec<Option<i32>>]) -> alloc::string::String {
    format_int_2d_text(rows)
}
pub fn format_bigint_2d_text_pub(rows: &[Vec<Option<i64>>]) -> alloc::string::String {
    format_bigint_2d_text(rows)
}
pub fn format_text_2d_text_pub(
    rows: &[Vec<Option<alloc::string::String>>],
) -> alloc::string::String {
    format_text_2d_text(rows)
}

/// v7.39 (read01 round 75) — `bool[][]` external form. A BOOL element prints as
/// `t` / `f` INSIDE an array (and `true` / `false` outside it) — the whole reason
/// this type exists.
#[must_use]
pub fn format_bool_2d_text_pub(rows: &[Vec<Option<bool>>]) -> alloc::string::String {
    use core::fmt::Write as _;
    let mut out = alloc::string::String::from("{");
    for (i, row) in rows.iter().enumerate() {
        if i > 0 {
            out.push(',');
        }
        out.push('{');
        for (j, cell) in row.iter().enumerate() {
            if j > 0 {
                out.push(',');
            }
            let _ = match cell {
                None => write!(out, "NULL"),
                Some(true) => write!(out, "t"),
                Some(false) => write!(out, "f"),
            };
        }
        out.push('}');
    }
    out.push('}');
    out
}

/// v7.17.0 Phase 3.P0-38 — parse a PG range literal of the form
/// `'[lo,up)'` / `'(lo,up]'` / `'[lo,up]'` / `'(lo,up)'` /
/// `'empty'`. Lower / upper may be empty (unbounded). Returns
/// `None` on any parse failure; caller surfaces as hard error.
/// v7.38 (read01 U26) — PG range canonicalization, shared by the
/// `int4range(...)` constructors and the `'...'::int4range` text-input
/// path so both agree. PG forces an infinite (missing) bound to be
/// exclusive, then for DISCRETE element kinds (int4/int8/date) rewrites
/// to the `[)` form: an exclusive lower bumps to inclusive lower+1, an
/// inclusive upper bumps to exclusive upper+1 — so `[1,3]` becomes
/// `[1,4)`. Continuous kinds (num/ts/tstz) keep their bounds. Returns
/// the canonical `(lower, upper, lower_inc, upper_inc, empty)`, or
/// v7.38 — the canonical `[)` form of a range's bounds:
/// `(lower, upper, lower_inc, upper_inc, empty)`.
pub(crate) type CanonRangeBounds = (
    Option<Value<'static>>,
    Option<Value<'static>>,
    bool,
    bool,
    bool,
);

/// `None` if a discrete successor overflows the element type.
pub(crate) fn canonicalize_range_bounds(
    kind: spg_storage::RangeKind,
    lower: Option<Value<'static>>,
    upper: Option<Value<'static>>,
    lower_inc: bool,
    upper_inc: bool,
) -> Option<CanonRangeBounds> {
    use spg_storage::RangeKind as K;
    // An infinite bound is always exclusive.
    let mut lower_inc = lower.is_some() && lower_inc;
    let mut upper_inc = upper.is_some() && upper_inc;
    let mut lower = lower;
    let mut upper = upper;
    if matches!(kind, K::Int4 | K::Int8 | K::Date) {
        fn succ(v: Value<'static>) -> Option<Value<'static>> {
            Some(match v {
                Value::Int(n) => Value::Int(n.checked_add(1)?),
                Value::BigInt(n) => Value::BigInt(n.checked_add(1)?),
                Value::Date(d) => Value::Date(d.checked_add(1)?),
                other => other,
            })
        }
        if let Some(l) = lower {
            lower = Some(if lower_inc { l } else { succ(l)? });
            lower_inc = true;
        }
        if let Some(u) = upper {
            upper = Some(if upper_inc { succ(u)? } else { u });
            upper_inc = false;
        }
    }
    // Equal bounds that don't include both ends collapse to 'empty'.
    let empty = match (&lower, &upper) {
        (Some(l), Some(u)) => l == u && !(lower_inc && upper_inc),
        _ => false,
    };
    Some((lower, upper, lower_inc, upper_inc, empty))
}

/// v7.39 (read01 rangetypes.c) — the two failure classes of range text
/// input, mapping to PG's distinct errors (22P02 malformed vs 22000
/// misordered bounds).
pub(crate) enum RangeParseError {
    Malformed,
    Misordered,
    /// v7.39 (round 256) — the bracket/comma STRUCTURE parsed, but a
    /// bound is not a value of the element type. PG reports the
    /// element's own input error here (`invalid input syntax for type
    /// integer: "a"`), reserving "malformed range literal" for a
    /// structural problem — probed live on both shapes.
    BadElement(alloc::string::String),
}

/// v7.39 (round 256) — the PG name of a range type's ELEMENT type, used
/// when a bound fails to parse (`invalid input syntax for type integer`).
fn range_element_type_name(kind: spg_storage::RangeKind) -> &'static str {
    match kind {
        spg_storage::RangeKind::Int4 => "integer",
        spg_storage::RangeKind::Int8 => "bigint",
        spg_storage::RangeKind::Num => "numeric",
        spg_storage::RangeKind::Ts => "timestamp",
        spg_storage::RangeKind::TsTz => "timestamp with time zone",
        spg_storage::RangeKind::Date => "date",
    }
}

/// True when both bounds are present and lower sorts after upper —
/// PG rejects the range before canonicalization.
pub(crate) fn range_bounds_misordered(
    lower: &Option<Value<'static>>,
    upper: &Option<Value<'static>>,
) -> bool {
    match (lower, upper) {
        (Some(l), Some(u)) => crate::orderby::value_cmp(l, u) == core::cmp::Ordering::Greater,
        _ => false,
    }
}

pub(crate) fn parse_range_str(
    s: &str,
    kind: spg_storage::RangeKind,
) -> Result<Value<'static>, RangeParseError> {
    let s = s.trim();
    if s.eq_ignore_ascii_case("empty") {
        return Ok(Value::Range {
            kind,
            lower: None,
            upper: None,
            lower_inc: false,
            upper_inc: false,
            empty: true,
        });
    }
    let bytes = s.as_bytes();
    if bytes.len() < 3 {
        return Err(RangeParseError::Malformed);
    }
    let lower_inc = match bytes[0] {
        b'[' => true,
        b'(' => false,
        _ => return Err(RangeParseError::Malformed),
    };
    let upper_inc = match bytes[bytes.len() - 1] {
        b']' => true,
        b')' => false,
        _ => return Err(RangeParseError::Malformed),
    };
    let inner = &s[1..s.len() - 1];
    let (lo_text, up_text) = inner.split_once(',').ok_or(RangeParseError::Malformed)?;
    let lower = if lo_text.is_empty() {
        None
    } else {
        Some(
            parse_range_element(lo_text, kind)
                .ok_or_else(|| RangeParseError::BadElement(lo_text.trim().into()))?,
        )
    };
    let upper = if up_text.is_empty() {
        None
    } else {
        Some(
            parse_range_element(up_text, kind)
                .ok_or_else(|| RangeParseError::BadElement(up_text.trim().into()))?,
        )
    };
    // v7.39 (read01 rangetypes.c) — PG rejects misordered bounds before
    // canonicalization ('[3,1]'::int4range).
    if range_bounds_misordered(&lower, &upper) {
        return Err(RangeParseError::Misordered);
    }
    // Canonicalize (discrete `[)` fold + infinite→exclusive) so text
    // input agrees with the constructor functions.
    let (lower, upper, lower_inc, upper_inc, empty) =
        canonicalize_range_bounds(kind, lower, upper, lower_inc, upper_inc)
            .ok_or(RangeParseError::Malformed)?;
    Ok(Value::Range {
        kind,
        lower: lower.map(alloc::boxed::Box::new),
        upper: upper.map(alloc::boxed::Box::new),
        lower_inc,
        upper_inc,
        empty,
    })
}

/// v7.37.5 δ — parse a PG multirange external form into a Vec of
/// `RangeSpan`. Grammar: `{}` empty, `{range1,range2,...}` with
/// each range in canonical `[/(/]/)` brackets. Empty subranges
/// (`empty`) are accepted but get dropped on round-trip per PG
/// semantics. The bounds parser reuses `parse_range_str` by
/// wrapping each subrange in the parent kind.
pub(crate) fn parse_multirange_str(
    s: &str,
    kind: spg_storage::RangeKind,
) -> Option<Vec<spg_storage::RangeSpan>> {
    let s = s.trim();
    let inner = s.strip_prefix('{').and_then(|x| x.strip_suffix('}'))?;
    let inner = inner.trim();
    if inner.is_empty() {
        return Some(Vec::new());
    }
    // Split the inner on commas that sit *between* ranges — not the
    // commas inside `[a,b)`. Walk depth: bump on `[` / `(`, drop on
    // `]` / `)`. Commas at depth 0 are range separators.
    let mut spans: Vec<spg_storage::RangeSpan> = Vec::new();
    let bytes = inner.as_bytes();
    let mut depth: i32 = 0;
    let mut start = 0usize;
    for i in 0..=bytes.len() {
        let cut = i == bytes.len() || (depth == 0 && bytes[i] == b',');
        if !cut {
            match bytes.get(i) {
                Some(b'[') | Some(b'(') => depth += 1,
                Some(b']') | Some(b')') => depth -= 1,
                _ => {}
            }
            continue;
        }
        let piece = inner[start..i].trim();
        if piece.is_empty() {
            return None;
        }
        let r = parse_range_str(piece, kind).ok()?;
        let Value::Range {
            lower,
            upper,
            lower_inc,
            upper_inc,
            empty,
            ..
        } = r
        else {
            return None;
        };
        spans.push(spg_storage::RangeSpan {
            lower,
            upper,
            lower_inc,
            upper_inc,
            empty,
        });
        start = i + 1;
    }
    Some(spans)
}

/// v7.17.0 Phase 3.P0-38 — parse a single range bound text into
/// the matching element Value for the RangeKind.
/// "+HH[:MM]" tail (without the sign, caller split on '+') → seconds east.
fn parse_hhmm_offset_secs(off: &str) -> Option<i32> {
    let (h, m) = match off.split_once(':') {
        Some((h, m)) => (h, m),
        None => (off, "0"),
    };
    let h: i32 = h.parse().ok()?;
    let m: i32 = m.parse().ok()?;
    if !(0..=15).contains(&h) || !(0..60).contains(&m) {
        return None;
    }
    Some(h * 3600 + m * 60)
}

/// v7.39 (read01 regproc.c) — builtin type name (or alias) → OID, the
/// resolve half of regtype input. Mirrors the scalar map format_type
/// renders; extend both together.
pub(crate) fn regtype_name_to_oid(name: &str) -> Option<i64> {
    // v7.39 (round 621) — `integer[]` resolves to its array OID. Without this
    // `'integer[]'::regtype` was refused as `invalid input syntax for type
    // oid`, the mirror of the OID-to-name gap above.
    if let Some(base) = name.trim().strip_suffix("[]") {
        return array_oid_for_element(regtype_name_to_oid(base)?);
    }
    Some(match name.trim() {
        "bool" | "boolean" => 16,
        "bytea" => 17,
        "name" => 19,
        "int8" | "bigint" => 20,
        "int2" | "smallint" => 21,
        "int4" | "int" | "integer" => 23,
        "text" => 25,
        "oid" => 26,
        "json" => 114,
        "xml" => 142,
        "float4" | "real" => 700,
        "float8" | "double precision" => 701,
        "cidr" => 650,
        "inet" => 869,
        "macaddr" => 829,
        "macaddr8" => 774,
        "money" => 790,
        "bpchar" | "char" | "character" => 1042,
        "varchar" | "character varying" => 1043,
        "date" => 1082,
        "time" | "time without time zone" => 1083,
        "timestamp" | "timestamp without time zone" => 1114,
        "timestamptz" | "timestamp with time zone" => 1184,
        "interval" => 1186,
        "timetz" | "time with time zone" => 1266,
        "numeric" | "decimal" => 1700,
        "uuid" => 2950,
        "jsonb" => 3802,
        "tsvector" => 3614,
        "tsquery" => 3615,
        "pg_lsn" => 3220,
        "regtype" => 2206,
        "regclass" => 2205,
        "regproc" => 24,
        // v7.39 (round 640) — `'xid'::regtype` answered `type "xid" does
        // not exist` while `NULL::xid` resolved, because the two go
        // through different tables. Same three row-header types
        // `pg_attribute` names.
        "xid" => 28,
        "xid8" => 5069,
        "tid" => 27,
        "cid" => 29,
        _ => return None,
    })
}

/// Type name (or alias) → PG's canonical spelling ('int4' → 'integer'),
/// via the two builtin OID maps; `None` when unknown. Handles a `[]`
/// array suffix.
pub(crate) fn regtype_canonical_name(name: &str) -> Option<alloc::string::String> {
    let t = name.trim();
    if let Some(base) = t.strip_suffix("[]") {
        let inner = regtype_canonical_name(base)?;
        return Some(alloc::format!("{inner}[]"));
    }
    // PG's internal array-type spelling ('_int4' = int4[]).
    if let Some(base) = t.strip_prefix('_') {
        let inner = regtype_canonical_name(base)?;
        return Some(alloc::format!("{inner}[]"));
    }
    let oid = regtype_name_to_oid(&t.to_lowercase())?;
    regtype_oid_to_name(oid).map(alloc::string::String::from)
}

pub(crate) fn parse_range_element(
    text: &str,
    kind: spg_storage::RangeKind,
) -> Option<Value<'static>> {
    let text = text.trim().trim_matches('"');
    use spg_storage::RangeKind as K;
    match kind {
        K::Int4 => text.parse::<i32>().ok().map(Value::Int),
        K::Int8 => text.parse::<i64>().ok().map(Value::BigInt),
        K::Num => {
            // Reuse the Numeric parse via the engine's text-coercion
            // path; bail to None on failure.
            let dot = text.find('.');
            let scale: u16 = dot.map_or(0, |p| (text.len() - p - 1) as u16);
            let digits: alloc::string::String = text
                .chars()
                .filter(|c| *c == '-' || c.is_ascii_digit())
                .collect();
            let scaled: i128 = digits.parse().ok()?;
            Some(Value::Numeric {
                scaled,
                scale,
                kind: spg_storage::NumericKind::Finite,
            })
        }
        K::Ts | K::TsTz => {
            // v7.39 (read01 rangetypes.c) — the timestamp parser handles
            // datetime[+offset]; a bare date with an offset suffix
            // ('2024-01-02+00', legal tstz input) parses as its midnight.
            crate::eval::parse_timestamp_literal(text)
                .or_else(|| {
                    let (date_part, off) = text.split_once(['+'])?;
                    if !off.chars().all(|c| c.is_ascii_digit() || c == ':') {
                        return None;
                    }
                    let d = crate::eval::parse_date_literal(date_part.trim())?;
                    let mut t = i64::from(d) * 86_400_000_000;
                    // Apply the offset (east-positive) back to UTC.
                    let secs = parse_hhmm_offset_secs(off)?;
                    t -= i64::from(secs) * 1_000_000;
                    Some(t)
                })
                .map(Value::Timestamp)
        }
        K::Date => crate::eval::parse_date_literal(text).map(Value::Date),
    }
}

/// v7.17.0 Phase 3.P0-38 — render a Range value as its canonical
/// PG text form. Re-exported via [`format_range_text`] for use
/// from spg-server's pgwire layer.
pub fn format_range_text(v: &Value) -> alloc::string::String {
    format_range_str(v)
}

pub(crate) fn format_range_str(v: &Value) -> alloc::string::String {
    let Value::Range {
        kind,
        lower,
        upper,
        lower_inc,
        upper_inc,
        empty,
    } = v
    else {
        return alloc::string::String::new();
    };
    if *empty {
        return "empty".into();
    }
    // v7.39 (read01 rangetypes.c) — tstzrange bounds render with the
    // session-UTC offset suffix, as PG's timestamptz_out does. (Named
    // session zones inside range elements are a recorded residual with
    // the per-value wire SessionTz channel.)
    let elem = |v: &Value| -> alloc::string::String {
        let base = format_range_element(v);
        if matches!(kind, spg_storage::RangeKind::TsTz) && matches!(v, Value::Timestamp(_)) {
            alloc::format!("{base}+00")
        } else {
            base
        }
    };
    let mut out = alloc::string::String::new();
    out.push(if *lower_inc { '[' } else { '(' });
    if let Some(l) = lower {
        out.push_str(&quote_range_bound(&elem(l)));
    }
    out.push(',');
    if let Some(u) = upper {
        out.push_str(&quote_range_bound(&elem(u)));
    }
    out.push(if *upper_inc { ']' } else { ')' });
    out
}

/// PG's `range_out` double-quotes a bound whose text is empty or
/// contains a range-syntax metacharacter (`"` `\` `(` `)` `[` `]` `,`)
/// or whitespace — so a timestamp bound `2020-01-01 10:00:00` prints
/// as `"2020-01-01 10:00:00"` inside the range. `"` and `\` are
/// backslash-escaped within the quotes. Numeric / date bounds (no
/// spaces) pass through unquoted, matching PG.
fn quote_range_bound(s: &str) -> alloc::string::String {
    let needs_quote = s.is_empty()
        || s.chars()
            .any(|c| matches!(c, '"' | '\\' | '(' | ')' | '[' | ']' | ',') || c.is_whitespace());
    if !needs_quote {
        return s.into();
    }
    let mut out = alloc::string::String::with_capacity(s.len() + 2);
    out.push('"');
    for c in s.chars() {
        if c == '"' || c == '\\' {
            out.push('\\');
        }
        out.push(c);
    }
    out.push('"');
    out
}

/// v7.37.5 ε — render a Point as PG canonical `(x,y)`.
pub fn format_point(p: spg_storage::Point2D) -> alloc::string::String {
    alloc::format!("({},{})", p.x, p.y)
}

/// v7.37.5 ε — render an Lseg as PG canonical `[(x1,y1),(x2,y2)]`.
pub fn format_lseg(p1: spg_storage::Point2D, p2: spg_storage::Point2D) -> alloc::string::String {
    alloc::format!("[({},{}),({},{})]", p1.x, p1.y, p2.x, p2.y)
}

/// v7.37.5 ε — render a Box as PG canonical `(ux,uy),(lx,ly)`.
/// PG normalises the corner order on input; we trust the engine's
/// constructor has already normalised so the field order here is
/// the canonical upper-right + lower-left.
pub fn format_pg_box(ur: spg_storage::Point2D, ll: spg_storage::Point2D) -> alloc::string::String {
    alloc::format!("({},{}),({},{})", ur.x, ur.y, ll.x, ll.y)
}

/// v7.37.5 ε — render a Line as PG canonical `{a,b,c}` (Ax+By+C=0).
pub fn format_line(a: f64, b: f64, c: f64) -> alloc::string::String {
    alloc::format!("{{{},{},{}}}", a, b, c)
}

/// v7.37.5 ε — render a Circle as PG canonical `<(x,y),r>`.
pub fn format_circle(center: spg_storage::Point2D, radius: f64) -> alloc::string::String {
    alloc::format!("<({},{}),{}>", center.x, center.y, radius)
}

/// v7.37.5 ε — render a Path as PG canonical `[(x,y),...]` open
/// or `((x,y),...)` closed.
pub fn format_path(points: &[spg_storage::Point2D], closed: bool) -> alloc::string::String {
    let (open, close) = if closed { ('(', ')') } else { ('[', ']') };
    let mut out = alloc::string::String::new();
    out.push(open);
    for (i, p) in points.iter().enumerate() {
        if i > 0 {
            out.push(',');
        }
        out.push_str(&alloc::format!("({},{})", p.x, p.y));
    }
    out.push(close);
    out
}

/// v7.37.5 ε — render a Polygon as PG canonical `((x,y),...)`.
pub fn format_polygon(points: &[spg_storage::Point2D]) -> alloc::string::String {
    let mut out = alloc::string::String::new();
    out.push('(');
    for (i, p) in points.iter().enumerate() {
        if i > 0 {
            out.push(',');
        }
        out.push_str(&alloc::format!("({},{})", p.x, p.y));
    }
    out.push(')');
    out
}

/// v7.37.5 ε — parse a single `(x,y)` or bare `x,y` Point text.
/// Surrounding whitespace OK. Returns `None` on malformed input.
fn parse_point(s: &str) -> Option<spg_storage::Point2D> {
    let s = s.trim();
    let inner = s
        .strip_prefix('(')
        .and_then(|x| x.strip_suffix(')'))
        .unwrap_or(s);
    let (xs, ys) = inner.split_once(',')?;
    let x: f64 = xs.trim().parse().ok()?;
    let y: f64 = ys.trim().parse().ok()?;
    Some(spg_storage::Point2D { x, y })
}

/// v7.37.5 ε — parse N points from a comma-separated PG point
/// list (`(x1,y1),(x2,y2),...`). Depth-aware split so the commas
/// inside each `(...)` aren't taken as separators. Returns `None`
/// on malformed input.
fn parse_point_list(s: &str) -> Option<Vec<spg_storage::Point2D>> {
    let bytes = s.as_bytes();
    let mut out: Vec<spg_storage::Point2D> = Vec::new();
    let mut depth: i32 = 0;
    let mut start = 0usize;
    for i in 0..=bytes.len() {
        let cut = i == bytes.len() || (depth == 0 && bytes[i] == b',');
        if !cut {
            match bytes.get(i) {
                Some(b'(') | Some(b'[') | Some(b'<') => depth += 1,
                Some(b')') | Some(b']') | Some(b'>') => depth -= 1,
                _ => {}
            }
            continue;
        }
        let piece = s[start..i].trim();
        if !piece.is_empty() {
            out.push(parse_point(piece)?);
        }
        start = i + 1;
    }
    Some(out)
}

/// v7.37.5 ε — parse Lseg text `[(x1,y1),(x2,y2)]`.
pub fn parse_lseg_text(s: &str) -> Option<(spg_storage::Point2D, spg_storage::Point2D)> {
    let s = s.trim();
    // PG accepts the bracketed `[(x1,y1),(x2,y2)]`, the fully-wrapped
    // `((x1,y1),(x2,y2))`, and the bare `(x1,y1),(x2,y2)` spellings.
    let inner = s
        .strip_prefix('[')
        .and_then(|x| x.strip_suffix(']'))
        .unwrap_or(s);
    let two_points = |v: Option<alloc::vec::Vec<spg_storage::Point2D>>| v.filter(|p| p.len() == 2);
    let pts = if let Some(p) = two_points(parse_point_list(inner)) {
        p
    } else {
        inner
            .strip_prefix('(')
            .and_then(|x| x.strip_suffix(')'))
            .and_then(|w| two_points(parse_point_list(w)))?
    };
    Some((pts[0], pts[1]))
}

/// v7.37.5 ε — parse Box text `(ux,uy),(lx,ly)`. PG normalises
/// any two-corner input into upper-right + lower-left; we do
/// the same.
pub fn parse_box_text(s: &str) -> Option<(spg_storage::Point2D, spg_storage::Point2D)> {
    // PG box input: `(x1,y1),(x2,y2)`, the fully-wrapped `((x1,y1),(x2,y2))`,
    // or the bare `x1,y1,x2,y2` (four raw numbers). Try the point-list form,
    // then the same list inside one stripped `(...)` layer, then four floats.
    let s = s.trim();
    let two_points = |v: Option<alloc::vec::Vec<spg_storage::Point2D>>| v.filter(|p| p.len() == 2);
    let pts = if let Some(p) = two_points(parse_point_list(s)) {
        p
    } else if let Some(p) = s
        .strip_prefix('(')
        .and_then(|x| x.strip_suffix(')'))
        .and_then(|inner| two_points(parse_point_list(inner)))
    {
        p
    } else {
        let nums: Option<alloc::vec::Vec<f64>> =
            s.split(',').map(|t| t.trim().parse::<f64>().ok()).collect();
        let nums = nums?;
        if nums.len() != 4 {
            return None;
        }
        alloc::vec![
            spg_storage::Point2D {
                x: nums[0],
                y: nums[1]
            },
            spg_storage::Point2D {
                x: nums[2],
                y: nums[3]
            },
        ]
    };
    if pts.len() != 2 {
        return None;
    }
    let (a, b) = (pts[0], pts[1]);
    // Normalise: upper-right has the larger x AND larger y.
    let ur = spg_storage::Point2D {
        x: a.x.max(b.x),
        y: a.y.max(b.y),
    };
    let ll = spg_storage::Point2D {
        x: a.x.min(b.x),
        y: a.y.min(b.y),
    };
    Some((ur, ll))
}

/// v7.37.5 ε — parse Line text `{a,b,c}`.
pub fn parse_line_text(s: &str) -> Option<(f64, f64, f64)> {
    let s = s.trim();
    if let Some(inner) = s.strip_prefix('{').and_then(|x| x.strip_suffix('}')) {
        let parts: Vec<&str> = inner.split(',').collect();
        if parts.len() != 3 {
            return None;
        }
        let a: f64 = parts[0].trim().parse().ok()?;
        let b: f64 = parts[1].trim().parse().ok()?;
        // PG rejects A = B = 0 (not a line).
        if a == 0.0 && b == 0.0 {
            return None;
        }
        let c: f64 = parts[2].trim().parse().ok()?;
        return Some((a, b, c));
    }
    // v7.39 (read01 geo_ops.c) — the two-point form `((x1,y1),(x2,y2))`
    // (or the lseg spellings): PG builds Ax+By+C=0 from the slope —
    // vertical is "x = C" (-1, 0, x), horizontal "y = C" (0, -1, y),
    // else (m, -1, y - m·x). Coincident points are not a line.
    let (p1, p2) = parse_lseg_text(s)?;
    if p1.x == p2.x && p1.y == p2.y {
        return None;
    }
    Some(line_from_points(p1, p2))
}

/// PG's line_construct from two points (geo_ops.c behavior).
pub fn line_from_points(p1: spg_storage::Point2D, p2: spg_storage::Point2D) -> (f64, f64, f64) {
    if p1.x == p2.x {
        (-1.0, 0.0, p1.x)
    } else if p1.y == p2.y {
        (0.0, -1.0, p1.y)
    } else {
        let m = (p1.y - p2.y) / (p1.x - p2.x);
        let c = p1.y - m * p1.x;
        (m, -1.0, if c == 0.0 { 0.0 } else { c })
    }
}

/// v7.37.5 ε — parse Circle text `<(x,y),r>` or `((x,y),r)`.
pub fn parse_circle_text(s: &str) -> Option<(spg_storage::Point2D, f64)> {
    let s = s.trim();
    // PG circle input: `<(x,y),r>`, `((x,y),r)`, `(x,y),r`, or bare `x,y,r`.
    let inner = if let Some(i) = s.strip_prefix('<').and_then(|x| x.strip_suffix('>')) {
        i
    } else if let Some(i) = s.strip_prefix('(').and_then(|x| x.strip_suffix(')')) {
        i
    } else {
        s
    };
    // The last comma at depth 0 splits the center from the radius.
    let bytes = inner.as_bytes();
    let mut depth = 0i32;
    let mut split_at: Option<usize> = None;
    for (i, &b) in bytes.iter().enumerate() {
        match b {
            b'(' | b'[' | b'<' => depth += 1,
            b')' | b']' | b'>' => depth -= 1,
            b',' if depth == 0 => split_at = Some(i),
            _ => {}
        }
    }
    let i = split_at?;
    let center = parse_point(&inner[..i])?;
    let radius: f64 = inner[i + 1..].trim().parse().ok()?;
    Some((center, radius))
}

/// v7.37.5 ε — parse Path text `[(x,y),...]` (open) or
/// `((x,y),...)` (closed). The leading bracket pins openness.
pub fn parse_path_text(s: &str) -> Option<(Vec<spg_storage::Point2D>, bool)> {
    let s = s.trim();
    // `[...]` = open path, `(...)` = closed. A bare point list (no brackets)
    // is a closed path in PG. Strip a wrapping layer only when it yields a
    // valid point list; otherwise parse the bare list directly as closed
    // (stripping unconditionally would mangle `(0,0),(1,1)` into `0,0),(1,1`).
    if let Some(i) = s.strip_prefix('[').and_then(|x| x.strip_suffix(']')) {
        if let Some(pts) = parse_point_list(i) {
            return Some((pts, false));
        }
    }
    if let Some(i) = s.strip_prefix('(').and_then(|x| x.strip_suffix(')')) {
        if let Some(pts) = parse_point_list(i) {
            return Some((pts, true));
        }
    }
    parse_point_list(s).map(|pts| (pts, true))
}

/// v7.37.5 ε — parse Polygon text `((x,y),...)` (implicit closed).
pub fn parse_polygon_text(s: &str) -> Option<Vec<spg_storage::Point2D>> {
    let s = s.trim();
    // The outer parens are optional in PG — `((0,0),(1,1))` and `(0,0),(1,1)`
    // both parse. Try stripping one wrapping layer first (the `((...))` form);
    // if that doesn't yield a valid point list, parse the bare list directly.
    if let Some(inner) = s.strip_prefix('(').and_then(|x| x.strip_suffix(')')) {
        if let Some(pts) = parse_point_list(inner) {
            return Some(pts);
        }
    }
    parse_point_list(s)
}

/// v7.37.5 ζ-A — render an INET/CIDR address as canonical PG text:
/// IPv4: `a.b.c.d/bits`; IPv6: `xxxx:xxxx:.../bits`. The mask is
/// elided when it equals the family default (32 for IPv4, 128 for
/// IPv6), per PG convention.
/// v7.38 (read01) — inet text with the mask ALWAYS shown (`192.168.1.0/32`),
/// as PG's `inet::text` / `::varchar` cast renders it (the default display and
/// concat omit `/32` and `/128`; this is the cast-path form).
pub fn format_inet_full(family: u8, bits: u8, addr: &[u8; 16]) -> alloc::string::String {
    let max = if family == 4 { 32 } else { 128 };
    let base = format_inet(family, max, addr);
    alloc::format!("{base}/{bits}")
}

pub fn format_inet(family: u8, bits: u8, addr: &[u8; 16]) -> alloc::string::String {
    match family {
        4 => {
            let s = alloc::format!("{}.{}.{}.{}", addr[0], addr[1], addr[2], addr[3]);
            if bits == 32 {
                s
            } else {
                alloc::format!("{s}/{bits}")
            }
        }
        6 => {
            // v7.38 (read01) — RFC 5952 canonical form: compress the longest
            // run of consecutive all-zero groups (leftmost among ties) to `::`,
            // but only when that run is ≥ 2 groups. PG always renders this form.
            let mut groups = [0u16; 8];
            for (i, g) in groups.iter_mut().enumerate() {
                *g = (u16::from(addr[i * 2]) << 8) | u16::from(addr[i * 2 + 1]);
            }
            // v7.38 (read01, T19) — IPv4-mapped IPv6 (`::ffff:0:0/96` range:
            // first five groups zero, sixth 0xffff) renders with a dotted-quad
            // tail, matching PG (independent of the input spelling).
            if groups[..5].iter().all(|&g| g == 0) && groups[5] == 0xffff {
                let s =
                    alloc::format!("::ffff:{}.{}.{}.{}", addr[12], addr[13], addr[14], addr[15]);
                return if bits == 128 {
                    s
                } else {
                    alloc::format!("{s}/{bits}")
                };
            }
            let (mut best_start, mut best_len) = (usize::MAX, 0usize);
            let mut i = 0;
            while i < 8 {
                if groups[i] == 0 {
                    let start = i;
                    while i < 8 && groups[i] == 0 {
                        i += 1;
                    }
                    if i - start > best_len {
                        best_start = start;
                        best_len = i - start;
                    }
                } else {
                    i += 1;
                }
            }
            let mut out = alloc::string::String::new();
            if best_len >= 2 {
                for (idx, g) in groups.iter().enumerate().take(best_start) {
                    if idx > 0 {
                        out.push(':');
                    }
                    out.push_str(&alloc::format!("{g:x}"));
                }
                out.push_str("::");
                for (idx, g) in groups.iter().enumerate().skip(best_start + best_len) {
                    if idx > best_start + best_len {
                        out.push(':');
                    }
                    out.push_str(&alloc::format!("{g:x}"));
                }
            } else {
                for (idx, g) in groups.iter().enumerate() {
                    if idx > 0 {
                        out.push(':');
                    }
                    out.push_str(&alloc::format!("{g:x}"));
                }
            }
            if bits == 128 {
                out
            } else {
                alloc::format!("{out}/{bits}")
            }
        }
        _ => alloc::format!("?invalid-inet-family-{family}"),
    }
}

/// v7.37.5 ζ-A — render a MACADDR (6 bytes) as `aa:bb:cc:dd:ee:ff`.
pub fn format_macaddr(m: &[u8; 6]) -> alloc::string::String {
    alloc::format!(
        "{:02x}:{:02x}:{:02x}:{:02x}:{:02x}:{:02x}",
        m[0],
        m[1],
        m[2],
        m[3],
        m[4],
        m[5]
    )
}

/// v7.37.5 ζ-A — render a MACADDR8 (8 bytes) as `aa:bb:cc:dd:ee:ff:00:11`.
pub fn format_macaddr8(m: &[u8; 8]) -> alloc::string::String {
    alloc::format!(
        "{:02x}:{:02x}:{:02x}:{:02x}:{:02x}:{:02x}:{:02x}:{:02x}",
        m[0],
        m[1],
        m[2],
        m[3],
        m[4],
        m[5],
        m[6],
        m[7]
    )
}

/// v7.37.5 ζ-A — render a BIT / BIT VARYING as a binary string of
/// `'0'` and `'1'` chars (PG canonical text form). Bytes are packed
/// big-endian within each byte: the most-significant bit of byte 0
/// is bit 0 of the bit string.
pub fn format_bit_string(nbits: u32, bytes: &[u8]) -> alloc::string::String {
    let mut out = alloc::string::String::with_capacity(nbits as usize);
    for i in 0..nbits as usize {
        let byte = bytes[i / 8];
        let bit = (byte >> (7 - (i % 8))) & 1;
        out.push(if bit == 1 { '1' } else { '0' });
    }
    out
}

/// MSB-first integer value of a bit string (PG `bit`/`varbit` → integer cast).
pub fn bit_string_to_i64(nbits: u32, bytes: &[u8]) -> i64 {
    let mut val: i64 = 0;
    for i in 0..nbits as usize {
        let byte = bytes.get(i / 8).copied().unwrap_or(0);
        val = (val << 1) | i64::from((byte >> (7 - (i % 8))) & 1);
    }
    val
}

/// v7.37.5 ζ-A — render a MONEY[] in PG external form. Each element
/// is the canonical `format_money` output; the array wrapper is
/// `{...}` with NULL elements as the literal token `NULL`.
pub fn format_money_array(items: &[Option<i64>]) -> alloc::string::String {
    let mut out = alloc::string::String::new();
    out.push('{');
    for (i, item) in items.iter().enumerate() {
        if i > 0 {
            out.push(',');
        }
        match item {
            None => out.push_str("NULL"),
            Some(c) => out.push_str(&crate::eval::format_money(*c)),
        }
    }
    out.push('}');
    out
}

/// v7.37.5 ζ-A — parse PG INET text. Accepts `a.b.c.d[/bits]`
/// (IPv4) or `xxxx:xxxx:.../[bits]` (IPv6 colon-separated). The
/// mask defaults to 32 (IPv4) / 128 (IPv6) when omitted. Returns
/// `(family, bits, addr16)`. `None` on malformed input.
pub fn parse_inet_text(s: &str) -> Option<(u8, u8, [u8; 16])> {
    let s = s.trim();
    let (addr_s, bits_s) = match s.split_once('/') {
        Some((a, b)) => (a, Some(b)),
        None => (s, None),
    };
    if addr_s.contains(':') {
        // IPv6 — colon-separated up to 8 × u16 hex with optional
        // `::` zero-compression. v7.37.5 ship triage broadened the
        // pre-7.37.10 8-group-only form to accept canonical PG
        // IPv6 abbreviations like `2001:db8::/32`.
        let (head, tail) = match addr_s.find("::") {
            Some(idx) => (&addr_s[..idx], Some(&addr_s[idx + 2..])),
            None => (addr_s, None),
        };
        let mut head_groups: alloc::vec::Vec<&str> = if head.is_empty() {
            alloc::vec::Vec::new()
        } else {
            head.split(':').collect()
        };
        let mut tail_groups: alloc::vec::Vec<&str> = match tail {
            Some(t) if !t.is_empty() => t.split(':').collect(),
            _ => alloc::vec::Vec::new(),
        };
        // v7.38 (read01, T19) — a trailing dotted-quad (IPv4-in-IPv6, e.g.
        // `::ffff:192.168.1.1`, `64:ff9b::192.0.2.1`) fills the last two 16-bit
        // words. It is always the final group overall.
        let mut dotted_words: Option<[u16; 2]> = None;
        if let Some(g) = tail_groups.last().or_else(|| head_groups.last()) {
            if g.contains('.') {
                let oct: alloc::vec::Vec<&str> = g.split('.').collect();
                if oct.len() != 4 {
                    return None;
                }
                let mut b = [0u8; 4];
                for (i, o) in oct.iter().enumerate() {
                    b[i] = o.parse::<u8>().ok()?;
                }
                dotted_words = Some([
                    (u16::from(b[0]) << 8) | u16::from(b[1]),
                    (u16::from(b[2]) << 8) | u16::from(b[3]),
                ]);
                if !tail_groups.is_empty() {
                    tail_groups.pop();
                } else {
                    head_groups.pop();
                }
            }
        }
        let dq = if dotted_words.is_some() { 2 } else { 0 };
        let head_len = head_groups.len();
        let tail_len = tail_groups.len();
        if tail.is_none() {
            if head_len + dq != 8 {
                return None;
            }
        } else if head_len + tail_len + dq > 7 {
            return None;
        }
        let mut words = [0u16; 8];
        for (i, g) in head_groups.iter().enumerate() {
            words[i] = u16::from_str_radix(g, 16).ok()?;
        }
        // The dotted-quad (if any) occupies the final two words; hex tail groups
        // sit just before it.
        let trailing_start = 8 - dq - tail_len;
        for (i, g) in tail_groups.iter().enumerate() {
            words[trailing_start + i] = u16::from_str_radix(g, 16).ok()?;
        }
        if let Some(dw) = dotted_words {
            words[6] = dw[0];
            words[7] = dw[1];
        }
        let mut addr = [0u8; 16];
        for (i, w) in words.iter().enumerate() {
            addr[i * 2] = (w >> 8) as u8;
            addr[i * 2 + 1] = (w & 0xff) as u8;
        }
        let bits = match bits_s {
            Some(b) => b.parse::<u8>().ok().filter(|&n| n <= 128)?,
            None => 128,
        };
        Some((6, bits, addr))
    } else {
        // IPv4 — `a.b.c.d`.
        let parts: alloc::vec::Vec<&str> = addr_s.split('.').collect();
        if parts.len() != 4 {
            return None;
        }
        let mut addr = [0u8; 16];
        for (i, p) in parts.iter().enumerate() {
            addr[i] = p.parse::<u8>().ok()?;
        }
        let bits = match bits_s {
            Some(b) => b.parse::<u8>().ok().filter(|&n| n <= 32)?,
            None => 32,
        };
        Some((4, bits, addr))
    }
}

/// v7.39 (read01 inet_net_pton.c) — parse CIDR text. Beyond the inet
/// grammar, cidr accepts ABBREVIATED IPv4 network forms (`10/8`,
/// `10.5/16`, `128.1`) zero-filling the missing octets; a missing
/// /width defaults to 8×(octets given) for IPv4 and 128 for IPv6.
/// Returns Err(()) for "bits set to right of mask" (PG's dedicated
/// invalid-cidr-value error), Ok(None) for a plain syntax error.
pub fn parse_cidr_text(s: &str) -> Result<Option<(u8, u8, [u8; 16])>, ()> {
    let s = s.trim();
    let parsed = if !s.contains(':') {
        let (addr_s, bits_s) = match s.split_once('/') {
            Some((a, b)) => (a, Some(b)),
            None => (s, None),
        };
        let parts: alloc::vec::Vec<&str> = addr_s.split('.').collect();
        if parts.is_empty() || parts.len() > 4 || parts.iter().any(|p| p.is_empty()) {
            return Ok(None);
        }
        let mut addr = [0u8; 16];
        for (i, p) in parts.iter().enumerate() {
            match p.parse::<u8>() {
                Ok(v) => addr[i] = v,
                Err(_) => return Ok(None),
            }
        }
        let bits = match bits_s {
            Some(b) => match b.parse::<u8>() {
                Ok(n) if n <= 32 => n,
                _ => return Ok(None),
            },
            None => (parts.len() as u8) * 8,
        };
        Some((4u8, bits, addr))
    } else {
        parse_inet_text(s).map(|(f, b, a)| {
            // cidr IPv6 without a /width is the full /128.
            (f, if s.contains('/') { b } else { 128 }, a)
        })
    };
    let Some((family, bits, addr)) = parsed else {
        return Ok(None);
    };
    // PG cidr_in rejects host bits to the right of the mask.
    let total = if family == 4 { 32u16 } else { 128 };
    let nbytes = if family == 4 { 4 } else { 16 };
    for byte in 0..nbytes {
        let bit_base = (byte as u16) * 8;
        let keep = (u16::from(bits)).saturating_sub(bit_base).min(8) as u8;
        let mask: u8 = if keep == 0 { 0 } else { 0xffu8 << (8 - keep) };
        if addr[byte] & !mask != 0 {
            return Err(());
        }
        if bit_base >= total {
            break;
        }
    }
    Ok(Some((family, bits, addr)))
}

/// v7.37.5 ζ-A — parse PG MACADDR text `aa:bb:cc:dd:ee:ff` (also
/// accepts `aa-bb-cc-dd-ee-ff` and unseparated `aabbccddeeff`).
pub fn parse_macaddr_text(s: &str) -> Option<[u8; 6]> {
    let s = s.trim();
    let cleaned: alloc::string::String = s.chars().filter(|c| c.is_ascii_hexdigit()).collect();
    if cleaned.len() != 12 {
        return None;
    }
    let mut out = [0u8; 6];
    for i in 0..6 {
        out[i] = u8::from_str_radix(&cleaned[i * 2..i * 2 + 2], 16).ok()?;
    }
    Some(out)
}

/// v7.37.5 ζ-A — parse PG MACADDR8 text.
/// v7.39 (read01 pg_lsn.c) — parse PG's `%X/%X` LSN form: two hex halves,
/// each at most 8 hex digits (u32), joined `hi << 32 | lo`.
/// v7.39 (read01 timestamp.c, sentinel audit) — date days → timestamp
/// microseconds with the ±infinity sentinels mapped through (the plain
/// multiply overflowed i64 and aborted debug builds).
#[must_use]
pub fn date_days_to_micros(d: i32) -> i64 {
    match d {
        i32::MAX => i64::MAX,
        i32::MIN => i64::MIN,
        _ => i64::from(d) * 86_400_000_000,
    }
}

pub fn parse_pg_lsn_text(s: &str) -> Option<u64> {
    let t = s.trim();
    let (hi, lo) = t.split_once('/')?;
    if hi.is_empty() || lo.is_empty() || hi.len() > 8 || lo.len() > 8 {
        return None;
    }
    let hi = u32::from_str_radix(hi, 16).ok()?;
    let lo = u32::from_str_radix(lo, 16).ok()?;
    Some((u64::from(hi) << 32) | u64::from(lo))
}

/// Render an LSN in PG's `%X/%X` form (uppercase hex, no zero-padding).
#[must_use]
pub fn format_pg_lsn(l: u64) -> alloc::string::String {
    alloc::format!("{:X}/{:X}", l >> 32, l & 0xFFFF_FFFF)
}

pub fn parse_macaddr8_text(s: &str) -> Option<[u8; 8]> {
    let s = s.trim();
    let cleaned: alloc::string::String = s.chars().filter(|c| c.is_ascii_hexdigit()).collect();
    // v7.39 (read01 mac8.c) — a 6-byte (EUI-48) input converts by
    // inserting ff:fe as the 4th/5th octets, like PG's macaddr8_in.
    if cleaned.len() == 12 {
        let mut six = [0u8; 6];
        for i in 0..6 {
            six[i] = u8::from_str_radix(&cleaned[i * 2..i * 2 + 2], 16).ok()?;
        }
        return Some([six[0], six[1], six[2], 0xff, 0xfe, six[3], six[4], six[5]]);
    }
    if cleaned.len() != 16 {
        return None;
    }
    let mut out = [0u8; 8];
    for i in 0..8 {
        out[i] = u8::from_str_radix(&cleaned[i * 2..i * 2 + 2], 16).ok()?;
    }
    Some(out)
}

/// v7.37.5 ζ-A — parse PG bit string text (a sequence of `'0'` and
/// `'1'` chars). Returns `(nbits, packed_bytes)` — bytes are
/// big-endian within each byte (PG canonical).
pub fn parse_bit_string_text(s: &str) -> Option<(u32, alloc::vec::Vec<u8>)> {
    let s = s.trim();
    let nbits = u32::try_from(s.len()).ok()?;
    let nbytes = (s.len()).div_ceil(8);
    let mut bytes = alloc::vec![0u8; nbytes];
    for (i, c) in s.chars().enumerate() {
        let bit = match c {
            '0' => 0u8,
            '1' => 1u8,
            _ => return None,
        };
        if bit == 1 {
            bytes[i / 8] |= 1 << (7 - (i % 8));
        }
    }
    Some((nbits, bytes))
}

/// v7.37.5 δ — render a Multirange in PG external form
/// `{[a,b),[c,d)}`. Empty multirange renders as `{}`. Each range
/// element is formatted with the same `[/(/]/)` bracket grammar
/// as scalar `Value::Range`. RangeSpan carries no `kind` (it
/// lives on the parent Multirange), so this routes element
/// formatting through `format_range_element` as Value::Range does.
pub fn format_multirange(ranges: &[spg_storage::RangeSpan]) -> alloc::string::String {
    let mut out = alloc::string::String::new();
    out.push('{');
    for (i, r) in ranges.iter().enumerate() {
        if i > 0 {
            out.push(',');
        }
        if r.empty {
            out.push_str("empty");
            continue;
        }
        out.push(if r.lower_inc { '[' } else { '(' });
        if let Some(l) = &r.lower {
            out.push_str(&quote_range_bound(&format_range_element(l)));
        }
        out.push(',');
        if let Some(u) = &r.upper {
            out.push_str(&quote_range_bound(&format_range_element(u)));
        }
        out.push(if r.upper_inc { ']' } else { ')' });
    }
    out.push('}');
    out
}

pub(crate) fn format_range_element(v: &Value) -> alloc::string::String {
    match v {
        Value::Int(n) => alloc::format!("{n}"),
        Value::BigInt(n) => alloc::format!("{n}"),
        Value::Date(d) => crate::eval::format_date(*d),
        Value::Timestamp(t) => crate::eval::format_timestamp(*t),
        Value::Numeric {
            scaled,
            scale,
            kind,
        } => crate::eval::format_numeric_kind(*kind, *scaled, *scale),
        other => alloc::format!("{other:?}"),
    }
}

/// v7.17.0 Phase 3.P0-35 — parse a PG `money` literal into i64
/// cents. Accepts:
///   * Optional leading `-` (negative)
///   * Optional `$` prefix
///   * Integer portion with optional `,` thousands separators
///   * Optional `.` followed by 1-2 digits (cents); 1 digit
///     auto-pads to 2 (`.5` → 50 cents).
///
/// Returns None on any parse failure — caller surfaces as hard
/// SQL error.
pub(crate) fn parse_money_str(s: &str) -> Option<i64> {
    // v7.39 (read01 utils/adt, cash.c) — PG's cash_in accepts the sign
    // and currency symbol before OR after the digits, accounting
    // parentheses for negative, and rounds the first digit past the
    // cent (C-locale: fpoint 2, '$', ',').
    let mut rest = s.trim();
    let mut neg = false;
    // Leading currency symbol / sign / accounting paren, in any order
    // with whitespace.
    loop {
        let before = rest;
        rest = rest.trim_start();
        if let Some(r) = rest.strip_prefix('$') {
            rest = r;
        } else if let Some(r) = rest.strip_prefix('-') {
            neg = true;
            rest = r;
        } else if let Some(r) = rest.strip_prefix('(') {
            neg = true;
            rest = r;
        } else if let Some(r) = rest.strip_prefix('+') {
            rest = r;
        }
        if rest == before {
            break;
        }
    }
    let (int_part, tail) = {
        let end = rest
            .find(|c: char| !(c.is_ascii_digit() || c == ','))
            .unwrap_or(rest.len());
        (&rest[..end], &rest[end..])
    };
    // Validate + strip commas from the integer portion.
    let mut int_digits = alloc::string::String::with_capacity(int_part.len());
    for b in int_part.bytes() {
        match b {
            b',' => {}
            b'0'..=b'9' => int_digits.push(b as char),
            _ => return None,
        }
    }
    if int_digits.is_empty() {
        return None;
    }
    let dollars: i64 = int_digits.parse().ok()?;
    // Fractional part: first two digits are cents, the third rounds.
    let (mut cents, tail) = match tail.strip_prefix('.') {
        None => (0i64, tail),
        Some(f) => {
            let end = f.find(|c: char| !c.is_ascii_digit()).unwrap_or(f.len());
            let (digits, rest_tail) = (&f[..end], &f[end..]);
            if digits.is_empty() {
                return None;
            }
            let b = digits.as_bytes();
            let mut c = i64::from(b[0] - b'0') * 10;
            if b.len() >= 2 {
                c += i64::from(b[1] - b'0');
            }
            if b.len() >= 3 && b[2] >= b'5' {
                c += 1;
            }
            (c, rest_tail)
        }
    };
    // Trailing whitespace / closing paren / sign / currency symbol.
    let mut tail = tail;
    while !tail.is_empty() {
        let t = tail.trim_start();
        if let Some(r) = t.strip_prefix(')') {
            tail = r;
        } else if let Some(r) = t.strip_prefix('-') {
            neg = true;
            tail = r;
        } else if let Some(r) = t.strip_prefix('+') {
            tail = r;
        } else if let Some(r) = t.strip_prefix('$') {
            tail = r;
        } else if t.is_empty() {
            break;
        } else {
            return None;
        }
    }
    // cents rounding can carry into the dollar (0.995 -> 1.00).
    let carry = cents / 100;
    cents %= 100;
    let total = dollars
        .checked_add(carry)?
        .checked_mul(100)?
        .checked_add(cents)?;
    Some(if neg { -total } else { total })
}

/// v7.17.0 Phase 3.P0-34 — parse a PG `timetz` literal
/// `HH:MM:SS[.fraction]±HH[:MM]` into (us, offset_secs).
///
/// The offset suffix is MANDATORY: SPG doesn't have a session TZ
/// wired into eval, so a bare `HH:MM:SS` literal would be
/// ambiguous. Returns None for any parse failure or out-of-range
/// component — caller surfaces as a hard SQL error.
///
/// Offset range: ±14 hours (±50400 seconds), matching PG's
/// internal limit.
pub(crate) fn parse_timetz_str(s: &str) -> Option<(i64, i32)> {
    let s = s.trim();
    // Find the offset sign — scan from right since the time part
    // never contains '+' / '-' (after the optional fractional dot
    // it's all digits and ':').
    let bytes = s.as_bytes();
    let sign_pos = bytes
        .iter()
        .enumerate()
        .rev()
        .find(|&(_, &b)| b == b'+' || b == b'-')
        .map(|(i, _)| i)?;
    if sign_pos == 0 {
        return None; // bare sign — no time component
    }
    let time_part = &s[..sign_pos];
    let offset_part = &s[sign_pos..];
    let us = parse_time_str(time_part)?;
    let sign: i32 = if offset_part.starts_with('+') { 1 } else { -1 };
    let offset_body = &offset_part[1..];
    // v7.39 (round 253) — PG accepts the compact offset spellings too
    // (probed live): `+0230` = 02:30, `+023` = 00:23.
    let (hh_str, mm_str) = match offset_body.split_once(':') {
        Some((h, m)) => (h, m),
        None if offset_body.len() == 4 => offset_body.split_at(2),
        None if offset_body.len() == 3 => offset_body.split_at(1),
        None => (offset_body, "0"),
    };
    let hh: i32 = hh_str.parse().ok()?;
    let mm: i32 = mm_str.parse().ok()?;
    if !(0..=14).contains(&hh) || !(0..=59).contains(&mm) {
        return None;
    }
    let total = sign * (hh * 3600 + mm * 60);
    if total.abs() > 50_400 {
        return None;
    }
    Some((us, total))
}

/// v7.17.0 Phase 3.P0-33 — funnel an integer literal through MySQL
/// YEAR range validation: 0 sentinel or 1901..=2155. Out-of-range
/// surfaces as a hard SQL error (no silent truncation, mirrors PG
/// `time_in` / `uuid_in` discipline).
pub(crate) fn coerce_int_to_year(n: i64, col_name: &str) -> Result<Value<'static>, EngineError> {
    if n == 0 || (1901..=2155).contains(&n) {
        // u16::try_from cannot fail in this range; the cast also
        // covers the 0 sentinel.
        return Ok(Value::Year(n as u16));
    }
    Err(EngineError::Eval(EvalError::TypeMismatch {
        detail: alloc::format!(
            "year value out of range: {n} (column `{col_name}`; \
             MySQL accepts 0 or 1901..=2155)"
        ),
    }))
}

/// v7.17.0 Phase 3.P0-32 — parse a PG `time` literal
/// `HH:MM:SS[.fraction]` into microseconds since 00:00:00.
///
/// Accepts:
///   * `HH:MM:SS`            — exact-second precision
///   * `HH:MM:SS.f` .. `.ffffff` — 1-6 fractional digits, right-padded
///     with zeros to microseconds
///
/// Range: hour 0..=24 (`24:00:00` is PG's day-end special, measured
/// round 764), minute 0..=59, second 0..=59. Anything else returns
/// None — caller surfaces as a hard SQL error (no silent truncation,
/// matches PG's `time_in` behaviour).
pub(crate) fn parse_time_str(s: &str) -> Option<i64> {
    let s = s.trim();
    // PG special TIME value: `allballs` is midnight (all zeros).
    if s.eq_ignore_ascii_case("allballs") {
        return Some(0);
    }
    let (hms, frac) = match s.split_once('.') {
        Some((h, f)) => (h, Some(f)),
        None => (s, None),
    };
    let mut parts = hms.split(':');
    let hh: u32 = parts.next()?.parse().ok()?;
    let mm: u32 = parts.next()?.parse().ok()?;
    // PG accepts the seconds-optional `HH:MM` form for TIME
    // (`'10:30'::time` → `10:30:00`); missing seconds default to 0.
    let ss: u32 = match parts.next() {
        Some(x) => x.parse().ok()?,
        None => 0,
    };
    if parts.next().is_some() {
        return None;
    }
    // PG accepts the end-of-day sentinel `24:00:00` (but nothing past it).
    if hh > 24 || mm > 59 || ss > 59 || (hh == 24 && (mm != 0 || ss != 0)) {
        return None;
    }
    let frac_us: i64 = match frac {
        None => 0,
        Some(f) => {
            if f.is_empty() || f.len() > 6 || !f.bytes().all(|b| b.is_ascii_digit()) {
                return None;
            }
            // Right-pad with zeros so '.5' = 500000 µsec.
            let mut padded = alloc::string::String::with_capacity(6);
            padded.push_str(f);
            while padded.len() < 6 {
                padded.push('0');
            }
            padded.parse().ok()?
        }
    };
    if hh == 24 && frac_us != 0 {
        return None;
    }
    Some(
        i64::from(hh) * 3_600_000_000
            + i64::from(mm) * 60_000_000
            + i64::from(ss) * 1_000_000
            + frac_us,
    )
}

/// v7.39 (round 272) — PG's declared-typmod bounds: precision 1..=1000
/// and scale -1000..=1000 (SPG does not carry a negative scale yet, so
/// the lower half is a recorded gap rather than an accepted range).
pub(crate) fn numeric_typmod_in_range(precision: u16, scale: i16) -> bool {
    (1..=1000).contains(&precision) && (-1000..=1000).contains(&scale)
}

/// PG's wording for a typmod outside those bounds, given the text
/// between the parentheses. `None` when the typmod is fine or the text
/// is not a numeric one.
pub(crate) fn numeric_typmod_error(name: &str) -> Option<alloc::string::String> {
    let lower = name.trim().to_ascii_lowercase();
    let (head, rest) = lower.split_once('(')?;
    if !matches!(head.trim(), "numeric" | "decimal") {
        return None;
    }
    let args = rest.strip_suffix(')')?;
    let mut it = args.split(',').map(str::trim);
    let p: i64 = it.next()?.parse().ok()?;
    if !(1..=1000).contains(&p) {
        return Some(alloc::format!(
            "NUMERIC precision {p} must be between 1 and 1000"
        ));
    }
    if let Some(s) = it.next() {
        let s: i64 = s.parse().ok()?;
        if !(-1000..=1000).contains(&s) {
            return Some(alloc::format!(
                "NUMERIC scale {s} must be between -1000 and 1000"
            ));
        }
    }
    None
}

/// v7.37.5 ship triage — string-form PG type name → `DataType`
/// lookup driving `CastTarget::Named` (the generic typed-cast
/// escape). Covers the v7.37.5 γ/δ/ε/ζ-A type-completeness work
/// that landed without per-type CastTarget variants. Returns
/// `None` for genuinely-unknown idents so the caller can surface
/// the existing "unsupported cast target" error.
pub(crate) fn type_name_to_data_type(name: &str) -> Option<DataType> {
    with_lower_name(name.trim(), type_name_to_data_type_lower)
}

/// v7.39 (round 607) — lowercase a type NAME without allocating.
///
/// A cast's target is fixed for the whole statement, but every helper that
/// reads it rebuilt its lowercase form for EVERY ROW. `id::REAL` cost 8
/// allocations a row where `id::FLOAT` — the same conversion, spelled with a
/// name the parser settles into a `CastTarget` variant instead of `Named` —
/// cost none, and ran 7.5 ms against 44.6 over 200k rows. Type names are
/// short, so the stack buffer covers every spelling that resolves; a longer
/// one still answers correctly through the owned path.
///
/// Only ASCII `A-Z` bytes change, and those never appear inside a multi-byte
/// UTF-8 sequence, so lowercasing in place leaves the slice valid UTF-8.
pub(crate) fn with_lower_name<R>(name: &str, f: impl FnOnce(&str) -> R) -> R {
    const CAP: usize = 64;
    if name.len() <= CAP {
        let mut buf = [0u8; CAP];
        buf[..name.len()].copy_from_slice(name.as_bytes());
        buf[..name.len()].make_ascii_lowercase();
        if let Ok(s) = core::str::from_utf8(&buf[..name.len()]) {
            return f(s);
        }
    }
    f(&name.to_ascii_lowercase())
}

fn type_name_to_data_type_lower(n: &str) -> Option<DataType> {
    // v7.37.5 ship triage — `numeric(p,s)` precision/scale params:
    // peel them off and route to a precision-bearing DataType.
    if let Some((head, paren)) = n.split_once('(')
        && let Some(args) = paren.strip_suffix(')')
    {
        // v7.39 (round 272) — parsed as u16. At u8 a typmod PG accepts
        // (`numeric(1000,999)`) failed to parse and `unwrap_or(0)`
        // turned it into the UNCONSTRAINED type, so the cast silently
        // did nothing at all rather than reporting anything.
        // v7.39 (round 607) — a fixed pair rather than two Vecs. No typmod
        // this resolves has a third argument, and both were built for every
        // row a `numeric(p,s)` cast touched.
        let mut wide: [Option<i32>; 2] = [None, None];
        for (slot, s) in wide.iter_mut().zip(args.split(',')) {
            *slot = s.trim().parse::<i32>().ok();
        }
        let nums: [u8; 2] = [
            wide[0].and_then(|v| u8::try_from(v).ok()).unwrap_or(0),
            wide[1].and_then(|v| u8::try_from(v).ok()).unwrap_or(0),
        ];
        match head {
            // v7.39 (round 281) — `bit(3)` / `varbit(3)` as cast targets.
            "bit" => {
                return Some(DataType::Bit(
                    u32::try_from(wide.first().copied().flatten()?).ok()?,
                ));
            }
            "varbit" | "bit varying" => {
                return Some(DataType::BitVarying(
                    u32::try_from(wide.first().copied().flatten()?).ok()?,
                ));
            }
            "numeric" | "decimal" => {
                let precision = u16::try_from(wide.first().copied().flatten()?).ok()?;
                // v7.39 (round 273) — the declared scale is signed.
                let scale = i16::try_from(wide.get(1).copied().flatten().unwrap_or(0)).ok()?;
                if !numeric_typmod_in_range(precision, scale) {
                    return None;
                }
                return Some(DataType::Numeric { precision, scale });
            }
            // `varchar(n)` / `char(n)` carry length caps; SPG stores
            // these as DataType::Varchar / Char(n). v7.37.5 cast
            // recognises both but the cast itself drops the cap
            // (Text widening at value time honours the per-row
            // length contract already in coerce_value).
            "varchar" => {
                return Some(DataType::Varchar(nums.first().copied().unwrap_or(0).into()));
            }
            "char" | "character" => {
                return Some(DataType::Char(nums.first().copied().unwrap_or(0).into()));
            }
            _ => {}
        }
    }
    Some(match n {
        "smallint" | "int2" => DataType::SmallInt,
        "numeric" | "decimal" => DataType::Numeric {
            precision: 0,
            scale: 0,
        },
        // Network/MAC/bit/XML/"char" — all first-class since
        // v7.37.5 ζ-A.
        "inet" => DataType::Inet,
        "cidr" => DataType::Cidr,
        "macaddr" => DataType::Macaddr,
        "macaddr8" => DataType::Macaddr8,
        "pg_lsn" => DataType::PgLsn,
        // v7.39 (read01 varbit.c) — the B'...' literal's internal target.
        "__bit_literal" => DataType::BitVarying(0),
        // v7.39 (round 640) — a transaction id has its own identity now.
        // The name used to resolve to `bigint`, which is why
        // `pg_typeof(NULL::xid)` said so, `pg_type` could not list oid
        // 28, and `CREATE TABLE t (a xid)` was an unknown type.
        "xid" => DataType::Xid,
        "xid8" => DataType::Xid8,
        "bit" => DataType::Bit(0),
        "varbit" | "bit varying" => DataType::BitVarying(0),
        "xml" => DataType::Xml,
        // v7.37 (round 894) — the four names a QUOTED cast could not
        // reach. `::tsvector` parses as a keyword arm and works;
        // `::"tsvector"` becomes `CastTarget::Named("tsvector")` and lands
        // here, where these four were absent, so PG18's own spelling
        // answered `type "tsvector" does not exist`. Everything a client
        // generates with quoted identifiers — ORMs, pg_dump output — takes
        // that path. Enumerated against PG18: of its 75 builtin scalar and
        // range types, PG accepts every one quoted and SPG rejected exactly
        // these.
        "tsvector" => DataType::TsVector,
        "tsquery" => DataType::TsQuery,
        // `regclass` / `regtype` are the other two PG18 accepts quoted and
        // SPG does not, but they have no `DataType` of their own — they
        // live as `Value::RegClass` / `Value::RegType` and their casts are
        // special-cased at value level. Routing them here would need that
        // path, not a name-to-DataType row, so they stay open rather than
        // guessed at.
        "money" => DataType::Money,
        "char1" => DataType::Char1,
        // Geometry (v7.37.5 ε).
        "point" => DataType::Point,
        "lseg" => DataType::Lseg,
        "path" => DataType::Path,
        "box" => DataType::PgBox,
        "polygon" => DataType::Polygon,
        "line" => DataType::Line,
        "circle" => DataType::Circle,
        // Multirange (v7.37.5 δ).
        "int4multirange" => DataType::Multirange(spg_storage::RangeKind::Int4),
        "int8multirange" => DataType::Multirange(spg_storage::RangeKind::Int8),
        "nummultirange" => DataType::Multirange(spg_storage::RangeKind::Num),
        "tsmultirange" => DataType::Multirange(spg_storage::RangeKind::Ts),
        "tstzmultirange" => DataType::Multirange(spg_storage::RangeKind::TsTz),
        "datemultirange" => DataType::Multirange(spg_storage::RangeKind::Date),
        // Range scalars(scaffolded in v7.17, casts join here).
        "int4range" => DataType::Range(spg_storage::RangeKind::Int4),
        "int8range" => DataType::Range(spg_storage::RangeKind::Int8),
        "numrange" => DataType::Range(spg_storage::RangeKind::Num),
        "tsrange" => DataType::Range(spg_storage::RangeKind::Ts),
        "tstzrange" => DataType::Range(spg_storage::RangeKind::TsTz),
        "daterange" => DataType::Range(spg_storage::RangeKind::Date),
        // Array forms — `::BOOL[]` etc. The parser canonicalises
        // postfix `[]` into the `_array` suffix; mirror PG's
        // builtin arrays so the cast lands on a typed array Value.
        "bool_array" | "boolean_array" => DataType::BoolArray,
        "smallint_array" | "int2_array" => DataType::SmallIntArray,
        "int_array" | "integer_array" | "int4_array" => DataType::IntArray,
        "bigint_array" | "int8_array" => DataType::BigIntArray,
        "float_array" | "double_array" | "real_array" | "float8_array" | "float4_array" => {
            DataType::FloatArray
        }
        // Width-suffixed float spellings — SPG has one float
        // representation.
        "float4" | "real" => DataType::Real,
        "float8" | "double precision" | "float" => DataType::Float,
        // v7.39 (round 667) — this said "OIDs are plain integers" and
        // mapped to BigInt, which is why `pg_typeof(1::oid)` answered
        // `bigint`. The VALUE is still a bigint; what changed is that the
        // declared type is no longer thrown away. See `DataType::Oid`.
        "oid" => DataType::Oid,
        // v7.39 (round 694) — the array forms of the system types. PG has
        // an array type for every scalar; these five were the ones a cast
        // could name and SPG could not answer. `regtype[]` and
        // `regclass[]` did not even parse (their scalars have dedicated
        // CastTarget variants, so they never reached the postfix `[]`
        // handling); `oid[]` and `name[]` parsed and then met `type
        // "oid_array" does not exist`.
        //
        // They land on TextArray rather than a variant apiece for the
        // reason the scalars do NOT: a reg* value renders as a NAME, and
        // TextArray already carries and renders names. `oid_array` is the
        // exception and takes BigIntArray, because an OID renders as its
        // number.
        "oid_array" => DataType::OidArray,
        "name_array" | "regtype_array" | "regclass_array" | "regproc_array" => DataType::TextArray,
        // TIME [WITHOUT TIME ZONE] — first-class since the codec
        // carries Value::Time; the coerce path parses HH:MM:SS.
        "time" | "time without time zone" => DataType::Time,
        "timetz" | "time with time zone" => DataType::TimeTz,
        // v7.39 (round 780, F31-D1) — `hstore` is a first-class SPG
        // type (parser, storage variant, codec and both text
        // conversions have existed since v7.17.0) but the type-NAME
        // map never listed it, so every wire spelling — a column
        // declared `hstore`, a `::hstore` cast — answered
        // 'type "hstore" does not exist'.
        "hstore" => DataType::Hstore,
        "numeric_array" | "decimal_array" => DataType::NumericArray,
        "varchar_array" | "character varying_array" | "char_array" | "bpchar_array" => {
            DataType::TextArray
        }
        "text_array" => DataType::TextArray,
        "date_array" => DataType::DateArray,
        "timestamp_array" => DataType::TimestampArray,
        "timestamptz_array" => DataType::TimestamptzArray,
        "uuid_array" => DataType::UuidArray,
        "json_array" => DataType::JsonArray,
        "jsonb_array" => DataType::JsonbArray,
        "bytea_array" => DataType::BytesArray,
        "interval_array" => DataType::IntervalArray,
        "money_array" => DataType::MoneyArray,
        // v7.38 (read01) — primitive scalar spellings. These reach here only
        // via CastTarget::Named (e.g. the function-style typecast `int4('5')` /
        // `text(42)` / `date('2024-01-15')`); the `expr::type` parser path maps
        // them to dedicated CastTarget variants and never touches this table.
        "int" | "int4" | "integer" => DataType::Int,
        "bigint" | "int8" => DataType::BigInt,
        "text" => DataType::Text,
        // v7.39 (round 291) — PG's identifier type. `CREATE TABLE t (a
        // name)` is legal SQL that SPG answered "type \"name\" does not
        // exist" to.
        "name" => DataType::Name,
        "varchar" | "character varying" => DataType::Varchar(0),
        // v7.39 (bpchar epic) — bare `char` / `character` is char(1) (SQL
        // standard, `'xyz'::char` = 'x'); bare `bpchar` is PG's unlimited
        // blank-trimmed type.
        "char" | "character" => DataType::Char(1),
        "bpchar" => DataType::Char(0),
        "bool" | "boolean" => DataType::Bool,
        "date" => DataType::Date,
        "timestamp" | "timestamp without time zone" => DataType::Timestamp,
        "timestamptz" | "timestamp with time zone" => DataType::Timestamptz,
        "uuid" => DataType::Uuid,
        "json" => DataType::Json,
        "jsonb" => DataType::Jsonb,
        "bytea" => DataType::Bytes,
        "interval" => DataType::Interval,
        _ => return None,
    })
}

pub(crate) const fn column_type_to_data_type(t: ColumnTypeName) -> DataType {
    match t {
        ColumnTypeName::SmallInt => DataType::SmallInt,
        ColumnTypeName::Int => DataType::Int,
        ColumnTypeName::BigInt => DataType::BigInt,
        ColumnTypeName::Float => DataType::Float,
        ColumnTypeName::Real => DataType::Real,
        ColumnTypeName::Text => DataType::Text,
        ColumnTypeName::Name => DataType::Name,
        ColumnTypeName::Xid => DataType::Xid,
        ColumnTypeName::Xid8 => DataType::Xid8,
        ColumnTypeName::Oid => DataType::Oid,
        ColumnTypeName::Varchar(n) => DataType::Varchar(n),
        ColumnTypeName::Char(n) => DataType::Char(n),
        ColumnTypeName::Bool => DataType::Bool,
        ColumnTypeName::Vector { dim, encoding } => DataType::Vector {
            dim,
            encoding: match encoding {
                SqlVecEncoding::F32 => VecEncoding::F32,
                SqlVecEncoding::Sq8 => VecEncoding::Sq8,
                SqlVecEncoding::F16 => VecEncoding::F16,
            },
        },
        ColumnTypeName::Numeric(precision, scale) => DataType::Numeric { precision, scale },
        ColumnTypeName::Date => DataType::Date,
        ColumnTypeName::Timestamp => DataType::Timestamp,
        ColumnTypeName::Timestamptz => DataType::Timestamptz,
        ColumnTypeName::Json => DataType::Json,
        ColumnTypeName::Jsonb => DataType::Jsonb,
        ColumnTypeName::Bytes => DataType::Bytes,
        ColumnTypeName::TextArray => DataType::TextArray,
        ColumnTypeName::IntArray => DataType::IntArray,
        ColumnTypeName::BigIntArray => DataType::BigIntArray,
        ColumnTypeName::TsVector => DataType::TsVector,
        ColumnTypeName::TsQuery => DataType::TsQuery,
        ColumnTypeName::Uuid => DataType::Uuid,
        ColumnTypeName::Time => DataType::Time,
        ColumnTypeName::Year => DataType::Year,
        ColumnTypeName::TimeTz => DataType::TimeTz,
        ColumnTypeName::Money => DataType::Money,
        ColumnTypeName::Range(k) => DataType::Range(match k {
            spg_sql::ast::RangeKindAst::Int4 => spg_storage::RangeKind::Int4,
            spg_sql::ast::RangeKindAst::Int8 => spg_storage::RangeKind::Int8,
            spg_sql::ast::RangeKindAst::Num => spg_storage::RangeKind::Num,
            spg_sql::ast::RangeKindAst::Ts => spg_storage::RangeKind::Ts,
            spg_sql::ast::RangeKindAst::TsTz => spg_storage::RangeKind::TsTz,
            spg_sql::ast::RangeKindAst::Date => spg_storage::RangeKind::Date,
        }),
        ColumnTypeName::Hstore => DataType::Hstore,
        ColumnTypeName::IntArray2D => DataType::IntArray2D,
        ColumnTypeName::BigIntArray2D => DataType::BigIntArray2D,
        ColumnTypeName::TextArray2D => DataType::TextArray2D,
        ColumnTypeName::BoolArray2D => DataType::BoolArray2D,
        ColumnTypeName::Interval => DataType::Interval,
        ColumnTypeName::IntervalArray => DataType::IntervalArray,
        ColumnTypeName::BoolArray => DataType::BoolArray,
        ColumnTypeName::SmallIntArray => DataType::SmallIntArray,
        ColumnTypeName::FloatArray => DataType::FloatArray,
        ColumnTypeName::NumericArray => DataType::NumericArray,
        ColumnTypeName::DateArray => DataType::DateArray,
        ColumnTypeName::TimestampArray => DataType::TimestampArray,
        ColumnTypeName::TimestamptzArray => DataType::TimestamptzArray,
        ColumnTypeName::UuidArray => DataType::UuidArray,
        ColumnTypeName::JsonArray => DataType::JsonArray,
        ColumnTypeName::JsonbArray => DataType::JsonbArray,
        ColumnTypeName::BytesArray => DataType::BytesArray,
        ColumnTypeName::VarcharArray => DataType::VarcharArray,
        ColumnTypeName::CharArray => DataType::CharArray,
        ColumnTypeName::Multirange(k) => DataType::Multirange(match k {
            spg_sql::ast::RangeKindAst::Int4 => spg_storage::RangeKind::Int4,
            spg_sql::ast::RangeKindAst::Int8 => spg_storage::RangeKind::Int8,
            spg_sql::ast::RangeKindAst::Num => spg_storage::RangeKind::Num,
            spg_sql::ast::RangeKindAst::Ts => spg_storage::RangeKind::Ts,
            spg_sql::ast::RangeKindAst::TsTz => spg_storage::RangeKind::TsTz,
            spg_sql::ast::RangeKindAst::Date => spg_storage::RangeKind::Date,
        }),
        ColumnTypeName::Point => DataType::Point,
        ColumnTypeName::Lseg => DataType::Lseg,
        ColumnTypeName::Path => DataType::Path,
        ColumnTypeName::PgBox => DataType::PgBox,
        ColumnTypeName::Polygon => DataType::Polygon,
        ColumnTypeName::Line => DataType::Line,
        ColumnTypeName::Circle => DataType::Circle,
        ColumnTypeName::Inet => DataType::Inet,
        ColumnTypeName::Cidr => DataType::Cidr,
        ColumnTypeName::Macaddr => DataType::Macaddr,
        ColumnTypeName::Macaddr8 => DataType::Macaddr8,
        ColumnTypeName::Bit(n) => DataType::Bit(n),
        ColumnTypeName::BitVarying(n) => DataType::BitVarying(n),
        ColumnTypeName::Xml => DataType::Xml,
        ColumnTypeName::Char1 => DataType::Char1,
        ColumnTypeName::MoneyArray => DataType::MoneyArray,
    }
}

/// Convert an INSERT VALUES expression to a storage Value. Supports literal
/// expressions, unary-minus over numeric literals, and pgvector-style
/// `'[..]'::vector` cast (v1.2). Anything more complex returns `Unsupported`.
pub(crate) fn literal_expr_to_value(expr: Expr) -> Result<Value<'static>, EngineError> {
    literal_expr_to_value_in(expr, None)
}

/// v7.39 (read01 round 55) — the catalog-aware form. `cast_value` cannot
/// resolve a user-named type (composite / domain / enum) or a regclass on its
/// own: those live in the catalog. Without it, `INSERT INTO t VALUES
/// (ROW(1,2)::pt)` failed with "unsupported cast target `::pt`" — the whole
/// INSERT, so the table stayed empty. Callers that HAVE a catalog pass it;
/// the ones that don't (DDL default folding, partition bounds) keep the old
/// literal-only behaviour.
pub(crate) fn literal_expr_to_value_in(
    expr: Expr,
    catalog: Option<&spg_storage::Catalog>,
) -> Result<Value<'static>, EngineError> {
    match expr {
        Expr::Literal(l) => Ok(literal_to_value(l)),
        Expr::Cast { expr, target } => {
            // A catalog-dependent cast target has to go through eval's
            // pre-hook, which is the only place that knows the user types.
            if catalog.is_some()
                && matches!(
                    target,
                    spg_sql::ast::CastTarget::Named(_) | spg_sql::ast::CastTarget::RegClass
                )
            {
                return eval_expr_with_catalog(Expr::Cast { expr, target }, catalog);
            }
            let inner_value = literal_expr_to_value_in(*expr, catalog)?;
            crate::eval::cast_value(inner_value, target).map_err(EngineError::Eval)
        }
        Expr::Unary {
            op: UnOp::Neg,
            expr,
        } => match *expr {
            Expr::Literal(Literal::Integer(n)) => {
                // Fold to i32 if it fits, else BigInt. Parser emits Integer(i64)
                // — overflow on negate of i64::MIN is the one edge case.
                let neg = n.checked_neg().ok_or_else(|| {
                    EngineError::Unsupported("integer literal overflow on negation".into())
                })?;
                Ok(int_value_for(neg))
            }
            Expr::Literal(Literal::Float(x)) => Ok(Value::Float(-x)),
            // v7.38 (read01) — a dotted literal is NUMERIC; negate the mantissa.
            Expr::Literal(Literal::Numeric { unscaled, scale }) => Ok(Value::Numeric {
                scaled: -unscaled,
                scale,
                kind: spg_storage::NumericKind::Finite,
            }),
            // v7.38 (read01, T3.C3) — a NUMERIC literal beyond i128; negate by
            // flipping the sign of the decimal string, then re-resolve.
            Expr::Literal(Literal::NumericBig(ref s)) => {
                let flipped = if let Some(rest) = s.strip_prefix('-') {
                    rest.to_string()
                } else {
                    alloc::format!("-{s}")
                };
                Ok(big_literal_to_value(&flipped))
            }
            // v7.37.5 ship triage — fold the unary minus through a
            // `Cast { Literal, target }` wrapper (`-2::smallint`,
            // `-3.14::numeric(10,2)`). We negate the inner literal,
            // re-wrap with the same cast, and re-enter the literal
            // resolver — the cast path handles the typed result.
            Expr::Cast {
                expr: inner,
                target,
            } => {
                let negated_inner = match *inner {
                    Expr::Literal(Literal::Integer(n)) => {
                        let neg = n.checked_neg().ok_or_else(|| {
                            EngineError::Unsupported("integer literal overflow on negation".into())
                        })?;
                        Expr::Literal(Literal::Integer(neg))
                    }
                    Expr::Literal(Literal::Float(x)) => Expr::Literal(Literal::Float(-x)),
                    Expr::Literal(Literal::Numeric { unscaled, scale }) => {
                        Expr::Literal(Literal::Numeric {
                            unscaled: -unscaled,
                            scale,
                        })
                    }
                    // v7.38 (read01, T3.C3) — big NUMERIC literal: flip its sign
                    // in the decimal string, re-wrap with the same cast.
                    Expr::Literal(Literal::NumericBig(ref s)) => {
                        let flipped = if let Some(rest) = s.strip_prefix('-') {
                            rest.to_string()
                        } else {
                            alloc::format!("-{s}")
                        };
                        Expr::Literal(Literal::NumericBig(flipped))
                    }
                    other => Expr::Unary {
                        op: spg_sql::ast::UnOp::Neg,
                        expr: alloc::boxed::Box::new(other),
                    },
                };
                literal_expr_to_value_in(
                    Expr::Cast {
                        expr: alloc::boxed::Box::new(negated_inner),
                        target,
                    },
                    catalog,
                )
            }
            other => Err(EngineError::Unsupported(alloc::format!(
                "unary minus over non-literal expression: {other:?}"
            ))),
        },
        // v7.10.10 — `ARRAY[lit, lit, …]` constructor accepted at
        // INSERT-time. Each element must reduce to a Value through
        // `literal_expr_to_value`; NULL elements become `None`.
        // v7.11.13 — deduce shape from element values: all Int →
        // IntArray; any BigInt → BigIntArray (widening); any Text
        // → TextArray. Cast targets (`ARRAY[]::INT[]`) flow through
        // the outer Cast arm before reaching here and re-coerce.
        Expr::Array(items) => {
            let mut materialised: alloc::vec::Vec<Value<'static>> =
                alloc::vec::Vec::with_capacity(items.len());
            for elem in &items {
                materialised.push(literal_expr_to_value_in(elem.clone(), catalog)?);
            }
            Ok(crate::describe::upgrade_timestamptz_array(
                array_literal_widen(materialised),
                &items,
                &[],
            ))
        }
        // Any other Expr shape — fall back to a general evaluation
        // against an empty row + empty schema. This unblocks the
        // app-common patterns where INSERT VALUES carries a
        // non-correlated function call:
        //   INSERT INTO t VALUES (concat('U-', 42))
        //   INSERT INTO t VALUES (now())
        //   INSERT INTO t VALUES (format('%s-%s', 'a', 'b'))
        // Any expression that references a column or `$N`
        // placeholder fails cleanly inside `eval_expr` with a
        // descriptive error; literals + casts + ARRAY[…] continue
        // to take the fast paths above so the hot INSERT path is
        // unchanged on the common case.
        other => eval_expr_with_catalog(other, catalog),
    }
}

/// v7.39 (read01 round 55) — evaluate a row-free expression, threading the
/// catalog when the caller has one so user-named casts resolve.
fn eval_expr_with_catalog(
    expr: Expr,
    catalog: Option<&spg_storage::Catalog>,
) -> Result<Value<'static>, EngineError> {
    let empty_schema: alloc::vec::Vec<spg_storage::ColumnSchema> = alloc::vec::Vec::new();
    let mut ctx = EvalContext::new(&empty_schema, None);
    if let Some(cat) = catalog {
        ctx = ctx.with_catalog(cat);
    }
    let empty_row = spg_storage::Row::new(alloc::vec::Vec::new());
    crate::eval::eval_expr(&expr, &empty_row, &ctx).map_err(EngineError::Eval)
}

pub(crate) fn literal_to_value(l: Literal) -> Value<'static> {
    match l {
        Literal::Integer(n) => int_value_for(n),
        Literal::Float(x) => Value::Float(x),
        Literal::Numeric { unscaled, scale } => Value::Numeric {
            scaled: unscaled,
            scale,
            kind: spg_storage::NumericKind::Finite,
        },
        Literal::NumericBig(s) => big_literal_to_value(&s),
        Literal::Timestamp { micros, .. } => Value::Timestamp(micros),
        Literal::Date { days, .. } => Value::Date(days),
        Literal::String(s) => Value::text(s),
        Literal::Bool(b) => Value::Bool(b),
        Literal::Null => Value::Null,
        Literal::Vector(v) => Value::vector(v),
        Literal::TextArray(items) => Value::TextArray(items),
        Literal::IntArray(items) => Value::IntArray(items),
        Literal::BigIntArray(items) => Value::BigIntArray(items),
        Literal::Interval {
            months,
            days,
            micros,
            ..
        } => Value::Interval {
            months,
            days,
            micros,
            kind: spg_storage::IntervalKind::Finite,
        },
    }
}

/// Pick `Int` (`i32`) when the literal fits, else `BigInt`. `INT` vs `BIGINT`
/// columns will still enforce the right tag downstream — this is just the
/// default we synthesise from an unannotated integer literal.
pub(crate) fn int_value_for(n: i64) -> Value<'static> {
    if let Ok(small) = i32::try_from(n) {
        Value::Int(small)
    } else {
        Value::BigInt(n)
    }
}

/// Widen / narrow `v` to fit `expected`. Numerics permit safe widening
/// (`Int → BigInt`, `Int/BigInt → Float`) and best-effort narrowing
/// (`BigInt → Int` succeeds only when the value fits in `i32`). Everything
/// else returns `TypeMismatch` carrying the column name for caller diagnostics.
/// `NULL` is always permitted; the nullability check happens later in storage.
/// v7.17.0 Phase 4.4 / v7.39 round 387 (type-fidelity epic P2) — enforce
/// the integer range a column's storage `DataType` is too wide to hold.
/// Two cases: an UNSIGNED column rejects negatives (Phase 4.4), and a
/// TINYINT / MEDIUMINT column (whose storage is the wider SmallInt / Int)
/// rejects values outside its real bounds — `INSERT 128 INTO TINYINT` was
/// stored silently where MariaDB strict raises ERROR 1264. Called after
/// `coerce_value` at each INSERT / UPDATE site. NULL / non-integer cells
/// pass through. SPG always presents STRICT_TRANS_TABLES, so out of range
/// is an error (the non-strict clamp is a later stage).
/// v7.39 (round 424, type-fidelity epic) — apply a MySQL temporal column's
/// declared fractional-seconds precision to a value on its way in. MariaDB
/// TRUNCATES toward zero to the declared digits — `DATETIME(1)` stores
/// `.256789` as `.2`, and a BARE `DATETIME` (precision 0) drops the fraction
/// entirely. Called next to `check_unsigned_range` at each INSERT / UPDATE
/// site; a column with no declared precision (every PG column) is untouched,
/// which is what keeps microsecond behaviour intact there.
pub(crate) fn truncate_to_column_fsp(v: Value<'static>, schema: &ColumnSchema) -> Value<'static> {
    let Some(fsp) = schema.mysql_fsp else {
        return v;
    };
    if fsp >= 6 {
        return v;
    }
    let scale = 10i64.pow(u32::from(6 - fsp));
    // Toward zero, so a negative TIME loses the same digits.
    let cut = |micros: i64| (micros / scale) * scale;
    match v {
        Value::Timestamp(m) => Value::Timestamp(cut(m)),
        Value::Time(m) => Value::Time(cut(m)),
        other => other,
    }
}

/// v7.39.3 — apply a MySQL `FLOAT(m,d)` / `DOUBLE(m,d)`'s declared pair
/// to a value on its way in.
///
/// The digits are NOT a display hint, which is what SPG's comment
/// claimed: measured on MySQL 9.7.2, `3.14159265358979` into either
/// `FLOAT(10,2)` or `DOUBLE(10,2)` stores `3.14`, and a value wider than
/// `m` is refused with errno 1264 rather than stored. SPG accepted the
/// syntax and kept the full double, so a column declared for money held
/// more precision than its schema said.
///
/// The tie rule is measured, not reasoned. At `d >= 1` it is
/// round-half-to-EVEN — eleven exactly-representable ties agree
/// (`0.25 -> 0.2`, `0.75 -> 0.8`, `1.25 -> 1.2`, `1.75 -> 1.8`, and the
/// negatives). At `d = 0` the ties go toward NEGATIVE INFINITY instead
/// (`0.5 -> 0`, `1.5 -> 1`, `2.5 -> 2`, `-0.5 -> -1`), which is a
/// different rule; non-ties round normally there (`1.6 -> 2`). Why MySQL
/// differs between the two is not explained here — this matches the
/// behaviour, which is what the rule about learning from PG and MySQL
/// asks for.
pub(crate) fn round_to_column_float_md(
    v: Value<'static>,
    schema: &ColumnSchema,
) -> Result<Value<'static>, EvalError> {
    let Some((m, d)) = schema.mysql_float_md else {
        return Ok(v);
    };
    let round = |x: f64| -> f64 {
        if d == 0 {
            // Measured: ties toward negative infinity, non-ties normal.
            let f = x.floor();
            return if x - f == 0.5 { f } else { x.round() };
        }
        // For d >= 1 the rule is round-half-to-even on the value's TRUE
        // binary magnitude, which is what formatting to `d` places does.
        // Scaling by `10^d` first does not: `2.3455 * 1000.0` lands just
        // ABOVE the tie in f64 and rounded up to 2.346, where MySQL
        // stores 2.345 because the value itself is just below it. The
        // differential said so on the first run.
        alloc::format!("{x:.*}", usize::from(d))
            .parse::<f64>()
            .unwrap_or(x)
    };
    // `m` counts ALL digits, so the integral part may hold `m - d` of
    // them. MySQL refuses a wider value outright (1264) rather than
    // storing a number the schema does not describe.
    let limit = 10f64.powi(i32::from(m.saturating_sub(d)));
    let checked = |x: f64| -> Result<f64, EvalError> {
        let r = round(x);
        if r.abs() >= limit {
            return Err(EvalError::TypeMismatch {
                detail: alloc::format!("Out of range value for column '{}' at row 1", schema.name),
            });
        }
        Ok(r)
    };
    match v {
        Value::Float(x) => Ok(Value::Float(checked(x)?)),
        #[allow(clippy::cast_possible_truncation)]
        Value::Real(x) => Ok(Value::Real(checked(f64::from(x))? as f32)),
        other => Ok(other),
    }
}

/// v7.39 (round 434) — the integer bounds a column actually accepts: the
/// declared MySQL width when one is annotated (TINYINT / MEDIUMINT store in a
/// wider tag), otherwise the storage type's own range. Shared by the strict
/// range check below and the `INSERT IGNORE` clamp.
fn column_int_bounds(schema: &ColumnSchema) -> Option<(i128, i128)> {
    if let Some(width) = schema.mysql_int_width {
        return Some(match (width, schema.is_unsigned) {
            (spg_storage::MysqlIntWidth::Tiny, false) => (-128, 127),
            (spg_storage::MysqlIntWidth::Tiny, true) => (0, 255),
            (spg_storage::MysqlIntWidth::Small, false) => (-32_768, 32_767),
            (spg_storage::MysqlIntWidth::Small, true) => (0, 65_535),
            (spg_storage::MysqlIntWidth::Medium, false) => (-8_388_608, 8_388_607),
            (spg_storage::MysqlIntWidth::Medium, true) => (0, 16_777_215),
            (spg_storage::MysqlIntWidth::Int, false) => (-2_147_483_648, 2_147_483_647),
            (spg_storage::MysqlIntWidth::Int, true) => (0, 4_294_967_295),
            // v7.39 (round 471, epic P4b) — the whole point of i128 bounds:
            // 18446744073709551615 does not fit the i64 these used to be.
            (spg_storage::MysqlIntWidth::Big, false) => {
                (i128::from(i64::MIN), i128::from(i64::MAX))
            }
            (spg_storage::MysqlIntWidth::Big, true) => (0, i128::from(u64::MAX)),
        });
    }
    let (lo, hi) = match schema.ty {
        DataType::SmallInt => (i128::from(i16::MIN), i128::from(i16::MAX)),
        DataType::Int => (i128::from(i32::MIN), i128::from(i32::MAX)),
        DataType::BigInt => (i128::from(i64::MIN), i128::from(i64::MAX)),
        _ => return None,
    };
    Some(if schema.is_unsigned {
        (0, hi)
    } else {
        (lo, hi)
    })
}

/// v7.39 (round 434) — bend a value so a MySQL `INSERT IGNORE` can store it.
///
/// MySQL's IGNORE does two things. Round 406 implemented the first: skip a
/// row that violates a unique key. This is the second: per-VALUE errors are
/// downgraded to coercions, so a bulk load never stops. Measured on
/// MariaDB 11 —
///   * a NULL into a NOT NULL column becomes the type's default (0 / '')
///   * an out-of-range integer clamps to the declared type's bound
///     (99999999999999 into INT → 2147483647)
///   * a non-numeric string into an integer column takes its leading numeric
///     prefix, or 0 when there is none ('12abc' → 12, 'abc' → 0)
///   * an over-long string truncates to the declared length
///
/// Anything this does not recognise is returned unchanged, so the ordinary
/// coercion path still raises its ordinary error. That is deliberate: where
/// SPG cannot represent MySQL's answer (a `'0000-00-00'` zero date, an ENUM's
/// empty error-member) the statement fails loudly rather than silently
/// storing a value MySQL would not have stored.
/// v7.38.18 (C12) — what MySQL would have said about a value that
/// [`mysql_ignore_fit`] bent, or `None` if it did not bend it.
///
/// Derived from the before/after pair rather than reported out of the
/// conversion, which stays a pure function of value and column.
///
/// Every code and every wording is from a MySQL 9.7.2 run, not from
/// documentation — an application switching on an errno has to see the
/// errno it would have seen:
///
/// ```text
/// INSERT INTO w VALUES (1,'toolong')     1265 Data truncated for column 's' at row 1
/// INSERT INTO w VALUES ('abc','ok')      1366 Incorrect integer value: 'abc' for column 'i' at row 1
/// INSERT INTO w VALUES (99999999999,..)  1264 Out of range value for column 'i' at row 1
/// INSERT INTO w (s) VALUES ('ok')        1364 Field 'i' doesn't have a default value
/// ```
/// v7.39 — what MySQL calls a value that does not fit, in the STRICT
/// case where it refuses rather than bends.
///
/// Same classification as `mysql_fit_warning` and derived from it, so
/// the two can never describe the same failure differently — which they
/// did: the warning path had learned MySQL's vocabulary and the error
/// path still spoke PostgreSQL's, so a MySQL client that let a value
/// through got `Out of range value for column 'n' at row 1` and one
/// that was refused got `integer out of range` with errno 1690.
///
/// The codes are NOT simply the warning's. Measured on MySQL 9.7.2,
/// `sql_mode='STRICT_TRANS_TABLES'` versus `sql_mode=''`:
///
/// ```text
///                     non-strict (warning)          strict (error)
///   TINYINT <- 999    1264 Out of range value…      1264 (22003) same wording
///   VARCHAR(3) <- …   1265 Data truncated…          1406 (22001) Data too long…
///   INT <- 'abc'      1366 Incorrect integer value  1366 (HY000) same wording
///   INT <- '12xy'     1265 Data truncated…          1265 (01000) same wording
///   omitted NOT NULL  1364 Field … no default       1364 (HY000) same wording
/// ```
///
/// The string case is the one that breaks the pattern: both the code and
/// the wording change. Assuming otherwise would have been wrong on
/// exactly the column type this version's worst defect lived in.
///
/// Returns the errno, the SQLSTATE MySQL pairs with it, and the message.
/// `None` when the value fits and there is nothing to refuse.
pub(crate) fn mysql_fit_error(
    before: &Value<'_>,
    after: &Value<'_>,
    schema: &ColumnSchema,
    row: usize,
    omitted: bool,
) -> Option<(u16, &'static str, alloc::string::String)> {
    let w = mysql_fit_warning(before, after, schema, row, omitted)?;
    let col = &schema.name;
    Some(match w.code {
        // A string that had to be cut is refused as "too long", not as
        // "truncated" — different errno AND different wording.
        1265 if matches!(
            schema.ty,
            DataType::Varchar(_) | DataType::Char(_) | DataType::Text
        ) =>
        {
            (
                1406,
                "22001",
                alloc::format!("Data too long for column '{col}' at row {row}"),
            )
        }
        1265 => (1265, "01000", w.message),
        1264 => (1264, "22003", w.message),
        1366 => (1366, "HY000", w.message),
        1364 => (1364, "HY000", w.message),
        other => (other, "HY000", w.message),
    })
}

pub(crate) fn mysql_fit_warning(
    before: &Value<'_>,
    after: &Value<'_>,
    schema: &ColumnSchema,
    row: usize,
    omitted: bool,
) -> Option<crate::MysqlWarning> {
    if before == after {
        return None;
    }
    let col = &schema.name;
    // An omitted NOT NULL column is a different complaint from a value
    // that would not fit.
    if omitted || before.is_null() {
        return Some(crate::MysqlWarning {
            level: "Warning",
            code: 1364,
            message: alloc::format!("Field '{col}' doesn't have a default value"),
        });
    }
    // v7.39 — `Numeric` belongs here. Without it a DECIMAL that would not
    // fit fell to the 1265 default and was reported as a truncation, where
    // MySQL calls it 1264 `Out of range value`.
    let numeric_col = matches!(
        schema.ty,
        DataType::SmallInt
            | DataType::Int
            | DataType::BigInt
            | DataType::Float
            | DataType::Real
            | DataType::Numeric { .. }
    );
    if numeric_col {
        // v7.39 — the NOUN is the column's type, and MySQL is not
        // consistent about its case. Measured on 9.7.2, one string into
        // each:
        //
        //     BIGINT   Incorrect integer value: 'abc' …
        //     DECIMAL  Incorrect decimal value: 'abc' …
        //     FLOAT    Incorrect FLOAT value: 'abc' …
        //     DOUBLE   Incorrect DOUBLE value: 'abc' …
        //
        // Two lower-case, two upper-case. A drop-in copies that rather
        // than tidying it.
        let noun = match schema.ty {
            DataType::Numeric { .. } => "decimal",
            DataType::Real => "FLOAT",
            DataType::Float => "DOUBLE",
            _ => "integer",
        };
        // A string given to a numeric column is 1366; a number that did
        // not fit its range is 1264.
        return Some(if matches!(before, Value::Text(_) | Value::BpChar(_)) {
            crate::MysqlWarning {
                level: "Warning",
                code: 1366,
                message: alloc::format!(
                    "Incorrect {noun} value: '{}' for column '{col}' at row {row}",
                    crate::eval::value_to_text(before)
                ),
            }
        } else {
            crate::MysqlWarning {
                level: "Warning",
                code: 1264,
                message: alloc::format!("Out of range value for column '{col}' at row {row}"),
            }
        });
    }
    Some(crate::MysqlWarning {
        level: "Warning",
        code: 1265,
        message: alloc::format!("Data truncated for column '{col}' at row {row}"),
    })
}

/// v7.39 — carry a value that this clamp does not model through
/// unchanged, so a shape nobody measured is not silently rewritten.
fn numeric_untouched(v: Value<'static>, _schema: &ColumnSchema) -> Value<'static> {
    v
}

/// v7.39 — restate a scaled integer at a different scale, rounding half
/// away from zero, which is what MySQL does going to a narrower scale
/// (measured: `DECIMAL(3,1) <- 1.26` stores `1.3`, not `1.2`).
fn restate_scaled(scaled: i128, from: u16, to: u16) -> i128 {
    if from == to {
        return scaled;
    }
    if to > from {
        let f = 10i128.checked_pow(u32::from(to - from)).unwrap_or(1);
        return scaled.saturating_mul(f);
    }
    let f = 10i128.checked_pow(u32::from(from - to)).unwrap_or(1);
    if f == 0 {
        return scaled;
    }
    let half = f / 2;
    if scaled >= 0 {
        (scaled + half) / f
    } else {
        (scaled - half) / f
    }
}

pub(crate) fn mysql_ignore_fit(v: Value<'static>, schema: &ColumnSchema) -> Value<'static> {
    if v.is_null() {
        if schema.nullable {
            return v;
        }
        // MySQL fills a NOT NULL column with its type's zero value.
        return match schema.ty {
            DataType::SmallInt | DataType::Int | DataType::BigInt => Value::BigInt(0),
            DataType::Float | DataType::Real => Value::Float(0.0),
            DataType::Text | DataType::Varchar(_) | DataType::Char(_) => Value::text(""),
            _ => v,
        };
    }
    // A string bound for an integer column: MySQL reads the leading numeric
    // prefix and calls the rest a truncation warning.
    if let Value::Text(ref s) = v
        && matches!(
            schema.ty,
            DataType::SmallInt | DataType::Int | DataType::BigInt
        )
        && s.trim().parse::<i64>().is_err()
    {
        return Value::BigInt(leading_numeric_prefix(s));
    }
    // v7.39 — a DECIMAL that will not fit clamps to the column's bound.
    //
    // Not a message fix: SPG REFUSED this even in a non-strict session,
    // where MySQL stores the bound and warns. Measured on 9.7.2 with
    // `sql_mode=''` and `DECIMAL(3,1)`:
    //
    //     9999  ->  99.9      -9999  ->  -99.9      1.26  ->  1.3
    //
    // The last is ordinary rounding to the declared scale and happens in
    // strict mode too; only the first two are the overflow. A bulk load
    // into a non-strict session stopped here on a row MySQL would have
    // taken.
    //
    // With the value bent, the STRICT path gets its answer from the same
    // classifier as every other type — `Out of range value for column 'd'
    // at row 1`, errno 1264 — instead of PostgreSQL's `numeric field
    // overflow`, which also carried its `DETAIL:` clause inline.
    if let DataType::Numeric { precision, scale } = schema.ty
        && precision != 0
        && scale >= 0
    {
        let col_scale = u16::try_from(scale).unwrap_or(0);
        let (scaled, val_scale) = match v {
            Value::Numeric {
                scaled,
                scale: vs,
                kind: spg_storage::NumericKind::Finite,
            } => (scaled, vs),
            Value::SmallInt(n) => (i128::from(n), 0),
            Value::Int(n) => (i128::from(n), 0),
            Value::BigInt(n) => (i128::from(n), 0),
            _ => return numeric_untouched(v, schema),
        };
        // Restate at the column's scale, rounding half away from zero as
        // MySQL does, then clamp to +/-(10^precision - 1) in those units.
        // Only a CLAMP is a value that would not fit. Rounding to the
        // declared scale is ordinary coercion and happens in strict mode
        // too: `DECIMAL(3,1) <- 1.26` stores `1.3` on both engines and in
        // both modes.
        //
        // The first cut returned the restated value unconditionally, and
        // the classifier — which decides "did not fit" by comparing before
        // with after — read that rounding as an overflow. Strict sessions
        // then REFUSED 1.26 with `Out of range value`, a value MySQL
        // takes. The six refusal probes were all green while that was
        // true; only checking the ACCEPTING side found it.
        let restated = restate_scaled(scaled, val_scale, col_scale);
        let limit = 10i128
            .checked_pow(u32::from(precision))
            .map_or(i128::MAX, |p| p - 1);
        if restated < -limit || restated > limit {
            return Value::numeric(restated.clamp(-limit, limit), col_scale);
        }
        return numeric_untouched(v, schema);
    }
    // An out-of-range integer clamps to the column's bound.
    let as_int = match v {
        Value::SmallInt(n) => Some(i128::from(n)),
        Value::Int(n) => Some(i128::from(n)),
        Value::BigInt(n) => Some(i128::from(n)),
        // v7.39 (round 471) — a BIGINT UNSIGNED cell arrives as Numeric.
        Value::Numeric {
            scaled, scale: 0, ..
        } => Some(scaled),
        _ => None,
    };
    if let Some(n) = as_int
        && let Some((lo, hi)) = column_int_bounds(schema)
        && (n < lo || n > hi)
    {
        return int_value_for_column(n.clamp(lo, hi));
    }
    // An over-long string truncates to the declared length.
    if let Value::Text(ref s) = v {
        let max = match schema.ty {
            DataType::Varchar(m) | DataType::Char(m) if m > 0 => m as usize,
            _ => return v,
        };
        if s.chars().count() > max {
            return Value::text(s.chars().take(max).collect::<alloc::string::String>());
        }
    }
    v
}

/// MySQL's string → integer coercion: take the longest leading numeric
/// prefix, read it as a double, and round half AWAY FROM ZERO. Measured on
/// MariaDB 11 — `'3.7abc'` → 4, `'2.4'` → 2, `'2.5'` → 3, `'-2.5'` → -3,
/// `'1e3x'` → 1000, `'0x10'` → 0 (the prefix is just the leading `0`),
/// `'abc'` / `'-'` / `''` → 0.
///
/// The prefix is a float, not an integer: reading only digits would answer 0
/// for `'.5'` where MySQL answers 1.
fn leading_numeric_prefix(s: &str) -> i64 {
    let t = s.trim_start();
    let b = t.as_bytes();
    let mut i = 0;
    if i < b.len() && (b[i] == b'-' || b[i] == b'+') {
        i += 1;
    }
    let int_start = i;
    while i < b.len() && b[i].is_ascii_digit() {
        i += 1;
    }
    let mut end = i;
    if i < b.len() && b[i] == b'.' {
        i += 1;
        while i < b.len() && b[i].is_ascii_digit() {
            i += 1;
        }
        // A lone "." after the sign is not a number; digits on either side
        // of it are.
        if i > int_start + 1 {
            end = i;
        }
    }
    // An exponent only counts when it has at least one digit AND a mantissa.
    if end > int_start && i < b.len() && (b[i] == b'e' || b[i] == b'E') {
        let mut j = i + 1;
        if j < b.len() && (b[j] == b'-' || b[j] == b'+') {
            j += 1;
        }
        let digits_start = j;
        while j < b.len() && b[j].is_ascii_digit() {
            j += 1;
        }
        if j > digits_start {
            end = j;
        }
    }
    let Ok(f) = t[..end].parse::<f64>() else {
        return 0;
    };
    // `f64::round` is already half-away-from-zero, which is MySQL's rule.
    let r = f.round();
    if r >= i64::MAX as f64 {
        i64::MAX
    } else if r <= i64::MIN as f64 {
        i64::MIN
    } else {
        r as i64
    }
}

/// v7.39 (round 471) — the Value an integer takes when it may exceed i64.
/// Mirrors `eval::u64_as_value`: BigInt while it fits, Numeric (scale 0)
/// past it, which is how a BIGINT UNSIGNED cell is stored.
fn int_value_for_column(n: i128) -> Value<'static> {
    match i64::try_from(n) {
        Ok(v) => Value::BigInt(v),
        Err(_) => Value::numeric(n, 0),
    }
}

pub(crate) fn check_unsigned_range(
    v: &Value,
    schema: &ColumnSchema,
    position: usize,
) -> Result<(), EngineError> {
    let n: i128 = match v {
        Value::SmallInt(x) => i128::from(*x),
        Value::Int(x) => i128::from(*x),
        Value::BigInt(x) => i128::from(*x),
        // v7.39 (round 471) — a BIGINT UNSIGNED cell arrives as Numeric,
        // which is the whole reason the bounds are i128 now.
        Value::Numeric { scaled, scale, .. } if *scale == 0 => *scaled,
        _ => return Ok(()), // non-integer cells (NULL, default) skip
    };
    // TINYINT / MEDIUMINT: the storage tag (SmallInt / Int) is wider than
    // the declared MySQL type, so the real bounds are enforced here. The
    // unsigned variant's 0 lower bound also covers the negative check.
    if let Some(width) = schema.mysql_int_width {
        // Small / Int are only ever set on an UNSIGNED column (a signed
        // SMALLINT / INT keeps its faithful storage tag and no marker); the
        // signed arms are unreachable but keep the match total.
        // v7.39 (round 471) — one bounds table, not two. The copy here
        // drifted out of reach the moment BIGINT UNSIGNED needed i128.
        let _ = width;
        let (lo, hi) = column_int_bounds(schema).unwrap_or((i128::MIN, i128::MAX));
        if n < lo || n > hi {
            // MariaDB's wording (SQLSTATE 22003); SPG tracks the column,
            // not the multi-row row number the "at row N" suffix carries.
            return Err(EngineError::Unsupported(alloc::format!(
                "Out of range value for column '{}'",
                schema.name
            )));
        }
        return Ok(());
    }
    // Other columns: reject a negative on any UNSIGNED column (Phase 4.4).
    if schema.is_unsigned && n < 0 {
        return Err(EngineError::Unsupported(alloc::format!(
            "column {:?} is UNSIGNED but got negative value {n} at position {position}",
            schema.name
        )));
    }
    Ok(())
}

/// Coerce a non-empty `TEXT[]` (how an array literal reaches a typed-array
/// cast) into a typed array by parsing each element through the existing
/// scalar `coerce_value` path. NULL elements pass through. Returns `None` for
/// array targets this helper does not cover (leaving the caller's other arms
/// or the final type-mismatch to handle it).
fn coerce_text_array_to(
    items: alloc::vec::Vec<Option<alloc::string::String>>,
    target: DataType,
    col: &str,
) -> Result<Option<Value<'static>>, EngineError> {
    let elem_dt = match target {
        DataType::BoolArray => DataType::Bool,
        DataType::NumericArray => DataType::Numeric {
            precision: 0,
            scale: 0,
        },
        DataType::DateArray => DataType::Date,
        DataType::TimestampArray => DataType::Timestamp,
        DataType::TimestamptzArray => DataType::Timestamptz,
        DataType::UuidArray => DataType::Uuid,
        // v7.39 (round 326, V43) — INTERVAL[] joined the covered set;
        // `'{1 day}'::interval[]` used to fail as a plain type mismatch.
        DataType::IntervalArray => DataType::Interval,
        _ => return Ok(None),
    };
    let mut scal: alloc::vec::Vec<Option<Value<'static>>> =
        alloc::vec::Vec::with_capacity(items.len());
    for item in items {
        match item {
            None => scal.push(None),
            Some(s) => scal.push(Some(coerce_value(Value::text(s), elem_dt, col, 0)?)),
        }
    }
    let out = match target {
        DataType::BoolArray => Value::BoolArray(
            scal.into_iter()
                .map(|o| o.map(|v| matches!(v, Value::Bool(true))))
                .collect(),
        ),
        DataType::NumericArray => Value::NumericArray(
            scal.into_iter()
                .map(|o| {
                    o.map(|v| match v {
                        Value::Numeric { scaled, scale, .. } => (scaled, scale),
                        _ => (0, 0),
                    })
                })
                .collect(),
        ),
        DataType::DateArray => Value::DateArray(
            scal.into_iter()
                .map(|o| {
                    o.map(|v| match v {
                        Value::Date(d) => d,
                        _ => 0,
                    })
                })
                .collect(),
        ),
        DataType::TimestampArray => Value::TimestampArray(
            scal.into_iter()
                .map(|o| {
                    o.map(|v| match v {
                        Value::Timestamp(t) => t,
                        _ => 0,
                    })
                })
                .collect(),
        ),
        DataType::TimestamptzArray => Value::TimestamptzArray(
            scal.into_iter()
                .map(|o| {
                    o.map(|v| match v {
                        Value::Timestamp(t) => t,
                        _ => 0,
                    })
                })
                .collect(),
        ),
        DataType::UuidArray => Value::UuidArray(
            scal.into_iter()
                .map(|o| {
                    o.map(|v| match v {
                        Value::Uuid(u) => u,
                        _ => [0u8; 16],
                    })
                })
                .collect(),
        ),
        DataType::IntervalArray => Value::IntervalArray(
            scal.into_iter()
                .map(|o| {
                    o.and_then(|v| match v {
                        Value::Interval {
                            months,
                            days,
                            micros,
                            kind,
                        } => Some(spg_storage::IntervalSpan {
                            months,
                            days,
                            micros,
                            kind,
                        }),
                        _ => None,
                    })
                })
                .collect(),
        ),
        _ => return Ok(None),
    };
    Ok(Some(out))
}

/// Parse a PG integer literal in text: decimal, plus the PG 16+ forms —
/// radix prefixes (`0x1F` hex / `0o17` octal / `0b101` binary) and `_` digit
/// separators (`1_000`). An optional leading sign applies to the magnitude.
/// Map a built-in type OID to its SQL-standard name, PG's `format_type`
/// / `oid::regtype` spelling (without the typmod). `None` for OIDs SPG
/// doesn't recognise (callers render the numeric OID, as PG does for an
/// unknown regtype). Shared by the `::regtype` cast and `format_type`.
/// v7.39 (read01 utils/adt, format_type.c) — the element OID for a
/// standard array type OID (PG's pg_type.typelem for the built-in `_x`
/// array types). format_type renders these as `<element>[]`.
pub(crate) fn array_oid_element(oid: i64) -> Option<i64> {
    Some(match oid {
        1000 => 16,   // _bool
        1001 => 17,   // _bytea
        1002 => 18,   // _char
        1003 => 19,   // _name
        1016 => 20,   // _int8
        1005 => 21,   // _int2
        1007 => 23,   // _int4
        1009 => 25,   // _text
        1028 => 26,   // _oid
        199 => 114,   // _json
        143 => 142,   // _xml
        651 => 650,   // _cidr
        1021 => 700,  // _float4
        1022 => 701,  // _float8
        775 => 774,   // _macaddr8
        791 => 790,   // _money
        1040 => 829,  // _macaddr
        1041 => 869,  // _inet
        1014 => 1042, // _bpchar
        1015 => 1043, // _varchar
        1182 => 1082, // _date
        1183 => 1083, // _time
        1115 => 1114, // _timestamp
        1185 => 1184, // _timestamptz
        1187 => 1186, // _interval
        1270 => 1266, // _timetz
        1561 => 1560, // _bit
        1563 => 1562, // _varbit
        1231 => 1700, // _numeric
        2951 => 2950, // _uuid
        3643 => 3614, // _tsvector
        3645 => 3615, // _tsquery
        3807 => 3802, // _jsonb
        _ => return None,
    })
}

/// v7.39 (round 621) — the OID of an ARRAY reads back as `<element>[]`.
///
/// `1007::regtype` rendered the number `1007` instead of `integer[]`, so a
/// column-type query — `atttypid::regtype`, the shape this cast exists for —
/// told an ORM the type of every array column was a bare number. The scalar
/// OIDs were all there; only the array half was missing, from both directions.
pub(crate) fn regtype_oid_to_name_owned(oid: i64) -> Option<alloc::string::String> {
    if let Some(scalar) = regtype_oid_to_name(oid) {
        return Some(alloc::string::String::from(scalar));
    }
    let (_, _, elem) = crate::system_catalog::ARRAY_TYPE_OIDS
        .iter()
        .find(|(arr, _, _)| *arr == oid)?;
    Some(alloc::format!("{}[]", regtype_oid_to_name(*elem)?))
}

/// The array OID whose element is `elem`, for the reverse direction.
pub(crate) fn array_oid_for_element(elem: i64) -> Option<i64> {
    crate::system_catalog::ARRAY_TYPE_OIDS
        .iter()
        .find(|(_, _, e)| *e == elem)
        .map(|(arr, _, _)| *arr)
}

pub(crate) fn regtype_oid_to_name(oid: i64) -> Option<&'static str> {
    Some(match oid {
        4600 => "pg_brin_bloom_summary",
        16 => "boolean",
        17 => "bytea",
        18 => "\"char\"",
        19 => "name",
        20 => "bigint",
        21 => "smallint",
        23 => "integer",
        25 => "text",
        26 => "oid",
        // v7.39 (round 640) — the row-header types.
        27 => "tid",
        28 => "xid",
        29 => "cid",
        5069 => "xid8",
        114 => "json",
        142 => "xml",
        650 => "cidr",
        700 => "real",
        701 => "double precision",
        774 => "macaddr8",
        790 => "money",
        829 => "macaddr",
        869 => "inet",
        1042 => "character",
        1043 => "character varying",
        1082 => "date",
        1083 => "time without time zone",
        1114 => "timestamp without time zone",
        1184 => "timestamp with time zone",
        1186 => "interval",
        1266 => "time with time zone",
        1560 => "bit",
        1562 => "bit varying",
        1700 => "numeric",
        2950 => "uuid",
        3614 => "tsvector",
        3615 => "tsquery",
        3802 => "jsonb",
        3904 => "int4range",
        3906 => "numrange",
        3908 => "tsrange",
        3910 => "tstzrange",
        3912 => "daterange",
        3926 => "int8range",
        _ => return None,
    })
}

pub(crate) fn parse_pg_int(s: &str) -> Option<i64> {
    let s = s.trim();
    let (neg, rest) = if let Some(r) = s.strip_prefix('-') {
        (true, r)
    } else if let Some(r) = s.strip_prefix('+') {
        (false, r)
    } else {
        (false, s)
    };
    // Split off an optional radix prefix (PG 16+: 0x / 0o / 0b), leaving
    // the digit portion. PG allows `_` group separators ONLY between two
    // digits — a leading/trailing/doubled underscore (`_5`, `5_`, `1__2`)
    // or one adjacent to the prefix is "invalid input syntax".
    let (radix, digits, has_prefix) =
        if let Some(h) = rest.strip_prefix("0x").or_else(|| rest.strip_prefix("0X")) {
            (16u32, h, true)
        } else if let Some(o) = rest.strip_prefix("0o").or_else(|| rest.strip_prefix("0O")) {
            (8, o, true)
        } else if let Some(b) = rest.strip_prefix("0b").or_else(|| rest.strip_prefix("0B")) {
            (2, b, true)
        } else {
            (10, rest, false)
        };
    let db = digits.as_bytes();
    // Reject a trailing or doubled underscore anywhere, and a leading
    // underscore unless it follows a radix prefix (PG accepts `0x_FF` but
    // not `_5` / `5_` / `1__2` / `0xFF_`).
    if db.last() == Some(&b'_')
        || digits.contains("__")
        || (!has_prefix && db.first() == Some(&b'_'))
    {
        return None;
    }
    let cleaned: alloc::string::String = digits.chars().filter(|&c| c != '_').collect();
    if cleaned.is_empty() {
        return None;
    }
    let mag = i64::from_str_radix(&cleaned, radix).ok()?;
    Some(if neg { mag.checked_neg()? } else { mag })
}

/// v7.38 (read01 P6.38) — well-formedness check for PG's `xml` CONTENT mode.
/// Verifies element tags are balanced and properly nested; comments (`<!-- -->`),
/// processing instructions (`<? ?>`), CDATA sections, `<!DOCTYPE …>`, plain
/// text, self-closing tags and multiple top-level elements are all accepted.
/// Attribute values are quote-aware so a `>` inside an attribute doesn't end a
/// tag early. This catches the common malformedness (unclosed / mismatched
/// tags) libxml2 rejects; deeper libxml2 checks (entity validity, duplicate
/// attributes, char legality) are a documented follow-up.
fn xml_content_is_well_formed(s: &str) -> bool {
    let b = s.as_bytes();
    let is_name =
        |c: u8| c.is_ascii_alphanumeric() || matches!(c, b'-' | b'_' | b'.' | b':') || c >= 0x80;
    let mut stack: alloc::vec::Vec<&[u8]> = alloc::vec::Vec::new();
    let mut i = 0;
    while i < b.len() {
        if b[i] != b'<' {
            i += 1;
            continue;
        }
        let rest = &s[i..];
        if rest.starts_with("<!--") {
            match rest.find("-->") {
                Some(p) => i += p + 3,
                None => return false,
            }
        } else if rest.starts_with("<![CDATA[") {
            match rest.find("]]>") {
                Some(p) => i += p + 3,
                None => return false,
            }
        } else if rest.starts_with("<?") {
            match rest.find("?>") {
                Some(p) => i += p + 2,
                None => return false,
            }
        } else if rest.starts_with("<!") {
            match rest.find('>') {
                Some(p) => i += p + 1,
                None => return false,
            }
        } else {
            // Element open / close / self-close tag.
            let close = i + 1 < b.len() && b[i + 1] == b'/';
            let name_start = if close { i + 2 } else { i + 1 };
            let mut j = name_start;
            while j < b.len() && is_name(b[j]) {
                j += 1;
            }
            if j == name_start {
                return false; // `<` not followed by a tag name
            }
            let name = &b[name_start..j];
            // Scan to the matching `>`, skipping quoted attribute values.
            let mut k = j;
            let mut quote = 0u8;
            let mut prev = 0u8;
            loop {
                if k >= b.len() {
                    return false; // unterminated tag
                }
                let c = b[k];
                if quote != 0 {
                    if c == quote {
                        quote = 0;
                    }
                } else if c == b'"' || c == b'\'' {
                    quote = c;
                } else if c == b'>' {
                    break;
                }
                prev = c;
                k += 1;
            }
            let self_closing = prev == b'/';
            i = k + 1;
            if close {
                match stack.pop() {
                    Some(top) if top == name => {}
                    _ => return false,
                }
            } else if !self_closing {
                stack.push(name);
            }
        }
    }
    stack.is_empty()
}

/// v7.38 (read01) — parse a float8 the way PG's `float8in` does: a
/// numeric literal that overflows to ±∞, or a nonzero magnitude that
/// underflows to 0, is "out of range" (returns None → the caller errors),
/// not a silent Infinity/0. The `inf`/`infinity`/`nan` spellings (a letter
/// after the optional sign) are the legitimate special values and pass.
pub(crate) fn parse_float8(s: &str) -> Option<f64> {
    let t = s.trim();
    let parsed = t.parse::<f64>().ok()?;
    let body = t.strip_prefix(['+', '-']).unwrap_or(t);
    let numeric_looking = body
        .bytes()
        .next()
        .is_some_and(|c| c.is_ascii_digit() || c == b'.');
    if numeric_looking {
        if parsed.is_infinite() {
            return None; // overflow
        }
        if parsed == 0.0 {
            // A mantissa with a nonzero digit that resolves to 0 underflowed.
            let mantissa = body.split(['e', 'E']).next().unwrap_or(body);
            if mantissa.bytes().any(|c| c.is_ascii_digit() && c != b'0') {
                return None;
            }
        }
    }
    Some(parsed)
}

/// v7.38 (read01) — decode PG's external array form (`{a,b,NULL}`) and coerce
/// each element to `elem` through `coerce_value`, so element semantics (bool
/// spellings, date formats, numeric parsing, float8 range) live in one place.
fn decode_array_elems(
    s: &str,
    elem: DataType,
    col_name: &str,
    position: usize,
) -> Result<Vec<Option<Value<'static>>>, EngineError> {
    // v7.39 (round 325, V57) — PG's wording. This path used to answer
    // `cannot parse "abc" as an array: TEXT[] literal must be enclosed in
    // '{...}'` — SPG's own phrasing, naming TEXT[] even for an INT[]
    // column, and differing from what the `::int[]` CAST path already
    // said for the very same input.
    let raw = decode_text_array_literal(s).map_err(|_| {
        EngineError::Eval(EvalError::TypeMismatch {
            detail: malformed_array_literal(s),
        })
    })?;
    let mut out = Vec::with_capacity(raw.len());
    for e in raw {
        match e {
            None => out.push(None),
            Some(t) => out.push(Some(coerce_value(
                Value::text(t),
                elem,
                col_name,
                position,
            )?)),
        }
    }
    Ok(out)
}

/// v7.39 (read01 round 54) — coerce a value whose `data_type()` is None (the
/// eval-only variants: RegClass carries an oid + name, Composite a field
/// tuple). They used to panic in `coerce_value`.
fn coerce_untyped_value(
    v: Value<'static>,
    expected: DataType,
    col_name: &str,
    position: usize,
) -> Result<Value<'static>, EngineError> {
    match (&v, expected) {
        // A regclass IS an oid — it coerces to any integer width, and to text
        // through its relation name.
        //
        // v7.39 (round 667) — `DataType::Oid` is listed with BigInt here and
        // is not optional. Giving the `oid` name its own DataType turned
        // `'text'::regtype::oid` from a coercion into a column of type
        // BIGINT into one of type OID, and nine catalog tests went red at
        // once. This is the THIRD list that has to name the reg* trio
        // together; the other two are the bigint materialiser below and the
        // classifier that decides a value is reg-shaped.
        (
            Value::RegClass(oid, _) | Value::RegProc(oid, _) | Value::RegType(oid, _),
            DataType::BigInt | DataType::Oid,
        ) => Ok(Value::BigInt(*oid)),
        (
            Value::RegClass(oid, _) | Value::RegProc(oid, _) | Value::RegType(oid, _),
            DataType::Int,
        ) => Ok(Value::Int(i32::try_from(*oid).unwrap_or(i32::MAX))),
        (
            Value::RegClass(_, name) | Value::RegProc(_, name) | Value::RegType(_, name),
            DataType::Text,
        ) => Ok(Value::text(alloc::string::String::from(name.as_ref()))),
        // v7.39 (read01 round 55) — SPG stores a composite-typed column as
        // JSON (an object keyed by field name), so a real Composite value —
        // which is what `ROW(1,2)::pt` now produces — coerces into it. Before
        // this the cast resolved but the INSERT died on "cannot coerce
        // Composite(...) to Jsonb".
        (Value::Composite(fields), DataType::Jsonb | DataType::Json) => {
            let mut obj = alloc::string::String::from("{");
            for (i, (name, val)) in fields.iter().enumerate() {
                if i > 0 {
                    obj.push(',');
                }
                // Reuse the JSON encoder for the key so escaping is identical.
                obj.push_str(&crate::json::value_to_json_text(&Value::text(
                    alloc::string::String::from(name.as_str()),
                )));
                obj.push(':');
                obj.push_str(&crate::json::value_to_json_text(val));
            }
            obj.push('}');
            Ok(Value::Json(alloc::borrow::Cow::Owned(obj)))
        }
        // …and its canonical PG text form for a text column.
        (Value::Composite(_), DataType::Text) => Ok(Value::text(crate::eval::value_to_text(&v))),
        _ => Err(EngineError::Unsupported(alloc::format!(
            "cannot coerce {:?} to {expected:?} for column {col_name:?} (position {position})",
            v
        ))),
    }
}

/// v7.39 (read01 round 90) — PG's 22P02 for a text value that will not parse as
/// the target type: `invalid input syntax for type <T>: "<value>"`. The type
/// word is PG's own spelling (`integer`, `double precision`, `boolean`, …).
fn invalid_input_syntax(ty: &str, value: &str) -> EngineError {
    EngineError::Eval(EvalError::TypeMismatch {
        detail: alloc::format!("invalid input syntax for type {ty}: \"{value}\""),
    })
}

/// v7.39 (round 269) — PG quotes the offending source when it has one:
/// `"1e40" is out of range for type real`.
fn real_out_of_range(value: &str) -> EngineError {
    float_out_of_range(value, "real")
}

/// v7.39 (round 270) — the same for either float width.
fn float_out_of_range(value: &str, ty: &str) -> EngineError {
    EngineError::Eval(EvalError::TypeMismatch {
        detail: alloc::format!("\"{value}\" is out of range for type {ty}"),
    })
}

/// v7.39 (round 270) — a float text that `parse_float8` rejected is
/// either not a number at all or a number outside the type's range, and
/// PG words the two differently. `parse_float8` already distinguishes
/// them internally (it returns None for a numeric-looking infinity or a
/// nonzero mantissa that underflowed to zero); this recovers which.
fn float_text_error(s: &str, ty: &str) -> EngineError {
    let t = s.trim();
    let body = t.strip_prefix(['+', '-']).unwrap_or(t);
    let numeric_looking = body
        .bytes()
        .next()
        .is_some_and(|c| c.is_ascii_digit() || c == b'.');
    if numeric_looking && t.parse::<f64>().is_ok() {
        float_out_of_range(t, ty)
    } else {
        invalid_input_syntax(ty, s)
    }
}

/// Whether a float text names a nonzero value: a mantissa carrying any
/// digit other than 0. Underflowing such a source to zero is an error
/// in PG, while `'0'` really is zero.
fn float_text_is_nonzero(t: &str) -> bool {
    let body = t.strip_prefix(['+', '-']).unwrap_or(t);
    let mantissa = body.split(['e', 'E']).next().unwrap_or(body);
    mantissa.bytes().any(|c| c.is_ascii_digit() && c != b'0')
}

/// Whether a float text literally spells an infinity, which PG accepts
/// as a value rather than treating as an overflow.
fn text_is_explicit_infinity(t: &str) -> bool {
    let t = t.trim_start_matches(['+', '-']);
    t.eq_ignore_ascii_case("inf") || t.eq_ignore_ascii_case("infinity")
}

/// v7.39 (read01 round 90) — PG splits a failed date/time text into two states:
/// a date-shaped string whose fields are out of range (month 13, day 30) is
/// 22008 `date/time field value out of range: "X"`; anything not date-shaped is
/// 22007 `invalid input syntax for type <T>: "X"`. SPG's parsers return a single
/// None, so classify by shape here — runs ONLY on an already-failed parse, so it
/// only ever picks between two error strings, never changes behaviour. A string
/// of date punctuation (digits, `- / : . space`, `+`, `T`) with at least one
/// digit is treated as "well-formed but out of range".
fn datetime_parse_error(ty: &str, s: &str) -> EngineError {
    let t = s.trim();
    let date_shaped = t.chars().any(|c| c.is_ascii_digit())
        && t.chars().all(|c| {
            c.is_ascii_digit() || matches!(c, '-' | '/' | ':' | '.' | ' ' | '+' | 'T' | 't')
        });
    let detail = if date_shaped {
        alloc::format!("date/time field value out of range: \"{t}\"")
    } else {
        alloc::format!("invalid input syntax for type {ty}: \"{t}\"")
    };
    EngineError::Eval(EvalError::TypeMismatch { detail })
}

/// v7.39 (read01 round 113) — the underlying scalar of a `jsonb` value being
/// cast to a numeric or boolean target. PG decodes it first: a JSON number
/// becomes an unconstrained NUMERIC (so int targets round half-away, matching
/// `2.5::numeric::int` = 3), true/false become bool, `null` becomes SQL NULL.
/// A JSON string / array / object is not castable to any scalar target.
pub(crate) enum JsonbScalar {
    Numeric(Value<'static>),
    Bool(bool),
    Null,
}

/// PG's "cannot cast jsonb <kind> to type <target>" (SQLSTATE 22023).
pub(crate) fn jsonb_cast_type_error(kind: &str, target: &str) -> EvalError {
    EvalError::TypeMismatch {
        detail: alloc::format!("cannot cast jsonb {kind} to type {target}"),
    }
}

/// Decode a serialized `jsonb` scalar for a numeric/bool cast. `target` names
/// the SQL type only for the error text on the non-scalar kinds.
pub(crate) fn jsonb_scalar_for_cast(s: &str, target: &str) -> Result<JsonbScalar, EvalError> {
    use crate::json::JsonValue;
    match crate::json::parse(s) {
        Ok(JsonValue::Null) => Ok(JsonbScalar::Null),
        Ok(JsonValue::Bool(b)) => Ok(JsonbScalar::Bool(b)),
        // Route the number through the unconstrained NUMERIC input path so the
        // integer targets inherit PG's numeric (half-away) rounding + range
        // errors, and scientific / big forms are handled once, centrally.
        Ok(JsonValue::Number(x)) => {
            let num = coerce_value(
                Value::text(alloc::format!("{x}")),
                DataType::Numeric {
                    precision: 0,
                    scale: 0,
                },
                "",
                0,
            )
            .map_err(|e| match e {
                EngineError::Eval(ev) => ev,
                _ => jsonb_cast_type_error("numeric", target),
            })?;
            Ok(JsonbScalar::Numeric(num))
        }
        Ok(JsonValue::NumberText(text)) => {
            let num = coerce_value(
                Value::text(text),
                DataType::Numeric {
                    precision: 0,
                    scale: 0,
                },
                "",
                0,
            )
            .map_err(|e| match e {
                EngineError::Eval(ev) => ev,
                _ => jsonb_cast_type_error("numeric", target),
            })?;
            Ok(JsonbScalar::Numeric(num))
        }
        Ok(JsonValue::String(_)) => Err(jsonb_cast_type_error("string", target)),
        Ok(JsonValue::Array(_)) => Err(jsonb_cast_type_error("array", target)),
        Ok(JsonValue::Object(_)) => Err(jsonb_cast_type_error("object", target)),
        Err(_) => Err(jsonb_cast_type_error("value", target)),
    }
}
/// v7.39 (round 263) — normalise a value being written into a COMPOSITE
/// column before the generic coercion runs.
///
/// A composite column stores JSON keyed by FIELD NAME, and the field
/// names are PG-observable (`row_to_json(col)` keys by them, probed).
/// Two inputs reached the column without ever being labelled by the
/// target type:
///   * `ROW('elm', 999)` carries the constructor's placeholder names
///     `f1`/`f2`, so the stored object had the wrong keys and the read
///     side — which looks fields up BY NAME — rebuilt an all-NULL
///     record: silent data loss, `(elm,999)` came back as `(,)`.
///   * a record TEXT literal (`'("oak ave",111)'`) was stored verbatim,
///     which is not JSON at all, so the read side's parse failed and
///     field access errored.
/// Relabelling through the declared type also COERCES each field to its
/// declared type, which is what refuses `ROW('x','notanint')::addr`.
/// Returns the value untouched for a non-composite column.
pub(crate) fn normalize_composite_for_column(
    v: Value<'static>,
    col: &ColumnSchema,
    catalog: Option<&spg_storage::Catalog>,
) -> Result<Value<'static>, EngineError> {
    let Some(tname) = col.user_composite_type.as_deref() else {
        return Ok(v);
    };
    if matches!(v, Value::Null) {
        return Ok(v);
    }
    // No catalog in scope degrades to the previous behaviour rather than
    // erroring, matching how the read-side rehydration handles it.
    let Some(def) = catalog.and_then(|c| c.composite_types().get(tname)) else {
        return Ok(v);
    };
    // An already-labelled Composite still goes through so its fields get
    // coerced; a Json value is already in storage form.
    if matches!(v, Value::Json(_)) {
        return Ok(v);
    }
    crate::eval::apply_composite_cast_pub(v, def, catalog).map_err(EngineError::Eval)
}

/// Coerce a `jsonb` value to a scalar numeric/bool `expected`. Returns `None`
/// when `expected` is not one of those targets (so the caller falls through to
/// the ordinary coercion table).
fn try_coerce_json_scalar(
    s: &str,
    expected: DataType,
    col_name: &str,
    position: usize,
) -> Option<Result<Value<'static>, EngineError>> {
    let target = match expected {
        DataType::Int => "integer",
        DataType::BigInt => "bigint",
        DataType::SmallInt => "smallint",
        DataType::Numeric { .. } => "numeric",
        DataType::Real => "real",
        DataType::Float => "double precision",
        DataType::Bool => "boolean",
        _ => return None,
    };
    Some(
        (|| match jsonb_scalar_for_cast(s, target).map_err(EngineError::Eval)? {
            JsonbScalar::Null => Ok(Value::Null),
            JsonbScalar::Bool(b) => {
                if matches!(expected, DataType::Bool) {
                    Ok(Value::Bool(b))
                } else {
                    Err(EngineError::Eval(jsonb_cast_type_error("boolean", target)))
                }
            }
            JsonbScalar::Numeric(n) => {
                if matches!(expected, DataType::Bool) {
                    Err(EngineError::Eval(jsonb_cast_type_error("numeric", target)))
                } else {
                    coerce_value(n, expected, col_name, position)
                }
            }
        })(),
    )
}

/// v7.39 (round 367, M20 P2) — in the MySQL dialect a binary-string
/// literal (`0x…` / `X'…'` / `b'…'`, backed by `Value::Bytes`) coerces to
/// the target column like MariaDB does: into a BINARY / BLOB column it
/// stays bytes (handled by `coerce_value` itself); into a NUMERIC column
/// it is the bytes' big-endian integer (`INSERT … VALUES (0x10)` stores
/// 16); into a CHAR / VARCHAR / TEXT column it is the bytes read as a
/// latin-1 string (`0x4546` → 'EF'). A PostgreSQL session never produces
/// a `Value::Bytes` from these literals, so this only fires under the
/// dialect and leaves every other value untouched.
pub(crate) fn mysql_bytes_for_column(
    v: Value<'static>,
    expected: DataType,
    mysql: bool,
) -> Value<'static> {
    if !mysql {
        return v;
    }
    let Value::Bytes(ref b) = v else {
        return v;
    };
    match expected {
        DataType::SmallInt
        | DataType::Int
        | DataType::BigInt
        | DataType::Float
        | DataType::Real
        | DataType::Numeric { .. } => {
            let start = b.len().saturating_sub(16);
            let acc = b[start..]
                .iter()
                .fold(0u128, |a, &x| (a << 8) | u128::from(x));
            if acc <= i64::MAX as u128 {
                #[allow(clippy::cast_possible_truncation)]
                Value::BigInt(acc as i64)
            } else {
                big_literal_to_value(&alloc::format!("{acc}"))
            }
        }
        DataType::Text | DataType::Varchar(_) | DataType::Char(_) => Value::text(
            b.iter()
                .map(|&x| x as char)
                .collect::<alloc::string::String>(),
        ),
        _ => v,
    }
}

/// v7.39 (round 544) — `timetz → time` and `interval → time`.
///
/// Measured on PG18:
///
/// ```text
///     '10:20:30.5'::timetz::time      10:20:30.5   (the zone is dropped,
///                                                   the wall clock kept)
///     '25:00:00'::interval::time      01:00:00     (modulo 24 hours)
///     '-1 hour'::interval::time       23:00:00     (and negatives wrap)
///     '1 day 02:00:00'::interval::time 02:00:00    (days do not count)
/// ```
///
/// `time → timetz` and `timestamp(tz) → time` are NOT here: the first
/// needs the session zone to attach, and the second needs to know which
/// of the two timestamp types the source was — `Value::Timestamp` is
/// the same variant for both, so answering from the value would be
/// right for `timestamp` and off by the session offset for
/// `timestamptz`. An error beats a silent wrong answer.
fn try_coerce_time_family(
    v: &Value<'static>,
    expected: DataType,
) -> Option<Result<Value<'static>, EngineError>> {
    const DAY_US: i64 = 86_400_000_000;
    if expected != DataType::Time {
        return None;
    }
    match v {
        Value::TimeTz { us, .. } => Some(Ok(Value::Time(*us))),
        Value::Interval { micros, .. } => Some(Ok(Value::Time(micros.rem_euclid(DAY_US)))),
        _ => None,
    }
}

/// Normalise a value into PG's `oid` domain, or `Ok(None)` when the value is
/// not something an oid can be made from.
///
/// v7.39 (round 667) — extracted rather than copied. The rules lived inline
/// in the `::oid` cast and were already right (a negative wraps the way C's
/// `(Oid)` cast does, past `u32::MAX` is "OID out of range", bad text is
/// PG's 22P02 wording). Assigning INTO an oid column needed the same rules,
/// and round 665 had just finished paying for four hand-copies of one
/// accumulator, so this is one function with two callers instead.
pub(crate) fn coerce_to_oid(v: &Value<'_>) -> Result<Option<Value<'static>>, EvalError> {
    let as_i64 = match v {
        Value::Null => return Ok(Some(Value::Null)),
        Value::SmallInt(n) => i64::from(*n),
        Value::Int(n) => i64::from(*n),
        Value::BigInt(n) => *n,
        Value::Text(t) => match t.trim().parse::<i64>() {
            Ok(n) => n,
            Err(_) => {
                return Err(EvalError::TypeMismatch {
                    detail: alloc::format!("invalid input syntax for type oid: {:?}", t.trim()),
                });
            }
        },
        _ => return Ok(None),
    };
    // 32-bit wrap for negatives (C cast semantics).
    if (-(1i64 << 31)..0).contains(&as_i64) {
        return Ok(Some(Value::BigInt(as_i64 + (1i64 << 32))));
    }
    if !(0..=i64::from(u32::MAX)).contains(&as_i64) {
        return Err(EvalError::TypeMismatch {
            detail: "OID out of range".into(),
        });
    }
    Ok(Some(Value::BigInt(as_i64)))
}

pub(crate) fn coerce_value(
    v: Value<'static>,
    expected: DataType,
    col_name: &str,
    position: usize,
) -> Result<Value<'static>, EngineError> {
    if v.is_null() {
        return Ok(Value::Null);
    }
    // v7.39 (read01 round 113) — a jsonb value cast to a scalar numeric/bool
    // target decodes its underlying JSON scalar first (PG's jsonb → int/bigint/
    // smallint/numeric/real/float8/bool casts). Json → Json still takes the
    // identity fast-path below; this only fires for the scalar targets.
    if let Value::Json(ref s) = v {
        if let Some(res) = try_coerce_json_scalar(s, expected, col_name, position) {
            return res;
        }
    }
    // v7.39 (round 544) — the temporal conversions PG performs and SPG
    // refused outright. Found by comparing a probe of SPG's own cast
    // function against PG18's pg_cast; see synth_pg_cast's note.
    if let Some(res) = try_coerce_time_family(&v, expected) {
        return res;
    }
    // v7.39 (read01 round 54) — `data_type()` is None for the eval-only
    // variants that carry no DataType (RegClass, Composite): they are NOT
    // NULL, so the old `.expect("non-null")` PANICKED on them. A regclass
    // reaching a coercion (e.g. `EXISTS (SELECT 1 WHERE oid_col = 't'::regclass)`,
    // which coerces the subquery's row) crashed the query with an
    // "internal error" instead of comparing by oid. Fall through to the
    // coercion table, which handles the shapes it knows and errors cleanly
    // on the rest.
    // v7.39 (round 254) — a NUMERIC special (NaN / ±Infinity) crossing a
    // cast: every arm below rebuilds its result from `scaled`/`scale`
    // with `kind: Finite`, which silently turned a special into 0
    // (`'Infinity'::numeric::float8` = 0). PG's table, probed live:
    // float8 / real pass the special through; the integer targets refuse
    // it; an unconstrained numeric keeps it, and a typmod'd numeric takes
    // NaN but overflows on an infinity.
    if let Value::Numeric { kind, .. } = v
        && kind != spg_storage::NumericKind::Finite
    {
        use spg_storage::NumericKind as K;
        let as_f64 = match kind {
            K::NaN => f64::NAN,
            K::PosInf => f64::INFINITY,
            K::NegInf => f64::NEG_INFINITY,
            K::Finite => unreachable!("checked above"),
        };
        // PG names any infinity "infinity" here, sign included.
        let what = if kind == K::NaN { "NaN" } else { "infinity" };
        let int_err = |target: &str| {
            Err(EngineError::Eval(EvalError::TypeMismatch {
                detail: alloc::format!("cannot convert {what} to {target}"),
            }))
        };
        match expected {
            DataType::Float => return Ok(Value::Float(as_f64)),
            #[allow(clippy::cast_possible_truncation)]
            DataType::Real => return Ok(Value::Real(as_f64 as f32)),
            DataType::Int => return int_err("integer"),
            DataType::BigInt => return int_err("bigint"),
            DataType::SmallInt => return int_err("smallint"),
            DataType::Numeric { precision, scale } => {
                // Unconstrained numeric (the 0/0 sentinel) keeps the
                // special; a declared precision overflows on an infinity
                // but still accepts NaN (PG: NaN has no magnitude).
                if precision != 0 && kind != K::NaN {
                    return Err(EngineError::Eval(EvalError::TypeMismatch {
                        detail: alloc::string::String::from("numeric field overflow"),
                    }));
                }
                let _ = scale;
                return Ok(v);
            }
            _ => {}
        }
    }
    // v7.39 (round 254) — the reverse direction: an IEEE special arriving
    // from float8 / real becomes the NUMERIC special (PG accepts it since
    // 14); the finite path below cannot represent one.
    if let DataType::Numeric { precision, .. } = expected {
        let f = match v {
            Value::Float(f) if !f.is_finite() => Some(f),
            #[allow(clippy::cast_lossless)]
            Value::Real(f) if !f.is_finite() => Some(f as f64),
            _ => None,
        };
        if let Some(f) = f {
            use spg_storage::NumericKind as K;
            if f.is_nan() {
                return Ok(Value::numeric_special(K::NaN));
            }
            if precision != 0 {
                return Err(EngineError::Eval(EvalError::TypeMismatch {
                    detail: alloc::string::String::from("numeric field overflow"),
                }));
            }
            return Ok(Value::numeric_special(if f > 0.0 {
                K::PosInf
            } else {
                K::NegInf
            }));
        }
    }
    let Some(actual) = v.data_type() else {
        return coerce_untyped_value(v, expected, col_name, position);
    };
    if actual == expected {
        return Ok(v);
    }
    // v7.38.8 — text reaching a json/jsonb column is validated here, the
    // way PG validates at its own input boundary, and reports what PG
    // reports when it will not parse.
    //
    // It was not validated at all, and the comment where the coercion
    // used to live said so outright: "no structural validation — the
    // responsibility for valid JSON lies with the producer". The jsonb
    // side went further and swallowed the parse error, storing the raw
    // text when canonicalisation failed. So `INSERT INTO t VALUES
    // ('{bad')` into a jsonb column was accepted where PG18 answers
    // `invalid input syntax for type json`, and every later read of
    // that row raised instead — including, in v7.38.7, one on the
    // checkpoint thread, which is the worst place for it: writes keep
    // being acknowledged while nothing reaches disk.
    //
    // Handled ahead of the match so the message names the real problem.
    // Reported through the generic path it read `expected Jsonb, actual
    // Text`, which describes a coercion that is ordinarily fine and
    // says nothing about the document being malformed.
    //
    // This boundary is also what the accessors now rest on: with it
    // enforced, a `Value::Json` is valid by construction, and `->>`
    // stops parsing the whole document once per row to find that out.
    if matches!(expected, DataType::Json | DataType::Jsonb)
        && let Value::Text(ref s) | Value::Json(ref s) = v
    {
        let bad = || {
            EngineError::Eval(crate::eval::EvalError::TypeMismatch {
                detail: alloc::string::String::from("invalid input syntax for type json"),
            })
        };
        return if expected == DataType::Jsonb {
            crate::json::canonicalize_jsonb(s.as_ref())
                .map(Value::json)
                .map_err(|_| bad())
        } else {
            crate::json::parse(s.as_ref())
                .map_err(|_| bad())
                .map(|_| Value::json(s.clone()))
        };
    }
    let coerced: Option<Value<'static>> = match (v, expected) {
        (Value::Int(n), DataType::BigInt) => Some(Value::BigInt(i64::from(n))),
        (Value::Int(n), DataType::Float) => Some(Value::Float(f64::from(n))),
        // v7.39 (read01 int.c) — a narrowing overflow is PG's typed
        // "smallint out of range" (22003), not a generic type mismatch.
        (Value::Int(n), DataType::SmallInt) => match i16::try_from(n) {
            Ok(v) => Some(Value::SmallInt(v)),
            Err(_) => {
                return Err(EngineError::Eval(EvalError::TypeMismatch {
                    detail: "smallint out of range".into(),
                }));
            }
        },
        (Value::Int(n), DataType::Numeric { precision, scale }) => Some(numeric_from_integer(
            i128::from(n),
            precision,
            scale,
            col_name,
        )?),
        (Value::SmallInt(n), DataType::Int) => Some(Value::Int(i32::from(n))),
        (Value::SmallInt(n), DataType::BigInt) => Some(Value::BigInt(i64::from(n))),
        (Value::SmallInt(n), DataType::Float) => Some(Value::Float(f64::from(n))),
        (Value::SmallInt(n), DataType::Numeric { precision, scale }) => Some(numeric_from_integer(
            i128::from(n),
            precision,
            scale,
            col_name,
        )?),
        (Value::BigInt(n), DataType::Int) => match i32::try_from(n) {
            Ok(v) => Some(Value::Int(v)),
            Err(_) => {
                return Err(EngineError::Eval(EvalError::TypeMismatch {
                    detail: "integer out of range".into(),
                }));
            }
        },
        (Value::BigInt(n), DataType::SmallInt) => match i16::try_from(n) {
            Ok(v) => Some(Value::SmallInt(v)),
            Err(_) => {
                return Err(EngineError::Eval(EvalError::TypeMismatch {
                    detail: "smallint out of range".into(),
                }));
            }
        },
        #[allow(clippy::cast_precision_loss)]
        (Value::BigInt(n), DataType::Float) => Some(Value::Float(n as f64)),
        (Value::BigInt(n), DataType::Numeric { precision, scale }) => Some(numeric_from_integer(
            i128::from(n),
            precision,
            scale,
            col_name,
        )?),
        (Value::Float(x), DataType::Numeric { precision, scale }) => {
            // Unconstrained `numeric` (precision 0 is the sentinel —
            // numeric(0,0) is invalid in PG) keeps the value's
            // natural scale instead of truncating to 0 decimals.
            // Route the float through its shortest round-trip decimal
            // text so `3.14::numeric` stays 3.14, not 3.
            if precision == 0 && scale == 0 && x.is_finite() {
                if let Some((mantissa, src_scale)) = parse_numeric_text(&alloc::format!("{x}")) {
                    Some(Value::Numeric {
                        scaled: mantissa,
                        scale: src_scale,
                        kind: spg_storage::NumericKind::Finite,
                    })
                } else {
                    Some(numeric_from_float(x, precision, scale, col_name)?)
                }
            } else {
                Some(numeric_from_float(x, precision, scale, col_name)?)
            }
        }
        // v7.39 (read01 round 110) — REAL (float4) → NUMERIC. Mirrors the
        // Float arm above; `real::numeric` used to have no arm at all, so the
        // value stayed a REAL and the column check rejected it. Format the f32
        // via its OWN shortest round-trip decimal (not through f64) so
        // `0.1::real::numeric` matches PG's float4 text.
        (Value::Real(x), DataType::Numeric { precision, scale }) => {
            if precision == 0 && scale == 0 && x.is_finite() {
                // v7.39 (round 662) — SIX significant digits, PG's `FLT_DIG`.
                // `format!("{x}")` is Rust's shortest round-trip, up to nine
                // digits for f32 — right for `real::text`, wrong here.
                // `real::numeric` is a different rule and PG measurably takes
                // the shorter one: `12345.678::real::numeric` is `12345.7`,
                // `1.23456789::real::numeric` is `1.23457`,
                // `123456789::real::numeric` is `123457000`. SPG answered
                // `12345.678`, `1.2345679`, `123456790` — more digits than a
                // float4 carries, presented as if it did.
                //
                // Found while adding `to_char(real, …)`: PG routes that
                // through numeric, not float8, so the missing overload was the
                // symptom and this cast was the cause.
                let six = alloc::format!("{:.5e}", x);
                let six: f64 = six.parse().unwrap_or_else(|_| f64::from(x));
                if let Some((mantissa, src_scale)) = parse_numeric_text(&alloc::format!("{six}")) {
                    Some(Value::Numeric {
                        scaled: mantissa,
                        scale: src_scale,
                        kind: spg_storage::NumericKind::Finite,
                    })
                } else {
                    Some(numeric_from_float(
                        f64::from(x),
                        precision,
                        scale,
                        col_name,
                    )?)
                }
            } else {
                Some(numeric_from_float(
                    f64::from(x),
                    precision,
                    scale,
                    col_name,
                )?)
            }
        }
        // v7.17.0 Phase 3.P0-67 — Text → NUMERIC. Parse a
        // canonical decimal text (`"-1234.56"` / `"42"` /
        // `"0.0001"`) into `(mantissa, source_scale)` and rescale
        // to the column's declared scale. Required for prepared
        // binds: `value_to_literal` flattens a Value::Numeric
        // into a TEXT literal because Literal carries no native
        // Numeric variant, so the placeholder substitution path
        // reaches coerce_value as Text → Numeric. Without this
        // arm the round-trip surfaces a TypeMismatch even though
        // the cell already left the engine as a valid Numeric.
        (Value::Text(s), DataType::Numeric { precision, scale }) => {
            // v7.38 (read01, T6) — PG's NUMERIC specials (`'NaN'`, `'Infinity'`,
            // `'-Infinity'`) parse before the ordinary decimal path.
            if let Some(kind) = crate::numeric::parse_numeric_special(&s) {
                return Ok(Value::numeric_special(kind));
            }
            let Some((mantissa, src_scale)) = parse_numeric_text(&s) else {
                // v7.39 (read01 numeric.c) — PG's numeric input accepts
                // scientific notation ('1e300'::numeric): expand the exponent
                // and re-enter this arm with the plain form (which no longer
                // contains an 'e', so this recurses at most once).
                match spg_sql::parser::expand_scientific_literal(&s) {
                    spg_sql::parser::SciExpanded::Expanded(plain) => {
                        return coerce_value(
                            Value::Text(plain.into()),
                            DataType::Numeric { precision, scale },
                            col_name,
                            position,
                        );
                    }
                    spg_sql::parser::SciExpanded::Overflow => {
                        return Err(EngineError::Eval(EvalError::TypeMismatch {
                            detail: "value overflows numeric format".into(),
                        }));
                    }
                    spg_sql::parser::SciExpanded::NotScientific => {}
                }
                // A plain decimal whose mantissa overflows i128 is still a
                // valid unconstrained NUMERIC — keep it exact as NumericBig.
                if precision == 0 && scale == 0 {
                    if let Some(b) = spg_storage::bignum::BigNumeric::from_decimal_str(&s) {
                        return Ok(Value::NumericBig(alloc::boxed::Box::new(b)));
                    }
                }
                return Err(EngineError::Eval(EvalError::TypeMismatch {
                    detail: alloc::format!("invalid input syntax for type numeric: \"{s}\""),
                }));
            };
            // Unconstrained `numeric` keeps the parsed scale as-is.
            if precision == 0 && scale == 0 {
                Some(Value::Numeric {
                    scaled: mantissa,
                    scale: src_scale,
                    kind: spg_storage::NumericKind::Finite,
                })
            } else {
                Some(numeric_rescale(
                    mantissa, src_scale, precision, scale, col_name,
                )?)
            }
        }
        // Text → DATE / TIMESTAMP: parse canonical text forms.
        (Value::Text(s), DataType::Date) => {
            // PG truncates a full timestamp string on the way into a
            // DATE column (verified vs live PG18.4: INSERT
            // '2020-01-01 12:00:00' into a date column stores
            // 2020-01-01). Try the plain date parser first, then fall
            // back to the timestamp parser (validates the time) floored
            // to the day — mirroring the ::date cast path.
            let d = eval::parse_date_literal(&s)
                .or_else(|| {
                    eval::parse_timestamp_literal(&s)
                        .and_then(|t| i32::try_from(t.div_euclid(86_400_000_000)).ok())
                })
                .ok_or_else(|| datetime_parse_error("date", &s))?;
            Some(Value::Date(d))
        }
        // v7.14.0 — MySQL DEFAULT clauses quote integer / float
        // / boolean literals (`DEFAULT '0'`, `DEFAULT '1'`,
        // `DEFAULT '3.14'`, `DEFAULT 'true'`). Coerce the text
        // form to the column's numeric / bool type at DEFAULT-
        // installation time so the storage check sees a typed
        // value. Parse failures fall through to TypeMismatch.
        // PG trims surrounding whitespace on numeric text input, so
        // `'  256  '::int2` / `'  3.14  '::float8` (both of which route
        // through this generic coerce path, unlike `::int` / `::float`
        // that trim in the CAST helper) parse rather than error.
        // v7.39 (read01 round 90) — a text value that fails to parse as the
        // target numeric type is PG's 22P02 `invalid input syntax for type
        // <T>: "<value>"`, not SPG's generic "type mismatch in column …". The
        // Numeric arm above already worded it this way; these matched it now.
        (Value::Text(s), DataType::SmallInt) => Some(Value::SmallInt(
            parse_pg_int(&s)
                .and_then(|n| i16::try_from(n).ok())
                .ok_or_else(|| invalid_input_syntax("smallint", &s))?,
        )),
        (Value::Text(s), DataType::Int) => Some(Value::Int(
            parse_pg_int(&s)
                .and_then(|n| i32::try_from(n).ok())
                .ok_or_else(|| invalid_input_syntax("integer", &s))?,
        )),
        (Value::Text(s), DataType::BigInt) => Some(Value::BigInt(
            parse_pg_int(&s).ok_or_else(|| invalid_input_syntax("bigint", &s))?,
        )),
        // v7.39 (round 640) — `INSERT INTO t(x) VALUES ('11')` into an
        // `xid` column, which is how PG takes one: the literal is
        // unknown-typed and the column's input function reads it. An
        // INTEGER in the same place is refused by both engines — PG
        // has no int-to-xid cast at all, measured.
        (Value::Text(s), DataType::Xid) => Some(Value::Xid(
            s.parse::<u32>()
                .map_err(|_| invalid_input_syntax("xid", &s))?,
        )),
        (Value::Xid(x), DataType::Xid) => Some(Value::Xid(x)),
        (Value::Text(s), DataType::Xid8) => Some(Value::BigInt(
            parse_pg_int(&s).ok_or_else(|| invalid_input_syntax("xid8", &s))?,
        )),
        // `'16'::xid8` evaluates to a BigInt — xid8 has a declared-type
        // identity but no value of its own, the way `xid` has
        // `Value::Xid`. The consequence is that SPG accepts a bigint
        // where PG refuses one ("column is of type xid8 but expression
        // is of type bigint"); closing that needs a `Value::Xid8`, which
        // is its own unit of work.
        (Value::BigInt(n), DataType::Xid8) => Some(Value::BigInt(n)),
        // v7.39 (round 667) — assigning into an OID column. PG takes an
        // integer here (and, measured, refuses the same integer for an xid
        // column); the range and wrap rules are the cast's, shared.
        (ref other, DataType::Oid) => coerce_to_oid(other)?,
        (Value::Text(s), DataType::Float) => {
            // v7.39 (round 270) — a numeric-looking text outside the
            // double range is "out of range", not "invalid input
            // syntax"; PG quotes the source either way.
            Some(Value::Float(
                parse_float8(&s).ok_or_else(|| float_text_error(&s, "double precision"))?,
            ))
        }
        // v7.38 (read01, T-float4) — coerce to REAL narrows to f32.
        (Value::Int(n), DataType::Real) => Some(Value::Real(n as f32)),
        (Value::SmallInt(n), DataType::Real) => Some(Value::Real(f32::from(n))),
        (Value::BigInt(n), DataType::Real) => Some(Value::Real(n as f32)),
        (Value::Float(x), DataType::Real) => {
            // v7.39 (round 269) — narrowing a finite f64 past the f32
            // range overflows; PG words this one "value out of range:
            // overflow" (it has no source text to quote).
            let narrowed = x as f32;
            if narrowed.is_infinite() && x.is_finite() {
                return Err(EngineError::Eval(EvalError::TypeMismatch {
                    detail: "value out of range: overflow".into(),
                }));
            }
            // v7.39 (round 270) — PG names the other end separately.
            if narrowed == 0.0 && x != 0.0 {
                return Err(EngineError::Eval(EvalError::TypeMismatch {
                    detail: "value out of range: underflow".into(),
                }));
            }
            Some(Value::Real(narrowed))
        }
        (
            Value::Numeric {
                scaled,
                scale,
                kind,
            },
            DataType::Real,
        ) => Some(Value::Real(match kind {
            spg_storage::NumericKind::NaN => f32::NAN,
            spg_storage::NumericKind::PosInf => f32::INFINITY,
            spg_storage::NumericKind::NegInf => f32::NEG_INFINITY,
            spg_storage::NumericKind::Finite => {
                let mut div = 1.0f64;
                for _ in 0..scale {
                    div *= 10.0;
                }
                let x = (scaled as f64 / div) as f32;
                // v7.39 (round 270) — same underflow rule at real's
                // (much nearer) bottom end.
                if x == 0.0 && scaled != 0 {
                    return Err(real_out_of_range(&crate::eval::format_numeric(
                        scaled, scale,
                    )));
                }
                x
            }
        })),
        (Value::Real(x), DataType::Float) => Some(Value::Float(f64::from(x))),
        // v7.39 (round 269) — overflowing the f32 range is an ERROR, not
        // an infinity. `parse::<f32>()` reports "1e40" as inf and this
        // used to hand that back, so a value PG rejects arrived as
        // Infinity and every later comparison against it was wrong. An
        // explicitly written infinity still passes; the test is whether
        // the SOURCE said infinity, not whether the result is one.
        (Value::Text(s), DataType::Real) => {
            let t = s.trim();
            let x = t
                .parse::<f32>()
                .ok()
                .ok_or_else(|| invalid_input_syntax("real", &s))?;
            if x.is_infinite() && !text_is_explicit_infinity(t) {
                return Err(real_out_of_range(t));
            }
            // v7.39 (round 270) — the other end: a nonzero source that
            // underflows to zero is an error too, not a silent 0.
            if x == 0.0 && float_text_is_nonzero(t) {
                return Err(real_out_of_range(t));
            }
            Some(Value::Real(x))
        }
        // PG boolin accepts any unambiguous prefix of true/false/yes/no,
        // plus on/off/1/0, case-insensitively with surrounding whitespace
        // trimmed. `o` alone is ambiguous (on vs off) → error.
        (Value::Text(s), DataType::Bool) => match s.trim().to_ascii_lowercase().as_str() {
            "0" | "f" | "fa" | "fal" | "fals" | "false" | "n" | "no" | "of" | "off" => {
                Some(Value::Bool(false))
            }
            "1" | "t" | "tr" | "tru" | "true" | "y" | "ye" | "yes" | "on" => {
                Some(Value::Bool(true))
            }
            _ => return Err(invalid_input_syntax("boolean", &s)),
        },
        // v7.17.0 Phase 3.P0-46 — MySQL TINYINT(1) (which Phase 4.3
        // classifies as DataType::Bool) is the storage shape every
        // mysqldump-restored boolean column lands in. mysqldump emits
        // the values as integer `0` / `1` literals, so int → bool
        // coerce on INSERT is required for a 0-change cutover. MySQL's
        // rule is "any non-zero is truthy"; we follow that for all
        // signed int widths so the same coerce path serves an
        // explicit `BOOLEAN` column too.
        (Value::Int(n), DataType::Bool) => Some(Value::Bool(n != 0)),
        (Value::SmallInt(n), DataType::Bool) => Some(Value::Bool(n != 0)),
        (Value::BigInt(n), DataType::Bool) => Some(Value::Bool(n != 0)),
        // v7.38.8 — text reaching a json/jsonb column is validated, the
        // way PG validates at its own input boundary.
        //
        // It was not, and the comment here said so: "no structural
        // validation — the responsibility for valid JSON lies with the
        // producer". The jsonb arm went further and swallowed the parse
        // error, storing the raw text when canonicalisation failed. So
        // `INSERT INTO t VALUES ('{bad')` into a jsonb column was
        // accepted (PG18: `invalid input syntax for type json`), and
        // every later read of that row raised instead — including, in
        // v7.38.7, one on the checkpoint thread, which is the worst
        // place for it because writes keep being acknowledged while
        // nothing reaches disk.
        //
        // Rejecting here is also what lets the accessors trust a
        // `Value::Json`: with the column boundary enforced, a value
        // that came out of storage IS valid, and `->>` no longer has
        // to parse the whole document per row to find that out.
        // Text → json/jsonb is handled before this match, so that a
        // document that will not parse reports PG's own message
        // instead of a type mismatch that misnames the problem.
        (Value::Json(s), DataType::Text) => Some(Value::text(s)),
        // v7.13.3 — mailrs round-7 S10. SPG's storage represents
        // both JSON and JSONB on-disk as `Value::json(String)` —
        // they share the underlying text payload. The cast
        // `'<text>'::jsonb` produces a Value::Json that needs to
        // satisfy a DataType::Jsonb column. Identity coerce in
        // both directions so JSON ↔ JSONB assignments work at all
        // INSERT / ALTER COLUMN TYPE / DEFAULT contexts.
        (Value::Json(s), DataType::Json) => Some(Value::json(s)),
        (Value::Json(s), DataType::Jsonb) => Some(Value::json(
            crate::json::canonicalize_jsonb(s.as_ref()).unwrap_or_else(|_| s.into_owned()),
        )),
        // v7.10.4 — Text → BYTEA. Decode PG-style literal forms:
        //   - Hex:    `\x48656c6c6f`  (case-insensitive hex pairs)
        //   - Escape: `Hello\\000world`  (backslash + octal triples)
        //   - Plain:  any string → raw UTF-8 bytes (PG also accepts)
        // Errors surface as TypeMismatch so the operator gets a
        // clear "this literal isn't a bytea literal" hint.
        (Value::Text(s), DataType::Bytes) => {
            let bytes = decode_bytea_literal(&s)
                .map_err(|e| EngineError::Eval(EvalError::TypeMismatch { detail: e }))?;
            Some(Value::bytes(bytes))
        }
        // v7.10.4 — BYTEA → Text round-trip uses the PG hex
        // output (lowercase, `\x` prefix). Important when a
        // SELECT pulls a bytea cell through a Text column path.
        (Value::Bytes(b), DataType::Text) => Some(Value::text(encode_bytea_hex(&b))),
        // v7.17.0 — Text → UUID. PG accepts canonical hyphenated,
        // unhyphenated, uppercase, and `{...}`-braced forms; we
        // funnel all four through `spg_storage::parse_uuid_str`.
        // A malformed literal surfaces as a SQL TypeMismatch
        // rather than silently inserting garbage — `0-change
        // cutover` requires that an app inserting bad UUID text
        // sees the same hard error PG would raise.
        (Value::Text(s), DataType::Uuid) => match spg_storage::parse_uuid_str(&s) {
            Some(b) => Some(Value::Uuid(b)),
            None => {
                return Err(EngineError::Eval(EvalError::TypeMismatch {
                    detail: alloc::format!("invalid input syntax for type uuid: {s:?}"),
                }));
            }
        },
        // v7.17.0 — UUID → Text canonical 8-4-4-4-12 lowercase.
        // Surfaces when a SELECT plucks a uuid cell through a
        // Text column path (e.g. INSERT INTO log SELECT id::text
        // FROM other_table).
        (Value::Uuid(b), DataType::Text) => Some(Value::text(spg_storage::format_uuid(&b))),
        // v7.17.0 Phase 3.P0-32 — Text → TIME. Accepts
        // `HH:MM:SS` and `HH:MM:SS.ffffff` (1-6 fractional digits).
        // Out-of-range hour/min/sec is a hard SQL error (no
        // silent truncation — same 0-change-cutover discipline
        // we apply to UUID).
        (Value::Text(s), DataType::Time) => match parse_time_str(&s) {
            Some(us) => Some(Value::Time(us)),
            None => {
                // v7.39 (round 764, F31 tranche 3 #81) — PG splits the
                // refusals: a time-SHAPED literal with an impossible
                // component (`25:00:00`, `10:61:00`) is "date/time
                // field value out of range" (22008-family), only junk
                // is "invalid input syntax" (PG18-measured).
                let time_shaped = {
                    let core = s.trim().split('.').next().unwrap_or("");
                    !core.is_empty()
                        && core.split(':').count() >= 2
                        && core
                            .split(':')
                            .all(|p| !p.is_empty() && p.chars().all(|c| c.is_ascii_digit()))
                };
                let detail = if time_shaped {
                    alloc::format!("date/time field value out of range: {s:?}")
                } else {
                    alloc::format!("invalid input syntax for type time: {s:?}")
                };
                return Err(EngineError::Eval(EvalError::TypeMismatch { detail }));
            }
        },
        // v7.17.0 Phase 3.P0-32 — TIME → Text canonical `HH:MM:SS[.ffffff]`.
        (Value::Time(us), DataType::Text) => Some(Value::text(eval::format_time(us))),
        // v7.17.0 Phase 3.P0-33 — int / bigint → YEAR. Range
        // check enforces the MySQL canonical 1901..=2155 + 0
        // sentinel; out-of-range is a hard SQL error (no silent
        // truncation, mirrors P0-32 / P0-25 discipline).
        (Value::SmallInt(n), DataType::Year) => Some(coerce_int_to_year(i64::from(n), col_name)?),
        (Value::Int(n), DataType::Year) => Some(coerce_int_to_year(i64::from(n), col_name)?),
        (Value::BigInt(n), DataType::Year) => Some(coerce_int_to_year(n, col_name)?),
        // Text → YEAR. Accepts the 4-digit decimal form only;
        // two-digit YEAR (`'99'` → 1999) was deprecated in MySQL
        // 5.7 and is out of scope for v7.17.0.
        (Value::Text(s), DataType::Year) => match s.trim().parse::<i64>() {
            Ok(n) => Some(coerce_int_to_year(n, col_name)?),
            Err(_) => {
                return Err(EngineError::Eval(EvalError::TypeMismatch {
                    detail: alloc::format!("invalid input syntax for type year: {s:?}"),
                }));
            }
        },
        // YEAR → Text 4-digit zero-padded.
        (Value::Year(y), DataType::Text) => Some(Value::text(alloc::format!("{y:04}"))),
        // v7.17.0 Phase 3.P0-34 — Text → TIMETZ.
        // v7.39 (round 761, F31 tranche 2 #59) — an offset-less
        // literal is accepted at the session offset, PG18-measured
        // (`INSERT '07:08:09'` into a TIMETZ column reads back
        // `07:08:09+00` in a UTC session). The old "mandatory signed
        // offset" rule refused what PG accepts; offset 0 is the same
        // session-zero assumption the time→timetz cast below carries.
        // v7.39 (round 634) — a time or a timestamp reaching `::TIMETZ`.
        // PG registers time -> timetz as IMPLICIT and timestamptz -> timetz
        // as an assignment cast; SPG answered "cannot cast time without
        // time zone to USER-DEFINED", the target having fallen through to
        // the user-type lookup. The session offset is zero here, which is
        // what SPG's timetz values already carry.
        (Value::Time(t), DataType::TimeTz) => Some(Value::TimeTz {
            us: t,
            offset_secs: 0,
        }),
        (Value::Timestamp(t), DataType::TimeTz) => Some(Value::TimeTz {
            us: t.rem_euclid(86_400_000_000),
            offset_secs: 0,
        }),
        (Value::Text(s), DataType::TimeTz) => {
            match parse_timetz_str(&s).or_else(|| parse_time_str(s.trim()).map(|us| (us, 0))) {
                Some((us, offset_secs)) => Some(Value::TimeTz { us, offset_secs }),
                None => {
                    return Err(EngineError::Eval(EvalError::TypeMismatch {
                        detail: alloc::format!(
                            "invalid input syntax for type time with time zone: \
                         {s:?}"
                        ),
                    }));
                }
            }
        }
        // TIMETZ → Text canonical `HH:MM:SS[.ffffff]±HH[:MM]`.
        (Value::TimeTz { us, offset_secs }, DataType::Text) => {
            Some(Value::text(eval::format_timetz(us, offset_secs)))
        }
        // v7.17.0 Phase 3.P0-35 — Text → MONEY. Accepts `$N.NN`,
        // `$N,NNN.NN`, optional leading `-`. Bare numeric literals
        // arrive via the Int/BigInt/Float/Numeric arms below.
        (Value::Text(s), DataType::Money) => match parse_money_str(&s) {
            Some(c) => Some(Value::Money(c)),
            None => {
                return Err(EngineError::Eval(EvalError::TypeMismatch {
                    detail: alloc::format!("invalid input syntax for type money: {s:?}"),
                }));
            }
        },
        // Int / BigInt / SmallInt / Float / Numeric → MONEY.
        // Bare numeric literal is interpreted as a major-unit
        // amount (matches PG: `100`::money → $100.00 = 10000 cents).
        (Value::SmallInt(n), DataType::Money) => {
            Some(Value::Money(i64::from(n).saturating_mul(100)))
        }
        (Value::Int(n), DataType::Money) => Some(Value::Money(i64::from(n).saturating_mul(100))),
        (Value::BigInt(n), DataType::Money) => Some(Value::Money(n.saturating_mul(100))),
        (Value::Float(x), DataType::Money) => {
            // Round half-away-from-zero to cents (no_std — no
            // `f64::round`, so hand-roll via biased truncation).
            let scaled = x * 100.0;
            let cents = if scaled >= 0.0 {
                (scaled + 0.5) as i64
            } else {
                (scaled - 0.5) as i64
            };
            Some(Value::Money(cents))
        }
        (Value::Numeric { scaled, scale, .. }, DataType::Money) => {
            // Convert exact decimal to cents (scale 2). If scale > 2,
            // round half-away-from-zero. If scale < 2, multiply up.
            let cents = if scale == 2 {
                scaled
            } else if scale < 2 {
                let mult = 10_i128.pow(u32::from(2 - scale));
                scaled.saturating_mul(mult)
            } else {
                let div = 10_i128.pow(u32::from(scale - 2));
                let half = div / 2;
                let bias = if scaled >= 0 { half } else { -half };
                (scaled + bias) / div
            };
            Some(Value::Money(i64::try_from(cents).unwrap_or(i64::MAX)))
        }
        // MONEY → Text canonical `$N,NNN.CC`.
        (Value::Money(c), DataType::Text) => Some(Value::text(eval::format_money(c))),
        // MONEY → NUMERIC: integer cents become a scale-2 decimal (dollars).
        (Value::Money(c), DataType::Numeric { .. }) => Some(Value::Numeric {
            scaled: i128::from(c),
            scale: 2,
            kind: spg_storage::NumericKind::Finite,
        }),
        // v7.17.0 Phase 3.P0-38 — Text → Range. Accepts canonical
        // PG forms: `'empty'`, `'[a,b)'`, `'(a,b]'`, `'[a,b]'`,
        // `'(a,b)'`, with empty lower or upper for unbounded.
        (Value::Text(s), DataType::Range(kind)) => match parse_range_str(&s, kind) {
            Ok(v) => Some(v),
            // v7.39 (read01 rangetypes.c) — PG's two distinct rejections.
            Err(RangeParseError::Misordered) => {
                return Err(EngineError::Eval(EvalError::TypeMismatch {
                    detail: alloc::string::String::from(
                        "range lower bound must be less than or equal to range upper bound",
                    ),
                }));
            }
            Err(RangeParseError::Malformed) => {
                return Err(EngineError::Eval(EvalError::TypeMismatch {
                    detail: alloc::format!("malformed range literal: \"{s}\""),
                }));
            }
            Err(RangeParseError::BadElement(bad)) => {
                return Err(EngineError::Eval(EvalError::TypeMismatch {
                    detail: alloc::format!(
                        "invalid input syntax for type {}: \"{bad}\"",
                        range_element_type_name(kind)
                    ),
                }));
            }
        },
        // Range → Text canonical form (`[a,b)`, `'empty'`, etc).
        (v @ Value::Range { .. }, DataType::Text) => Some(Value::text(format_range_str(&v))),
        // v7.37.5 ζ-A — Text → network / bit / xml / "char" / money[].
        (Value::Text(s), DataType::Inet) => match parse_inet_text(&s) {
            Some((family, bits, addr)) => Some(Value::Inet { family, bits, addr }),
            None => {
                // v7.39 (round 262) — PG's wording: the lowercase type
                // name and no column suffix (the cidr arm below already
                // had it right).
                return Err(EngineError::Eval(EvalError::TypeMismatch {
                    detail: alloc::format!("invalid input syntax for type inet: {s:?}"),
                }));
            }
        },
        // v7.39 (round 262) — the inet <-> cidr casts, probed live:
        // `inet::cidr` keeps the mask length (defaulting to the family's
        // full width) and ZEROES the host bits, so `192.168.1.5/24`
        // becomes `192.168.1.0/24`; `cidr::inet` passes through
        // unchanged. Neither existed, so both raised a storage type
        // mismatch on perfectly ordinary SQL.
        (Value::Inet { family, bits, addr }, DataType::Cidr) => {
            let full = if family == 6 { 128 } else { 32 };
            let bits = if bits > full { full } else { bits };
            let mut masked = addr;
            for i in 0..16usize {
                let bit_start = i * 8;
                if bit_start >= usize::from(bits) {
                    masked[i] = 0;
                } else if bit_start + 8 > usize::from(bits) {
                    let keep = usize::from(bits) - bit_start;
                    masked[i] &= 0xffu8 << (8 - keep);
                }
            }
            Some(Value::Cidr {
                family,
                bits,
                addr: masked,
            })
        }
        (Value::Cidr { family, bits, addr }, DataType::Inet) => {
            Some(Value::Inet { family, bits, addr })
        }
        (Value::Text(s), DataType::Cidr) => match parse_cidr_text(&s) {
            Ok(Some((family, bits, addr))) => Some(Value::Cidr { family, bits, addr }),
            Err(()) => {
                return Err(EngineError::Eval(EvalError::TypeMismatch {
                    detail: alloc::format!(
                        "invalid cidr value: {s:?} DETAIL: Value has bits set to right of mask."
                    ),
                }));
            }
            Ok(None) => {
                return Err(EngineError::Eval(EvalError::TypeMismatch {
                    detail: alloc::format!("invalid input syntax for type cidr: {s:?}"),
                }));
            }
        },
        // INSERT / assignment of a text literal into an INTERVAL column
        // parses it, matching the `::interval` cast (mirrors macaddr/inet).
        (Value::Text(s), DataType::Interval) => match spg_sql::parser::parse_interval_text(&s) {
            Some((months, days, micros)) => Some(Value::Interval {
                months,
                days,
                micros,
                kind: spg_storage::IntervalKind::from_fields(months, days, micros),
            }),
            None => {
                return Err(EngineError::Eval(EvalError::TypeMismatch {
                    detail: alloc::format!("invalid input syntax for type interval: {s:?}"),
                }));
            }
        },
        (Value::Text(s), DataType::Macaddr) => match parse_macaddr_text(&s) {
            Some(m) => Some(Value::Macaddr(m)),
            None => {
                return Err(EngineError::Eval(EvalError::TypeMismatch {
                    detail: alloc::format!("invalid input syntax for type macaddr: {s:?}"),
                }));
            }
        },
        // v7.39 (read01 pg_lsn.c) — `XX/XX` hex pair, each half <= u32.
        (Value::Text(s), DataType::PgLsn) => match parse_pg_lsn_text(&s) {
            Some(l) => Some(Value::PgLsn(l)),
            None => {
                return Err(EngineError::Eval(EvalError::TypeMismatch {
                    detail: alloc::format!("invalid input syntax for type pg_lsn: \"{s}\""),
                }));
            }
        },
        (Value::Text(s), DataType::Macaddr8) => match parse_macaddr8_text(&s) {
            Some(m) => Some(Value::Macaddr8(m)),
            None => {
                return Err(EngineError::Eval(EvalError::TypeMismatch {
                    detail: alloc::format!("invalid input syntax for type macaddr8: {s:?}"),
                }));
            }
        },
        // v7.37.5 ship triage — `Value::BitString` self-reports as
        // `DataType::BitVarying(0)` (see `Value::data_type`), so an
        // INSERT into a `BIT` column triggered a spurious type
        // mismatch. Accept BitString into either.
        //
        // v7.39 (round 281) — and enforce the declared length, which
        // used to be parsed and dropped so `bit(3)` took a five-bit
        // string. PG's two types differ: BIT is FIXED (a shorter value
        // is an error too) while BIT VARYING is a maximum. An explicit
        // CAST still pads or truncates — the same assignment-enforces /
        // cast-adjusts split the varchar arms below already model.
        (Value::BitString { nbits, bytes }, DataType::Bit(n)) => {
            // A bare `bit` is `bit(1)` in PG.
            let want = if n == 0 { 1 } else { n };
            if nbits != want {
                return Err(EngineError::Unsupported(alloc::format!(
                    "bit string length {nbits} does not match type bit({want})"
                )));
            }
            Some(Value::BitString { nbits, bytes })
        }
        (Value::BitString { nbits, bytes }, DataType::BitVarying(n)) => {
            if n != 0 && nbits > n {
                return Err(EngineError::Unsupported(alloc::format!(
                    "bit string too long for type bit varying({n})"
                )));
            }
            Some(Value::BitString { nbits, bytes })
        }
        (Value::Text(s), bit_ty @ (DataType::Bit(_) | DataType::BitVarying(_))) => {
            match parse_bit_string_text(&s) {
                Some((nbits, bytes)) => {
                    // v7.39 (round 325, V57) — the DECLARED width applies to a
                    // string literal too. It was checked only on the
                    // `B'…'` bit-literal path, so `INSERT INTO t(b)
                    // VALUES ('10')` into a `BIT(3)` column was accepted and
                    // stored two bits wide — a column that promises a fixed
                    // width silently holding another one. PG 18.4:
                    // `bit string length 2 does not match type bit(3)`, and
                    // `bit string too long for type bit varying(3)` past a
                    // varying cap.
                    match bit_ty {
                        // A bare `bit` is `bit(1)` in PG, as the arm above.
                        DataType::Bit(n) => {
                            let want = if n == 0 { 1 } else { n };
                            if nbits != want {
                                return Err(EngineError::Unsupported(alloc::format!(
                                    "bit string length {nbits} does not match type bit({want})"
                                )));
                            }
                        }
                        DataType::BitVarying(n) if n != 0 && nbits > n => {
                            return Err(EngineError::Unsupported(alloc::format!(
                                "bit string too long for type bit varying({n})"
                            )));
                        }
                        _ => {}
                    }
                    Some(Value::bit_string(nbits, bytes))
                }
                None => {
                    // v7.39 (read01 varbit.c) — PG names the first bad digit.
                    let bad = s.chars().find(|c| *c != '0' && *c != '1');
                    return Err(EngineError::Eval(EvalError::TypeMismatch {
                        detail: match bad {
                            Some(c) => {
                                alloc::format!("\"{c}\" is not a valid binary digit")
                            }
                            None => alloc::format!("invalid input syntax for BIT: {s:?}"),
                        },
                    }));
                }
            }
        }
        (Value::Text(s), DataType::Xml) => {
            // v7.38 (read01 P6.38) — `::xml` (PG's CONTENT mode) requires the
            // text to be well-formed: element tags must be balanced and
            // properly nested. Plain text, multiple top-level elements,
            // comments/PIs/CDATA and self-closing tags are all fine.
            if !xml_content_is_well_formed(&s) {
                return Err(EngineError::Eval(EvalError::TypeMismatch {
                    detail: alloc::format!("invalid XML content: {s:?}"),
                }));
            }
            Some(Value::xml(s))
        }
        // v7.39 (round 634) — the bpchar forms of two casts the Text arms
        // above already have. `'ab'::CHAR(4)::"char"` answered "cannot cast
        // character to \"char\"" and `::XML` likewise, while the same value
        // as TEXT worked: the cast path never normalises a bpchar the way
        // the function dispatch does. PG answers `a` and `ab` — the text
        // form of a bpchar drops its padding.
        (Value::BpChar(s), DataType::Char1) => {
            Some(Value::Char1(s.as_bytes().first().copied().unwrap_or(0)))
        }
        (Value::BpChar(s), DataType::Xml) => {
            let stripped = s.trim_end_matches(' ');
            if !xml_content_is_well_formed(stripped) {
                return Err(EngineError::Eval(EvalError::TypeMismatch {
                    detail: alloc::format!("invalid XML content: {stripped:?}"),
                }));
            }
            Some(Value::xml(alloc::string::String::from(stripped)))
        }
        // v7.39 (round 634) — bytea to an integer reads the bytes
        // BIG-ENDIAN, all of them, and errors when the result does not fit.
        // Measured on PG: `'\x3132'` is 12594, a single `'\x31'` is 49, an
        // empty bytea is 0, and three bytes into a smallint is
        // "smallint out of range".
        (Value::Bytes(b), DataType::SmallInt | DataType::Int | DataType::BigInt) => {
            let mut acc: i128 = 0;
            for byte in b.iter() {
                acc = acc.saturating_mul(256).saturating_add(i128::from(*byte));
            }
            let (fits, made) = match expected {
                DataType::SmallInt => (
                    i16::try_from(acc).is_ok(),
                    i16::try_from(acc).map(Value::SmallInt).ok(),
                ),
                DataType::Int => (
                    i32::try_from(acc).is_ok(),
                    i32::try_from(acc).map(Value::Int).ok(),
                ),
                _ => (
                    i64::try_from(acc).is_ok(),
                    i64::try_from(acc).map(Value::BigInt).ok(),
                ),
            };
            if !fits {
                return Err(EngineError::Eval(EvalError::TypeMismatch {
                    detail: alloc::format!("{} out of range", pg_type_name_for_error(expected)),
                }));
            }
            made
        }
        // v7.39 (read01 char.c) — an integer coerces to "char" by its
        // low byte (65::"char" = 'A'; PG's i2char/int4char).
        (Value::Int(n), DataType::Char1) => Some(Value::Char1((n & 0xff) as u8)),
        (Value::SmallInt(n), DataType::Char1) => Some(Value::Char1((n & 0xff) as u8)),
        (Value::BigInt(n), DataType::Char1) => Some(Value::Char1((n & 0xff) as u8)),
        (Value::Text(s), DataType::Char1) => {
            // v7.39 (read01 utils/adt, char.c) — charin accepts the
            // `\ooo` octal form charout produces for high bytes
            // ('\101'::"char" = 'A'); otherwise the FIRST byte, with
            // any remainder silently discarded (PG's compatibility
            // provision); empty = 0x00.
            let bytes = s.as_bytes();
            if bytes.len() == 4
                && bytes[0] == b'\\'
                && bytes[1..].iter().all(|b| (b'0'..=b'7').contains(b))
            {
                let v = ((bytes[1] - b'0') << 6) | ((bytes[2] - b'0') << 3) | (bytes[3] - b'0');
                Some(Value::Char1(v))
            } else {
                let b = s.bytes().next().unwrap_or(0);
                Some(Value::Char1(b))
            }
        }
        // v7.37.5 ζ-A — inverse coerces.
        (Value::Inet { family, bits, addr }, DataType::Text) => {
            // v7.39 (read01 inet family) — PG's text(inet) ALWAYS carries
            // the /netmask (192.168.1.5 -> "192.168.1.5/32"), unlike the
            // display form which suppresses a full-length mask.
            let base = format_inet(family, bits, &addr);
            Some(Value::text(if base.contains('/') {
                base
            } else {
                alloc::format!("{base}/{bits}")
            }))
        }
        (Value::Cidr { family, bits, addr }, DataType::Text) => {
            Some(Value::text(format_inet(family, bits, &addr)))
        }
        (Value::Macaddr(m), DataType::Text) => Some(Value::text(format_macaddr(&m))),
        (Value::Macaddr8(m), DataType::Text) => Some(Value::text(format_macaddr8(&m))),
        (Value::PgLsn(l), DataType::Text) => Some(Value::text(format_pg_lsn(l))),
        // MACADDR → MACADDR8: PG widens EUI-48 to EUI-64 by inserting the
        // `ff:fe` marker in the middle (08:00:2b:01:02:03 → 08:00:2b:ff:fe:01:02:03).
        (Value::Macaddr(m), DataType::Macaddr8) => Some(Value::Macaddr8([
            m[0], m[1], m[2], 0xff, 0xfe, m[3], m[4], m[5],
        ])),
        (Value::BitString { nbits, bytes }, DataType::Text) => {
            Some(Value::text(format_bit_string(nbits, &bytes)))
        }
        // BIT → integer: MSB-first bit value (PG bit→int cast).
        #[allow(clippy::cast_possible_truncation)]
        (Value::BitString { nbits, bytes }, DataType::SmallInt) => {
            Some(Value::SmallInt(bit_string_to_i64(nbits, &bytes) as i16))
        }
        #[allow(clippy::cast_possible_truncation)]
        (Value::BitString { nbits, bytes }, DataType::Int) => {
            Some(Value::Int(bit_string_to_i64(nbits, &bytes) as i32))
        }
        (Value::BitString { nbits, bytes }, DataType::BigInt) => {
            Some(Value::BigInt(bit_string_to_i64(nbits, &bytes)))
        }
        (Value::Xml(s), DataType::Text) => Some(Value::text(s)),
        (Value::Char1(b), DataType::Text) => Some(Value::text((b as char).to_string())),
        // v7.37.5 ε — Text → geometry coerce. Each parser returns
        // None on malformed input; we surface a TypeMismatch with
        // the column name so the engine error is debuggable.
        (Value::Text(s), DataType::Point) => match parse_point(&s) {
            Some(p) => Some(Value::Point(p)),
            None => {
                return Err(EngineError::Eval(EvalError::TypeMismatch {
                    detail: alloc::format!("invalid input syntax for type point: {s:?}"),
                }));
            }
        },
        (Value::Text(s), DataType::Lseg) => match parse_lseg_text(&s) {
            Some((p1, p2)) => Some(Value::Lseg(p1, p2)),
            None => {
                return Err(EngineError::Eval(EvalError::TypeMismatch {
                    detail: alloc::format!("invalid input syntax for type lseg: {s:?}"),
                }));
            }
        },
        (Value::Text(s), DataType::PgBox) => match parse_box_text(&s) {
            Some((ur, ll)) => Some(Value::PgBox(ur, ll)),
            None => {
                return Err(EngineError::Eval(EvalError::TypeMismatch {
                    detail: alloc::format!("invalid input syntax for type box: {s:?}"),
                }));
            }
        },
        (Value::Text(s), DataType::Line) => match parse_line_text(&s) {
            Some((a, b, c)) => Some(Value::Line { a, b, c }),
            None => {
                // v7.39 (round 775, F31 J6) — the degenerate `{0,0,C}`
                // form gets PG's OWN sentence (measured), not the
                // generic syntax one.
                let zero_ab = s
                    .trim()
                    .strip_prefix('{')
                    .and_then(|x| x.strip_suffix('}'))
                    .map(|inner| inner.split(',').collect::<alloc::vec::Vec<_>>())
                    .is_some_and(|parts| {
                        parts.len() == 3
                            && parts[0].trim().parse::<f64>() == Ok(0.0)
                            && parts[1].trim().parse::<f64>() == Ok(0.0)
                            && parts[2].trim().parse::<f64>().is_ok()
                    });
                let detail = if zero_ab {
                    alloc::string::String::from(
                        "invalid line specification: A and B cannot both be zero",
                    )
                } else {
                    alloc::format!("invalid input syntax for type line: {s:?}")
                };
                return Err(EngineError::Eval(EvalError::TypeMismatch { detail }));
            }
        },
        (Value::Text(s), DataType::Circle) => match parse_circle_text(&s) {
            Some((center, radius)) => Some(Value::Circle { center, radius }),
            None => {
                return Err(EngineError::Eval(EvalError::TypeMismatch {
                    detail: alloc::format!("invalid input syntax for type circle: {s:?}"),
                }));
            }
        },
        (Value::Text(s), DataType::Path) => match parse_path_text(&s) {
            Some((points, closed)) => Some(Value::Path { points, closed }),
            None => {
                return Err(EngineError::Eval(EvalError::TypeMismatch {
                    detail: alloc::format!("invalid input syntax for type path: {s:?}"),
                }));
            }
        },
        // v7.39 (read01 geo_ops.c) — box_poly: a box converts to its
        // 4-corner polygon (low, (low.x, high.y), high, (high.x, low.y)).
        (Value::PgBox(a, b), DataType::Polygon) => {
            let (hx, hy) = (a.x.max(b.x), a.y.max(b.y));
            let (lx, ly) = (a.x.min(b.x), a.y.min(b.y));
            let p = |x: f64, y: f64| spg_storage::Point2D { x, y };
            Some(Value::Polygon(alloc::vec![
                p(lx, ly),
                p(lx, hy),
                p(hx, hy),
                p(hx, ly),
            ]))
        }
        (Value::Text(s), DataType::Polygon) => match parse_polygon_text(&s) {
            Some(points) => Some(Value::Polygon(points)),
            None => {
                return Err(EngineError::Eval(EvalError::TypeMismatch {
                    detail: alloc::format!("invalid input syntax for type polygon: {s:?}"),
                }));
            }
        },
        // v7.37.5 ε — geometry → Text canonical forms.
        (Value::Point(p), DataType::Text) => Some(Value::text(format_point(p))),
        (Value::Lseg(p1, p2), DataType::Text) => Some(Value::text(format_lseg(p1, p2))),
        (Value::PgBox(ur, ll), DataType::Text) => Some(Value::text(format_pg_box(ur, ll))),
        (Value::Line { a, b, c }, DataType::Text) => Some(Value::text(format_line(a, b, c))),
        (Value::Circle { center, radius }, DataType::Text) => {
            Some(Value::text(format_circle(center, radius)))
        }
        (Value::Path { points, closed }, DataType::Text) => {
            Some(Value::text(format_path(&points, closed)))
        }
        (Value::Polygon(points), DataType::Text) => Some(Value::text(format_polygon(&points))),
        // v7.37.5 δ — Text → Multirange. Accepts `{}` empty and
        // `{[a,b),[c,d),...}` comma-separated ranges; each
        // subrange parses with the parent kind.
        // v7.39 (round 256) — `range::<type>multirange`: PG casts a range
        // to the one-element multirange containing it (an empty range
        // gives the empty multirange).
        (ref rv @ Value::Range { kind: rk, .. }, DataType::Multirange(kind)) => {
            if rk != kind {
                return Err(EngineError::Eval(EvalError::TypeMismatch {
                    detail: alloc::format!(
                        "cannot cast type {} to {}",
                        DataType::Range(rk),
                        DataType::Multirange(kind)
                    ),
                }));
            }
            crate::eval::binop::range_as_multirange(rv)
        }
        (Value::Text(s), DataType::Multirange(kind)) => match parse_multirange_str(&s, kind) {
            // v7.39 (round 231) — a multirange is normalized whatever built
            // it. The constructor function already sorted / merged / dropped
            // empties; the text cast kept the literal's spans verbatim, so
            // `'{[1,3),[3,5)}'::int4multirange` printed back two adjacent
            // spans where PG prints the merged `{[1,5)}`.
            Some(ranges) => Some(Value::Multirange {
                kind,
                ranges: crate::eval::binop::normalize_multirange_spans(kind, &ranges),
            }),
            None => {
                return Err(EngineError::Eval(EvalError::TypeMismatch {
                    detail: alloc::format!("invalid input syntax for multirange type: {s:?}"),
                }));
            }
        },
        // Multirange → Text canonical form (`{[a,b),[c,d)}`).
        (Value::Multirange { ranges, .. }, DataType::Text) => {
            Some(Value::text(format_multirange(&ranges)))
        }
        // v7.17.0 Phase 3.P0-39 — Text → Hstore.
        (Value::Text(s), DataType::Hstore) => match parse_hstore_str(&s) {
            Some(pairs) => Some(Value::Hstore(pairs)),
            None => {
                return Err(EngineError::Eval(EvalError::TypeMismatch {
                    detail: alloc::format!("invalid input syntax for type hstore: {s:?}"),
                }));
            }
        },
        // Hstore → Text canonical `"k"=>"v"` form.
        (Value::Hstore(pairs), DataType::Text) => Some(Value::text(format_hstore_str(&pairs))),
        // v7.17.0 Phase 3.P0-40 — Text → 2D arrays via PG
        // external `'{{a,b},{c,d}}'` literal.
        (Value::Text(s), DataType::IntArray2D) => match parse_int_2d_literal(&s) {
            Ok(m) => Some(Value::IntArray2D(m)),
            Err(e) => {
                return Err(EngineError::Eval(EvalError::TypeMismatch {
                    detail: alloc::format!("invalid input syntax for INT[][]: {s:?}: {e}"),
                }));
            }
        },
        (Value::Text(s), DataType::BigIntArray2D) => match parse_bigint_2d_literal(&s) {
            Ok(m) => Some(Value::BigIntArray2D(m)),
            Err(e) => {
                return Err(EngineError::Eval(EvalError::TypeMismatch {
                    detail: alloc::format!("invalid input syntax for BIGINT[][]: {s:?}: {e}"),
                }));
            }
        },
        (Value::Text(s), DataType::TextArray2D) => match parse_text_2d_literal(&s) {
            Ok(m) => Some(Value::TextArray2D(m)),
            Err(e) => {
                return Err(EngineError::Eval(EvalError::TypeMismatch {
                    detail: alloc::format!("invalid input syntax for TEXT[][]: {s:?}: {e}"),
                }));
            }
        },
        // 2D arrays → Text canonical nested form.
        (Value::IntArray2D(rows), DataType::Text) => Some(Value::text(format_int_2d_text(&rows))),
        (Value::BigIntArray2D(rows), DataType::Text) => {
            Some(Value::text(format_bigint_2d_text(&rows)))
        }
        (Value::TextArray2D(rows), DataType::Text) => Some(Value::text(format_text_2d_text(&rows))),
        // v7.10.11 — Text → TEXT[]. Decode PG's external array
        // form `'{a,b,NULL}'`. NULL element token (case-insensitive)
        // is the literal `NULL`; everything else is a quoted or
        // unquoted text element. mailrs `'{label1,label2}'::TEXT[]`.
        (Value::Text(s), DataType::TextArray) => {
            // v7.39 (round 325, V57) — PG's wording (and the same message
            // the CAST path gives for the identical input; this one used to
            // name TEXT[] whatever the column's element type was).
            let arr = decode_text_array_literal(&s).map_err(|_| {
                EngineError::Eval(EvalError::TypeMismatch {
                    detail: malformed_array_literal(&s),
                })
            })?;
            Some(Value::TextArray(arr))
        }
        // v7.16.0 — Text → IntArray / BigIntArray for the
        // spg-sqlx Bind path. Decode the PG external form
        // `{1,2,3}` as a TEXT array first, then parse each
        // element as int. Same shape as the TextArray decode
        // above with an element-wise narrow.
        (Value::Text(s), DataType::IntArray) => {
            // v7.39 (round 325, V57) — PG's wording (and the same message
            // the CAST path gives for the identical input; this one used to
            // name TEXT[] whatever the column's element type was).
            let arr = decode_text_array_literal(&s).map_err(|_| {
                EngineError::Eval(EvalError::TypeMismatch {
                    detail: malformed_array_literal(&s),
                })
            })?;
            let mut out: Vec<Option<i32>> = Vec::with_capacity(arr.len());
            for elem in arr {
                match elem {
                    None => out.push(None),
                    Some(t) => {
                        let n: i32 = t.parse().map_err(|_| {
                            EngineError::Eval(EvalError::TypeMismatch {
                                detail: alloc::format!(
                                    "invalid input syntax for type integer: {t:?}"
                                ),
                            })
                        })?;
                        out.push(Some(n));
                    }
                }
            }
            Some(Value::IntArray(out))
        }
        // v7.38 (read01) — the remaining Text → typed-array casts
        // (`'{1.5}'::numeric[]`, `'{t}'::bool[]`, `'{2020-01-01}'::date[]`, …),
        // which previously errored while `::int[]` / `::text[]` worked.
        (Value::Text(s), DataType::SmallIntArray) => Some(Value::SmallIntArray(
            decode_array_elems(&s, DataType::SmallInt, col_name, position)?
                .into_iter()
                .map(|o| match o {
                    Some(Value::SmallInt(n)) => Some(n),
                    _ => None,
                })
                .collect(),
        )),
        (Value::Text(s), DataType::BoolArray) => {
            // v7.39 (read01 round 92) — a 2-D bool literal `{{t,f},{f,t}}`
            // becomes a BoolArray2D (the ::int[]/::text[] cast path learned this
            // separately; the typed-array coerce path routes here). 1-D stays a
            // BoolArray.
            if let Some(rows) = crate::eval::values::split_2d_rows(&s) {
                let mut row_vals: Vec<Value<'static>> = Vec::with_capacity(rows.len());
                for r in &rows {
                    let bools: Vec<Option<bool>> =
                        decode_array_elems(r, DataType::Bool, col_name, position)?
                            .into_iter()
                            .map(|o| match o {
                                Some(Value::Bool(b)) => Some(b),
                                _ => None,
                            })
                            .collect();
                    row_vals.push(Value::BoolArray(bools));
                }
                return crate::eval::values::build_2d_from_rows(&row_vals).ok_or_else(|| {
                    EngineError::Eval(EvalError::TypeMismatch {
                        detail: malformed_array_literal(&s),
                    })
                });
            }
            Some(Value::BoolArray(
                decode_array_elems(&s, DataType::Bool, col_name, position)?
                    .into_iter()
                    .map(|o| match o {
                        Some(Value::Bool(b)) => Some(b),
                        _ => None,
                    })
                    .collect(),
            ))
        }
        (Value::Text(s), DataType::FloatArray) => Some(Value::FloatArray(
            decode_array_elems(&s, DataType::Float, col_name, position)?
                .into_iter()
                .map(|o| match o {
                    Some(Value::Float(f)) => Some(f),
                    _ => None,
                })
                .collect(),
        )),
        (Value::Text(s), DataType::NumericArray) => Some(Value::NumericArray(
            decode_array_elems(
                &s,
                DataType::Numeric {
                    precision: 0,
                    scale: 0,
                },
                col_name,
                position,
            )?
            .into_iter()
            .map(|o| match o {
                Some(Value::Numeric { scaled, scale, .. }) => Some((scaled, scale)),
                _ => None,
            })
            .collect(),
        )),
        (Value::Text(s), DataType::DateArray) => Some(Value::DateArray(
            decode_array_elems(&s, DataType::Date, col_name, position)?
                .into_iter()
                .map(|o| match o {
                    Some(Value::Date(d)) => Some(d),
                    _ => None,
                })
                .collect(),
        )),
        (Value::Text(s), DataType::UuidArray) => Some(Value::UuidArray(
            decode_array_elems(&s, DataType::Uuid, col_name, position)?
                .into_iter()
                .map(|o| match o {
                    Some(Value::Uuid(u)) => Some(u),
                    _ => None,
                })
                .collect(),
        )),
        // v7.39 (round 694) — `oid[]` decodes exactly as `bigint[]` does;
        // the variant exists to keep the DECLARED type, not to change the
        // body. Listed here rather than mapped to BigIntArray upstream
        // because mapping it upstream is what made `pg_typeof('{1,2}'::oid[])`
        // answer `bigint[]`, which is the defect round 667 closed for the
        // scalar.
        (Value::Text(s), DataType::BigIntArray | DataType::OidArray) => {
            // v7.39 (round 325, V57) — PG's wording (and the same message
            // the CAST path gives for the identical input; this one used to
            // name TEXT[] whatever the column's element type was).
            let arr = decode_text_array_literal(&s).map_err(|_| {
                EngineError::Eval(EvalError::TypeMismatch {
                    detail: malformed_array_literal(&s),
                })
            })?;
            let mut out: Vec<Option<i64>> = Vec::with_capacity(arr.len());
            for elem in arr {
                match elem {
                    None => out.push(None),
                    Some(t) => {
                        let n: i64 = t.parse().map_err(|_| {
                            EngineError::Eval(EvalError::TypeMismatch {
                                detail: alloc::format!(
                                    "invalid input syntax for type bigint: {t:?}"
                                ),
                            })
                        })?;
                        out.push(Some(n));
                    }
                }
            }
            Some(Value::BigIntArray(out))
        }
        // v7.10.11 — TEXT[] → Text round-trip uses PG's
        // external array form (`{a,b,NULL}`). Lets a SELECT
        // pull an array column through any Text-side codepath.
        (Value::TextArray(items), DataType::Text) => Some(Value::text(encode_text_array(&items))),
        // v7.37.5 ship triage — empty `ARRAY[]` literal lands as
        // `Value::TextArray(vec![])`. Allow widening to the typed
        // array sibling so `ARRAY[]::BOOL[]` / `::FLOAT[]` etc.
        // round-trip through INSERT into the typed column. Only
        // empty contents go through silently — non-empty TextArray
        // must round-trip via per-element parsing(handled by the
        // existing element-specific coercion paths above).
        (Value::TextArray(items), DataType::BoolArray) if items.is_empty() => {
            Some(Value::BoolArray(alloc::vec::Vec::new()))
        }
        (Value::TextArray(items), DataType::SmallIntArray) if items.is_empty() => {
            Some(Value::SmallIntArray(alloc::vec::Vec::new()))
        }
        (Value::TextArray(items), DataType::IntArray) if items.is_empty() => {
            Some(Value::IntArray(alloc::vec::Vec::new()))
        }
        (Value::TextArray(items), DataType::BigIntArray) if items.is_empty() => {
            Some(Value::BigIntArray(alloc::vec::Vec::new()))
        }
        (Value::TextArray(items), DataType::FloatArray) if items.is_empty() => {
            Some(Value::FloatArray(alloc::vec::Vec::new()))
        }
        // `expr::float8[]` — an array literal reaches here as TEXT[] (elements
        // rendered to text); parse each element to f64. NULLs pass through.
        (Value::TextArray(items), DataType::FloatArray) => {
            let mut out = alloc::vec::Vec::with_capacity(items.len());
            let mut ok = true;
            for item in items {
                match item {
                    None => out.push(None),
                    Some(s) => match s.trim().parse::<f64>() {
                        Ok(x) => out.push(Some(x)),
                        Err(_) => {
                            ok = false;
                            break;
                        }
                    },
                }
            }
            if ok {
                Some(Value::FloatArray(out))
            } else {
                None
            }
        }
        // Identity for an already-float array, and widen integer arrays
        // element-wise (PG accepts `ARRAY[1,2]::float8[]`).
        (Value::FloatArray(items), DataType::FloatArray) => Some(Value::FloatArray(items)),
        #[allow(clippy::cast_precision_loss)]
        (Value::IntArray(items), DataType::FloatArray) => Some(Value::FloatArray(
            items.into_iter().map(|o| o.map(|n| f64::from(n))).collect(),
        )),
        #[allow(clippy::cast_precision_loss)]
        (Value::BigIntArray(items), DataType::FloatArray) => Some(Value::FloatArray(
            items.into_iter().map(|o| o.map(|n| n as f64)).collect(),
        )),
        // v7.38 (read01) — widen a NUMERIC[] into float8[] element-wise (PG
        // accepts `ARRAY[1.5::numeric]::float8[]` and coerces a numeric array
        // into a float8[] column on INSERT). Mirrors the scalar Numeric→Float.
        #[allow(clippy::cast_precision_loss)]
        (Value::NumericArray(items), DataType::FloatArray) => Some(Value::FloatArray(
            items
                .into_iter()
                .map(|o| {
                    o.map(|(scaled, scale)| {
                        crate::eval::format_numeric(scaled, scale)
                            .parse()
                            .unwrap_or(f64::NAN)
                    })
                })
                .collect(),
        )),
        // v7.38 (read01) — the rest of the numeric-array coercion matrix PG
        // accepts on INSERT / cast. Widening int→bigint / int·bigint→numeric /
        // float→numeric never fails; narrowing bigint→int fails the whole
        // coercion (→ None) if any element overflows i32.
        (Value::IntArray(items), DataType::BigIntArray) => Some(Value::BigIntArray(
            items.into_iter().map(|o| o.map(i64::from)).collect(),
        )),
        (Value::BigIntArray(items), DataType::IntArray) => {
            let mut out = alloc::vec::Vec::with_capacity(items.len());
            let mut ok = true;
            for o in items {
                match o {
                    None => out.push(None),
                    Some(n) => match i32::try_from(n) {
                        Ok(v) => out.push(Some(v)),
                        Err(_) => {
                            ok = false;
                            break;
                        }
                    },
                }
            }
            if ok { Some(Value::IntArray(out)) } else { None }
        }
        (Value::IntArray(items), DataType::NumericArray) => Some(Value::NumericArray(
            items
                .into_iter()
                .map(|o| o.map(|n| (i128::from(n), 0_u16)))
                .collect(),
        )),
        (Value::BigIntArray(items), DataType::NumericArray) => Some(Value::NumericArray(
            items
                .into_iter()
                .map(|o| o.map(|n| (i128::from(n), 0_u16)))
                .collect(),
        )),
        (Value::FloatArray(items), DataType::NumericArray) => {
            let mut out = alloc::vec::Vec::with_capacity(items.len());
            let mut ok = true;
            for o in items {
                match o {
                    None => out.push(None),
                    Some(x) => match parse_numeric_text(&alloc::format!("{x}")) {
                        Some((mantissa, scale)) => out.push(Some((mantissa, scale))),
                        None => {
                            ok = false;
                            break;
                        }
                    },
                }
            }
            if ok {
                Some(Value::NumericArray(out))
            } else {
                None
            }
        }
        // v7.38 (read01) — narrow a NUMERIC[] into int[] / bigint[] element-wise,
        // rounding half away from zero (PG) like the scalar Numeric→Int coercion.
        // An out-of-range element fails the whole coercion (→ None).
        (Value::NumericArray(items), DataType::IntArray) => {
            let mut out = alloc::vec::Vec::with_capacity(items.len());
            let mut ok = true;
            for o in items {
                match o {
                    None => out.push(None),
                    Some((scaled, scale)) => {
                        match i32::try_from(numeric_round_to_integer(scaled, scale)) {
                            Ok(v) => out.push(Some(v)),
                            Err(_) => {
                                ok = false;
                                break;
                            }
                        }
                    }
                }
            }
            if ok { Some(Value::IntArray(out)) } else { None }
        }
        (Value::NumericArray(items), DataType::BigIntArray) => {
            let mut out = alloc::vec::Vec::with_capacity(items.len());
            let mut ok = true;
            for o in items {
                match o {
                    None => out.push(None),
                    Some((scaled, scale)) => {
                        match i64::try_from(numeric_round_to_integer(scaled, scale)) {
                            Ok(v) => out.push(Some(v)),
                            Err(_) => {
                                ok = false;
                                break;
                            }
                        }
                    }
                }
            }
            if ok {
                Some(Value::BigIntArray(out))
            } else {
                None
            }
        }
        // v7.38 (read01, T2) — float8[] → int[] / bigint[], rounding each element
        // half-to-even (PG's float→int rule, distinct from numeric's half-away).
        // A non-finite / out-of-range element fails the whole coercion.
        #[allow(clippy::cast_possible_truncation)]
        (Value::FloatArray(items), DataType::IntArray) => {
            let mut out = alloc::vec::Vec::with_capacity(items.len());
            let mut ok = true;
            for o in items {
                match o {
                    None => out.push(None),
                    Some(x) if x.is_finite() => {
                        let r = crate::eval::math::f64_round_half_even(x);
                        if r >= f64::from(i32::MIN) && r <= f64::from(i32::MAX) {
                            out.push(Some(r as i32));
                        } else {
                            ok = false;
                            break;
                        }
                    }
                    Some(_) => {
                        ok = false;
                        break;
                    }
                }
            }
            if ok { Some(Value::IntArray(out)) } else { None }
        }
        #[allow(clippy::cast_possible_truncation)]
        (Value::FloatArray(items), DataType::BigIntArray) => {
            let mut out = alloc::vec::Vec::with_capacity(items.len());
            let mut ok = true;
            for o in items {
                match o {
                    None => out.push(None),
                    Some(x) if x.is_finite() => {
                        out.push(Some(crate::eval::math::f64_round_half_even(x) as i64));
                    }
                    Some(_) => {
                        ok = false;
                        break;
                    }
                }
            }
            if ok {
                Some(Value::BigIntArray(out))
            } else {
                None
            }
        }
        (Value::TextArray(items), DataType::NumericArray) if items.is_empty() => {
            Some(Value::NumericArray(alloc::vec::Vec::new()))
        }
        (Value::TextArray(items), DataType::DateArray) if items.is_empty() => {
            Some(Value::DateArray(alloc::vec::Vec::new()))
        }
        (Value::TextArray(items), DataType::TimestampArray) if items.is_empty() => {
            Some(Value::TimestampArray(alloc::vec::Vec::new()))
        }
        (Value::TextArray(items), DataType::TimestamptzArray) if items.is_empty() => {
            Some(Value::TimestamptzArray(alloc::vec::Vec::new()))
        }
        (Value::TextArray(items), DataType::UuidArray) if items.is_empty() => {
            Some(Value::UuidArray(alloc::vec::Vec::new()))
        }
        (Value::TextArray(items), DataType::JsonArray) if items.is_empty() => {
            Some(Value::JsonArray(alloc::vec::Vec::new()))
        }
        (Value::TextArray(items), DataType::JsonbArray) if items.is_empty() => {
            Some(Value::JsonbArray(alloc::vec::Vec::new()))
        }
        (Value::TextArray(items), DataType::BytesArray) if items.is_empty() => {
            Some(Value::BytesArray(alloc::vec::Vec::new()))
        }
        (Value::TextArray(items), DataType::IntervalArray) if items.is_empty() => {
            Some(Value::IntervalArray(alloc::vec::Vec::new()))
        }
        // Non-empty `TEXT[]` → typed array (`ARRAY[..]::bool[]`, `::numeric[]`,
        // `::date[]`, `::timestamp[]`, `::uuid[]`): parse each element via the
        // scalar path. Empty arrays are handled by the arms above.
        (
            Value::TextArray(items),
            dt @ (DataType::BoolArray
            | DataType::NumericArray
            | DataType::DateArray
            | DataType::TimestampArray
            | DataType::TimestamptzArray
            | DataType::IntervalArray
            | DataType::UuidArray),
        ) => coerce_text_array_to(items, dt, col_name)?,
        // v7.39 (round 326, V43) — the same targets from a STRING LITERAL.
        // `'{1,2}'::int[]` had a Text arm and worked; `'{…}'::timestamp[]`,
        // `::timestamptz[]` and `::interval[]` had none, so the literal
        // stayed TEXT and the cast died as a plain type mismatch — a whole
        // literal form that simply did not exist for the temporal arrays.
        (
            Value::Text(s),
            dt @ (DataType::TimestampArray | DataType::TimestamptzArray | DataType::IntervalArray),
        ) => {
            let items = decode_text_array_literal(&s).map_err(|_| {
                EngineError::Eval(EvalError::TypeMismatch {
                    detail: malformed_array_literal(&s),
                })
            })?;
            coerce_text_array_to(items, dt, col_name)?
        }
        (Value::TextArray(items), DataType::MoneyArray) if items.is_empty() => {
            Some(Value::MoneyArray(alloc::vec::Vec::new()))
        }
        // v7.37.5 ship triage — IntArray(empty) widens to
        // SmallIntArray for the `INSERT INTO t (xs) VALUES
        // (ARRAY[1::smallint, …])` path where the array literal
        // collected mixed int widths into IntArray.
        (Value::IntArray(items), DataType::SmallIntArray) => {
            let mut out = alloc::vec::Vec::with_capacity(items.len());
            let mut ok = true;
            for item in items {
                match item {
                    None => out.push(None),
                    Some(n) => match i16::try_from(n) {
                        Ok(x) => out.push(Some(x)),
                        Err(_) => {
                            ok = false;
                            break;
                        }
                    },
                }
            }
            if ok {
                Some(Value::SmallIntArray(out))
            } else {
                None
            }
        }
        // v7.17.0 Phase 3.P0-68 — Text → VECTOR auto-coerce.
        // Matches the existing Text → TsVector arm and the
        // `::vector` cast: PG-canonical pgvector external form
        // (`'[1, 2, -3]'`) becomes a typed Vector value at the
        // column boundary. Dim mismatch surfaces as TypeMismatch.
        // For SQ8 / HALF encodings we chain through the standard
        // quantise helpers so the storage shape matches the
        // declared encoding without a second coerce pass.
        (Value::Text(s), DataType::Vector { dim, encoding }) => {
            let parsed = eval::parse_vector_text(&s).ok_or_else(|| {
                EngineError::Eval(EvalError::TypeMismatch {
                    detail: alloc::format!("cannot parse {s:?} as VECTOR"),
                })
            })?;
            if parsed.len() != dim as usize {
                return Err(EngineError::Eval(EvalError::TypeMismatch {
                    detail: alloc::format!(
                        "VECTOR({dim}) column `{col_name}` rejects literal of length {}",
                        parsed.len()
                    ),
                }));
            }
            Some(match encoding {
                VecEncoding::F32 => Value::vector(parsed),
                VecEncoding::Sq8 => Value::Sq8Vector(spg_storage::quantize::quantize(&parsed)),
                VecEncoding::F16 => {
                    Value::HalfVector(spg_storage::halfvec::HalfVector::from_f32_slice(&parsed))
                }
            })
        }
        // v7.16.1 — Text → TSVECTOR auto-coerce for the
        // INSERT-side wire path (mailrs round-9 A.2.a). PG
        // implicitly promotes the TEXT literal at INSERT into a
        // TSVECTOR column; SPG previously rejected with a hard
        // type mismatch, blocking 23,276 pg_dump rows into
        // `messages.search_vector`. We route through the same
        // `decode_tsvector_external` the `::tsvector` cast
        // already uses, so PG-canonical forms (`'word'`,
        // `'word:1A,2B'`, multi-lexeme, empty `''`) all parse.
        (Value::Text(s), DataType::TsVector) => {
            let lexs = eval::decode_tsvector_external(&s).map_err(|e| {
                EngineError::Eval(EvalError::TypeMismatch {
                    detail: alloc::format!("cannot parse {s:?} as TSVECTOR: {e}"),
                })
            })?;
            Some(Value::TsVector(lexs))
        }
        (Value::Text(s), DataType::Timestamp | DataType::Timestamptz) => {
            let t = eval::parse_timestamp_literal(&s)
                .ok_or_else(|| datetime_parse_error("timestamp", &s))?;
            Some(Value::Timestamp(t))
        }
        // DATE ↔ TIMESTAMP convertibility (DATE → midnight,
        // TIMESTAMP → day truncation).
        (Value::Date(i32::MAX), DataType::Timestamp | DataType::Timestamptz) => {
            Some(Value::Timestamp(i64::MAX))
        }
        (Value::Date(i32::MIN), DataType::Timestamp | DataType::Timestamptz) => {
            Some(Value::Timestamp(i64::MIN))
        }
        (Value::Date(d), DataType::Timestamp | DataType::Timestamptz) => {
            Some(Value::Timestamp(i64::from(d) * 86_400_000_000))
        }
        // v7.9.21 — Value::Timestamp lands in either Timestamp
        // or Timestamptz columns; the on-disk layout is the
        // same i64 microseconds UTC.
        (Value::Timestamp(t), DataType::Timestamptz) => Some(Value::Timestamp(t)),
        (Value::Timestamp(t), DataType::Date) => {
            let days = t.div_euclid(86_400_000_000);
            i32::try_from(days).ok().map(Value::Date)
        }
        // v7.39 (round 633) — the time of day out of a timestamp.
        //
        // `TIMESTAMP '2020-01-02 03:04:05'::TIME` answered "cannot cast
        // timestamp without time zone to time without time zone"; PG
        // answers `03:04:05`, and has the cast registered as an assignment
        // one. `rem_euclid` rather than `%` so a pre-epoch timestamp gives
        // a time in [0, 24h) instead of a negative one. A timestamptz value
        // is carried in the same variant, so it comes through here too.
        (Value::Timestamp(t), DataType::Time) => Some(Value::Time(t.rem_euclid(86_400_000_000))),
        // v7.39 (read01 numeric.c) — a NumericBig is already an unconstrained
        // NUMERIC ('…0.5::numeric' where the mantissa exceeds i128); pass it
        // through. A declared numeric(p, s) still falls to the typed error.
        (
            Value::NumericBig(b),
            DataType::Numeric {
                precision: 0,
                scale: 0,
            },
        ) => Some(Value::NumericBig(b)),
        (
            Value::Numeric {
                scaled,
                scale: src_scale,
                ..
            },
            DataType::Numeric { precision, scale },
        ) => {
            // v7.38 (read01) — the unconstrained `::numeric` sentinel (0, 0)
            // keeps the value's natural scale, matching the Float/Text→Numeric
            // arms above; only a declared numeric(p, s) rescales. Without this,
            // casting an existing NUMERIC through unconstrained numeric
            // (`n::numeric(5,2)::numeric`) rounded it to scale 0.
            if precision == 0 && scale == 0 {
                Some(Value::Numeric {
                    scaled,
                    scale: src_scale,
                    kind: spg_storage::NumericKind::Finite,
                })
            } else {
                Some(numeric_rescale(
                    scaled, src_scale, precision, scale, col_name,
                )?)
            }
        }
        // v7.39 (round 272) — an arbitrary-precision value cast to a
        // DECLARED numeric had no arm at all, so a 47-digit literal
        // going into numeric(50,2) — a column PG accepts — reported an
        // internal storage type mismatch.
        (Value::NumericBig(b), DataType::Numeric { precision, scale }) => {
            if precision == 0 && scale == 0 {
                Some(Value::NumericBig(b))
            } else {
                #[allow(clippy::cast_sign_loss)]
                let rounded = if scale < 0 {
                    // Round to the multiple of 10^|scale| and land at 0.
                    b.round_to(0)
                } else {
                    b.round_to(scale as u16)
                };
                let out = crate::eval::binop::bignum_to_value(rounded);
                // The declared precision still binds; check it on the
                // decimal text, which both forms can produce.
                crate::numeric::check_precision_text(&out, precision, scale, col_name)?;
                Some(out)
            }
        }
        #[allow(clippy::cast_precision_loss)]
        (Value::Numeric { scaled, scale, .. }, DataType::Float) => {
            // v7.39 (round 271) — parse the decimal text rather than
            // dividing by a power built with repeated multiplication.
            // With scale widened to u16 that loop both accumulated
            // rounding error (1e-300 came out 9.999999999999999e-301)
            // and ran to infinity for a large enough scale, which then
            // looked like an underflow.
            let text = crate::eval::format_numeric(scaled, scale);
            let x: f64 = text.parse().unwrap_or(f64::NAN);
            // v7.39 (round 270) — a nonzero NUMERIC that underflows the
            // double range is an error in PG, quoting the decimal
            // expansion. It used to arrive as a silent zero.
            if x == 0.0 && scaled != 0 {
                return Err(float_out_of_range(
                    &crate::eval::format_numeric(scaled, scale),
                    "double precision",
                ));
            }
            Some(Value::Float(x))
        }
        // v7.39 (read01 numeric.c) — a big NUMERIC (`3.14e100` literal) casts
        // to float8 through its decimal text; a value beyond the double range
        // errors like PG ("value out of range: overflow").
        // v7.39 (round 269) — the same route to real. Without this arm a
        // NUMERIC literal past the i128 range (1.8e38 and up) never
        // reached a real cast at all and surfaced an internal
        // "expected REAL, got NUMERIC(0)" storage mismatch.
        (Value::NumericBig(b), DataType::Real) => {
            let text = b.to_decimal_str();
            let x: f32 = text.parse().map_err(|_| real_out_of_range(&text))?;
            if !x.is_finite() || (x == 0.0 && float_text_is_nonzero(&text)) {
                return Err(real_out_of_range(&text));
            }
            Some(Value::Real(x))
        }
        (Value::NumericBig(b), DataType::Float) => {
            // v7.39 (round 270) — PG quotes the decimal expansion here
            // rather than saying "value out of range: overflow", which
            // it reserves for narrowing a double.
            let text = b.to_decimal_str();
            let x: f64 = text
                .parse()
                .map_err(|_| float_out_of_range(&text, "double precision"))?;
            if !x.is_finite() || (x == 0.0 && float_text_is_nonzero(&text)) {
                return Err(float_out_of_range(&text, "double precision"));
            }
            Some(Value::Float(x))
        }
        // v7.38 (read01) — coercing NUMERIC into an integer column rounds half
        // away from zero (PG assignment cast: `1.5 → 2`), matching the `::int`
        // cast path; it previously truncated (`1.7 → 1`).
        // v7.39 (read01 float.c) — float → integer coercion (int4()/int8()/
        // int2() function casts, INSERT float into int column): PG rounds
        // half-to-even and errors on a non-finite / out-of-range value
        // rather than saturating.
        (Value::Float(x), DataType::Int) => {
            let r = crate::eval::math::f64_round_half_even(x);
            if !r.is_finite() || !(-2_147_483_648.0..=2_147_483_647.0).contains(&r) {
                return Err(EngineError::Eval(EvalError::TypeMismatch {
                    detail: "integer out of range".into(),
                }));
            }
            #[allow(clippy::cast_possible_truncation)]
            Some(Value::Int(r as i32))
        }
        (Value::Float(x), DataType::BigInt) => {
            let r = crate::eval::math::f64_round_half_even(x);
            if !r.is_finite()
                || !(-9.223_372_036_854_776e18..=9.223_372_036_854_776e18).contains(&r)
            {
                return Err(EngineError::Eval(EvalError::TypeMismatch {
                    detail: "bigint out of range".into(),
                }));
            }
            #[allow(clippy::cast_possible_truncation)]
            Some(Value::BigInt(r as i64))
        }
        (Value::Float(x), DataType::SmallInt) => {
            let r = crate::eval::math::f64_round_half_even(x);
            if !r.is_finite() || !(-32768.0..=32767.0).contains(&r) {
                return Err(EngineError::Eval(EvalError::TypeMismatch {
                    detail: "smallint out of range".into(),
                }));
            }
            #[allow(clippy::cast_possible_truncation)]
            Some(Value::SmallInt(r as i16))
        }
        // v7.39 (read01 round 112) — REAL (float4) → integer types. Mirrors the
        // float8 arms above (round half-to-even, PG's rule); these had no arm at
        // all, so `real::int` errored "cannot cast Real to int".
        (Value::Real(x), DataType::Int) => {
            let r = crate::eval::math::f64_round_half_even(f64::from(x));
            if !r.is_finite() || !(-2_147_483_648.0..=2_147_483_647.0).contains(&r) {
                return Err(EngineError::Eval(EvalError::TypeMismatch {
                    detail: "integer out of range".into(),
                }));
            }
            #[allow(clippy::cast_possible_truncation)]
            Some(Value::Int(r as i32))
        }
        (Value::Real(x), DataType::BigInt) => {
            let r = crate::eval::math::f64_round_half_even(f64::from(x));
            if !r.is_finite()
                || !(-9.223_372_036_854_776e18..=9.223_372_036_854_776e18).contains(&r)
            {
                return Err(EngineError::Eval(EvalError::TypeMismatch {
                    detail: "bigint out of range".into(),
                }));
            }
            #[allow(clippy::cast_possible_truncation)]
            Some(Value::BigInt(r as i64))
        }
        (Value::Real(x), DataType::SmallInt) => {
            let r = crate::eval::math::f64_round_half_even(f64::from(x));
            if !r.is_finite() || !(-32768.0..=32767.0).contains(&r) {
                return Err(EngineError::Eval(EvalError::TypeMismatch {
                    detail: "smallint out of range".into(),
                }));
            }
            #[allow(clippy::cast_possible_truncation)]
            Some(Value::SmallInt(r as i16))
        }
        (Value::Numeric { scaled, scale, .. }, DataType::Int) => {
            let rounded = numeric_round_to_integer(scaled, scale);
            i32::try_from(rounded).ok().map(Value::Int)
        }
        (Value::Numeric { scaled, scale, .. }, DataType::BigInt) => {
            let rounded = numeric_round_to_integer(scaled, scale);
            i64::try_from(rounded).ok().map(Value::BigInt)
        }
        (Value::Numeric { scaled, scale, .. }, DataType::SmallInt) => {
            let rounded = numeric_round_to_integer(scaled, scale);
            i16::try_from(rounded).ok().map(Value::SmallInt)
        }
        // VARCHAR(n) enforces an upper bound on character count. A bare
        // `varchar` (no typmod) is modelled as `Varchar(0)` and, like PG, holds
        // a string of any length — `'a'::varchar` must not read as VARCHAR(0).
        // v7.39 (round 291) — `name` is text truncated to NAMEDATALEN-1
        // (63) bytes. PG truncates silently rather than erroring, which
        // is the behaviour a catalog identifier column needs.
        (Value::Text(s), DataType::Name) => {
            let mut cut = s.into_owned();
            if cut.len() > 63 {
                let mut idx = 63;
                while !cut.is_char_boundary(idx) {
                    idx -= 1;
                }
                cut.truncate(idx);
            }
            Some(Value::text(cut))
        }
        (Value::Text(s), DataType::Varchar(max)) => {
            if max == 0 || u32::try_from(s.chars().count()).unwrap_or(u32::MAX) <= max {
                Some(Value::text(s))
            } else {
                // v7.39 (bpchar epic) — overflow that is only trailing
                // blanks is cut AT the limit (PG keeps 'abcd ' from
                // 'abcd  ' in varchar(5) — not a full strip); anything
                // else is 22001 with PG's phrasing.
                let excess_all_blanks = s.chars().skip(max as usize).all(|c| c == ' ');
                if excess_all_blanks {
                    Some(Value::text(
                        s.chars()
                            .take(max as usize)
                            .collect::<alloc::string::String>(),
                    ))
                } else {
                    return Err(EngineError::Unsupported(alloc::format!(
                        "value too long for type character varying({max})"
                    )));
                }
            }
        }
        // v6.0.1: f32 → SQ8 INSERT-time quantisation. Triggered
        // when the column declares `VECTOR(N) USING SQ8` and
        // the INSERT VALUES expression yields a raw f32 vector
        // (the normal pgvector-shape literal). Dim mismatch
        // falls through the `_ => None` arm and surfaces as
        // `TypeMismatch` with the expected SQ8 column type —
        // matching the F32 path's existing error.
        (
            Value::Vector(v),
            DataType::Vector {
                dim,
                encoding: VecEncoding::Sq8,
            },
        ) if v.len() == dim as usize => Some(Value::Sq8Vector(spg_storage::quantize::quantize(&v))),
        // v6.0.3: f32 → f16 INSERT-time conversion for HALF
        // columns. Bit-exact at the storage layer (modulo
        // half-precision rounding); no rerank pass needed at
        // search time.
        (
            Value::Vector(v),
            DataType::Vector {
                dim,
                encoding: VecEncoding::F16,
            },
        ) if v.len() == dim as usize => Some(Value::HalfVector(
            spg_storage::halfvec::HalfVector::from_f32_slice(&v),
        )),
        // CHAR(n) right-pads with U+0020 to exactly n chars. Overflow that
        // is only trailing blanks is trimmed to fit (PG: 'abcd  ' fits
        // CHAR(5)); real overflow is 22001.
        (Value::Text(s), DataType::Char(size)) => {
            // v7.39 (bpchar epic) — bare `bpchar` (no length) is PG's
            // unlimited blank-trimmed character type: store stripped,
            // no pad, no length check.
            if size == 0 {
                return Ok(Value::BpChar(alloc::borrow::Cow::Owned(
                    s.trim_end_matches(' ').to_string(),
                )));
            }
            let len = u32::try_from(s.chars().count()).unwrap_or(u32::MAX);
            let body = if len > size {
                let trimmed = s.trim_end_matches(' ');
                let tlen = u32::try_from(trimmed.chars().count()).unwrap_or(u32::MAX);
                if tlen > size {
                    return Err(EngineError::Unsupported(alloc::format!(
                        "value too long for type character({size})"
                    )));
                }
                trimmed.to_string()
            } else {
                s.into_owned()
            };
            let need = (size as usize) - body.chars().count();
            let mut padded = body;
            padded.reserve(need);
            for _ in 0..need {
                padded.push(' ');
            }
            // v7.38 (read01, T11) — CHAR(n) is bpchar: blank-padded, and
            // length / comparison / ::text ignore the padding (handled at those
            // sites).
            Some(Value::BpChar(alloc::borrow::Cow::Owned(padded)))
        }
        _ => None,
    };
    coerced.ok_or_else(|| {
        EngineError::Storage(StorageError::TypeMismatch {
            column: col_name.into(),
            expected,
            actual,
            position,
        })
    })
}

/// v7.38 (read01, T3.C3) — a lexer-validated big decimal literal → NumericBig,
/// demoted to a plain Numeric if its mantissa happens to fit i128.
pub(crate) fn big_literal_to_value(s: &str) -> Value<'static> {
    let b = spg_storage::bignum::BigNumeric::from_decimal_str(s).expect("lexer-validated decimal");
    match b.to_i128() {
        Some(scaled) => Value::Numeric {
            scaled,
            scale: b.scale(),
            kind: spg_storage::NumericKind::Finite,
        },
        None => Value::NumericBig(alloc::boxed::Box::new(b)),
    }
}

/// v7.39 (round 233 / round 236) — do two types share a PG type category,
/// so a set operation, an ARRAY constructor, a VALUES list, CASE, COALESCE
/// or GREATEST/LEAST can resolve them to one result type? Same type
/// always does; otherwise PG unifies within the numeric, string and
/// date/time families and refuses across them (probed against 18.4:
/// int ∪ bigint → bigint, text ∪ varchar → text, date ∪ timestamp →
/// timestamp, but int ∪ boolean, int ∪ text and text ∪ date are all
/// refused).
pub(crate) fn types_unify(a: DataType, b: DataType) -> bool {
    fn category(t: DataType) -> Option<u8> {
        Some(match t {
            DataType::SmallInt
            | DataType::Int
            | DataType::BigInt
            | DataType::Numeric { .. }
            | DataType::Real
            | DataType::Float => 1,
            DataType::Text | DataType::Varchar(_) | DataType::Char(_) => 2,
            DataType::Date | DataType::Timestamp | DataType::Timestamptz => 3,
            _ => return None,
        })
    }
    if a == b {
        return true;
    }
    match (category(a), category(b)) {
        (Some(x), Some(y)) => x == y,
        // Outside the families a set operation needs the exact same type;
        // `a == b` above already covered that.
        _ => false,
    }
}

/// v7.39 (round 236) — the type name PG puts in a "types X and Y cannot be
/// matched" message. `pg_data_type_text` answers for
/// `information_schema.columns.data_type`, where every array is the
/// pseudo-name `ARRAY`; an error message names the real thing
/// (`integer[]`).
/// v7.39 (round 622, S05a) — the `Option<DataType>` form, which is what
/// `Value::data_type()` returns and therefore what every "got X" error had.
///
/// Those errors printed it with `{:?}`, so a user asking for `upper(1)` was
/// told the argument was `Some(Int)` — Rust's Debug for an Option wrapping an
/// internal enum. 421 sites did this. `None` is the eval-only variants that
/// carry no storage type (RegClass, Composite); PG calls an untyped value
/// `unknown`, and that is what it becomes here.
pub(crate) fn pg_type_name_for_error_opt(t: Option<DataType>) -> alloc::string::String {
    match t {
        Some(t) => pg_type_name_for_error(t),
        None => alloc::string::String::from("unknown"),
    }
}

pub(crate) fn pg_type_name_for_error(t: DataType) -> alloc::string::String {
    use spg_storage::DataType as D;
    let elem = match t {
        D::TextArray => Some(D::Text),
        D::IntArray => Some(D::Int),
        D::BigIntArray => Some(D::BigInt),
        D::SmallIntArray => Some(D::SmallInt),
        D::FloatArray => Some(D::Float),
        D::NumericArray => Some(D::Numeric {
            precision: 0,
            scale: 0,
        }),
        D::BoolArray => Some(D::Bool),
        D::DateArray => Some(D::Date),
        D::TimestampArray => Some(D::Timestamp),
        D::TimestamptzArray => Some(D::Timestamptz),
        D::IntervalArray => Some(D::Interval),
        D::UuidArray => Some(D::Uuid),
        D::JsonArray | D::JsonbArray => Some(D::Jsonb),
        D::BytesArray => Some(D::Bytes),
        D::MoneyArray => Some(D::Money),
        _ => None,
    };
    match elem {
        Some(e) => alloc::format!("{}[]", crate::system_catalog::pg_data_type_text(e)),
        None => crate::system_catalog::pg_data_type_text(t),
    }
}