Skip to main content

spg_engine/eval/
cast.rs

1//! `expr::TYPE` CAST evaluation (cut 29 — extracted from `eval.rs`).
2//!
3//! Implements PG-style runtime coercion: the giant `cast_value`
4//! dispatcher plus its per-target helpers (numeric / bool / array /
5//! date / timestamp / interval / vector). Date and timestamp casts
6//! defer to the calendar parsers (`parse_date_literal` /
7//! `parse_timestamp_literal`) that stay in `eval.rs`; tsvector /
8//! tsquery casts defer to the FTS codecs re-exported from
9//! `eval::textsearch`.
10
11use alloc::format;
12use alloc::string::{String, ToString};
13use alloc::vec::Vec;
14
15use spg_sql::ast::CastTarget;
16use spg_storage::Value;
17
18use super::math::{f64_powi, f64_round_half_even};
19use super::{
20    EvalError, decode_tsquery_external, decode_tsvector_external, parse_date_literal,
21    parse_timestamp_literal, value_to_text,
22};
23
24/// v7.39 (read01 regproc.c) — the reg* input types SPG carries as text:
25/// resolve the name and return its canonical rendering, with PG's
26/// distinct not-found / ambiguous errors.
27/// v7.39 (round 515) — the type names each family below accepts, declared
28/// ONCE so the value path and the NULL-target check cannot drift apart.
29///
30/// They already had, three times. Round 509 added a check that a cast target
31/// names something, with its own hand-kept list; round 512 then added
32/// `Value::Cid` without registering `cid` there, round 514 found that and
33/// `xid`, and round 515 found six more — `'english'::regconfig` resolved
34/// while `NULL::regconfig` did not. The duplication was the defect, so it
35/// is gone rather than patched a third time.
36pub(crate) const REG_MISC_TYPES: &[&str] = &[
37    "regproc",
38    "regprocedure",
39    "regoper",
40    "regoperator",
41    "regconfig",
42    "regdictionary",
43    "regcollation",
44];
45
46/// The catalog-shaped scalars — see `cast_catalog_scalar`.
47pub(crate) const CATALOG_SCALAR_TYPES: &[&str] = &[
48    "cid",
49    "xid",
50    "oidvector",
51    "int2vector",
52    "aclitem",
53    "refcursor",
54    "pg_snapshot",
55    "txid_snapshot",
56    "jsonpath",
57];
58
59/// The pseudotypes and the statistics / GiST internals: a NULL is NULL, a
60/// value is "cannot accept a value of type X".
61pub(crate) const OPAQUE_TYPES: &[&str] = &[
62    "anyarray",
63    "anyelement",
64    "anyenum",
65    "anyrange",
66    "anymultirange",
67    "anynonarray",
68    "anycompatible",
69    "anycompatiblearray",
70    "anycompatiblenonarray",
71    "anycompatiblerange",
72    "anycompatiblemultirange",
73    "any",
74    "trigger",
75    "event_trigger",
76    "internal",
77    "language_handler",
78    "fdw_handler",
79    "pg_ddl_command",
80    "pg_node_tree",
81    "pg_ndistinct",
82    "pg_mcv_list",
83    "pg_dependencies",
84    "pg_brin_minmax_multi_summary",
85    "pg_brin_bloom_summary",
86    "gtsvector",
87];
88
89fn cast_reg_misc(kind: &str, s: &str) -> Result<Value<'static>, EvalError> {
90    let bare = s
91        .strip_prefix("pg_catalog.")
92        .unwrap_or(s)
93        .trim()
94        .to_string();
95    match kind {
96        "regconfig" => {
97            const CONFIGS: &[&str] = &[
98                "simple",
99                "arabic",
100                "armenian",
101                "basque",
102                "catalan",
103                "danish",
104                "dutch",
105                "english",
106                "finnish",
107                "french",
108                "german",
109                "greek",
110                "hindi",
111                "hungarian",
112                "indonesian",
113                "irish",
114                "italian",
115                "lithuanian",
116                "nepali",
117                "norwegian",
118                "portuguese",
119                "romanian",
120                "russian",
121                "serbian",
122                "spanish",
123                "swedish",
124                "tamil",
125                "turkish",
126                "yiddish",
127            ];
128            if CONFIGS.contains(&bare.as_str()) {
129                Ok(Value::text(bare))
130            } else {
131                Err(EvalError::TypeMismatch {
132                    detail: alloc::format!("text search configuration \"{s}\" does not exist"),
133                })
134            }
135        }
136        // v7.39 (round 513) — `regcollation`. PG lowercases an UNQUOTED
137        // identifier before it looks, which is why `'C'::regcollation` is
138        // "collation \"c\" for encoding \"UTF8\" does not exist" there while
139        // `'\"C\"'::regcollation` resolves — measured, and the reason the
140        // quoted form is the one anybody writes. The rendering keeps the
141        // quotes PG puts back on a name that needs them.
142        "regcollation" => {
143            let quoted = bare.starts_with('"') && bare.ends_with('"') && bare.len() >= 2;
144            let name = if quoted {
145                bare[1..bare.len() - 1].to_string()
146            } else {
147                bare.to_ascii_lowercase()
148            };
149            const COLLATIONS: &[&str] = &["C", "POSIX", "default", "ucs_basic"];
150            match COLLATIONS.iter().find(|c| **c == name) {
151                // PG re-quotes anything that is not a plain lowercase word.
152                Some(c) => Ok(Value::text(
153                    if c.chars().all(|ch| ch.is_ascii_lowercase() || ch == '_') && *c != "default" {
154                        (*c).to_string()
155                    } else {
156                        alloc::format!("\"{c}\"")
157                    },
158                )),
159                None => Err(EvalError::TypeMismatch {
160                    detail: alloc::format!(
161                        "collation \"{name}\" for encoding \"UTF8\" does not exist"
162                    ),
163                }),
164            }
165        }
166        "regdictionary" => {
167            if bare == "simple" || bare.ends_with("_stem") {
168                Ok(Value::text(bare))
169            } else {
170                Err(EvalError::TypeMismatch {
171                    detail: alloc::format!("text search dictionary \"{s}\" does not exist"),
172                })
173            }
174        }
175        "regproc" => {
176            let hits = crate::system_catalog::PG_PROC_FUNCS
177                .iter()
178                .filter(|(_, n, ..)| *n == bare)
179                .count();
180            match hits {
181                0 => Err(EvalError::TypeMismatch {
182                    detail: alloc::format!("function \"{s}\" does not exist"),
183                }),
184                1 => Ok(Value::text(bare)),
185                _ => Err(EvalError::TypeMismatch {
186                    detail: alloc::format!("more than one function named \"{bare}\""),
187                }),
188            }
189        }
190        "regprocedure" => {
191            // `name(argtype, ...)` — resolve the name, canonicalize each
192            // argument type, and re-render.
193            let Some((fname, rest)) = bare.split_once('(') else {
194                return Err(EvalError::TypeMismatch {
195                    detail: alloc::format!("expected a left parenthesis in \"{s}\""),
196                });
197            };
198            let Some(args_txt) = rest.strip_suffix(')') else {
199                return Err(EvalError::TypeMismatch {
200                    detail: alloc::format!("expected a right parenthesis in \"{s}\""),
201                });
202            };
203            let fname = fname.trim().to_ascii_lowercase();
204            let args: Vec<String> = if args_txt.trim().is_empty() {
205                Vec::new()
206            } else {
207                args_txt
208                    .split(',')
209                    .map(|a| {
210                        crate::conversions::regtype_canonical_name(a.trim()).ok_or_else(|| {
211                            EvalError::TypeMismatch {
212                                detail: alloc::format!("type \"{}\" does not exist", a.trim()),
213                            }
214                        })
215                    })
216                    .collect::<Result<_, _>>()?
217            };
218            let nargs = args.len() as i32;
219            let known = crate::system_catalog::PG_PROC_FUNCS
220                .iter()
221                .any(|(_, n, _, na, _)| *n == fname && *na == nargs);
222            if !known {
223                return Err(EvalError::TypeMismatch {
224                    detail: alloc::format!("function \"{s}\" does not exist"),
225                });
226            }
227            Ok(Value::text(alloc::format!("{fname}({})", args.join(","))))
228        }
229        // regoper / regoperator: SPG has no operator catalog; every core
230        // operator symbol is multiply overloaded in PG, so a known symbol
231        // reports PG's ambiguity and anything else does not exist.
232        _ => {
233            let sym: String = bare.chars().filter(|c| !c.is_whitespace()).collect();
234            let core_op = !sym.is_empty() && sym.chars().all(|c| "+-*/<>=~!@#%^&|`?".contains(c));
235            if core_op {
236                Err(EvalError::TypeMismatch {
237                    detail: alloc::format!("more than one operator named {sym}"),
238                })
239            } else {
240                Err(EvalError::TypeMismatch {
241                    detail: alloc::format!("operator does not exist: {s}"),
242                })
243            }
244        }
245    }
246}
247
248/// v7.39 (round 355, M13) — `BINARY expr` / `CAST(expr AS BINARY[(n)])`.
249fn cast_mysql_binary(v: Value<'static>, name: &str) -> Result<Value<'static>, EvalError> {
250    let limit: Option<usize> = name
251        .split_once('(')
252        .and_then(|(_, rest)| rest.trim_end_matches(')').trim().parse().ok());
253    let text = match &v {
254        Value::Null => return Ok(Value::Null),
255        Value::Text(t) => t.to_string(),
256        Value::BpChar(t) => t.to_string(),
257        other => crate::eval::values::value_to_text(other),
258    };
259    Ok(Value::text(match limit {
260        // Byte-wise, which is the point of the type.
261        Some(n) if text.len() > n => {
262            let mut cut = n;
263            while cut > 0 && !text.is_char_boundary(cut) {
264                cut -= 1;
265            }
266            text[..cut].to_string()
267        }
268        _ => text,
269    }))
270}
271
272/// v7.39 (round 352, M8) — `CAST(x AS SIGNED)` / `CAST(x AS UNSIGNED)`.
273fn cast_mysql_integer(v: Value<'static>, unsigned: bool) -> Result<Value<'static>, EvalError> {
274    // v7.39 (round 527) — an EXACT integer source must not round-trip
275    // through f64. It loses precision above 2^53, and the float→int cast
276    // SATURATES, so `CAST(18446744073709551615 AS UNSIGNED)` answered
277    // 9223372036854775807 — a different number, with nothing to say so.
278    // The value stores, compares and sums correctly at full width
279    // (measured against MariaDB 11); only the cast reduced it.
280    let exact: Option<i128> = match &v {
281        Value::Bool(b) => Some(i128::from(u8::from(*b))),
282        Value::SmallInt(x) => Some(i128::from(*x)),
283        Value::Int(x) => Some(i128::from(*x)),
284        Value::BigInt(x) => Some(i128::from(*x)),
285        Value::Numeric {
286            scaled,
287            scale: 0,
288            kind: spg_storage::NumericKind::Finite,
289        } => Some(*scaled),
290        _ => None,
291    };
292    let rounded: i128 = match exact {
293        Some(n) => n,
294        None => {
295            let n: f64 = match &v {
296                Value::Null => return Ok(Value::Null),
297                Value::Float(x) => *x,
298                Value::Real(x) => f64::from(*x),
299                #[allow(clippy::cast_precision_loss)]
300                Value::Numeric { scaled, scale, .. } => {
301                    *scaled as f64 / 10_f64.powi(i32::from(*scale))
302                }
303                Value::Text(t) | Value::BpChar(t) => crate::eval::mysql_leading_number(t),
304                other => {
305                    return Err(EvalError::TypeMismatch {
306                        detail: alloc::format!(
307                            "cannot cast {} to integer",
308                            crate::conversions::pg_type_name_for_error_opt(other.data_type())
309                        ),
310                    });
311                }
312            };
313            // Half away from zero, which is what MariaDB does
314            // (2.5 → 3, -2.5 → -3).
315            let r = if n >= 0.0 {
316                (n + 0.5).floor()
317            } else {
318                (n - 0.5).ceil()
319            };
320            #[allow(clippy::cast_possible_truncation)]
321            let as_i64 = r as i64;
322            i128::from(as_i64)
323        }
324    };
325    if unsigned {
326        // MariaDB wraps a negative through the full u64 range.
327        let wrapped: u64 = if rounded < 0 {
328            #[allow(clippy::cast_sign_loss, clippy::cast_possible_truncation)]
329            {
330                rounded as i64 as u64
331            }
332        } else {
333            u64::try_from(rounded).unwrap_or(u64::MAX)
334        };
335        // Above i64::MAX the value only fits the numeric carrier, which
336        // is the same one a BIGINT UNSIGNED column already uses.
337        return Ok(if wrapped > i64::MAX as u64 {
338            Value::Numeric {
339                scaled: i128::from(wrapped),
340                scale: 0,
341                kind: spg_storage::NumericKind::Finite,
342            }
343        } else {
344            #[allow(clippy::cast_possible_wrap)]
345            Value::BigInt(wrapped as i64)
346        });
347    }
348    Ok(Value::BigInt(
349        i64::try_from(rounded).unwrap_or(if rounded < 0 { i64::MIN } else { i64::MAX }),
350    ))
351}
352
353/// v7.39 (round 544) — the integer a bytea's bytes spell: big-endian,
354/// right-aligned into `width` bytes, sign-extended from the leading
355/// byte when the value already fills the width. Measured on PG18:
356/// `'\x05'::bytea::int8` is 5, `'\x'::bytea::int4` is 0,
357/// `'\xffffffff'::bytea::int4` is -1.
358#[inline(never)]
359fn bytea_to_integer(v: &Value<'static>, width: usize) -> Result<Value<'static>, EvalError> {
360    let Value::Bytes(b) = v else {
361        return Err(EvalError::TypeMismatch {
362            detail: alloc::string::String::from("expected bytea"),
363        });
364    };
365    if b.len() > width {
366        return Err(EvalError::TypeMismatch {
367            detail: alloc::format!("bytea of {} bytes is too wide for the target", b.len()),
368        });
369    }
370    let negative = b.len() == width && b.first().is_some_and(|f| *f & 0x80 != 0);
371    let mut acc: i64 = if negative { -1 } else { 0 };
372    for byte in b.iter() {
373        acc = (acc << 8) | i64::from(*byte);
374    }
375    Ok(if width == 4 {
376        Value::Int(i32::try_from(acc).unwrap_or(0))
377    } else {
378        Value::BigInt(acc)
379    })
380}
381
382/// Round a numeric operand (`scaled` × 10^-`scale`) to the nearest
383/// integer, half-away-from-zero — PG's `numeric → int` coercion rule.
384fn numeric_round_to_i128(scaled: i128, scale: u16) -> i128 {
385    let factor = 10_i128.pow(u32::from(scale));
386    let neg = scaled < 0;
387    let abs = scaled.unsigned_abs() as i128;
388    let q = abs / factor;
389    let r = abs % factor;
390    let mag = if 2 * r >= factor { q + 1 } else { q };
391    if neg { -mag } else { mag }
392}
393
394/// PG-style `expr::TYPE` coercion. NULL always casts as NULL.
395pub fn cast_value(v: Value<'static>, target: CastTarget) -> Result<Value<'static>, EvalError> {
396    cast_value_in(v, target, false)
397}
398
399/// v7.39 (round 352, M8) — `cast_value` with the session dialect, for the
400/// targets the two disagree about (`SIGNED` / `UNSIGNED` exist only in
401/// MySQL: PG says `type "signed" does not exist`, measured).
402pub fn cast_value_in(
403    v: Value<'static>,
404    target: CastTarget,
405    mysql: bool,
406) -> Result<Value<'static>, EvalError> {
407    cast_value_ref_in(v, &target, mysql)
408}
409
410/// v7.39 (round 607) — the same dispatch, taking the target by REFERENCE.
411///
412/// `eval_cast_arm` cloned the target for every row. For the settled variants
413/// that clone is free, which is why `id::FLOAT` allocated nothing a row while
414/// `id::REAL` — the same conversion under a name the parser leaves as
415/// `Named(String)` — allocated one just to hand the name over, and seven more
416/// re-deriving its lowercase form inside.
417pub fn cast_value_ref_in(
418    v: Value<'static>,
419    target: &CastTarget,
420    mysql: bool,
421) -> Result<Value<'static>, EvalError> {
422    // v7.37 (round 896) — the quoted spelling of these two arrives as
423    // `Named`, the bare one as its own variant, and only the variant had an
424    // arm. `::regclass` worked and `::"regclass"` answered `type "regclass"
425    // does not exist` — and quoted identifiers are what an ORM or pg_dump
426    // writes. Folding here rather than adding a second arm keeps one
427    // implementation: whatever the variant does, the quoted form now does.
428    // Round 894 fixed `tsvector` / `tsquery` the same way round and left
429    // these open because this path had not been read; it has now.
430    if let CastTarget::Named(n) = target {
431        if n.eq_ignore_ascii_case("regclass") {
432            return cast_value_ref_in(v, &CastTarget::RegClass, mysql);
433        }
434        if n.eq_ignore_ascii_case("regtype") {
435            return cast_value_ref_in(v, &CastTarget::RegType, mysql);
436        }
437    }
438    // v7.39 (round 509) — PG validates the cast TARGET whatever the operand
439    // is: `NULL::nosuchtype` is an error there, not NULL. This returned early
440    // before ever looking at the target, so a misspelt type name silently
441    // produced NULL and `pg_typeof(NULL::nosuchtype)` answered `unknown`. A
442    // value operand DID error, so the gap was exactly the NULL case, in both
443    // spellings (`::t` and `CAST(… AS t)`).
444    //
445    // Only `Named` can fail to resolve; every other CastTarget is a variant
446    // the parser already settled. So a NULL keeps its short-circuit
447    // everywhere else and a Named target runs the real path, which is the
448    // only thing that knows every name that resolves. Writing a second
449    // resolver to check the name against looked simpler and was wrong: it
450    // missed `::binary` (the MySQL prefix's desugar), a table's row type,
451    // and the pseudotypes, all of which resolve further down this arm.
452    if matches!(v, Value::Null) {
453        return Ok(Value::Null);
454    }
455    match target {
456        CastTarget::Vector => cast_to_vector(v),
457        // v7.38 (read01) — the inet/cidr ::text cast shows the mask even for
458        // /32 and /128 (PG's cast-path form, unlike the display default).
459        CastTarget::Text => Ok(Value::text(match &v {
460            Value::Inet { family, bits, addr } | Value::Cidr { family, bits, addr } => {
461                crate::conversions::format_inet_full(*family, *bits, addr)
462            }
463            // v7.38 (read01, T11) — bpchar → text strips the trailing blanks
464            // (unlike the padded wire display).
465            Value::BpChar(s) => s.trim_end_matches(' ').to_string(),
466            // v7.39 (read01 ruleutils.c) — regclass::text is the name.
467            Value::RegClass(_, name) | Value::RegProc(_, name) | Value::RegType(_, name) => {
468                name.to_string()
469            }
470            _ => value_to_text(&v),
471        })),
472        // v7.39 (round 254) — the integer targets refuse a NUMERIC special
473        // outright (PG: `cannot convert NaN to integer`); without this the
474        // arms below read the special's canonical mantissa and answered 0.
475        // The float / numeric targets pass it through instead — handled in
476        // their own arms, which now consult `kind`.
477        // v7.39 (round 343) — an OID-typed reference casts to an integer
478        // the way PG's do (`'t'::regclass::bigint` is 27830 there). SPG
479        // reported `cannot cast None to bigint`: the integer path read the
480        // value's storage DataType, which these two deliberately do not
481        // have, and the message leaked that `None` to the client.
482        CastTarget::BigInt | CastTarget::Int
483            if matches!(
484                v,
485                Value::RegClass(..) | Value::RegProc(..) | Value::RegType(..)
486            ) =>
487        {
488            let (Value::RegClass(oid, _) | Value::RegProc(oid, _) | Value::RegType(oid, _)) = v
489            else {
490                unreachable!("guarded above")
491            };
492            Ok(if matches!(target, CastTarget::BigInt) {
493                Value::BigInt(oid)
494            } else {
495                Value::Int(i32::try_from(oid).unwrap_or(i32::MAX))
496            })
497        }
498        // v7.39 (round 544) — bytea reads back as the integer its bytes
499        // spell, big-endian and right-aligned. Measured on PG18:
500        // '\x05'::bytea::int8 is 5, '\x'::bytea::int4 is 0.
501        CastTarget::Int if matches!(v, Value::Bytes(_)) => bytea_to_integer(&v, 4),
502        CastTarget::BigInt if matches!(v, Value::Bytes(_)) => bytea_to_integer(&v, 8),
503        CastTarget::Int => {
504            cast_numeric_special_reject(&v, "integer").unwrap_or_else(|| cast_numeric_to_int(v))
505        }
506        CastTarget::BigInt => {
507            cast_numeric_special_reject(&v, "bigint").unwrap_or_else(|| cast_numeric_to_bigint(v))
508        }
509        CastTarget::Float => cast_numeric_to_float(v),
510        CastTarget::Bool => cast_to_bool(v),
511        CastTarget::Date => cast_to_date(v),
512        // TIMESTAMP and TIMESTAMPTZ share a runtime representation
513        // (i64 microseconds UTC) but NOT an input rule, and conflating
514        // the two silently stored the wrong instant. `::timestamp`
515        // keeps the wall clock a literal's offset was written against
516        // (round 289); `::timestamptz` CONVERTS by it. The evaluator's
517        // context-aware arm intercepts timestamptz before this point,
518        // so the difference only showed on the paths that reach here
519        // directly — INSERT VALUES folds its literals through
520        // `literal_expr_to_value_in`, and stored 10:00 for
521        // `'2020-01-01 10:00:00+02'::timestamptz` where PG stores
522        // 08:00 (round 310).
523        // v7.39 (round 423) — `CAST(x AS DATETIME)` / `AS TIMESTAMP` reach
524        // here as the dedicated variant rather than `Named`, so the MySQL
525        // "bare temporal type has fractional precision 0" rule has to be
526        // applied here too: MariaDB drops the fraction, PG keeps every
527        // microsecond.
528        CastTarget::Timestamp => {
529            let out = cast_to_timestamp(v)?;
530            Ok(if mysql {
531                round_temporal_to_precision(out, 0, true)
532            } else {
533                out
534            })
535        }
536        CastTarget::Timestamptz => cast_to_timestamptz(v),
537        // v7.9.25 — `expr::INTERVAL`. Currently only TEXT → Interval
538        // is supported (the mailrs idiom: `$1::INTERVAL` where the
539        // bound param is a string like `'7 days'`).
540        // v7.39 (round 544) — a time-of-day IS an interval of that
541        // length. Measured: '10:20:30.123456'::time::interval reads
542        // 10:20:30.123456 on PG18, fractional seconds and all.
543        CastTarget::Interval => match v {
544            Value::Time(us) => Ok(Value::Interval {
545                months: 0,
546                days: 0,
547                micros: us,
548            }),
549            other => cast_to_interval(other),
550        },
551        // v7.9.25 — `::json` keeps the input text verbatim (PG's json
552        // type preserves whitespace / key order / duplicates).
553        CastTarget::Json => match v {
554            Value::Json(s) => Ok(Value::json(s)),
555            Value::Text(s) => Ok(Value::json(s)),
556            other => Err(EvalError::TypeMismatch {
557                detail: alloc::format!(
558                    "::json only accepts TEXT-shape inputs, got {}",
559                    crate::conversions::pg_type_name_for_error_opt(other.data_type())
560                ),
561            }),
562        },
563        // v7.38 (read01) — `::jsonb` canonicalises like PG: object keys
564        // sorted (length, then bytes) + duplicates collapsed last-wins,
565        // `, ` / `: ` whitespace, and numbers normalised. Invalid JSON
566        // falls back to the verbatim text (validation stays a separate
567        // concern from this representation fix).
568        CastTarget::Jsonb => match v {
569            // v7.39 (read01 jsonb) — the explicit ::jsonb cast validates:
570            // invalid tokens (NaN / Infinity / malformed) error like PG
571            // instead of passing the raw text through.
572            Value::Json(s) | Value::Text(s) => match crate::json::canonicalize_jsonb(s.as_ref()) {
573                Ok(c) => Ok(Value::json(c)),
574                Err(_) => Err(EvalError::TypeMismatch {
575                    detail: alloc::string::String::from("invalid input syntax for type json"),
576                }),
577            },
578            other => Err(EvalError::TypeMismatch {
579                detail: alloc::format!(
580                    "::jsonb only accepts TEXT-shape inputs, got {}",
581                    crate::conversions::pg_type_name_for_error_opt(other.data_type())
582                ),
583            }),
584        },
585        // v7.17.0 Phase 5.3 — `::regtype` / `::regclass`. PG
586        // semantics: each is a textual catalog-name surfacing as
587        // a numeric OID at the wire layer that renders back as
588        // the original name. SPG has no OID space, but pg_dump /
589        // mailrs / Django code uses the cast purely for textual
590        // round-trip — feeding `'public.t'::regclass::text` into
591        // a downstream `format(…)` or string concat. We map to
592        // that textual contract: Text in → Text out (the schema-
593        // qualifier `public.` is stripped to match PG's default
594        // search_path-aware rendering); numeric in → re-cast to
595        // Text as best-effort; anything else errors.
596        //
597        // Pre-3.3 / pre-5.3 (v7.9.26) the cast surfaced a clean
598        // error; this lifts to accept-and-textify so the dominant
599        // dump-loader pattern unblocks. SPG-shaped queries that
600        // genuinely need an OID for runtime joins are still
601        // documented as unsupported.
602        // v7.39 (round 694) — `'{text,int4}'::regtype[]`. PG canonicalises
603        // every ELEMENT (`int4` → `integer`) and rejects an unknown one, so
604        // the array runs the scalar's own name resolution per member rather
605        // than keeping the literal. `regclass[]` keeps its names — a
606        // relation name is already what PG prints.
607        CastTarget::Named(n) if n.eq_ignore_ascii_case("regtype_array") => {
608            let Value::Text(s) = &v else {
609                return Ok(v);
610            };
611            let body = s.trim();
612            let inner = body
613                .strip_prefix('{')
614                .and_then(|b| b.strip_suffix('}'))
615                .unwrap_or(body);
616            let mut out: Vec<Option<alloc::string::String>> = Vec::new();
617            for part in inner.split(',') {
618                let t = part.trim();
619                if t.is_empty() {
620                    continue;
621                }
622                if t.eq_ignore_ascii_case("NULL") {
623                    out.push(None);
624                    continue;
625                }
626                let bare = t.rsplit('.').next().unwrap_or(t);
627                match crate::conversions::regtype_canonical_name(bare) {
628                    Some(c) => out.push(Some(c)),
629                    None => {
630                        return Err(EvalError::TypeMismatch {
631                            detail: alloc::format!("type \"{t}\" does not exist"),
632                        });
633                    }
634                }
635            }
636            Ok(Value::TextArray(out))
637        }
638        CastTarget::RegType | CastTarget::RegClass => match v {
639            Value::Text(s) => {
640                // Strip an optional `<schema>.` prefix — PG's
641                // regclass render drops it when the schema is on
642                // the search_path; SPG is single-schema so
643                // dropping is always safe.
644                let bare = s.rsplit('.').next().unwrap_or(&s).to_string();
645                // v7.39 (read01 regproc.c) — regtype canonicalizes the
646                // name ('int4' → 'integer') and rejects unknown types
647                // (PG 42704).
648                if matches!(target, CastTarget::RegType) {
649                    // v7.39 (round 648) — carry the OID as well as the
650                    // name, the way `::regclass` and `::regproc` already
651                    // do. As a plain Text this rendered correctly and
652                    // then failed everything downstream: `'text'::regtype
653                    // ::oid` parsed the NAME as a number and answered
654                    // `invalid input syntax for type oid: "text"` where
655                    // PG answers 25, and `pg_typeof` said `text`.
656                    return match crate::conversions::regtype_canonical_name(&bare) {
657                        Some(c) => {
658                            let oid =
659                                crate::conversions::regtype_name_to_oid(&c.to_ascii_lowercase())
660                                    .unwrap_or(0);
661                            Ok(Value::RegType(oid, c.into_boxed_str()))
662                        }
663                        None => Err(EvalError::TypeMismatch {
664                            detail: alloc::format!("type \"{s}\" does not exist"),
665                        }),
666                    };
667                }
668                Ok(Value::text(bare))
669            }
670            // A numeric OID → its type name for `::regtype` (the common
671            // `atttypid::regtype` column-type-name shape). `::regclass`
672            // needs a catalog reverse-lookup for user relations, which
673            // this cast has no access to, so it keeps rendering the OID.
674            Value::Int(_) | Value::BigInt(_) => {
675                let n = match v {
676                    Value::Int(n) => i64::from(n),
677                    Value::BigInt(n) => n,
678                    _ => unreachable!(),
679                };
680                if matches!(target, CastTarget::RegType)
681                    && let Some(name) = crate::conversions::regtype_oid_to_name_owned(n)
682                {
683                    Ok(Value::RegType(n, name.into_boxed_str()))
684                } else {
685                    Ok(Value::text(alloc::format!("{n}")))
686                }
687            }
688            other => Err(EvalError::TypeMismatch {
689                detail: alloc::format!(
690                    "::regtype / ::regclass accepts TEXT (name) or integer (oid), got {}",
691                    crate::conversions::pg_type_name_for_error_opt(other.data_type())
692                ),
693            }),
694        },
695        // v7.10.11 — `::TEXT[]`. Decode PG external array form
696        // when input is Text; pass through unchanged when it is
697        // already TextArray. Anything else is a type mismatch.
698        CastTarget::TextArray => match v {
699            Value::TextArray(items) => Ok(Value::TextArray(items)),
700            Value::Text(s) => {
701                if let Some(r) = try_cast_2d_array(&s, |row| {
702                    decode_text_array_external(row).map(Value::TextArray)
703                }) {
704                    return r;
705                }
706                decode_text_array_external(&s).map(Value::TextArray)
707            }
708            // Other scalar arrays cast element-wise, each element
709            // rendered as its own text (NULLs preserved). PG allows
710            // `ARRAY[1,2,3]::text[]`.
711            Value::IntArray(items) => Ok(Value::TextArray(
712                items
713                    .into_iter()
714                    .map(|o| o.map(|n| alloc::format!("{n}")))
715                    .collect(),
716            )),
717            Value::BigIntArray(items) => Ok(Value::TextArray(
718                items
719                    .into_iter()
720                    .map(|o| o.map(|n| alloc::format!("{n}")))
721                    .collect(),
722            )),
723            Value::SmallIntArray(items) => Ok(Value::TextArray(
724                items
725                    .into_iter()
726                    .map(|o| o.map(|n| alloc::format!("{n}")))
727                    .collect(),
728            )),
729            Value::BoolArray(items) => Ok(Value::TextArray(
730                items
731                    .into_iter()
732                    .map(|o| o.map(|b| String::from(if b { "t" } else { "f" })))
733                    .collect(),
734            )),
735            Value::FloatArray(items) => Ok(Value::TextArray(
736                items
737                    .into_iter()
738                    .map(|o| o.map(|x| value_to_text(&Value::Float(x))))
739                    .collect(),
740            )),
741            other => Err(EvalError::TypeMismatch {
742                detail: alloc::format!(
743                    "::TEXT[] only accepts TEXT / array inputs, got {}",
744                    crate::conversions::pg_type_name_for_error_opt(other.data_type())
745                ),
746            }),
747        },
748        // v7.11.13 — `::INT[]` / `::BIGINT[]`. Decode PG external
749        // form `{1,2,3}` when input is Text; widen TextArray /
750        // IntArray as appropriate.
751        CastTarget::IntArray => cast_to_int_array(v),
752        CastTarget::BigIntArray => cast_to_bigint_array(v),
753        // v7.12.0 — `::tsvector` / `::tsquery`. Decodes PG external
754        // form when input is Text; passes through unchanged when the
755        // input is already the target type. Other inputs are a type
756        // mismatch. Lexer / Porter stemmer arrive in v7.12.1; the
757        // external-form cast at v7.12.0 is the path pg_dump and
758        // direct-literal callers use.
759        CastTarget::TsVector => match v {
760            Value::TsVector(items) => Ok(Value::TsVector(items)),
761            Value::Text(s) => decode_tsvector_external(&s).map(Value::TsVector),
762            other => Err(EvalError::TypeMismatch {
763                detail: alloc::format!(
764                    "::tsvector only accepts TEXT / tsvector inputs, got {}",
765                    crate::conversions::pg_type_name_for_error_opt(other.data_type())
766                ),
767            }),
768        },
769        CastTarget::TsQuery => match v {
770            Value::TsQuery(ast) => Ok(Value::TsQuery(ast)),
771            Value::Text(s) => decode_tsquery_external(&s).map(Value::TsQuery),
772            other => Err(EvalError::TypeMismatch {
773                detail: alloc::format!(
774                    "::tsquery only accepts TEXT / tsquery inputs, got {}",
775                    crate::conversions::pg_type_name_for_error_opt(other.data_type())
776                ),
777            }),
778        },
779        // v7.17.0 — `::uuid`. Identity for `uuid → uuid`; parse
780        // text via the shared `parse_uuid_str`. Anything else is a
781        // type mismatch — PG also rejects e.g. INT → UUID without
782        // an explicit text bridge.
783        CastTarget::Uuid => match v {
784            Value::Uuid(b) => Ok(Value::Uuid(b)),
785            Value::Text(s) => match spg_storage::parse_uuid_str(&s) {
786                Some(b) => Ok(Value::Uuid(b)),
787                None => Err(EvalError::TypeMismatch {
788                    detail: alloc::format!("invalid input syntax for type uuid: {s:?}"),
789                }),
790            },
791            other => Err(EvalError::TypeMismatch {
792                detail: alloc::format!(
793                    "::uuid only accepts TEXT / uuid inputs, got {}",
794                    crate::conversions::pg_type_name_for_error_opt(other.data_type())
795                ),
796            }),
797        },
798        // v7.18 — `::bytea`. Identity for `Bytes → Bytes`; decode
799        // Text via the engine's PG-format bytea decoder (`\x`
800        // hex form + `\NNN` escape form). Anything else is a type
801        // mismatch — same shape as PG's contract. Closes the
802        // mailrs D-pre #3 reverse-acceptance gap.
803        CastTarget::Bytea => match v {
804            Value::Bytes(b) => Ok(Value::bytes(b)),
805            Value::Text(s) => match crate::conversions::decode_bytea_literal(&s) {
806                Ok(b) => Ok(Value::bytes(b)),
807                Err(msg) => Err(EvalError::TypeMismatch {
808                    detail: alloc::format!("invalid input syntax for type bytea: {msg}"),
809                }),
810            },
811            // v7.39 (round 544) — an integer's two's-complement bytes,
812            // big-endian, at the source type's width. Measured on PG18:
813            // 5::int2 -> \x0005, 5::int4 -> \x00000005,
814            // 5::int8 -> \x0000000000000005, (-1)::int4 -> \xffffffff.
815            Value::SmallInt(n) => Ok(Value::bytes(n.to_be_bytes().to_vec())),
816            Value::Int(n) => Ok(Value::bytes(n.to_be_bytes().to_vec())),
817            Value::BigInt(n) => Ok(Value::bytes(n.to_be_bytes().to_vec())),
818            other => Err(EvalError::TypeMismatch {
819                detail: alloc::format!(
820                    "::bytea only accepts TEXT / bytea / integer inputs, got {}",
821                    crate::conversions::pg_type_name_for_error_opt(other.data_type())
822                ),
823            }),
824        },
825        CastTarget::Named(name) => {
826            // v7.39 (round 777, F31-E1) — a typmod'd ARRAY cast:
827            // `::numeric(3,1)[]` arrives as Named("numeric(3,1)_array")
828            // and fell through to the user-type lookup ('type
829            // "numeric(3,1)_array" does not exist'). PG applies the
830            // modifier per element ({1.5, 2.3}, measured). Cast to the
831            // bare base array first, then run every element through the
832            // scalar typmod cast.
833            if let Some(base_paren) = name.strip_suffix("_array")
834                && base_paren.ends_with(')')
835                && let Some(popen) = base_paren.find('(')
836            {
837                let base = &base_paren[..popen];
838                let arr = cast_value_ref_in(
839                    v,
840                    &CastTarget::Named(alloc::format!("{base}_array")),
841                    mysql,
842                )?;
843                let scalar = CastTarget::Named(alloc::string::String::from(base_paren));
844                return match arr {
845                    Value::NumericArray(items) => {
846                        let mut out = alloc::vec::Vec::with_capacity(items.len());
847                        for it in items {
848                            out.push(match it {
849                                None => None,
850                                Some((scaled, scale)) => {
851                                    match cast_value_ref_in(
852                                        Value::Numeric { scaled, scale, kind: spg_storage::NumericKind::Finite },
853                                        &scalar,
854                                        mysql,
855                                    )? {
856                                        Value::Numeric { scaled, scale, .. } => {
857                                            Some((scaled, scale))
858                                        }
859                                        Value::NumericBig(b) => {
860                                            return Err(EvalError::TypeMismatch {
861                                                detail: alloc::format!(
862                                                    "numeric value too large for {base_paren}[]: {b:?}"
863                                                ),
864                                            });
865                                        }
866                                        Value::Null => None,
867                                        other => {
868                                            return Err(EvalError::TypeMismatch {
869                                                detail: alloc::format!(
870                                                    "unexpected element cast result {other:?}"
871                                                ),
872                                            });
873                                        }
874                                    }
875                                }
876                            });
877                        }
878                        Ok(Value::NumericArray(out))
879                    }
880                    other => Ok(other),
881                };
882            }
883            // v7.39 (round 613) — a plain scalar spelling goes straight to
884            // the tail. See `PLAIN_NAMED_TARGETS` for why that is the same
885            // thing as walking the arm, and the pin for the check that says
886            // so mechanically.
887            if let Some(dt) = plain_named_target(name) {
888                return finish_named_cast(v, dt, name, None, mysql);
889            }
890            // v7.38 (read01) — a temporal type with a fractional-seconds
891            // precision (`time(3)`, `timestamp(0)`, `timestamptz(2)`) rounds the
892            // sub-second field to that many digits, like PG. Resolve against the
893            // base type (`type_name_to_data_type` does not know the `(N)` form)
894            // and round the coerced result below.
895            // v7.38 (read01, T20) — an integer casts to `bit(n)` as the low n
896            // bits of its two's-complement representation (PG; int→varbit is
897            // rejected there, so only fixed-length `bit` is handled here).
898            if matches!(v, Value::Int(_) | Value::BigInt(_) | Value::SmallInt(_)) {
899                if let Some(width) = bit_cast_width(name) {
900                    return int_to_bit_string(v, width.0);
901                }
902            }
903            // v7.39 (read01 varbit.c) — internal exact-length form for
904            // B'...' literals (an explicit ::bit means bit(1) below).
905            if name == "__bit_literal" {
906                return match &v {
907                    Value::Null => Ok(Value::Null),
908                    Value::Text(s) => match crate::conversions::parse_bit_string_text(s) {
909                        Some((nb, by)) => Ok(Value::bit_string(nb, by)),
910                        None => Err(EvalError::TypeMismatch {
911                            detail: alloc::format!("invalid input syntax for type bit: \"{s}\""),
912                        }),
913                    },
914                    Value::BitString { .. } => Ok(v),
915                    other => Err(EvalError::TypeMismatch {
916                        detail: alloc::format!(
917                            "cannot cast {} to bit",
918                            crate::conversions::pg_type_name_for_error_opt(other.data_type())
919                        ),
920                    }),
921                };
922            }
923            // v7.39 (read01 varbit.c) — `bit(n)` over a bit string (or a
924            // '0101' text form) zero-extends on the RIGHT or truncates to
925            // n (PG's bit() cast, unlike the input-time exact-length rule).
926            let bit_src: Option<Value<'static>> = match &v {
927                Value::BitString { .. } => Some(v.clone()),
928                Value::Text(s) if bit_cast_width(name).is_some() => {
929                    match crate::conversions::parse_bit_string_text(s) {
930                        Some((nb, by)) => Some(Value::bit_string(nb, by)),
931                        None => {
932                            let bad = s.chars().find(|c| *c != '0' && *c != '1');
933                            return Err(EvalError::TypeMismatch {
934                                detail: match bad {
935                                    Some(c) => {
936                                        alloc::format!("\"{c}\" is not a valid binary digit")
937                                    }
938                                    None => {
939                                        alloc::format!("invalid input syntax for type bit: \"{s}\"")
940                                    }
941                                },
942                            });
943                        }
944                    }
945                }
946                _ => None,
947            };
948            if let Some(Value::BitString { nbits, bytes }) = &bit_src {
949                if let Some((width, pads)) = bit_cast_width(name) {
950                    // varbit truncates but never pads.
951                    if !pads && *nbits <= width {
952                        return Ok(Value::BitString {
953                            nbits: *nbits,
954                            bytes: alloc::borrow::Cow::Owned(bytes.to_vec()),
955                        });
956                    }
957                    let mut bits: alloc::vec::Vec<bool> = (0..*nbits as usize)
958                        .map(|i| bytes[i / 8] & (0x80 >> (i % 8)) != 0)
959                        .collect();
960                    bits.resize(width as usize, false);
961                    let mut out = alloc::vec![0u8; width.div_ceil(8) as usize];
962                    for (i, b) in bits.iter().enumerate() {
963                        if *b {
964                            out[i / 8] |= 0x80 >> (i % 8);
965                        }
966                    }
967                    return Ok(Value::BitString {
968                        nbits: width,
969                        bytes: alloc::borrow::Cow::Owned(out),
970                    });
971                }
972            }
973            // v7.39 (read01 oid.c) — OID is unsigned 32-bit: a negative
974            // integer wraps (PG's (Oid) cast semantics: -1 -> 4294967295),
975            // beyond u32 errors "OID out of range", bad text is 22P02.
976            // v7.39 (read01 oid.c) — OID is unsigned 32-bit: a negative
977            // integer wraps (PG's (Oid) cast semantics: -1 -> 4294967295),
978            // beyond u32 errors "OID out of range", bad text is 22P02.
979            //
980            // Round 667 moved the rules to `conversions::coerce_to_oid` so
981            // the column-assignment path shares them instead of growing a
982            // second copy.
983            if name.eq_ignore_ascii_case("oid")
984                && let Some(out) = crate::conversions::coerce_to_oid(&v)?
985            {
986                return Ok(out);
987            }
988            // v7.39 (read01 mac8.c) — macaddr8 -> macaddr requires the
989            // EUI-64 ff:fe infix; anything else is PG's dedicated error.
990            if name.eq_ignore_ascii_case("macaddr") {
991                if let Value::Macaddr8(b) = &v {
992                    if b[3] == 0xff && b[4] == 0xfe {
993                        return Ok(Value::Macaddr([b[0], b[1], b[2], b[5], b[6], b[7]]));
994                    }
995                    return Err(EvalError::TypeMismatch {
996                        detail: "macaddr8 data out of range to convert to macaddr".into(),
997                    });
998                }
999            }
1000            // v7.39 (read01 regproc.c) — the remaining reg* input types.
1001            // SPG carries them as their canonical text rendering; name
1002            // resolution runs against the static pg_proc table / the FTS
1003            // configuration list.
1004            // v7.39 (round 607) — matched against the static list rather than
1005            // through an owned lowercase copy. The copy was built for every
1006            // row and thrown away on every row that is not one of these.
1007            if let Some(lower_name) = REG_MISC_TYPES
1008                .iter()
1009                .copied()
1010                .find(|k| name.eq_ignore_ascii_case(k))
1011            {
1012                let s = match &v {
1013                    Value::Null => return Ok(Value::Null),
1014                    Value::Text(s) => s.as_ref().trim().to_string(),
1015                    // v7.39 (round 634) — an OID reaches these types too.
1016                    // PG registers int2/int4/int8/oid -> regproc as IMPLICIT
1017                    // casts and renders an oid with no matching entry as the
1018                    // number itself: `1::INT::REGPROC` is `1`, and
1019                    // `1247::OID::REGPROC` is `1247`. SPG refused the whole
1020                    // integer family with "accepts TEXT".
1021                    Value::SmallInt(n) => return Ok(Value::text(n.to_string())),
1022                    Value::Int(n) => return Ok(Value::text(n.to_string())),
1023                    Value::BigInt(n) => return Ok(Value::text(n.to_string())),
1024                    other => {
1025                        return Err(EvalError::TypeMismatch {
1026                            detail: alloc::format!(
1027                                "::{lower_name} accepts TEXT, got {}",
1028                                crate::conversions::pg_type_name_for_error_opt(other.data_type())
1029                            ),
1030                        });
1031                    }
1032                };
1033                return cast_reg_misc(lower_name, &s);
1034            }
1035            // v7.39 (round 514) — the remaining catalog-shaped types. Each
1036            // validates its own text form and keeps it, which is what PG's
1037            // input functions do; the wordings below are PG18 readings.
1038            if let Some(out) = cast_catalog_scalar(name, &v)? {
1039                return Ok(out);
1040            }
1041            // v7.39 (round 511) — `'(0,1)'::tid`, so a caller can name a row
1042            // it read a ctid from earlier. PG's text form is the only input
1043            // shape it has.
1044            if name.eq_ignore_ascii_case("tid") {
1045                return match &v {
1046                    Value::Tid(..) => Ok(v),
1047                    Value::Text(t) => parse_tid_text(t).ok_or_else(|| EvalError::TypeMismatch {
1048                        detail: alloc::format!("invalid input syntax for type tid: \"{t}\""),
1049                    }),
1050                    other => Err(EvalError::TypeMismatch {
1051                        detail: alloc::format!(
1052                            "cannot cast type {} to tid",
1053                            crate::eval::strings::pg_typeof_name(other)
1054                        ),
1055                    }),
1056                };
1057            }
1058            // v7.39 (read01 pseudotypes.c) — casting a value INTO a
1059            // pseudotype hits PG's dummy input functions (0A000).
1060            if let Some(lower) = OPAQUE_TYPES
1061                .iter()
1062                .copied()
1063                .find(|k| name.eq_ignore_ascii_case(k))
1064            {
1065                // v7.39 (round 509) — a pseudotype is a REAL type name,
1066                // so `NULL::anyarray` is NULL on PG, not an error. Only a
1067                // VALUE hits the dummy input function. Before this the
1068                // NULL case fell through to the type table below, which
1069                // does not carry the pseudotypes, and once NULL stopped
1070                // short-circuiting the whole cast it started reporting
1071                // them as unknown types.
1072                return if matches!(v, Value::Null) {
1073                    Ok(Value::Null)
1074                } else {
1075                    Err(EvalError::TypeMismatch {
1076                        detail: alloc::format!("cannot accept a value of type {lower}"),
1077                    })
1078                };
1079            }
1080            // v7.39 (read01 pseudotypes.c) — `::cstring` is PG's I/O-form
1081            // pseudotype: text in, text out (cstring_in/out are identity).
1082            // SPG carries it as text; pg_typeof(cstring) reading "text" is
1083            // a recorded delta alongside the literal projection OIDs.
1084            // v7.39 (read01 xid8funcs.c) — `::xid` (32-bit, wrapping) and
1085            // `::xid8` (64-bit, full) parse an integer text and render it
1086            // back verbatim. SPG carries them as BigInt.
1087            if name.eq_ignore_ascii_case("xid") || name.eq_ignore_ascii_case("xid8") {
1088                return Ok(match v {
1089                    Value::Null => Value::Null,
1090                    Value::SmallInt(n) => Value::BigInt(i64::from(n)),
1091                    Value::Int(n) => Value::BigInt(i64::from(n)),
1092                    Value::BigInt(n) => Value::BigInt(n),
1093                    Value::Text(s) => {
1094                        let t = s.trim();
1095                        match t.parse::<u64>() {
1096                            Ok(n) => Value::BigInt(n as i64),
1097                            Err(_) => {
1098                                return Err(EvalError::TypeMismatch {
1099                                    detail: alloc::format!(
1100                                        "invalid input syntax for type {}: \"{s}\"",
1101                                        name.to_ascii_lowercase()
1102                                    ),
1103                                });
1104                            }
1105                        }
1106                    }
1107                    other => {
1108                        return Err(EvalError::TypeMismatch {
1109                            detail: alloc::format!(
1110                                "cannot cast {} to {name}",
1111                                crate::conversions::pg_type_name_for_error_opt(other.data_type())
1112                            ),
1113                        });
1114                    }
1115                });
1116            }
1117            // v7.39 (read01 varchar.c) — `::name` is text truncated to
1118            // NAMEDATALEN-1 (63) bytes.
1119            if name.eq_ignore_ascii_case("name") {
1120                return Ok(match v {
1121                    Value::Null => Value::Null,
1122                    other => {
1123                        let t = match other {
1124                            Value::Text(s) => s.into_owned(),
1125                            o => value_to_text(&o),
1126                        };
1127                        let mut cut = t;
1128                        if cut.len() > 63 {
1129                            let mut idx = 63;
1130                            while !cut.is_char_boundary(idx) {
1131                                idx -= 1;
1132                            }
1133                            cut.truncate(idx);
1134                        }
1135                        Value::text(cut)
1136                    }
1137                });
1138            }
1139            if name.eq_ignore_ascii_case("cstring") {
1140                return Ok(match v {
1141                    Value::Null => Value::Null,
1142                    Value::Text(s) => Value::Text(s),
1143                    other => Value::text(value_to_text(&other)),
1144                });
1145            }
1146            // v7.39 (read01 jsonpath.c) — `::jsonpath` parses and prints
1147            // the canonical form (PG's jsonpath type; SPG carries it as
1148            // text — the wire OID is a recorded residual with the other
1149            // literal projection OIDs).
1150            if name.eq_ignore_ascii_case("jsonpath") {
1151                return match v {
1152                    Value::Null => Ok(Value::Null),
1153                    Value::Text(s) => Ok(Value::text(crate::json::jsonpath_canonical(s.as_ref())?)),
1154                    other => Err(EvalError::TypeMismatch {
1155                        detail: alloc::format!(
1156                            "cannot cast {} to jsonpath",
1157                            crate::conversions::pg_type_name_for_error_opt(other.data_type())
1158                        ),
1159                    }),
1160                };
1161            }
1162            // v7.39 (round 355, M13) — MySQL's `BINARY` / `BINARY(n)`.
1163            // It is a COLLATION coercion, not a type change: MariaDB
1164            // renders `BINARY 'abc'` as `abc` (and `HEX()` of it as
1165            // 616263), so the value passes through unchanged; `(n)`
1166            // truncates to n bytes (`CAST('abc' AS BINARY(2))` is `ab`,
1167            // measured). What it really buys is byte-wise comparison,
1168            // which `compare_is_case_insensitive` now refuses to fold.
1169            if mysql
1170                && (name.eq_ignore_ascii_case("binary")
1171                    || name.to_ascii_lowercase().starts_with("binary("))
1172            {
1173                return cast_mysql_binary(v, name);
1174            }
1175            // v7.39 (round 352, M8) — MySQL's SIGNED / UNSIGNED targets.
1176            // Measured on MariaDB 11: a string gives its LEADING number
1177            // (`'12abc'` → 12, `'abc'` → 0); a fractional value ROUNDS
1178            // half-away-from-zero (1.5 → 2, 2.5 → 3, -2.5 → -3) rather
1179            // than truncating; and UNSIGNED wraps a negative through u64
1180            // (`-1` → 18446744073709551615).
1181            // PG has no such type — `type "signed" does not exist` — so the
1182            // reading is gated on the dialect, not just on the spelling.
1183            if mysql
1184                && (name.eq_ignore_ascii_case("signed") || name.eq_ignore_ascii_case("unsigned"))
1185            {
1186                return cast_mysql_integer(v, name.eq_ignore_ascii_case("unsigned"));
1187            }
1188            // v7.39 (round 423) — a bare MySQL temporal type carries
1189            // fractional precision 0, so `CAST(x AS DATETIME)` drops the
1190            // fraction (measured on MariaDB 11). PG's `::timestamp` keeps
1191            // every microsecond, so the default is dialect-gated.
1192            let temporal_prec = temporal_typmod(name)
1193                .or_else(|| (mysql && is_bare_temporal_type(name)).then_some(0));
1194            let resolve_name: alloc::borrow::Cow<'_, str> = if temporal_prec.is_some() {
1195                alloc::borrow::Cow::Owned(name.split('(').next().unwrap_or(name).trim().to_string())
1196            } else {
1197                alloc::borrow::Cow::Borrowed(name.as_str())
1198            };
1199            // v7.37.5 ship triage — generic typed-cast dispatch.
1200            // Resolve the ident to a `DataType` and route the value
1201            // through the existing `coerce_value` text-decoder for
1202            // every v7.37.5 γ/δ/ε/ζ-A type that already speaks
1203            // Text→typed via codec.
1204            let dt =
1205                crate::conversions::type_name_to_data_type(&resolve_name).ok_or_else(|| {
1206                    // v7.39 (round 272) — a numeric typmod outside PG's
1207                    // bounds gets PG's own wording rather than being
1208                    // reported as an unknown type.
1209                    // v7.39 (round 620) — and an unknown one is PG's
1210                    // wording, which also earns it PG's SQLSTATE (42704
1211                    // UNDEFINED_OBJECT; `unsupported cast target` fell
1212                    // through to the generic 42000).
1213                    EvalError::TypeMismatch {
1214                        detail: crate::conversions::numeric_typmod_error(&resolve_name)
1215                            .unwrap_or_else(|| unknown_type_error_text(name)),
1216                    }
1217                })?;
1218            finish_named_cast(v, dt, &resolve_name, temporal_prec, mysql)
1219        }
1220    }
1221}
1222
1223/// v7.39 (round 613) — the tail of the `Named` arm: stringify for the text
1224/// targets, coerce, and round a temporal precision. Split out so the fast
1225/// path below reaches exactly this code rather than a copy of it.
1226/// v7.39 (round 722) — the compiled `Step::CastPlain` entry: same tail,
1227/// name pre-resolved at compile time.
1228pub(crate) fn finish_named_cast_plain(
1229    v: Value<'static>,
1230    dt: spg_storage::DataType,
1231    resolve_name: &str,
1232    mysql: bool,
1233) -> Result<Value<'static>, EvalError> {
1234    finish_named_cast(v, dt, resolve_name, None, mysql)
1235}
1236
1237fn finish_named_cast(
1238    v: Value<'static>,
1239    dt: spg_storage::DataType,
1240    resolve_name: &str,
1241    temporal_prec: Option<u8>,
1242    mysql: bool,
1243) -> Result<Value<'static>, EvalError> {
1244    // PG semantics: any value casts to varchar(n) / char(n) through its text
1245    // representation (`99::char(2)` → '99'), and an EXPLICIT cast truncates
1246    // to n characters — only column assignment errors on overflow. Stringify
1247    // a non-text source first, then truncate up front so the coerce path's
1248    // length contract never fires here.
1249    let v = match (&dt, v) {
1250        // v7.38 (read01) — an explicit cast to TEXT stringifies any
1251        // value (`text(42)` → '42'), matching `42::text`. (coerce_value
1252        // deliberately rejects a bare INT→TEXT so INSERT stays strict.)
1253        (spg_storage::DataType::Text, v) => match v {
1254            Value::Text(s) => Value::Text(s),
1255            // v7.39 (read01 inet family) — inet/cidr ::text carries
1256            // the mask even at full length (cast-path form).
1257            Value::Inet { family, bits, addr } | Value::Cidr { family, bits, addr } => {
1258                Value::text(crate::conversions::format_inet_full(family, bits, &addr))
1259            }
1260            other => Value::text(value_to_text(&other)),
1261        },
1262        (spg_storage::DataType::Varchar(n) | spg_storage::DataType::Char(n), v) => {
1263            // v7.37 D.36 — previously only `Value::Text` was handled, so
1264            // `99::char(2)` reached coerce_value as an INT and hit a
1265            // CHAR/INT storage type-mismatch.
1266            // v7.39 (bpchar epic) — a bpchar source enters through its
1267            // text cast (trailing blanks stripped): `::varchar` keeps
1268            // the stripped form, `::char(m)` re-pads in coerce_value.
1269            let s = match v {
1270                Value::Text(s) => s.into_owned(),
1271                Value::BpChar(s) => s.trim_end_matches(' ').to_string(),
1272                other => value_to_text(&other),
1273            };
1274            let s = if *n > 0 && s.chars().count() > *n as usize {
1275                s.chars()
1276                    .take(*n as usize)
1277                    .collect::<alloc::string::String>()
1278            } else {
1279                s
1280            };
1281            Value::text(s)
1282        }
1283        (_, v) => v,
1284    };
1285    let coerced =
1286        crate::conversions::coerce_value(v, dt, resolve_name, 0).map_err(|e| match e {
1287            // v7.39 (read01 round 113) — pass an already-classed engine
1288            // error through unchanged. Re-stringifying via Display would
1289            // double the "eval: type mismatch: " class prefix (the wire
1290            // strips only the outermost one), leaking it into the message
1291            // — visible now that jsonb → numeric casts error with PG's
1292            // exact "cannot cast jsonb string to type numeric" wording.
1293            crate::EngineError::Eval(ev) => ev,
1294            // v7.39 (round 622, S05a) — `coerce_value` is the INSERT-time
1295            // COLUMN coercion, and a cast borrows it. Its rejection is
1296            // phrased for a column, so `SELECT 1::INET` answered
1297            //
1298            //   type mismatch in column "inet" (position 0): expected INET,
1299            //   got INT
1300            //
1301            // naming a column that does not exist, at a position that means
1302            // nothing, in the storage layer's own vocabulary. PG says
1303            // `cannot cast type integer to inet`. The column phrasing stays
1304            // where it belongs — an INSERT still says which column — and a
1305            // failed cast now says what it failed to cast, like every other
1306            // arm in this file already did.
1307            //
1308            // The two type names come off the error itself, which already
1309            // carries them as `DataType`. Naming them BEFORE the call — the
1310            // obvious way to write this, since the value and the target both
1311            // move into it — costs two `String`s on every SUCCESSFUL cast,
1312            // and the panel caught exactly that: `id::NUMERIC` 23.75 ->
1313            // 55.48 ms, `id::REAL` 21.55 -> 50.48. This is the same eager
1314            // error construction round 614 removed from 28 call sites,
1315            // rebuilt by hand a round later.
1316            crate::EngineError::Storage(spg_storage::StorageError::TypeMismatch {
1317                expected,
1318                actual,
1319                ..
1320            }) => EvalError::TypeMismatch {
1321                detail: alloc::format!(
1322                    "cannot cast {} to {}",
1323                    crate::conversions::pg_type_name_for_error(actual),
1324                    crate::conversions::pg_type_name_for_error(expected)
1325                ),
1326            },
1327            other => EvalError::TypeMismatch {
1328                detail: alloc::format!("{other}"),
1329            },
1330        })?;
1331    Ok(match temporal_prec {
1332        Some(prec) => round_temporal_to_precision(coerced, prec, mysql),
1333        None => coerced,
1334    })
1335}
1336
1337/// v7.39 (round 613) — the plain scalar spellings, with the type each one
1338/// resolves to.
1339///
1340/// Round 612 measured the `Named` arm re-deriving everything for every row:
1341/// `s::VARCHAR` cost 30.6 ms over 200k rows where `s::TEXT` — the identical
1342/// conversion, under a spelling the parser settles into a `CastTarget`
1343/// variant — cost 11.2, and probes split the difference across the whole arm
1344/// rather than any one place in it. These names reach the tail directly.
1345///
1346/// Both halves of that shortcut are checked mechanically by the pin, not by
1347/// eye: every entry's type is asserted to equal `type_name_to_data_type`'s
1348/// answer, and every entry is asserted absent from each arm above the
1349/// resolve (the reg-misc / catalog-scalar / opaque lists, `tid`, `xid`,
1350/// `xid8`, `jsonpath`, the MySQL `binary` / `signed` / `unsigned` names, and
1351/// the bit and temporal spellings). A name that grows a special case has to
1352/// leave this table, and the pin says so.
1353const PLAIN_NAMED_TARGETS: &[(&str, spg_storage::DataType)] = &[
1354    ("text", spg_storage::DataType::Text),
1355    ("varchar", spg_storage::DataType::Varchar(0)),
1356    ("character varying", spg_storage::DataType::Varchar(0)),
1357    (
1358        "numeric",
1359        spg_storage::DataType::Numeric {
1360            precision: 0,
1361            scale: 0,
1362        },
1363    ),
1364    (
1365        "decimal",
1366        spg_storage::DataType::Numeric {
1367            precision: 0,
1368            scale: 0,
1369        },
1370    ),
1371    ("real", spg_storage::DataType::Real),
1372    ("float4", spg_storage::DataType::Real),
1373    ("float8", spg_storage::DataType::Float),
1374    ("double precision", spg_storage::DataType::Float),
1375    ("int2", spg_storage::DataType::SmallInt),
1376    ("smallint", spg_storage::DataType::SmallInt),
1377    ("int4", spg_storage::DataType::Int),
1378    ("integer", spg_storage::DataType::Int),
1379    ("int8", spg_storage::DataType::BigInt),
1380    ("bool", spg_storage::DataType::Bool),
1381    ("boolean", spg_storage::DataType::Bool),
1382    ("date", spg_storage::DataType::Date),
1383    ("bytea", spg_storage::DataType::Bytes),
1384    ("uuid", spg_storage::DataType::Uuid),
1385];
1386
1387/// v7.39 (round 613) — the heads that may carry a typmod and are still
1388/// plain: `varchar(20)`, `char(4)`, `numeric(10,2)`. The type comes from
1389/// `type_name_to_data_type` over the WHOLE name, so the typmod is parsed
1390/// exactly where it always was; only the walk down the arm is skipped. The
1391/// pin checks each head against every arm above the resolve, and that none
1392/// of them is a bit or temporal spelling.
1393pub(crate) const PLAIN_NAMED_HEADS: &[&str] = &[
1394    "varchar",
1395    "character varying",
1396    "char",
1397    "character",
1398    "bpchar",
1399    "numeric",
1400    "decimal",
1401];
1402
1403/// The type a plain scalar spelling resolves to, or `None` when the name
1404/// needs the whole arm.
1405pub(crate) fn plain_named_target(name: &str) -> Option<spg_storage::DataType> {
1406    if let Some(dt) = PLAIN_NAMED_TARGETS
1407        .iter()
1408        .find(|(k, _)| name.eq_ignore_ascii_case(k))
1409        .map(|(_, dt)| *dt)
1410    {
1411        return Some(dt);
1412    }
1413    let head = name.split('(').next()?.trim();
1414    if name.len() == head.len()
1415        || !PLAIN_NAMED_HEADS
1416            .iter()
1417            .any(|k| head.eq_ignore_ascii_case(k))
1418    {
1419        return None;
1420    }
1421    crate::conversions::type_name_to_data_type(name)
1422}
1423
1424/// The scalar type names the `Named` arm resolves without a catalog.
1425fn is_known_scalar_name(lower: &str) -> bool {
1426    REG_MISC_TYPES.contains(&lower)
1427        || CATALOG_SCALAR_TYPES.contains(&lower)
1428        || OPAQUE_TYPES.contains(&lower)
1429        || matches!(
1430            lower,
1431            "tid"
1432                | "record"
1433                | "cstring"
1434                | "regnamespace"
1435                | "regrole"
1436                // Round 896 — the target validator runs before the arm, so
1437                // the quoted spelling has to be a known name here too or it
1438                // is rejected before the fold above ever sees it.
1439                | "regclass"
1440                | "regtype"
1441        )
1442}
1443
1444/// v7.39 (round 509) — does this name a type at all?
1445///
1446/// PG validates the cast TARGET whatever the operand is: `NULL::nosuchtype`
1447/// is an error there, not NULL. `cast_value_in` short-circuits a NULL before
1448/// it ever looks at the target, so the check has to happen in the caller —
1449/// and `eval_cast_arm` is the caller that has a catalog, which is what
1450/// enums, domains, composites and table row types need.
1451///
1452/// This lists what the `Named` arm below resolves WITHOUT a catalog. Keeping
1453/// the two in step is a real hazard: a first cut of this check missed three
1454/// live spellings — `::binary` (the MySQL prefix's desugar), a table's row
1455/// type, and the pseudotypes — and the e2e suite caught every one. It is the
1456/// check on this function.
1457/// v7.39 (round 620) — PG's wording for a cast target that names no type.
1458///
1459/// SPG said ``unsupported cast target `::nosuchtype` ``, which reads as "SPG
1460/// has not got round to that one" when what happened is that no such type
1461/// exists anywhere. PG says `type "nosuchtype" does not exist`, and because
1462/// the wire classifies by message text, saying it also moves the code off the
1463/// generic 42000 onto 42704 UNDEFINED_OBJECT.
1464pub(crate) fn unknown_type_error_text(name: &str) -> alloc::string::String {
1465    alloc::format!("type \"{name}\" does not exist")
1466}
1467
1468pub(crate) fn builtin_target_resolves(name: &str, mysql: bool) -> bool {
1469    if name == "__bit_literal" || bit_cast_width(name).is_some() {
1470        return true;
1471    }
1472    crate::conversions::with_lower_name(name, |lower| {
1473        builtin_target_resolves_lower(name, lower, mysql)
1474    })
1475}
1476
1477fn builtin_target_resolves_lower(name: &str, lower: &str, mysql: bool) -> bool {
1478    // The three families, read from the same declarations the value path
1479    // dispatches on — see their doc comment for why that matters.
1480    if is_known_scalar_name(lower) {
1481        return true;
1482    }
1483    // v7.39 (round 515) — `<element>[]`, which this parser names
1484    // `<element>_array`. PG has an array type for every scalar, so the rule
1485    // is the stem's: `NULL::cstring[]`, `NULL::aclitem[]` and
1486    // `NULL::"char"[]` all resolve there. A general rule rather than three
1487    // entries, because the next scalar added would otherwise need a fourth.
1488    if let Some(stem) = lower.strip_suffix("_array")
1489        && (is_known_scalar_name(stem)
1490            || crate::conversions::type_name_to_data_type(stem).is_some())
1491    {
1492        return true;
1493    }
1494    if mysql && matches!(lower, "binary" | "signed" | "unsigned") {
1495        return true;
1496    }
1497    let base = if temporal_typmod(name).is_some() || (mysql && is_bare_temporal_type(name)) {
1498        name.split('(').next().unwrap_or(name).trim()
1499    } else {
1500        name
1501    };
1502    crate::conversions::type_name_to_data_type(base).is_some()
1503        || crate::conversions::numeric_typmod_error(base).is_some()
1504}
1505
1506/// v7.39 (round 511) — PG's `(block,offset)` text form for a tid.
1507fn parse_tid_text(t: &str) -> Option<Value<'static>> {
1508    let inner = t.trim().strip_prefix('(')?.strip_suffix(')')?;
1509    let (b, o) = inner.split_once(',')?;
1510    Some(Value::Tid(
1511        b.trim().parse::<u32>().ok()?,
1512        o.trim().parse::<u32>().ok()?,
1513    ))
1514}
1515
1516/// v7.39 (round 514) — the catalog-shaped scalar types: the ids, the oid
1517/// vectors, an ACL item, a cursor name and a transaction snapshot.
1518///
1519/// `Some` when `name` is one of them, so the caller can fall through to
1520/// everything else. Every error wording is a PG18 reading — they differ per
1521/// type and per ELEMENT (`::oidvector` complains about `oid`,
1522/// `::int2vector` about `smallint`), which is why they are spelled out
1523/// rather than shared.
1524fn cast_catalog_scalar(name: &str, v: &Value<'_>) -> Result<Option<Value<'static>>, EvalError> {
1525    crate::conversions::with_lower_name(name, |lower| cast_catalog_scalar_lower(lower, v))
1526}
1527
1528fn cast_catalog_scalar_lower(
1529    lower: &str,
1530    v: &Value<'_>,
1531) -> Result<Option<Value<'static>>, EvalError> {
1532    // v7.39 (round 515) — `<element>[]` runs the element's own check over
1533    // each member and keeps the literal, which is what PG does: measured,
1534    // `'{a,b}'::aclitem[]` is "unrecognized key word: \"a\"".
1535    if let Some(stem) = lower.strip_suffix("_array")
1536        && (CATALOG_SCALAR_TYPES.contains(&stem) || OPAQUE_TYPES.contains(&stem))
1537    {
1538        let Value::Text(t) = v else {
1539            return Ok(None);
1540        };
1541        let body = t.trim();
1542        let inner = body
1543            .strip_prefix('{')
1544            .and_then(|b| b.strip_suffix('}'))
1545            .unwrap_or(body);
1546        for part in inner.split(',').filter(|p| !p.trim().is_empty()) {
1547            cast_catalog_scalar(stem, &Value::text(part.trim().to_string()))?;
1548        }
1549        return Ok(Some(Value::text(body.to_string())));
1550    }
1551    if !CATALOG_SCALAR_TYPES.contains(&lower) {
1552        return Ok(None);
1553    }
1554    let text = match v {
1555        Value::Text(t) => t.to_string(),
1556        Value::Cid(c) if lower == "cid" => return Ok(Some(Value::Cid(*c))),
1557        Value::Xid(x) if lower == "xid" => return Ok(Some(Value::Xid(*x))),
1558        // v7.39 (round 641) — PG has no cast between an integer and a
1559        // transaction id in either direction: `5::xid` is "cannot cast
1560        // type integer to xid" and `'5'::xid::int` is the mirror of it,
1561        // measured. The unknown-literal spelling `'5'::xid` is a
1562        // different thing — that is the type's input function, and it is
1563        // the Text arm above. Only `xid` is carved out here; `cid`,
1564        // `oid` and the vector types keep taking an integer.
1565        Value::SmallInt(_) | Value::Int(_) | Value::BigInt(_) if lower == "xid" => {
1566            return Err(EvalError::TypeMismatch {
1567                detail: alloc::format!(
1568                    "cannot cast type {} to xid",
1569                    crate::eval::strings::pg_typeof_name(v)
1570                ),
1571            });
1572        }
1573        Value::SmallInt(n) => alloc::format!("{n}"),
1574        Value::Int(n) => alloc::format!("{n}"),
1575        Value::BigInt(n) => alloc::format!("{n}"),
1576        other => {
1577            return Err(EvalError::TypeMismatch {
1578                detail: alloc::format!(
1579                    "cannot cast type {} to {lower}",
1580                    crate::eval::strings::pg_typeof_name(other)
1581                ),
1582            });
1583        }
1584    };
1585    let t = text.trim();
1586    let bad = |ty: &str, what: &str| EvalError::TypeMismatch {
1587        detail: alloc::format!("invalid input syntax for type {ty}: \"{what}\""),
1588    };
1589    let out = match lower {
1590        "cid" => Value::Cid(t.parse::<u32>().map_err(|_| bad("cid", t))?),
1591        "xid" => Value::Xid(t.parse::<u32>().map_err(|_| bad("xid", t))?),
1592        // Space-separated element lists, validated element by element and
1593        // kept in their own spelling.
1594        "oidvector" | "int2vector" => {
1595            let elem_ty = if lower == "oidvector" {
1596                "oid"
1597            } else {
1598                "smallint"
1599            };
1600            for part in t.split_whitespace() {
1601                let ok = if elem_ty == "oid" {
1602                    part.parse::<u32>().is_ok()
1603                } else {
1604                    part.parse::<i16>().is_ok()
1605                };
1606                if !ok {
1607                    return Err(bad(elem_ty, part));
1608                }
1609            }
1610            Value::text(t.to_string())
1611        }
1612        // `grantee=privileges/grantor`, and PG checks the key word first:
1613        // anything before the `=` that is not a role name, `group` or
1614        // `user` is "unrecognized key word".
1615        "aclitem" => {
1616            let Some((who, rest)) = t.split_once('=') else {
1617                return Err(EvalError::TypeMismatch {
1618                    detail: alloc::format!("unrecognized key word: \"{t}\""),
1619                });
1620            };
1621            if !rest.contains('/') {
1622                return Err(EvalError::TypeMismatch {
1623                    detail: alloc::format!("a name must follow the \"/\" sign"),
1624                });
1625            }
1626            let _ = who;
1627            Value::text(t.to_string())
1628        }
1629        // A cursor name is just a name.
1630        "refcursor" => Value::text(t.to_string()),
1631        // `xmin:xmax:xip_list` — two numbers and a comma-separated tail.
1632        "pg_snapshot" | "txid_snapshot" => {
1633            let parts: alloc::vec::Vec<&str> = t.splitn(3, ':').collect();
1634            let shaped = parts.len() == 3
1635                && parts[0].parse::<u64>().is_ok()
1636                && parts[1].parse::<u64>().is_ok()
1637                && (parts[2].is_empty() || parts[2].split(',').all(|x| x.parse::<u64>().is_ok()));
1638            if !shaped {
1639                return Err(bad(lower, t));
1640            }
1641            Value::text(t.to_string())
1642        }
1643        // PG normalises a path on input: `$.a` reads back `$."a"`. The
1644        // engine already has the parser its operators use.
1645        "jsonpath" => Value::text(crate::json::jsonpath_canonical(t)?),
1646        _ => unreachable!("guarded above"),
1647    };
1648    Ok(Some(out))
1649}
1650
1651/// v7.38 (read01, T20) — width of a `bit` cast target: bare `bit` is `bit(1)`,
1652/// `bit(N)` is N. `None` for `varbit` / `bit varying` (PG rejects int→varbit) and
1653/// any non-bit name.
1654fn bit_cast_width(name: &str) -> Option<(u32, bool)> {
1655    crate::conversions::with_lower_name(name, bit_cast_width_lower)
1656}
1657
1658fn bit_cast_width_lower(lower: &str) -> Option<(u32, bool)> {
1659    let trimmed = lower.trim();
1660    if trimmed == "bit" {
1661        return Some((1, true));
1662    }
1663    // v7.39 (round 281) — `varbit(n)` / `bit varying(n)` adjust on an
1664    // explicit cast too, but only DOWN: PG truncates a too-long value
1665    // and leaves a shorter one alone, where `bit(n)` also pads.
1666    for (prefix, pads) in [("varbit", false), ("bit varying", false), ("bit", true)] {
1667        if let Some(rest) = trimmed.strip_prefix(prefix) {
1668            let rest = rest.trim_start();
1669            if let Some(inner) = rest.strip_prefix('(').and_then(|r| r.strip_suffix(')'))
1670                && let Ok(n) = inner.trim().parse::<u32>()
1671            {
1672                return Some((n, pads));
1673            }
1674        }
1675    }
1676    None
1677}
1678
1679/// v7.38 (read01, T20) — build a `bit(width)` value from an integer: the low
1680/// `width` bits of the two's-complement, packed MSB-first / left-aligned (the
1681/// on-wire bit layout). Widths past 64 sign-extend.
1682fn int_to_bit_string(v: Value<'static>, width: u32) -> Result<Value<'static>, EvalError> {
1683    let n: i64 = match v {
1684        Value::Int(x) => i64::from(x),
1685        Value::BigInt(x) => x,
1686        Value::SmallInt(x) => i64::from(x),
1687        _ => {
1688            return Err(EvalError::TypeMismatch {
1689                detail: "int_to_bit_string: non-integer source".into(),
1690            });
1691        }
1692    };
1693    let w = width as usize;
1694    let mut bytes = alloc::vec![0u8; w.div_ceil(8)];
1695    for i in 0..w {
1696        let p = w - 1 - i; // bit position counted from the LSB
1697        let bit = if p >= 64 {
1698            u8::from(n < 0) // sign-extend beyond the integer's width
1699        } else {
1700            ((n >> p) & 1) as u8
1701        };
1702        if bit != 0 {
1703            bytes[i / 8] |= 1 << (7 - (i % 8));
1704        }
1705    }
1706    Ok(Value::bit_string(width, bytes))
1707}
1708
1709/// Extract the fractional-seconds precision from a temporal cast name like
1710/// `time(3)` / `timestamp(0)` / `timestamptz(2)`; `None` for any non-temporal
1711/// type or a bare temporal type with no `(N)`.
1712fn temporal_typmod(name: &str) -> Option<u8> {
1713    crate::conversions::with_lower_name(name, |lower| {
1714        let (base, rest) = lower.split_once('(')?;
1715        if !matches!(
1716            base.trim(),
1717            "time" | "timetz" | "timestamp" | "timestamptz" | "datetime"
1718        ) {
1719            return None;
1720        }
1721        let digits = rest.trim_start();
1722        let end = digits
1723            .find(|c: char| !c.is_ascii_digit())
1724            .unwrap_or(digits.len());
1725        digits[..end].parse::<u8>().ok()
1726    })
1727}
1728
1729/// Round a TIME / TIMESTAMP value's microsecond field to `prec` fractional-
1730/// second digits (`prec` 0..=6), half-away-from-zero as PG's AdjustTimestamp.
1731/// v7.39 (round 423) — `truncate` selects MySQL's reduction mode. PG's
1732/// AdjustTimestamp ROUNDS half-away-from-zero (`::timestamp(1)` of `.256` is
1733/// `.3`); MariaDB TRUNCATES toward zero (`.2`, measured). Same function, one
1734/// flag, because everything else about the reduction is identical.
1735fn round_temporal_to_precision(v: Value<'static>, prec: u8, truncate: bool) -> Value<'static> {
1736    if prec >= 6 {
1737        return v;
1738    }
1739    let scale = 10i64.pow(u32::from(6 - prec));
1740    let reduce = |micros: i64| -> i64 {
1741        if truncate {
1742            // Toward zero, so a negative time-of-day loses the same digits.
1743            (micros / scale) * scale
1744        } else {
1745            let half = scale / 2;
1746            if micros >= 0 {
1747                ((micros + half) / scale) * scale
1748            } else {
1749                -(((-micros + half) / scale) * scale)
1750            }
1751        }
1752    };
1753    match v {
1754        Value::Timestamp(m) => Value::Timestamp(reduce(m)),
1755        Value::Time(m) => Value::Time(reduce(m)),
1756        other => other,
1757    }
1758}
1759
1760/// v7.39 (round 423) — is `name` a bare temporal type (no `(N)` modifier)?
1761/// MySQL gives those fractional precision ZERO — `CAST(x AS DATETIME)` drops
1762/// the fraction entirely — where PG's `::timestamp` keeps full microseconds.
1763fn is_bare_temporal_type(name: &str) -> bool {
1764    let t = name.trim();
1765    ["time", "timestamp", "datetime"]
1766        .iter()
1767        .any(|k| t.eq_ignore_ascii_case(k))
1768}
1769
1770fn cast_to_int_array(v: Value) -> Result<Value, EvalError> {
1771    match v {
1772        Value::IntArray(items) => Ok(Value::IntArray(items)),
1773        Value::BigIntArray(items) => {
1774            let mut out: Vec<Option<i32>> = Vec::with_capacity(items.len());
1775            for item in items {
1776                match item {
1777                    None => out.push(None),
1778                    Some(n) => match i32::try_from(n) {
1779                        Ok(x) => out.push(Some(x)),
1780                        Err(_) => {
1781                            return Err(EvalError::TypeMismatch {
1782                                detail: alloc::format!("::INT[] element {n} overflows i32"),
1783                            });
1784                        }
1785                    },
1786                }
1787            }
1788            Ok(Value::IntArray(out))
1789        }
1790        Value::Text(s) => {
1791            if let Some(r) = try_cast_2d_array(&s, |row| {
1792                decode_int_array_external(row).map(Value::IntArray)
1793            }) {
1794                return r;
1795            }
1796            decode_int_array_external(&s).map(Value::IntArray)
1797        }
1798        Value::TextArray(items) => {
1799            let mut out: Vec<Option<i32>> = Vec::with_capacity(items.len());
1800            for item in items {
1801                match item {
1802                    None => out.push(None),
1803                    Some(s) => match s.parse::<i32>() {
1804                        Ok(n) => out.push(Some(n)),
1805                        Err(_) => {
1806                            return Err(EvalError::TypeMismatch {
1807                                detail: alloc::format!("::INT[] cannot parse {s:?}"),
1808                            });
1809                        }
1810                    },
1811                }
1812            }
1813            Ok(Value::IntArray(out))
1814        }
1815        other => Err(EvalError::TypeMismatch {
1816            detail: alloc::format!(
1817                "::INT[] does not accept {}",
1818                crate::conversions::pg_type_name_for_error_opt(other.data_type())
1819            ),
1820        }),
1821    }
1822}
1823
1824fn cast_to_bigint_array(v: Value) -> Result<Value, EvalError> {
1825    match v {
1826        Value::BigIntArray(items) => Ok(Value::BigIntArray(items)),
1827        Value::IntArray(items) => Ok(Value::BigIntArray(
1828            items.into_iter().map(|x| x.map(i64::from)).collect(),
1829        )),
1830        Value::Text(s) => {
1831            if let Some(r) = try_cast_2d_array(&s, |row| {
1832                decode_bigint_array_external(row).map(Value::BigIntArray)
1833            }) {
1834                return r;
1835            }
1836            decode_bigint_array_external(&s).map(Value::BigIntArray)
1837        }
1838        Value::TextArray(items) => {
1839            let mut out: Vec<Option<i64>> = Vec::with_capacity(items.len());
1840            for item in items {
1841                match item {
1842                    None => out.push(None),
1843                    Some(s) => match s.parse::<i64>() {
1844                        Ok(n) => out.push(Some(n)),
1845                        Err(_) => {
1846                            return Err(EvalError::TypeMismatch {
1847                                detail: alloc::format!("::BIGINT[] cannot parse {s:?}"),
1848                            });
1849                        }
1850                    },
1851                }
1852            }
1853            Ok(Value::BigIntArray(out))
1854        }
1855        other => Err(EvalError::TypeMismatch {
1856            detail: alloc::format!(
1857                "::BIGINT[] does not accept {}",
1858                crate::conversions::pg_type_name_for_error_opt(other.data_type())
1859            ),
1860        }),
1861    }
1862}
1863
1864/// Cast a possibly-2-D array literal: parse each top-level row with `elem` (the
1865/// 1-D element decoder) and fold into a 2-D value; `None` when the literal is 1-D.
1866fn try_cast_2d_array(
1867    s: &str,
1868    elem: impl Fn(&str) -> Result<Value<'static>, EvalError>,
1869) -> Option<Result<Value<'static>, EvalError>> {
1870    let rows = crate::eval::values::split_2d_rows(s)?;
1871    let mut row_vals: alloc::vec::Vec<Value<'static>> = alloc::vec::Vec::with_capacity(rows.len());
1872    for r in &rows {
1873        match elem(r) {
1874            Ok(v) => row_vals.push(v),
1875            Err(e) => return Some(Err(e)),
1876        }
1877    }
1878    Some(
1879        crate::eval::values::build_2d_from_rows(&row_vals).ok_or_else(|| EvalError::TypeMismatch {
1880            detail: crate::conversions::malformed_array_literal(s),
1881        }),
1882    )
1883}
1884
1885fn decode_int_array_external(s: &str) -> Result<Vec<Option<i32>>, EvalError> {
1886    let trimmed = s.trim();
1887    // v7.39 (read01 jsonfuncs.c) — the json_to_record/populate desugar
1888    // routes JSON array text ("[1,2]") through this cast; accept the
1889    // bracket form alongside PG's brace form.
1890    let inner = trimmed
1891        .strip_prefix('{')
1892        .and_then(|x| x.strip_suffix('}'))
1893        .or_else(|| trimmed.strip_prefix('[').and_then(|x| x.strip_suffix(']')))
1894        .ok_or_else(|| EvalError::TypeMismatch {
1895            detail: crate::conversions::malformed_array_literal(s),
1896        })?;
1897    if inner.trim().is_empty() {
1898        return Ok(Vec::new());
1899    }
1900    inner
1901        .split(',')
1902        .map(|part| {
1903            let p = part.trim();
1904            if p.eq_ignore_ascii_case("NULL") {
1905                Ok(None)
1906            } else {
1907                p.parse::<i32>()
1908                    .map(Some)
1909                    .map_err(|_| EvalError::TypeMismatch {
1910                        detail: alloc::format!("invalid input syntax for type integer: {p:?}"),
1911                    })
1912            }
1913        })
1914        .collect()
1915}
1916
1917fn decode_bigint_array_external(s: &str) -> Result<Vec<Option<i64>>, EvalError> {
1918    let trimmed = s.trim();
1919    let inner = trimmed
1920        .strip_prefix('{')
1921        .and_then(|x| x.strip_suffix('}'))
1922        .or_else(|| trimmed.strip_prefix('[').and_then(|x| x.strip_suffix(']')))
1923        .ok_or_else(|| EvalError::TypeMismatch {
1924            // v7.39 (round 325) — was "BIGmalformed array literal", a
1925            // stray edit that shipped: the message a client saw for
1926            // `'abc'::bigint[]` began with three letters of BIGINT.
1927            detail: crate::conversions::malformed_array_literal(s),
1928        })?;
1929    if inner.trim().is_empty() {
1930        return Ok(Vec::new());
1931    }
1932    inner
1933        .split(',')
1934        .map(|part| {
1935            let p = part.trim();
1936            if p.eq_ignore_ascii_case("NULL") {
1937                Ok(None)
1938            } else {
1939                p.parse::<i64>()
1940                    .map(Some)
1941                    .map_err(|_| EvalError::TypeMismatch {
1942                        detail: alloc::format!("invalid input syntax for type bigint: {p:?}"),
1943                    })
1944            }
1945        })
1946        .collect()
1947}
1948
1949/// v7.10.11 — same decoder as `decode_text_array_literal` in
1950/// `lib.rs`, but lives here so the eval-time cast path stays
1951/// inside `spg-engine::eval`. Kept in lock-step with the engine
1952/// `coerce_value` decoder by tests.
1953fn decode_text_array_external(s: &str) -> Result<Vec<Option<String>>, EvalError> {
1954    let trimmed = s.trim();
1955    let inner = trimmed
1956        .strip_prefix('{')
1957        .and_then(|x| x.strip_suffix('}'))
1958        .or_else(|| trimmed.strip_prefix('[').and_then(|x| x.strip_suffix(']')))
1959        .ok_or_else(|| EvalError::TypeMismatch {
1960            detail: alloc::format!("TEXT[] literal {s:?} must be enclosed in '{{...}}'"),
1961        })?;
1962    let mut out: Vec<Option<String>> = Vec::new();
1963    if inner.trim().is_empty() {
1964        return Ok(out);
1965    }
1966    let bytes = inner.as_bytes();
1967    let mut i = 0;
1968    while i <= bytes.len() {
1969        while i < bytes.len() && (bytes[i] == b' ' || bytes[i] == b'\t') {
1970            i += 1;
1971        }
1972        if i < bytes.len() && bytes[i] == b'"' {
1973            i += 1;
1974            let mut buf = String::new();
1975            while i < bytes.len() && bytes[i] != b'"' {
1976                if bytes[i] == b'\\' && i + 1 < bytes.len() {
1977                    buf.push(bytes[i + 1] as char);
1978                    i += 2;
1979                } else {
1980                    buf.push(bytes[i] as char);
1981                    i += 1;
1982                }
1983            }
1984            if i >= bytes.len() {
1985                return Err(EvalError::TypeMismatch {
1986                    detail: "unterminated quoted element in TEXT[] literal".into(),
1987                });
1988            }
1989            i += 1;
1990            out.push(Some(buf));
1991        } else {
1992            let start = i;
1993            while i < bytes.len() && bytes[i] != b',' {
1994                i += 1;
1995            }
1996            let raw = inner[start..i].trim();
1997            if raw.eq_ignore_ascii_case("NULL") {
1998                out.push(None);
1999            } else {
2000                out.push(Some(raw.to_string()));
2001            }
2002        }
2003        while i < bytes.len() && (bytes[i] == b' ' || bytes[i] == b'\t') {
2004            i += 1;
2005        }
2006        if i >= bytes.len() {
2007            break;
2008        }
2009        if bytes[i] != b',' {
2010            return Err(EvalError::TypeMismatch {
2011                detail: "expected ',' between TEXT[] elements".into(),
2012            });
2013        }
2014        i += 1;
2015    }
2016    Ok(out)
2017}
2018
2019fn cast_to_interval(v: Value) -> Result<Value, EvalError> {
2020    match v {
2021        Value::Interval {
2022            months,
2023            days,
2024            micros,
2025        } => Ok(Value::Interval {
2026            months,
2027            days,
2028            micros,
2029        }),
2030        Value::Text(s) => {
2031            let (months, days, micros) =
2032                spg_sql::parser::parse_interval_text(&s).ok_or_else(|| {
2033                    EvalError::TypeMismatch {
2034                        // v7.39 (round 324, V42) — PG's wording.
2035                        detail: alloc::format!("invalid input syntax for type interval: \"{s}\""),
2036                    }
2037                })?;
2038            Ok(Value::Interval {
2039                months,
2040                days,
2041                micros,
2042            })
2043        }
2044        other => Err(EvalError::TypeMismatch {
2045            detail: alloc::format!(
2046                "::INTERVAL only accepts TEXT-shape inputs, got {}",
2047                crate::conversions::pg_type_name_for_error_opt(other.data_type())
2048            ),
2049        }),
2050    }
2051}
2052
2053fn cast_to_date(v: Value) -> Result<Value, EvalError> {
2054    match v {
2055        Value::Date(d) => Ok(Value::Date(d)),
2056        // Integer literals carry days since the Unix epoch — used by
2057        // the `CURRENT_DATE` AST rewrite to inject the wall clock.
2058        Value::Int(n) => Ok(Value::Date(n)),
2059        Value::BigInt(n) => {
2060            i32::try_from(n)
2061                .map(Value::Date)
2062                .map_err(|_| EvalError::TypeMismatch {
2063                    detail: "bigint days-since-epoch out of DATE range".into(),
2064                })
2065        }
2066        // Timestamp truncates to its day boundary.
2067        Value::Timestamp(t) => {
2068            let days = t.div_euclid(86_400_000_000);
2069            i32::try_from(days)
2070                .map(Value::Date)
2071                .map_err(|_| EvalError::TypeMismatch {
2072                    detail: "timestamp out of DATE range".into(),
2073                })
2074        }
2075        Value::Text(s) => {
2076            if let Some(d) = parse_date_literal(&s) {
2077                return Ok(Value::Date(d));
2078            }
2079            // PG accepts a full timestamp string in a DATE cast and
2080            // truncates to the day (verified vs live PG18.4:
2081            // `'2020-01-01 12:00:00'::date` → 2020-01-01; a bad time
2082            // like `'... 25:00:00'` still raises). Reuse the timestamp
2083            // parser — it validates the time-of-day + optional TZ — then
2084            // floor to the date via the same path as the Timestamp arm.
2085            if let Some(t) = parse_timestamp_literal(&s) {
2086                let days = t.div_euclid(86_400_000_000);
2087                return i32::try_from(days)
2088                    .map(Value::Date)
2089                    .map_err(|_| EvalError::TypeMismatch {
2090                        detail: "timestamp out of DATE range".into(),
2091                    });
2092            }
2093            // PG error split: numeric-shaped input whose field values
2094            // fail the calendar checks is "out of range" (plus PG's
2095            // DateStyle hint); anything else is an input-syntax error.
2096            if super::format::date_text_is_field_shaped(&s) {
2097                return Err(EvalError::TypeMismatch {
2098                    detail: format!(
2099                        "date/time field value out of range: {s:?}\n\
2100                         HINT:  Perhaps you need a different \"DateStyle\" setting."
2101                    ),
2102                });
2103            }
2104            Err(EvalError::TypeMismatch {
2105                detail: format!("invalid input syntax for type date: {s:?}"),
2106            })
2107        }
2108        other => Err(EvalError::TypeMismatch {
2109            detail: format!(
2110                "cannot cast {} to DATE",
2111                crate::conversions::pg_type_name_for_error_opt(other.data_type())
2112            ),
2113        }),
2114    }
2115}
2116
2117fn cast_to_timestamp(v: Value) -> Result<Value, EvalError> {
2118    match v {
2119        Value::Timestamp(t) => Ok(Value::Timestamp(t)),
2120        // Int / BigInt carry microseconds since the Unix epoch — used
2121        // by the `NOW()` / `CURRENT_TIMESTAMP` AST rewrite to inject
2122        // the wall clock as a plain integer literal.
2123        Value::Int(n) => Ok(Value::Timestamp(i64::from(n))),
2124        Value::BigInt(n) => Ok(Value::Timestamp(n)),
2125        // DATE → TIMESTAMP picks midnight on the date.
2126        // v7.39 (read01 timestamp.c) — sentinel-aware (the plain multiply
2127        // overflowed on ±infinity dates).
2128        Value::Date(d) => Ok(Value::Timestamp(crate::conversions::date_days_to_micros(d))),
2129        Value::Text(s) => {
2130            // v7.39 (round 289) — the target has no zone, so PG ignores
2131            // any the literal carries: `'…+02'::timestamp` keeps the
2132            // wall clock rather than converting to UTC.
2133            crate::eval::format::parse_timestamp_literal_wall_ordered(
2134                &s,
2135                crate::eval::format::DateOrder::Mdy,
2136            )
2137            .map(Value::Timestamp)
2138            .ok_or_else(|| EvalError::TypeMismatch {
2139                // v7.39 (round 324, V42) — PG's wording, and PG's split
2140                // between "invalid input syntax" and "date/time field
2141                // value out of range".
2142                detail: crate::eval::format::datetime_input_error_text(&s, "timestamp"),
2143            })
2144        }
2145        other => Err(EvalError::TypeMismatch {
2146            detail: format!(
2147                "cannot cast {} to TIMESTAMP",
2148                crate::conversions::pg_type_name_for_error_opt(other.data_type())
2149            ),
2150        }),
2151    }
2152}
2153
2154/// v7.39 (round 310) — `::timestamptz` from text: an offset in the
2155/// literal is APPLIED, unlike the zone-less sibling which discards it.
2156/// Naive input (no offset) is read as UTC, which is what the
2157/// context-aware arm already assumed when it fell through to here.
2158fn cast_to_timestamptz(v: Value) -> Result<Value, EvalError> {
2159    let Value::Text(s) = &v else {
2160        return cast_to_timestamp(v);
2161    };
2162    crate::eval::format::parse_timestamp_literal_tz_ordered(s, crate::eval::format::DateOrder::Mdy)
2163        .map(|(micros, _had_tz)| Value::Timestamp(micros))
2164        .ok_or_else(|| EvalError::TypeMismatch {
2165            // v7.39 (round 324, V42) — and with the RIGHT type name: this arm
2166            // used to report `TIMESTAMP` for a `::timestamptz` cast.
2167            detail: crate::eval::format::datetime_input_error_text(s, "timestamp with time zone"),
2168        })
2169}
2170
2171/// v7.39 (round 254) — PG refuses to cast a NUMERIC special into any
2172/// integer type: `cannot convert NaN to integer` / `cannot convert
2173/// infinity to bigint` (an infinity is named without its sign, probed
2174/// live). Returns `None` for an ordinary value so the caller runs its
2175/// normal conversion.
2176fn cast_numeric_special_reject(
2177    v: &Value,
2178    target: &str,
2179) -> Option<Result<Value<'static>, EvalError>> {
2180    let Value::Numeric { kind, .. } = v else {
2181        return None;
2182    };
2183    if *kind == spg_storage::NumericKind::Finite {
2184        return None;
2185    }
2186    let what = if *kind == spg_storage::NumericKind::NaN {
2187        "NaN"
2188    } else {
2189        "infinity"
2190    };
2191    Some(Err(EvalError::TypeMismatch {
2192        detail: alloc::format!("cannot convert {what} to {target}"),
2193    }))
2194}
2195
2196fn cast_numeric_to_int(v: Value) -> Result<Value, EvalError> {
2197    match v {
2198        // v7.39 (round 633) — SMALLINT. `1::SMALLINT::INT` answered
2199        // "cannot cast smallint to int": the arm was simply absent, next to
2200        // the Int and BigInt ones. Widening a smallint is about as ordinary
2201        // as a cast gets, and PG has it registered as an IMPLICIT cast.
2202        // Same omission shape as the sum accumulator missing SmallInt in
2203        // round 626 — a variant list written out by hand, one entry short.
2204        Value::SmallInt(n) => Ok(Value::Int(i32::from(n))),
2205        Value::Int(n) => Ok(Value::Int(n)),
2206        Value::BigInt(n) => i32::try_from(n)
2207            .map(Value::Int)
2208            // v7.39 (read01 round 79) — PG's wording, which the Float arm two
2209            // arms down was already using: "integer out of range". Drivers match
2210            // on it. Three arms of one function had two different messages.
2211            .map_err(|_| EvalError::TypeMismatch {
2212                detail: "integer out of range".into(),
2213            }),
2214        // PG rounds (half-to-even) coercing a real number to an integer, and
2215        // errors on a non-finite or out-of-range value (`'inf'::int`,
2216        // `1e20::int`) rather than saturating.
2217        #[allow(clippy::cast_possible_truncation)]
2218        Value::Float(x) => {
2219            let r = f64_round_half_even(x);
2220            if !r.is_finite() || !(-2_147_483_648.0..=2_147_483_647.0).contains(&r) {
2221                return Err(EvalError::TypeMismatch {
2222                    detail: "integer out of range".into(),
2223                });
2224            }
2225            Ok(Value::Int(r as i32))
2226        }
2227        // v7.39 (read01 round 112) — `real` (float4) rounds/range-checks the
2228        // same way float8 does; only the float8 arm existed.
2229        #[allow(clippy::cast_possible_truncation)]
2230        Value::Real(x) => {
2231            let r = f64_round_half_even(f64::from(x));
2232            if !r.is_finite() || !(-2_147_483_648.0..=2_147_483_647.0).contains(&r) {
2233                return Err(EvalError::TypeMismatch {
2234                    detail: "integer out of range".into(),
2235                });
2236            }
2237            Ok(Value::Int(r as i32))
2238        }
2239        Value::Numeric { scaled, scale, .. } => {
2240            let rounded = numeric_round_to_i128(scaled, scale);
2241            i32::try_from(rounded)
2242                .map(Value::Int)
2243                .map_err(|_| EvalError::TypeMismatch {
2244                    detail: "integer out of range".into(),
2245                })
2246        }
2247        Value::Text(s) => crate::conversions::parse_pg_int(&s)
2248            .and_then(|n| i32::try_from(n).ok())
2249            .map(Value::Int)
2250            .ok_or_else(|| EvalError::TypeMismatch {
2251                detail: format!("invalid input syntax for type integer: {s:?}"),
2252            }),
2253        Value::Bool(b) => Ok(Value::Int(i32::from(b))),
2254        // v7.39 (read01 char.c) — ("char")::int is the byte value.
2255        Value::Char1(b) => Ok(Value::Int(i32::from(b))),
2256        // PG `bit`/`varbit` → int is the MSB-first bit value.
2257        #[allow(clippy::cast_possible_truncation)]
2258        Value::BitString { nbits, bytes } => Ok(Value::Int(crate::conversions::bit_string_to_i64(
2259            nbits, &bytes,
2260        ) as i32)),
2261        // v7.39 (read01 round 113) — jsonb → int: decode the JSON scalar, then
2262        // round via the numeric arm above. String/array/object/boolean error.
2263        Value::Json(s) => match crate::conversions::jsonb_scalar_for_cast(&s, "integer")? {
2264            crate::conversions::JsonbScalar::Numeric(n) => cast_numeric_to_int(n),
2265            crate::conversions::JsonbScalar::Bool(_) => Err(
2266                crate::conversions::jsonb_cast_type_error("boolean", "integer"),
2267            ),
2268            crate::conversions::JsonbScalar::Null => Ok(Value::Null),
2269        },
2270        other => Err(EvalError::TypeMismatch {
2271            detail: format!(
2272                "cannot cast {} to int",
2273                crate::conversions::pg_type_name_for_error_opt(other.data_type())
2274            ),
2275        }),
2276    }
2277}
2278
2279fn cast_numeric_to_bigint(v: Value) -> Result<Value, EvalError> {
2280    match v {
2281        Value::Int(n) => Ok(Value::BigInt(i64::from(n))),
2282        // v7.39 (round 633) — SMALLINT, missing here for the same reason.
2283        Value::SmallInt(n) => Ok(Value::BigInt(i64::from(n))),
2284        Value::BigInt(n) => Ok(Value::BigInt(n)),
2285        // PG rounds (half-to-even) coercing a real number to bigint, and errors
2286        // on a non-finite or out-of-range value rather than saturating.
2287        #[allow(clippy::cast_possible_truncation)]
2288        Value::Float(x) => {
2289            let r = f64_round_half_even(x);
2290            if !r.is_finite()
2291                || !(-9_223_372_036_854_775_808.0..9_223_372_036_854_775_808.0).contains(&r)
2292            {
2293                return Err(EvalError::TypeMismatch {
2294                    detail: "bigint out of range".into(),
2295                });
2296            }
2297            Ok(Value::BigInt(r as i64))
2298        }
2299        // v7.39 (read01 round 112) — `real` (float4) → bigint, matching float8.
2300        #[allow(clippy::cast_possible_truncation)]
2301        Value::Real(x) => {
2302            let r = f64_round_half_even(f64::from(x));
2303            if !r.is_finite()
2304                || !(-9_223_372_036_854_775_808.0..9_223_372_036_854_775_808.0).contains(&r)
2305            {
2306                return Err(EvalError::TypeMismatch {
2307                    detail: "bigint out of range".into(),
2308                });
2309            }
2310            Ok(Value::BigInt(r as i64))
2311        }
2312        Value::Numeric { scaled, scale, .. } => {
2313            let rounded = numeric_round_to_i128(scaled, scale);
2314            i64::try_from(rounded)
2315                .map(Value::BigInt)
2316                .map_err(|_| EvalError::TypeMismatch {
2317                    detail: format!("numeric {rounded} does not fit in bigint"),
2318                })
2319        }
2320        Value::Text(s) => crate::conversions::parse_pg_int(&s)
2321            .map(Value::BigInt)
2322            .ok_or_else(|| EvalError::TypeMismatch {
2323                // v7.39 (round 324, V42) — PG's wording.
2324                detail: format!("invalid input syntax for type bigint: \"{s}\""),
2325            }),
2326        Value::Bool(b) => Ok(Value::BigInt(i64::from(b))),
2327        // PG `bit`/`varbit` → bigint is the MSB-first bit value.
2328        Value::BitString { nbits, bytes } => Ok(Value::BigInt(
2329            crate::conversions::bit_string_to_i64(nbits, &bytes),
2330        )),
2331        // v7.39 (read01 round 113) — jsonb → bigint.
2332        Value::Json(s) => match crate::conversions::jsonb_scalar_for_cast(&s, "bigint")? {
2333            crate::conversions::JsonbScalar::Numeric(n) => cast_numeric_to_bigint(n),
2334            crate::conversions::JsonbScalar::Bool(_) => Err(
2335                crate::conversions::jsonb_cast_type_error("boolean", "bigint"),
2336            ),
2337            crate::conversions::JsonbScalar::Null => Ok(Value::Null),
2338        },
2339        other => Err(EvalError::TypeMismatch {
2340            detail: format!(
2341                "cannot cast {} to bigint",
2342                crate::conversions::pg_type_name_for_error_opt(other.data_type())
2343            ),
2344        }),
2345    }
2346}
2347
2348fn cast_numeric_to_float(v: Value) -> Result<Value, EvalError> {
2349    match v {
2350        Value::Int(n) => Ok(Value::Float(f64::from(n))),
2351        #[allow(clippy::cast_precision_loss)]
2352        Value::BigInt(n) => Ok(Value::Float(n as f64)),
2353        Value::Float(x) => Ok(Value::Float(x)),
2354        // PG's numeric→double precision is an implicit cast; a
2355        // `Value::Numeric` (from `::numeric`, a numeric column, or
2356        // numeric arithmetic) must convert to f64, not error.
2357        #[allow(clippy::cast_precision_loss)]
2358        // v7.39 (round 254) — a special crosses to its IEEE twin.
2359        Value::Numeric { kind, .. } if kind != spg_storage::NumericKind::Finite => {
2360            Ok(Value::Float(match kind {
2361                spg_storage::NumericKind::NaN => f64::NAN,
2362                spg_storage::NumericKind::PosInf => f64::INFINITY,
2363                _ => f64::NEG_INFINITY,
2364            }))
2365        }
2366        Value::Numeric { scaled, scale, .. } => Ok(Value::Float(
2367            (scaled as f64) / f64_powi(10.0, i32::from(scale)),
2368        )),
2369        Value::Text(s) => {
2370            let t = s.trim();
2371            // Unparseable → invalid syntax; parseable-but-out-of-range (overflow
2372            // to ±∞ / nonzero underflow to 0) → out of range, the way PG's
2373            // float8in does, rather than silently yielding Infinity/0. Shared
2374            // with the Named-cast coerce path so `::float` and `::float8` agree.
2375            if t.parse::<f64>().is_err() {
2376                return Err(EvalError::TypeMismatch {
2377                    detail: format!("cannot parse {s:?} as float"),
2378                });
2379            }
2380            crate::conversions::parse_float8(t)
2381                .map(Value::Float)
2382                .ok_or_else(|| EvalError::TypeMismatch {
2383                    detail: format!("\"{t}\" is out of range for type double precision"),
2384                })
2385        }
2386        // v7.39 (read01 round 113) — jsonb → double precision.
2387        Value::Json(s) => {
2388            match crate::conversions::jsonb_scalar_for_cast(&s, "double precision")? {
2389                crate::conversions::JsonbScalar::Numeric(n) => cast_numeric_to_float(n),
2390                crate::conversions::JsonbScalar::Bool(_) => Err(
2391                    crate::conversions::jsonb_cast_type_error("boolean", "double precision"),
2392                ),
2393                crate::conversions::JsonbScalar::Null => Ok(Value::Null),
2394            }
2395        }
2396        other => Err(EvalError::TypeMismatch {
2397            detail: format!(
2398                "cannot cast {} to float",
2399                crate::conversions::pg_type_name_for_error_opt(other.data_type())
2400            ),
2401        }),
2402    }
2403}
2404
2405fn cast_to_bool(v: Value) -> Result<Value, EvalError> {
2406    match v {
2407        Value::Bool(b) => Ok(Value::Bool(b)),
2408        Value::Int(n) => Ok(Value::Bool(n != 0)),
2409        Value::BigInt(n) => Ok(Value::Bool(n != 0)),
2410        Value::Text(s) => {
2411            // PG boolin accepts any unambiguous prefix of true/false/yes/no
2412            // plus on/off/1/0 (case-insensitive, trimmed); `o` alone is
2413            // ambiguous (on vs off) and errors.
2414            let lo = s.trim().to_ascii_lowercase();
2415            match lo.as_str() {
2416                "1" | "t" | "tr" | "tru" | "true" | "y" | "ye" | "yes" | "on" => {
2417                    Ok(Value::Bool(true))
2418                }
2419                "0" | "f" | "fa" | "fal" | "fals" | "false" | "n" | "no" | "of" | "off" => {
2420                    Ok(Value::Bool(false))
2421                }
2422                _ => Err(EvalError::TypeMismatch {
2423                    detail: format!("invalid input syntax for type boolean: {:?}", s.trim()),
2424                }),
2425            }
2426        }
2427        // v7.39 (read01 round 113) — jsonb → boolean accepts only JSON
2428        // true/false; a JSON number/string/array/object errors.
2429        Value::Json(s) => match crate::conversions::jsonb_scalar_for_cast(&s, "boolean")? {
2430            crate::conversions::JsonbScalar::Bool(b) => Ok(Value::Bool(b)),
2431            crate::conversions::JsonbScalar::Numeric(_) => Err(
2432                crate::conversions::jsonb_cast_type_error("numeric", "boolean"),
2433            ),
2434            crate::conversions::JsonbScalar::Null => Ok(Value::Null),
2435        },
2436        other => Err(EvalError::TypeMismatch {
2437            detail: format!(
2438                "cannot cast {} to bool",
2439                crate::conversions::pg_type_name_for_error_opt(other.data_type())
2440            ),
2441        }),
2442    }
2443}
2444
2445/// Parse a `Value::text("[1.0, 2.0, 3.0]")` into a `Value::vector(..)`. Mirrors
2446/// pgvector's `'[..]'::vector` cast. NULL casts as NULL.
2447pub fn cast_to_vector(v: Value) -> Result<Value<'static>, EvalError> {
2448    match v {
2449        Value::Null => Ok(Value::Null),
2450        Value::Vector(v) => Ok(Value::vector(v.into_owned())),
2451        Value::Text(s) => {
2452            parse_vector_text(&s)
2453                .map(Value::vector)
2454                .ok_or_else(|| EvalError::TypeMismatch {
2455                    detail: format!("cannot parse {s:?} as a vector literal"),
2456                })
2457        }
2458        other => Err(EvalError::TypeMismatch {
2459            detail: format!(
2460                "::vector requires text input, got {}",
2461                crate::conversions::pg_type_name_for_error_opt(other.data_type())
2462            ),
2463        }),
2464    }
2465}
2466
2467/// Parse `"[1.0, 2.0, -3]"` into `Vec<f32>`. Returns `None` on malformed input.
2468pub fn parse_vector_text(s: &str) -> Option<Vec<f32>> {
2469    let trimmed = s.trim();
2470    let inner = trimmed.strip_prefix('[')?.strip_suffix(']')?;
2471    let trimmed_inner = inner.trim();
2472    if trimmed_inner.is_empty() {
2473        return Some(Vec::new());
2474    }
2475    let mut out = Vec::new();
2476    for part in trimmed_inner.split(',') {
2477        let f: f32 = part.trim().parse().ok()?;
2478        out.push(f);
2479    }
2480    Some(out)
2481}
2482
2483#[cfg(test)]
2484mod round613_plain_named_targets {
2485    use super::*;
2486
2487    /// v7.39 (round 613) — the shortcut is only equivalent to walking the
2488    /// arm while these hold. Checked here rather than by eye, so a name that
2489    /// grows a special case above the resolve fails the gate instead of
2490    /// silently taking the wrong path.
2491    fn assert_no_arm_above_the_resolve_claims(name: &str) {
2492        assert!(
2493            !REG_MISC_TYPES.iter().any(|k| name.eq_ignore_ascii_case(k)),
2494            "{name} is a reg-misc type"
2495        );
2496        assert!(
2497            !CATALOG_SCALAR_TYPES
2498                .iter()
2499                .any(|k| name.eq_ignore_ascii_case(k)),
2500            "{name} is a catalog scalar"
2501        );
2502        assert!(
2503            !OPAQUE_TYPES.iter().any(|k| name.eq_ignore_ascii_case(k)),
2504            "{name} is a pseudotype"
2505        );
2506        for special in [
2507            "__bit_literal",
2508            "tid",
2509            "xid",
2510            "xid8",
2511            "jsonpath",
2512            "binary",
2513            "signed",
2514            "unsigned",
2515        ] {
2516            assert!(
2517                !name.eq_ignore_ascii_case(special),
2518                "{name} has its own arm ({special})"
2519            );
2520        }
2521        assert!(bit_cast_width(name).is_none(), "{name} is a bit spelling");
2522        assert!(
2523            temporal_typmod(name).is_none(),
2524            "{name} carries a temporal precision"
2525        );
2526        assert!(
2527            !is_bare_temporal_type(name),
2528            "{name} is a bare temporal type"
2529        );
2530        assert!(
2531            matches!(cast_catalog_scalar(name, &Value::text("x")), Ok(None)),
2532            "{name} is claimed by the catalog-scalar arm"
2533        );
2534    }
2535
2536    #[test]
2537    fn every_plain_target_resolves_to_the_type_the_table_claims() {
2538        for (name, dt) in PLAIN_NAMED_TARGETS {
2539            assert_eq!(
2540                crate::conversions::type_name_to_data_type(name),
2541                Some(*dt),
2542                "{name} does not resolve to the type the table gives it"
2543            );
2544            assert_eq!(plain_named_target(name), Some(*dt));
2545            // The spelling is matched without regard to case.
2546            assert_eq!(plain_named_target(&name.to_uppercase()), Some(*dt));
2547            assert_no_arm_above_the_resolve_claims(name);
2548        }
2549    }
2550
2551    #[test]
2552    fn every_typmod_head_is_plain_and_resolves_through_the_type_table() {
2553        for head in PLAIN_NAMED_HEADS {
2554            assert_no_arm_above_the_resolve_claims(head);
2555            for spelled in [alloc::format!("{head}(4)"), alloc::format!("{head}(10,2)")] {
2556                assert_no_arm_above_the_resolve_claims(&spelled);
2557                assert_eq!(
2558                    plain_named_target(&spelled),
2559                    crate::conversions::type_name_to_data_type(&spelled),
2560                    "{spelled} takes a different type through the shortcut"
2561                );
2562            }
2563            // A bare head with no typmod only shortcuts when it is in the
2564            // exact table; the head list alone must not claim it.
2565            let bare = plain_named_target(head);
2566            let exact = PLAIN_NAMED_TARGETS
2567                .iter()
2568                .find(|(k, _)| head.eq_ignore_ascii_case(k))
2569                .map(|(_, dt)| *dt);
2570            assert_eq!(bare, exact, "{head} bare");
2571        }
2572    }
2573
2574    #[test]
2575    fn a_name_with_its_own_arm_is_not_shortcut() {
2576        for name in [
2577            "regproc",
2578            "aclitem",
2579            "anyarray",
2580            "tid",
2581            "xid",
2582            "jsonpath",
2583            "bit",
2584            "bit(4)",
2585            "timestamp",
2586            "timestamp(2)",
2587            "time(3)",
2588            "nosuchtype",
2589            "int4range",
2590        ] {
2591            assert_eq!(plain_named_target(name), None, "{name} was shortcut");
2592        }
2593    }
2594}