uqa-sql 0.5.3

//
// Unified Query Algebra
//
// Copyright (c) 2023-2026 Cognica, Inc.
//

//! Catalog-aware cast selection shares controlled type, array and value constructors.

use uqa_core::{
    memory::{Produced, ProductionControl, ProductionString, ProductionVec},
    ArrayValue, Value,
};

use super::super::casting::{cast_value_from_with_control, parse_pg_array_literal_with_control};
use super::{format_regtype_elements_with_control, format_regtype_value_with_control, EngineHook};
use crate::{
    ast::ColumnType,
    error::{Result, SQLError},
};

mod datestyle;
mod timezone;

/// Catalog input needed after analysis selects a type for an unknown literal. Domain array input includes each element's constraints, so the caller must retain this value rather than read the text again at execution.
pub trait CatalogInputFunctions {
    fn read_unknown_input(&self, text: &str, target: &ColumnType) -> Result<Value>;
}

pub(crate) fn requires_domain_array_input(target: &ColumnType) -> bool {
    matches!(target, ColumnType::Array(element) if matches!(array_leaf_type(element), ColumnType::Domain { .. }))
}

/// Input constants containing named composites or domain array elements must be read while the definition's catalog descriptor is available.
pub(crate) fn requires_catalog_constant_input(target: &ColumnType) -> bool {
    requires_domain_array_input(target)
        || match target {
            ColumnType::Composite(_) => true,
            ColumnType::Array(element) | ColumnType::Domain { base: element, .. } => {
                requires_catalog_constant_input(element)
            }
            _ => false,
        }
}

/// Read a catalog-dependent constant once, retaining its typed datum in stored syntax.
pub fn read_catalog_input(
    text: &str,
    target: &ColumnType,
    engine: &dyn EngineHook,
) -> Result<Value> {
    if requires_domain_array_input(target) {
        return read_catalog_array_input(text, target, engine);
    }
    if !requires_catalog_constant_input(target) {
        return Err(SQLError::Internal(
            "catalog input requires a composite or domain array".into(),
        ));
    }
    super::cast_value_with_type_resolution(
        &Value::Str(text.into()),
        Some("unknown"),
        &target.catalog_name(),
        Some(engine),
    )
}

/// Read an array whose elements are domains with the same input and constraint path as a catalog-aware cast. The resolved target preserves domain identity; a scalar outer domain remains the surrounding expression's runtime coercion.
pub fn read_catalog_array_input(
    text: &str,
    target: &ColumnType,
    engine: &dyn EngineHook,
) -> Result<Value> {
    if !requires_domain_array_input(target) {
        return Err(SQLError::Internal(
            "catalog array input requires domain elements".into(),
        ));
    }
    let control = ProductionControl::uncontrolled();
    let parsed = parse_pg_array_literal_with_control(text, &control)?
        .into_uncontrolled()
        .map_err(|_| SQLError::Internal("ordinary catalog array input owner".into()))?;
    cast_catalog_array(&Value::Array(parsed), None, target, engine, &control)?
        .into_uncontrolled()
        .map_err(|_| SQLError::Internal("ordinary catalog array input owner".into()))
}

#[must_use]
pub fn coercion_type_name(ty: &ColumnType) -> String {
    coercion_type_name_with_control(ty, &ProductionControl::uncontrolled())
        .expect("ordinary coercion type name")
        .into_uncontrolled()
        .expect("ordinary coercion type name owner")
}

fn coercion_type_name_with_control(
    ty: &ColumnType,
    control: &ProductionControl<'_>,
) -> Result<Produced<String>> {
    control.check()?;
    match ty {
        ColumnType::Domain { base, .. } => coercion_type_name_with_control(base, control),
        ColumnType::Array(element) => {
            let element = coercion_type_name_with_control(element, control)?;
            let mut name = ProductionString::new(*control);
            name.push_str(&element)?;
            name.push_str("[]")?;
            Ok(name.finish()?)
        }
        _ => Ok(ty.sql_name_with_control(control)?),
    }
}

/// The types of `PostgreSQL`'s string category, which any type casts to through its output function.
fn is_string_type(ty: &ColumnType) -> bool {
    matches!(
        ty,
        ColumnType::Text | ColumnType::Name | ColumnType::Varchar(_) | ColumnType::Character(_)
    )
}

fn regrole_array_type(ty: &ColumnType) -> bool {
    match ty {
        ColumnType::Array(element) => {
            matches!(element.as_ref(), ColumnType::Regrole) || regrole_array_type(element)
        }
        _ => false,
    }
}

fn array_leaf_type(ty: &ColumnType) -> &ColumnType {
    match ty {
        ColumnType::Array(element) => array_leaf_type(element),
        _ => ty,
    }
}

fn optional_type_name(
    name: &str,
    control: &ProductionControl<'_>,
) -> Result<Option<Produced<ColumnType>>> {
    match ColumnType::from_sql_name_with_control(name, control) {
        Ok(ty) => Ok(Some(ty)),
        Err(error) if matches!(error.sqlstate(), Some("53200" | "57014")) => Err(error),
        Err(_) => Ok(None),
    }
}

/// Cast a value after resolving catalog-owned types and enforcing domain constraints before exposing a base-type carrier.
pub fn cast_value_with_type_resolution(
    value: &Value,
    source_ty: Option<&str>,
    target_ty: &str,
    engine: Option<&dyn EngineHook>,
) -> Result<Value> {
    cast_value_with_type_resolution_with_control(
        value,
        source_ty,
        target_ty,
        engine,
        &ProductionControl::uncontrolled(),
    )?
    .into_uncontrolled()
    .map_err(|_| SQLError::Internal("ordinary catalog cast owner".into()))
}

/// The text a value's output function spells, and the type to cast it from, when a cast goes through output: an enum label, `record_out` of a composite record cast to a string type, and a container whose enum fields and elements spell their current labels.
fn output_for_cast<'a>(
    engine: &dyn EngineHook,
    value: &Value,
    source_ty: Option<&'a str>,
    resolved_source_type: Option<&ColumnType>,
    target: &ColumnType,
) -> Result<Option<(Value, Option<&'a str>)>> {
    if let Value::Enum(label) = value {
        let output =
            super::super::enums::enum_output_for_cast(engine.enum_labels(), label, target)?;
        return Ok(Some((output, Some("text"))));
    }
    if let Value::Record(record) = value {
        if let Some(oid) = record.type_oid().filter(|_| is_string_type(target)) {
            let ty = engine
                .resolve_type_oid(oid)
                .map_err(SQLError::Internal)?
                .ok_or_else(|| SQLError::Routine {
                    sqlstate: "42704".into(),
                    message: format!("type with OID {oid} does not exist"),
                })?;
            return Ok(Some((
                Value::Str(crate::result::format_postgres_text(
                    value,
                    &ty,
                    Some(engine),
                )?),
                Some("text"),
            )));
        }
    }
    // `record_out` spells each field through its attribute type's output function.
    if let (Value::Record(_), Some(source @ ColumnType::Composite(_))) =
        (value, resolved_source_type)
    {
        if is_string_type(target) {
            let text = crate::result::format_postgres_text(value, source, Some(engine))?;
            return Ok(Some((Value::Str(text), Some("text"))));
        }
    }
    if matches!(value, Value::Record(_) | Value::Row(_))
        && is_string_type(target)
        && super::super::enums::contains_enum_carrier(value)
    {
        let rendered = super::super::enums::render_enum_labels(engine.enum_labels(), value)?;
        return Ok(Some((rendered, source_ty)));
    }
    Ok(None)
}

/// Array output borrows the input and retains both intermediate labels and final text under the caller's production budget.
fn catalog_array_output(
    value: &Value,
    source_ty: Option<&str>,
    target_ty: &str,
    target: &ColumnType,
    engine: &dyn EngineHook,
    control: &ProductionControl<'_>,
) -> Result<Option<Produced<Value>>> {
    if !matches!(value, Value::Array(_))
        || !is_string_type(target)
        || !(super::super::enums::contains_enum_carrier(value)
            || super::super::datums::contains_datum(value))
    {
        return Ok(None);
    }
    let actual_source;
    let source_ty = if let Value::Array(array) = value {
        actual_source = array
            .element_type_oid()
            .map(|oid| engine.resolve_type_oid(oid))
            .transpose()
            .map_err(SQLError::Internal)?
            .flatten()
            .map(|element| control.copy_text(&ColumnType::Array(Box::new(element)).catalog_name()))
            .transpose()?;
        actual_source.as_deref().map(String::as_str).or(source_ty)
    } else {
        source_ty
    };
    let text_array = ColumnType::Array(Box::new(ColumnType::Text));
    let labels = cast_catalog_array(value, source_ty, &text_array, engine, control)?;
    cast_value_from_with_control(&labels, target_ty, Some("text[]"), control).map(Some)
}

fn physical_output_cast(
    value: &Value,
    source_ty: Option<&str>,
    target_ty: &str,
    target: &ColumnType,
    engine: &dyn EngineHook,
    control: &ProductionControl<'_>,
) -> Result<Option<Produced<Value>>> {
    let Value::Datum(datum) = value else {
        return Ok(None);
    };
    if !is_string_type(target) {
        return Ok(None);
    }
    let decoded =
        super::super::datums::read_with_catalog_and_control(datum, Some(engine), control)?;
    cast_value_with_type_resolution_with_control(
        &decoded,
        source_ty,
        target_ty,
        Some(engine),
        control,
    )
    .map(Some)
}

/// Resolve catalog inputs at their external handoff, then admit SQL-owned names, element conversions and output before constructing them.
pub fn cast_value_with_type_resolution_with_control(
    value: &Value,
    source_ty: Option<&str>,
    target_ty: &str,
    engine: Option<&dyn EngineHook>,
    control: &ProductionControl<'_>,
) -> Result<Produced<Value>> {
    control.check()?;
    let resolved_target = resolve_catalog_type_with_control(target_ty, engine, control)?;
    control.check()?;
    let resolved_source_type = source_ty
        .map(|source| resolve_catalog_type_with_control(source, engine, control))
        .transpose()?
        .flatten();
    let parsed_target = if resolved_target.is_some() {
        None
    } else {
        optional_type_name(target_ty, control)?
    };
    let target_column_type = resolved_target.as_deref().or(parsed_target.as_deref());
    let _date_order = datestyle::input_scope(value, target_column_type, engine, control)?;
    if let (Some(engine), Some(target)) = (engine, resolved_target.as_deref()) {
        if let Some(value) = record_identity_cast(
            value,
            resolved_source_type.as_deref(),
            target,
            engine,
            control,
        )? {
            return Ok(value);
        }
        if let Some(output) =
            physical_output_cast(value, source_ty, target_ty, target, engine, control)?
        {
            return Ok(output);
        }
        if let Some(value) = engine.cast_domain(value, source_ty, target)? {
            return Ok(control.retain_external_value(value)?);
        }
        if let Some(value) = super::super::enums::cast_to_enum(
            engine.enum_labels(),
            value,
            resolved_source_type.as_deref(),
            target,
        )? {
            return Ok(control.retain_external_value(value)?);
        }
        if let Some(value) = super::super::composites::cast_to_composite(
            engine,
            value,
            resolved_source_type.as_deref(),
            target,
        )? {
            return Ok(control.retain_external_value(value)?);
        }
        if let Some(output) =
            catalog_array_output(value, source_ty, target_ty, target, engine, control)?
        {
            return Ok(output);
        }
        if let Some((output, output_source)) = output_for_cast(
            engine,
            value,
            source_ty,
            resolved_source_type.as_deref(),
            target,
        )? {
            return cast_value_with_type_resolution_with_control(
                &output,
                output_source,
                target_ty,
                Some(engine),
                control,
            );
        }
        control.check()?;
        if matches!(target, ColumnType::Array(_))
            && (requires_catalog_array_cast(target)
                || super::super::enums::contains_enum_carrier(value)
                || resolved_source_type
                    .as_deref()
                    .is_some_and(super::super::enums::is_enum_bearing))
        {
            return cast_catalog_array(value, source_ty, target, engine, control);
        }
    }
    let resolved_source = resolved_source_type
        .map(|ty| coercion_type_name_with_control(&ty, control))
        .transpose()?;
    let source_ty = resolved_source.as_deref().map(String::as_str).or(source_ty);
    let target_name = resolved_target
        .as_ref()
        .map(|ty| coercion_type_name_with_control(ty, control))
        .transpose()?;
    let target_ty = target_name.as_ref().map_or(target_ty, |name| name.as_str());
    cast_resolved_value(
        value,
        source_ty,
        target_ty,
        target_column_type,
        engine,
        control,
    )
}

fn resolve_catalog_type_with_control(
    name: &str,
    engine: Option<&dyn EngineHook>,
    control: &ProductionControl<'_>,
) -> Result<Option<Produced<ColumnType>>> {
    engine
        .map(|engine| engine.resolve_type_name(name))
        .transpose()
        .map_err(SQLError::Internal)?
        .flatten()
        .map(|ty| {
            ty.retain_external_with_control(control)
                .map_err(SQLError::from)
        })
        .transpose()
}

fn record_identity_cast(
    value: &Value,
    resolved_source_type: Option<&ColumnType>,
    target: &ColumnType,
    engine: &dyn EngineHook,
    control: &ProductionControl<'_>,
) -> Result<Option<Produced<Value>>> {
    if matches!(target, ColumnType::Record) {
        // Composite-to-record coercion is a binary relabel: even malformed physical fields remain unread.
        if let Value::Datum(datum) = value {
            let ty = engine
                .resolve_type_oid(datum.type_oid())
                .map_err(SQLError::Internal)?;
            if matches!(ty, Some(ColumnType::Composite(_))) {
                return Ok(Some(control.copy_value(value)?));
            }
        }
        if let (Value::Record(record), Some(ColumnType::Composite(source))) =
            (value, resolved_source_type)
        {
            let (value, memory) = control.copy_value(value)?.into_parts();
            let Value::Record(value) = value else {
                unreachable!();
            };
            return Ok(Some(control.finish(
                Value::Record(value.with_type_oid(Some(record.type_oid().unwrap_or(source.oid)))),
                memory,
            )?));
        }
    }

    Ok(None)
}

fn cast_resolved_value(
    value: &Value,
    source_ty: Option<&str>,
    target_ty: &str,
    target_column_type: Option<&ColumnType>,
    engine: Option<&dyn EngineHook>,
    control: &ProductionControl<'_>,
) -> Result<Produced<Value>> {
    if target_column_type.is_some_and(regrole_array_type) {
        let source_column_type = source_ty
            .map(|name| optional_type_name(name, control))
            .transpose()?
            .flatten();
        let source_name = source_column_type
            .as_deref()
            .map(array_leaf_type)
            .map(|ty| ty.sql_name_with_control(control))
            .transpose()?;
        return cast_array(
            value,
            source_name.as_ref().map(|name| name.as_str()),
            "regrole",
            "regrole[]",
            engine,
            control,
        );
    }
    // `find_coercion_pathway` casts an OID alias to a string type through its output function, as `CoerceViaIO`, and the string type then applies its own modifier; an array does so element by element.
    if let Some(ColumnType::Array(element)) = target_column_type {
        if is_string_type(element) {
            if let Some(source_ty) = source_ty
                .map(|source| optional_type_name(source, control))
                .transpose()?
                .flatten()
            {
                if let Some(elements) =
                    format_regtype_elements_with_control(value, &source_ty, engine, control)?
                {
                    let (elements, _memory) = elements.into_parts();
                    return cast_value_from_with_control(
                        &Value::Array(elements),
                        target_ty,
                        Some("text[]"),
                        control,
                    );
                }
            }
        }
    }
    let text_target = target_ty.eq_ignore_ascii_case("text");
    if text_target || target_column_type.is_some_and(is_string_type) {
        if let Some(source_ty) = source_ty
            .map(|source| optional_type_name(source, control))
            .transpose()?
            .flatten()
        {
            if let Some(text) =
                format_regtype_value_with_control(value, &source_ty, engine, control)?
            {
                let (text, memory) = text.into_parts();
                if text_target {
                    return Ok(control.finish(Value::Str(text), memory)?);
                }
                return cast_value_from_with_control(
                    &Value::Str(text),
                    target_ty,
                    Some("text"),
                    control,
                );
            }
        }
    }
    if let (Some(engine), Value::Str(name) | Value::FixedChar(name)) = (engine, value) {
        let oid = resolve_regobject_input(name, target_ty, target_column_type, engine, control)?;
        if let Some(oid) = oid {
            return Ok(control.finish(Value::Int(oid), control.empty_reservation())?);
        }
    }
    let zoned = timezone::cast_local_timestamp(value, target_column_type, engine, control)?;
    cast_value_from_with_control(
        zoned.as_ref().unwrap_or(value),
        target_ty,
        source_ty,
        control,
    )
}

fn resolve_regobject_input(
    name: &str,
    target_ty: &str,
    target_column_type: Option<&ColumnType>,
    engine: &dyn EngineHook,
    control: &ProductionControl<'_>,
) -> Result<Option<i64>> {
    enum ObjectKind {
        Relation,
        RoutineName,
        Routine,
        Role,
        Namespace,
        Collation,
        Type,
    }
    let kind = if target_ty.eq_ignore_ascii_case("regclass") {
        ObjectKind::Relation
    } else if matches!(target_column_type, Some(ColumnType::Regproc))
        || target_ty.eq_ignore_ascii_case("regproc")
    {
        ObjectKind::RoutineName
    } else if target_ty.eq_ignore_ascii_case("regprocedure") {
        ObjectKind::Routine
    } else if target_ty.eq_ignore_ascii_case("regrole") {
        ObjectKind::Role
    } else if matches!(target_column_type, Some(ColumnType::Regnamespace)) {
        ObjectKind::Namespace
    } else if matches!(target_column_type, Some(ColumnType::Regcollation)) {
        ObjectKind::Collation
    } else if matches!(target_column_type, Some(ColumnType::Regtype)) {
        ObjectKind::Type
    } else {
        return Ok(None);
    };
    let oid = match kind {
        ObjectKind::Relation => engine.resolve_regclass_input(name)?,
        // `regprocin` reports its own missing and ambiguous names.
        ObjectKind::RoutineName => return engine.resolve_regproc(name),
        ObjectKind::Routine => engine.resolve_regprocedure_input(name)?,
        ObjectKind::Role => engine.resolve_regrole(name)?,
        ObjectKind::Namespace => engine.resolve_regnamespace(name)?,
        ObjectKind::Collation => engine.resolve_regcollation(name)?,
        ObjectKind::Type => engine.resolve_regtype_input(name)?,
    };
    control.check()?;
    if oid.is_some() || matches!(kind, ObjectKind::Type) {
        return Ok(oid);
    }
    let (sqlstate, message) = match kind {
        ObjectKind::Relation => ("42P01", format!("relation \"{name}\" does not exist")),
        ObjectKind::Routine => ("42883", format!("function \"{name}\" does not exist")),
        ObjectKind::Role => ("42704", format!("role \"{name}\" does not exist")),
        ObjectKind::Namespace => ("3F000", format!("schema \"{name}\" does not exist")),
        ObjectKind::Collation => (
            "42704",
            format!("collation \"{name}\" for encoding \"UTF8\" does not exist"),
        ),
        ObjectKind::Type | ObjectKind::RoutineName => {
            unreachable!("regtype and regproc input return before reporting a missing object")
        }
    };
    Err(SQLError::Routine {
        sqlstate: sqlstate.into(),
        message,
    })
}

/// Whether an array cast needs the catalog for its elements: domains, enums, composites and the OID alias types whose input functions read catalog names. `regrole[]` has its own path above.
fn requires_catalog_array_cast(ty: &ColumnType) -> bool {
    match ty {
        ColumnType::Domain { .. }
        | ColumnType::Regtype
        | ColumnType::Regclass
        | ColumnType::Regproc
        | ColumnType::Regprocedure
        | ColumnType::Regcollation
        | ColumnType::Regnamespace
        | ColumnType::Enum(_)
        | ColumnType::Composite(_)
        | ColumnType::TimestampTz
        | ColumnType::TimestampTzPrecision(_) => true,
        ColumnType::Array(element) => requires_catalog_array_cast(element),
        _ => false,
    }
}

fn cast_catalog_array(
    value: &Value,
    source: Option<&str>,
    target: &ColumnType,
    engine: &dyn EngineHook,
    control: &ProductionControl<'_>,
) -> Result<Produced<Value>> {
    if matches!(value, Value::Null) {
        return Ok(control.finish(Value::Null, control.empty_reservation())?);
    }
    let source_element = source.map(|name| name.trim_end_matches("[]"));
    let leaf = array_leaf_type(target);
    let target_element = control.copy_text(&leaf.catalog_name())?;
    let target_name = target.sql_name_with_control(control)?;
    let output = cast_array(
        value,
        source_element,
        &target_element,
        &target_name,
        Some(engine),
        control,
    )?;
    let (Value::Array(array), memory) = output.into_parts() else {
        return Err(SQLError::Internal(
            "catalog array cast returned a non-array".into(),
        ));
    };
    let oid = crate::catalog::type_metadata::pg_type_oid(leaf) as u32;
    Ok(control.finish(Value::Array(array.with_element_type_oid(Some(oid))), memory)?)
}

fn cast_array(
    value: &Value,
    source: Option<&str>,
    target_element: &str,
    target_name: &str,
    engine: Option<&dyn EngineHook>,
    control: &ProductionControl<'_>,
) -> Result<Produced<Value>> {
    let parsed;
    let array = match value {
        Value::Array(array) => array,
        Value::Str(text) => {
            parsed = parse_pg_array_literal_with_control(text, control)?;
            &parsed
        }
        other => {
            return Err(SQLError::TypeMismatch(format!(
                "CAST AS {target_name}: expected array, got {other:?}"
            )))
        }
    };
    let elements = cast_array_elements(array.elements(), source, target_element, engine, control)?;
    let array = rebuild_array(array, elements, control)?
        .ok_or_else(|| SQLError::TypeMismatch("array dimensions changed during cast".into()))?;
    let (array, memory) = array.into_parts();
    let oid = if target_element == "regrole" {
        Some(crate::catalog::type_metadata::pg_type_oid(&ColumnType::Regrole) as u32)
    } else {
        array.element_type_oid()
    };
    Ok(control.finish(Value::Array(array.with_element_type_oid(oid)), memory)?)
}

fn cast_array_elements(
    values: &[Value],
    source: Option<&str>,
    target: &str,
    engine: Option<&dyn EngineHook>,
    control: &ProductionControl<'_>,
) -> Result<Produced<Vec<Value>>> {
    let mut output = ProductionVec::new(*control);
    output.reserve(values.len())?;
    for value in values {
        let value = match value {
            Value::List(values) => {
                let (values, memory) =
                    cast_array_elements(values, source, target, engine, control)?.into_parts();
                control.finish(Value::List(values), memory)?
            }
            value => cast_value_with_type_resolution_with_control(
                value, source, target, engine, control,
            )?,
        };
        output.push_produced(value)?;
    }
    Ok(output.finish()?)
}

pub(super) fn rebuild_array(
    source: &ArrayValue,
    elements: Produced<Vec<Value>>,
    control: &ProductionControl<'_>,
) -> Result<Option<Produced<ArrayValue>>> {
    let mut bounds = ProductionVec::new(*control);
    bounds.reserve(source.lower_bounds().len())?;
    for bound in source.lower_bounds() {
        bounds.push_copy(*bound)?;
    }
    ArrayValue::with_lower_bounds_with_control(elements, bounds.finish()?, control)?
        .map(|array| {
            let (array, memory) = array.into_parts();
            control
                .finish(
                    array.with_element_type_oid(source.element_type_oid()),
                    memory,
                )
                .map_err(Into::into)
        })
        .transpose()
}

#[cfg(test)]
mod tests;