//! Expression translation from IR to SQL AST.
//!
//! This module translates `IRExpression` (which wraps `TypedExpression`) to `SQLExpression`.
//! The translation uses the `TranslationRegistry` for function calls and handles all
//! expression types defined in `TypedExpressionKind`.

use hamelin_lib::err::{TranslationError, TranslationErrors};
use hamelin_lib::sql::expression::apply::Lambda as SQLLambda;
use hamelin_lib::sql::expression::apply::{BinaryOperatorApply, FunctionCallApply};
use hamelin_lib::sql::expression::extract::{ExtractField, ExtractFunction};
use hamelin_lib::sql::expression::identifier::{
    Identifier as SQLIdentifier, SimpleIdentifier as SQLSimpleIdentifier,
};
use hamelin_lib::sql::expression::literal::StringLiteral;
use hamelin_lib::sql::expression::literal::{
    ArrayLiteral as SQLArrayLiteral, RowLiteral as SQLRowLiteral,
};
use hamelin_lib::sql::expression::literal::{
    BinaryLiteral as SQLBinaryLiteral, BooleanLiteral as SQLBooleanLiteral,
    ColumnReference as SQLColumnReference, DecimalLiteral as SQLDecimalLiteral,
    IntegerLiteral as SQLIntegerLiteral, IntervalLiteral as SQLIntervalLiteral,
    NullLiteral as SQLNullLiteral, ScientificLiteral as SQLScientificLiteral,
    StringLiteral as SQLStringLiteral, Unit as SQLIntervalUnit,
};
use hamelin_lib::sql::expression::operator::Operator;
use hamelin_lib::sql::expression::Case as SQLCase;
use hamelin_lib::sql::expression::Cast as SQLCast;
use hamelin_lib::sql::expression::Dot;
use hamelin_lib::sql::expression::IndexLookup;
use hamelin_lib::sql::expression::{SQLExpression, TryCast};
use hamelin_lib::sql::types::{
    SQLAnonRowType, SQLArrayType, SQLBaseType, SQLDecimalType, SQLMapType, SQLRowType,
    SQLTimestampTzType, SQLType,
};
use hamelin_lib::translation::ExpressionTranslation;
use hamelin_lib::tree::ast::expression::{ExpressionKind, IntervalUnit, TruncUnit};
use hamelin_lib::tree::ast::identifier::ParsedSimpleIdentifier;
use hamelin_lib::tree::ast::identifier::SimpleIdentifier;
use hamelin_lib::tree::ast::node::Spannable;
use hamelin_lib::tree::typed_ast::expression::{
    CastKind, FieldAccess, FieldReferenceResolution, TypedApply, TypedArrayLiteral, TypedCast,
    TypedExpressionKind, TypedFieldLookup, TypedFieldReference, TypedLambda, TypedStructLiteral,
    TypedTsTrunc, TypedTupleLiteral, TypedVariantIndexAccess, VariantCastKind,
};
use hamelin_lib::types::Type;

use hamelin_translation::IRExpression;

use crate::context::TranslationContext;

/// Result type for expression translation.
pub type ExprTranslationResult = Result<SQLExpression, TranslationErrors>;

/// Translate an IR expression to a SQL expression.
///
/// This is the main entry point for expression translation. It recursively
/// translates the expression tree, handling all `TypedExpressionKind` variants.
pub fn translate_expression(
    ctx: &TranslationContext,
    expr: &IRExpression,
) -> ExprTranslationResult {
    let typed_expr = expr.inner();

    match &typed_expr.kind {
        TypedExpressionKind::Leaf => translate_leaf(ctx, expr),

        TypedExpressionKind::FieldReference(col_ref) => {
            translate_field_reference(ctx, col_ref, expr)
        }

        TypedExpressionKind::Cast(cast) => translate_cast(ctx, cast, expr),

        TypedExpressionKind::TsTrunc(ts_trunc) => translate_ts_trunc(ctx, ts_trunc),

        TypedExpressionKind::ArrayLiteral(arr) => translate_array_literal(ctx, arr),

        TypedExpressionKind::TupleLiteral(tuple) => translate_tuple_literal(ctx, tuple),

        TypedExpressionKind::StructLiteral(strct) => {
            translate_struct_literal(ctx, strct, expr.resolved_type(), expr)
        }

        TypedExpressionKind::FieldLookup(field_lookup) => {
            translate_field_lookup(ctx, field_lookup, expr)
        }

        TypedExpressionKind::VariantIndexAccess(variant_access) => {
            translate_variant_index_access(ctx, variant_access, expr)
        }

        TypedExpressionKind::Lambda(lambda) => translate_lambda(ctx, lambda),

        TypedExpressionKind::Apply(apply) => {
            translate_apply(ctx, apply, expr.resolved_type(), expr)
        }

        TypedExpressionKind::BroadcastApply(_) => {
            // BroadcastApply should be lowered by the lower_broadcast_apply normalization pass
            // before reaching translation. If we encounter it here, it's a bug in the pipeline.
            Err(TranslationError::msg(
                expr,
                "BroadcastApply should be lowered before translation (run lower_broadcast_apply pass)",
            )
            .single())
        }

        TypedExpressionKind::Error(err) => Err(err.error.as_ref().clone().single()),
    }
}

/// Translate a field reference expression.
fn translate_field_reference(
    ctx: &TranslationContext,
    col_ref: &TypedFieldReference,
    span: &impl Spannable,
) -> ExprTranslationResult {
    match &col_ref.field_name {
        ParsedSimpleIdentifier::Valid(simple) => {
            if col_ref.resolution == FieldReferenceResolution::ScalarDef {
                if let Some(inlined) = ctx.scalar_def_bindings.get(simple.as_str()) {
                    return translate_expression(ctx, inlined);
                }
                return Err(TranslationError::msg(
                    span,
                    &format!(
                        "missing scalar DEF binding for `{}` in SQL translation context",
                        simple.as_str()
                    ),
                )
                .single());
            }
            let sql_simple = SQLSimpleIdentifier::new(simple.as_str());
            let sql_ident: SQLIdentifier = sql_simple.into();
            Ok(SQLColumnReference::new(sql_ident).into())
        }
        ParsedSimpleIdentifier::Error(err) => Err(err.as_ref().clone().single()),
    }
}

/// Translate a cast expression.
fn translate_cast(
    ctx: &TranslationContext,
    cast: &TypedCast,
    span: &impl Spannable,
) -> ExprTranslationResult {
    // Translate the inner expression
    let inner_expr = IRExpression::new(cast.value.clone());
    let inner_sql = translate_expression(ctx, &inner_expr)?;

    match &cast.cast_kind {
        // Identity cast - no-op
        CastKind::Identity => Ok(inner_sql),

        // Simple casts - use TRY_CAST
        CastKind::IntToDouble => Ok(try_cast(inner_sql, SQLBaseType::Double.into())),
        CastKind::IntToDecimal => {
            let sql_type = hamelin_type_to_sql_type(&cast.target_type, span)?;
            Ok(try_cast(inner_sql, sql_type))
        }
        CastKind::DoubleToInt => Ok(try_cast(inner_sql, SQLBaseType::BigInt.into())),
        CastKind::DoubleToDecimal => {
            let sql_type = hamelin_type_to_sql_type(&cast.target_type, span)?;
            Ok(try_cast(inner_sql, sql_type))
        }
        CastKind::DecimalToInt => Ok(try_cast(inner_sql, SQLBaseType::BigInt.into())),
        CastKind::DecimalToDouble => Ok(try_cast(inner_sql, SQLBaseType::Double.into())),
        CastKind::DecimalToDecimal => {
            let sql_type = hamelin_type_to_sql_type(&cast.target_type, span)?;
            Ok(try_cast(inner_sql, sql_type))
        }

        // Boolean casts
        CastKind::IntToBoolean => Ok(try_cast(inner_sql, SQLBaseType::Boolean.into())),
        CastKind::BooleanToInt => Ok(try_cast(inner_sql, SQLBaseType::BigInt.into())),
        CastKind::BooleanToDouble => Ok(try_cast(inner_sql, SQLBaseType::Double.into())),
        CastKind::BooleanToDecimal => {
            let sql_type = hamelin_type_to_sql_type(&cast.target_type, span)?;
            Ok(try_cast(inner_sql, sql_type))
        }
        CastKind::DoubleToBoolean | CastKind::DecimalToBoolean => {
            Ok(try_cast(inner_sql, SQLBaseType::Boolean.into()))
        }

        // String casts - to string
        CastKind::ToStringFromInt
        | CastKind::ToStringFromDouble
        | CastKind::ToStringFromBoolean
        | CastKind::ToStringFromTimestamp
        | CastKind::ToStringFromBinary
        | CastKind::ToStringFromDecimal
        | CastKind::ToStringFromInterval
        | CastKind::ToStringFromCalendarInterval => {
            Ok(try_cast(inner_sql, SQLBaseType::VarChar.into()))
        }

        // String casts - from string
        CastKind::StringToInt => Ok(try_cast(inner_sql, SQLBaseType::BigInt.into())),
        CastKind::StringToDouble => Ok(try_cast(inner_sql, SQLBaseType::Double.into())),
        CastKind::StringToBoolean => Ok(try_cast(inner_sql, SQLBaseType::Boolean.into())),
        CastKind::StringToTimestamp => Ok(try_cast(inner_sql, SQLTimestampTzType::new(6).into())),
        CastKind::StringToDecimal => {
            let sql_type = hamelin_type_to_sql_type(&cast.target_type, span)?;
            Ok(try_cast(inner_sql, sql_type))
        }

        // Null to type - just cast null
        CastKind::NullToType => {
            let sql_type = hamelin_type_to_sql_type(&cast.target_type, span)?;
            Ok(try_cast(inner_sql, sql_type))
        }

        // Variant casts
        CastKind::ToVariant(_) => {
            // Cast to JSON
            Ok(try_cast(inner_sql, SQLBaseType::Json.into()))
        }

        CastKind::FromVariant(variant_kind) => {
            translate_from_variant(inner_sql, variant_kind, span)
        }

        // Complex casts - delegate to apply_cast_kind which handles them recursively
        CastKind::ArrayElementCast(_)
        | CastKind::TupleToStruct(_)
        | CastKind::RangeElementCast(_)
        | CastKind::IntervalToTimestampRange
        | CastKind::TimestampToTimestampRange
        | CastKind::IntervalRangeToTimestampRange
        | CastKind::StructExpansion(_) => {
            apply_cast_kind(inner_sql, &cast.cast_kind, &cast.target_type, span)
        }
    }
}

/// Create a TRY_CAST SQL expression.
fn try_cast(expr: SQLExpression, to: SQLType) -> SQLExpression {
    TryCast::new(expr, to).into()
}

/// Translate a FromVariant cast.
///
/// Converts a JSON/variant value to a typed value. For primitive types,
/// this uses `json_extract_scalar` which returns VARCHAR, then casts to the target type.
/// Type guards ensure no cross-type coercion: the JSON runtime type must match the target.
/// For complex types (Array, Struct, Map), this uses specialized extraction.
fn translate_from_variant(
    inner_sql: SQLExpression,
    variant_kind: &VariantCastKind,
    span: &impl Spannable,
) -> ExprTranslationResult {
    match variant_kind {
        // Trino's native `TRY_CAST(JSON AS T)` implements the boxed-type rule for
        // these scalar targets: numeric/boolean variants unbox to their value,
        // string variants double-cast-parse (null on parse failure), and
        // object/array/null variants become null.
        VariantCastKind::Int => Ok(try_cast(inner_sql, SQLBaseType::BigInt.into())),
        VariantCastKind::Double => Ok(try_cast(inner_sql, SQLBaseType::Double.into())),
        VariantCastKind::Decimal { precision, scale } => Ok(try_cast(
            inner_sql,
            SQLDecimalType::new(*precision as u64, *scale as u64).into(),
        )),
        VariantCastKind::Boolean => Ok(try_cast(inner_sql, SQLBaseType::Boolean.into())),
        VariantCastKind::String => Ok(try_cast(inner_sql, SQLBaseType::VarChar.into())),

        // Trino refuses `TRY_CAST(JSON AS TIMESTAMP)` and its `TRY_CAST(varchar AS
        // TIMESTAMP)` does not accept ISO 8601. Compose: guard on "variant runtime
        // type is string" (json_format starts with `"`), extract the bare string,
        // and parse via `try(from_iso8601_timestamp(...))`. Other runtime types
        // become null, matching the rule: only variant-string gets the double-cast.
        VariantCastKind::Timestamp => Ok(translate_variant_to_timestamp(inner_sql)),

        VariantCastKind::Array(element_kind) => {
            translate_variant_to_array(inner_sql, element_kind, span)
        }

        VariantCastKind::Struct(fields) => translate_variant_to_struct(inner_sql, fields, span),

        VariantCastKind::Map(key_kind, value_kind) => {
            translate_variant_to_map(inner_sql, key_kind, value_kind, span)
        }

        VariantCastKind::Unknown => Ok(SQLNullLiteral {}.into()),

        VariantCastKind::Variant => Ok(inner_sql),
    }
}

/// Build the SQL for variant → timestamp on Trino.
///
/// Trino refuses `TRY_CAST(JSON AS TIMESTAMP)` directly, and its
/// `TRY_CAST(varchar AS TIMESTAMP)` does not accept ISO 8601 (`2024-01-15T14:30:45Z`).
/// `from_iso8601_timestamp` does accept it, but it also accepts bare numbers like
/// `'42'` by interpreting them as year-only partials — which would violate the
/// Hamelin rule that only variant-string runtime types can reach timestamp.
///
/// Produce: `CASE WHEN json_format(x) LIKE '"%' THEN
///              try(from_iso8601_timestamp(json_extract_scalar(x, '$')))
///            ELSE NULL END`
///
/// The `json_format LIKE '"%'` guard keeps non-string variants out; `try(...)`
/// turns parse failures into null.
fn translate_variant_to_timestamp(inner_sql: SQLExpression) -> SQLExpression {
    let json_format: SQLExpression =
        FunctionCallApply::with_one("json_format", inner_sql.clone()).into();
    let is_string_variant: SQLExpression = BinaryOperatorApply::new(
        Operator::Like,
        json_format,
        SQLStringLiteral::new("\"%").into(),
    )
    .into();
    let scalar = json_extract_scalar(inner_sql);
    let parsed: SQLExpression =
        FunctionCallApply::with_one("from_iso8601_timestamp", scalar).into();
    let wrapped: SQLExpression = FunctionCallApply::with_one("try", parsed).into();
    SQLCase::new(&[(is_string_variant, wrapped)], None).into()
}

/// Create a json_extract_scalar call with path '$'.
///
/// If the input is already a json_extract call, we convert it to json_extract_scalar
/// with the same arguments (preserving the path).
fn json_extract_scalar(expr: SQLExpression) -> SQLExpression {
    // Check if the expression is already a json_extract call
    if let Some((base, path)) = try_extract_json_extract_args(&expr) {
        // Convert json_extract to json_extract_scalar with the same arguments
        FunctionCallApply::with_two("json_extract_scalar", base, path).into()
    } else {
        // Wrap in json_extract_scalar with path '$'
        FunctionCallApply::with_two("json_extract_scalar", expr, StringLiteral::new("$").into())
            .into()
    }
}

/// Try to extract the arguments from a json_extract function call.
/// Returns Some((base_expr, path_expr)) if the expression is a json_extract call.
fn try_extract_json_extract_args(expr: &SQLExpression) -> Option<(SQLExpression, SQLExpression)> {
    // Match against the FunctionCallApply structure
    if let SQLExpression::FunctionCallApply(func) = expr {
        if func.function_name == "json_extract" && func.arguments.len() == 2 {
            return Some((func.arguments[0].clone(), func.arguments[1].clone()));
        }
    }
    None
}

/// Translate a VariantToArray cast.
///
/// Pattern: transform(TRY_CAST(expr AS ARRAY(JSON)), e -> TRY_CAST(e AS element_type))
///
/// First casts the JSON to ARRAY(JSON), then transforms each element to the target type.
fn translate_variant_to_array(
    inner_sql: SQLExpression,
    element_kind: &VariantCastKind,
    span: &impl Spannable,
) -> ExprTranslationResult {
    // First, cast to ARRAY(JSON)
    let array_of_json: SQLType = SQLArrayType::new(SQLBaseType::Json.into()).into();
    let cast_to_array_json: SQLExpression = TryCast::new(inner_sql, array_of_json).into();

    // Build the element transformation lambda: e -> TRY_CAST(e AS element_type)
    let lambda_param = SQLSimpleIdentifier::new("e");
    let param_ref: SQLExpression = SQLColumnReference::new(lambda_param.clone().into()).into();

    // Get the element cast expression
    let element_cast = translate_from_variant(param_ref, element_kind, span)?;

    // Build: transform(array, e -> element_cast)
    let lambda: SQLExpression = SQLLambda::new(vec![lambda_param], element_cast).into();

    Ok(FunctionCallApply::with_two("transform", cast_to_array_json, lambda).into())
}

/// Translate a VariantToStruct cast.
///
/// Fast path: Trino's native `TRY_CAST(json AS ROW(...))` handles JSON → struct
/// efficiently for most field types.
///
/// Slow path: when any field (recursively) is a Timestamp, Trino refuses the
/// direct cast ("Cannot cast json to row(... timestamp ...)"), so we extract
/// each field individually via `json_extract` and apply the field's own variant
/// cast translation, then assemble into a ROW.
fn translate_variant_to_struct(
    inner_sql: SQLExpression,
    fields: &[(String, VariantCastKind)],
    span: &impl Spannable,
) -> ExprTranslationResult {
    // Per-field extraction is forced whenever a field's cast needs Trino-
    // specific fallback handling that its native TRY_CAST(JSON AS ROW(...))
    // wouldn't provide:
    //   - Timestamp anywhere: Trino refuses JSON→ROW with timestamp fields.
    //   - Variant anywhere: Trinos JSON→ROW fails the whole row when a field
    //     has the wrong JSON shape, violating the boxed-type rule that per-
    //     field shape mismatches should null just that field.
    let needs_per_field = fields.iter().any(|(_, kind)| {
        variant_cast_kind_contains_timestamp(kind) || variant_cast_kind_contains_variant(kind)
    });

    let cast_expr = if !needs_per_field {
        let mut row_type_bindings = ordermap::OrderMap::new();
        for (field_name, field_kind) in fields {
            let sql_name = SQLSimpleIdentifier::new(field_name);
            let sql_type = variant_cast_kind_to_sql_type(field_kind, span)?;
            row_type_bindings.insert(sql_name, sql_type);
        }
        let row_type: SQLType = SQLRowType::new(row_type_bindings).into();
        try_cast(inner_sql.clone(), row_type)
    } else {
        // Per-field extraction: CAST(ROW(cast1, cast2, ...) AS ROW(name1 t1, ...))
        let mut extracted_values: Vec<SQLExpression> = Vec::with_capacity(fields.len());
        let mut row_type_bindings = ordermap::OrderMap::new();
        for (field_name, field_kind) in fields {
            let path = format!("$[\"{}\"]", field_name.replace('"', "\\\""));
            let field_json: SQLExpression = FunctionCallApply::with_two(
                "json_extract",
                inner_sql.clone(),
                SQLStringLiteral::new(&path).into(),
            )
            .into();
            let field_expr = translate_from_variant(field_json, field_kind, span)?;
            extracted_values.push(field_expr);

            let sql_name = SQLSimpleIdentifier::new(field_name);
            let sql_type = variant_cast_kind_to_sql_type(field_kind, span)?;
            row_type_bindings.insert(sql_name, sql_type);
        }

        let row_type: SQLType = SQLRowType::new(row_type_bindings).into();
        let row_literal: SQLExpression =
            FunctionCallApply::with_positional("row", extracted_values).into();
        SQLCast::new(row_literal, row_type).into()
    };

    // Trino's TRY_CAST(JSON AS ROW(...)) happily absorbs a JSON array into a
    // row by spreading elements into fields, violating the boxed-type rule
    // which requires null on shape mismatch. Guard with a json_format LIKE '{%'
    // check so only actual JSON objects reach the cast.
    let json_format: SQLExpression = FunctionCallApply::with_one("json_format", inner_sql).into();
    let is_object: SQLExpression = BinaryOperatorApply::new(
        Operator::Like,
        json_format,
        SQLStringLiteral::new("{%").into(),
    )
    .into();
    Ok(SQLCase::new(&[(is_object, cast_expr)], None).into())
}

/// Recursively check whether a VariantCastKind mentions a Timestamp target
/// anywhere (directly or inside Array/Struct/Map). Used by the Trino struct
/// translator to decide between the fast JSON→ROW cast and per-field extraction.
fn variant_cast_kind_contains_timestamp(kind: &VariantCastKind) -> bool {
    match kind {
        VariantCastKind::Timestamp => true,
        VariantCastKind::Array(element) => variant_cast_kind_contains_timestamp(element),
        VariantCastKind::Map(key, value) => {
            variant_cast_kind_contains_timestamp(key) || variant_cast_kind_contains_timestamp(value)
        }
        VariantCastKind::Struct(fields) => fields
            .iter()
            .any(|(_, k)| variant_cast_kind_contains_timestamp(k)),
        _ => false,
    }
}

/// Recursively check whether a VariantCastKind mentions a Variant target
/// anywhere. Used alongside `contains_timestamp` to force per-field extraction
/// on Trino, so a shape mismatch on one field nulls just that field instead
/// of failing the whole row.
fn variant_cast_kind_contains_variant(kind: &VariantCastKind) -> bool {
    match kind {
        VariantCastKind::Variant => true,
        VariantCastKind::Array(element) => variant_cast_kind_contains_variant(element),
        VariantCastKind::Map(key, value) => {
            variant_cast_kind_contains_variant(key) || variant_cast_kind_contains_variant(value)
        }
        VariantCastKind::Struct(fields) => fields
            .iter()
            .any(|(_, k)| variant_cast_kind_contains_variant(k)),
        _ => false,
    }
}

/// Translate a VariantToMap cast.
///
/// Fast path: Trino's native `TRY_CAST(json AS MAP(k, v))` is the common case.
///
/// Per-value path: when the value cast needs Trino-specific fallback handling,
/// cast to `MAP(k, JSON)` and `transform_values` to apply each value's own
/// translation. This ensures a per-value shape mismatch nulls just that entry
/// rather than failing the whole map, matching the boxed-type rule.
///
/// Per-key path: Trino's `TRY_CAST(JSON AS MAP(k, ...))` cannot parse JSON
/// object keys into Timestamp (or Variant), so when the key kind needs
/// per-element handling we cast to `MAP(VARCHAR, JSON)` and rebuild the key
/// via `transform_keys`, wrapping each VARCHAR key as a JSON string with
/// `CAST(k AS JSON)` before running the normal variant-string double-cast
/// path. Per-key and per-value paths compose: when both apply, the values
/// also flow through `transform_values`.
///
/// Either fallback fires whenever the relevant kind is itself a compound
/// target (another map/struct/array) or contains a Timestamp/Variant — all
/// cases where Trino's native cast would fail the whole map on a shape
/// mismatch.
fn translate_variant_to_map(
    inner_sql: SQLExpression,
    key_kind: &VariantCastKind,
    value_kind: &VariantCastKind,
    span: &impl Spannable,
) -> ExprTranslationResult {
    let needs_per_value = matches!(
        value_kind,
        VariantCastKind::Array(_) | VariantCastKind::Struct(_) | VariantCastKind::Map(_, _)
    ) || variant_cast_kind_contains_timestamp(value_kind)
        || variant_cast_kind_contains_variant(value_kind);

    let needs_per_key = matches!(
        key_kind,
        VariantCastKind::Array(_) | VariantCastKind::Struct(_) | VariantCastKind::Map(_, _)
    ) || variant_cast_kind_contains_timestamp(key_kind)
        || variant_cast_kind_contains_variant(key_kind);

    if !needs_per_value && !needs_per_key {
        let key_type = variant_cast_kind_to_sql_type(key_kind, span)?;
        let value_type = variant_cast_kind_to_sql_type(value_kind, span)?;
        let map_type: SQLType = SQLMapType::new(key_type, value_type).into();
        return Ok(try_cast(inner_sql, map_type));
    }

    // Slow path. Choose the outer cast key type to keep keys parseable:
    // - If the key needs per-key handling, cast to MAP(VARCHAR, JSON) and
    //   re-apply the key cast via transform_keys (wrapping the VARCHAR key
    //   back into a JSON string so translate_from_variant can run).
    // - Otherwise the key is a primitive Trino can parse from JSON keys
    //   directly (BIGINT, BOOLEAN, etc.), so keep it as-is in the outer cast.
    let outer_key_type: SQLType = if needs_per_key {
        SQLBaseType::VarChar.into()
    } else {
        variant_cast_kind_to_sql_type(key_kind, span)?
    };
    let map_json_type: SQLType = SQLMapType::new(outer_key_type, SQLBaseType::Json.into()).into();
    let mut map_expr: SQLExpression = TryCast::new(inner_sql, map_json_type).into();

    if needs_per_key {
        let key_param = SQLSimpleIdentifier::new("k");
        let value_param = SQLSimpleIdentifier::new("v");
        let key_ref: SQLExpression = SQLColumnReference::new(key_param.clone().into()).into();
        // VARCHAR → JSON string literal so translate_from_variant sees a
        // proper variant-string and can run its double-cast (e.g. ISO 8601
        // → timestamp).
        let key_as_json: SQLExpression = SQLCast::new(key_ref, SQLBaseType::Json.into()).into();
        let key_cast = translate_from_variant(key_as_json, key_kind, span)?;
        let lambda: SQLExpression = SQLLambda::new(vec![key_param, value_param], key_cast).into();
        map_expr = FunctionCallApply::with_two("transform_keys", map_expr, lambda).into();
    }

    // Values are always JSON-typed in the slow path (regardless of which
    // sub-path triggered it), so always re-apply the value cast here.
    let key_param = SQLSimpleIdentifier::new("k");
    let value_param = SQLSimpleIdentifier::new("v");
    let value_ref: SQLExpression = SQLColumnReference::new(value_param.clone().into()).into();
    let value_cast = translate_from_variant(value_ref, value_kind, span)?;
    let lambda: SQLExpression = SQLLambda::new(vec![key_param, value_param], value_cast).into();
    map_expr = FunctionCallApply::with_two("transform_values", map_expr, lambda).into();

    Ok(map_expr)
}

/// Check if a cast kind can be translated with a direct SQL CAST on the array.
///
/// Primitive casts (numeric conversions, string conversions, etc.) can use
/// `CAST(array AS ARRAY(target_type))` directly, avoiding lambda overhead.
fn is_primitive_cast(cast_kind: &CastKind) -> bool {
    matches!(
        cast_kind,
        CastKind::Identity
            | CastKind::IntToDouble
            | CastKind::IntToDecimal
            | CastKind::DoubleToInt
            | CastKind::DoubleToDecimal
            | CastKind::DecimalToInt
            | CastKind::DecimalToDouble
            | CastKind::DecimalToDecimal
            | CastKind::IntToBoolean
            | CastKind::BooleanToInt
            | CastKind::BooleanToDouble
            | CastKind::BooleanToDecimal
            | CastKind::DoubleToBoolean
            | CastKind::DecimalToBoolean
            | CastKind::ToStringFromInt
            | CastKind::ToStringFromDouble
            | CastKind::ToStringFromBoolean
            | CastKind::ToStringFromTimestamp
            | CastKind::ToStringFromBinary
            | CastKind::ToStringFromDecimal
            | CastKind::ToStringFromInterval
            | CastKind::ToStringFromCalendarInterval
            | CastKind::StringToInt
            | CastKind::StringToDouble
            | CastKind::StringToBoolean
            | CastKind::StringToTimestamp
            | CastKind::StringToDecimal
            | CastKind::NullToType
    )
}

/// Translate an ArrayElementCast.
///
/// For primitive inner casts, uses direct `CAST(array AS ARRAY(target_type))`.
/// For complex inner casts (StructExpansion, nested ArrayElementCast), uses
/// `transform(arr, x -> apply_cast_kind(x))`.
fn translate_array_element_cast(
    inner_sql: SQLExpression,
    inner_cast_kind: &CastKind,
    target_type: &Type,
    span: &impl Spannable,
) -> ExprTranslationResult {
    // Get the element type from target_type (which should be Array)
    let element_type = match target_type {
        Type::Array(arr) => &arr.element_type,
        _ => {
            return Err(TranslationError::msg(
                span,
                "ArrayElementCast target type is not an array",
            )
            .single());
        }
    };

    // For primitive casts, use direct CAST on the array
    if is_primitive_cast(inner_cast_kind) {
        let sql_type = hamelin_type_to_sql_type(target_type, span)?;
        return Ok(try_cast(inner_sql, sql_type));
    }

    // For complex casts, use transform(arr, x -> apply_cast_kind(x))
    let lambda_param = SQLSimpleIdentifier::new("x");
    let param_ref: SQLExpression = SQLColumnReference::new(lambda_param.clone().into()).into();

    // Build the element cast using the recursive apply_cast_kind
    let element_cast = apply_cast_kind(param_ref, inner_cast_kind, element_type, span)?;

    // Build: transform(array, x -> element_cast)
    let lambda: SQLExpression = SQLLambda::new(vec![lambda_param], element_cast).into();

    Ok(FunctionCallApply::with_two("transform", inner_sql, lambda).into())
}

/// Translate a cast applied to an expression.
///
/// This is the recursive workhorse for applying CastKind to SQL expressions.
/// Used by array element casts, struct field casts, range element casts, etc.
fn apply_cast_kind(
    expr: SQLExpression,
    cast_kind: &CastKind,
    target_type: &Type,
    span: &impl Spannable,
) -> ExprTranslationResult {
    match cast_kind {
        CastKind::Identity => Ok(expr),

        // Simple primitive casts - use TRY_CAST
        CastKind::IntToDouble
        | CastKind::IntToDecimal
        | CastKind::DoubleToInt
        | CastKind::DoubleToDecimal
        | CastKind::DecimalToInt
        | CastKind::DecimalToDouble
        | CastKind::DecimalToDecimal
        | CastKind::IntToBoolean
        | CastKind::BooleanToInt
        | CastKind::BooleanToDouble
        | CastKind::BooleanToDecimal
        | CastKind::DoubleToBoolean
        | CastKind::DecimalToBoolean
        | CastKind::ToStringFromInt
        | CastKind::ToStringFromDouble
        | CastKind::ToStringFromBoolean
        | CastKind::ToStringFromTimestamp
        | CastKind::ToStringFromBinary
        | CastKind::ToStringFromDecimal
        | CastKind::ToStringFromInterval
        | CastKind::ToStringFromCalendarInterval
        | CastKind::StringToInt
        | CastKind::StringToDouble
        | CastKind::StringToBoolean
        | CastKind::StringToTimestamp
        | CastKind::StringToDecimal
        | CastKind::NullToType => {
            let sql_type = hamelin_type_to_sql_type(target_type, span)?;
            Ok(try_cast(expr, sql_type))
        }

        // Array element cast - use transform(arr, x -> apply_cast_kind(x))
        CastKind::ArrayElementCast(inner) => {
            translate_array_element_cast(expr, inner, target_type, span)
        }

        // Struct expansion - recursively apply field casts
        CastKind::StructExpansion(field_casts) => {
            apply_struct_expansion(expr, field_casts, target_type, span)
        }

        // Tuple to struct - cast to named ROW type
        CastKind::TupleToStruct(_) => {
            let row_type = hamelin_type_to_sql_type(target_type, span)?;
            Ok(SQLCast::new(expr, row_type).into())
        }

        // Range element cast - cast begin and end fields
        CastKind::RangeElementCast(inner) => {
            let element_type = match target_type {
                Type::Range(range) | Type::RangeInclusive(range) => &range.of,
                _ => {
                    return Err(TranslationError::msg(
                        span,
                        "RangeElementCast target type is not a range",
                    )
                    .single());
                }
            };

            let element_sql_type = hamelin_type_to_sql_type(element_type, span)?;

            let begin_ref: SQLExpression =
                Dot::new(expr.clone(), SQLSimpleIdentifier::new("begin").into()).into();
            let end_ref: SQLExpression =
                Dot::new(expr, SQLSimpleIdentifier::new("end").into()).into();

            let begin_cast = apply_cast_kind(begin_ref, inner, element_type, span)?;
            let end_cast = apply_cast_kind(end_ref, inner, element_type, span)?;

            let row_literal: SQLExpression = SQLRowLiteral::new(vec![begin_cast, end_cast]).into();
            let mut row_type_bindings = ordermap::OrderMap::new();
            row_type_bindings.insert(SQLSimpleIdentifier::new("begin"), element_sql_type.clone());
            row_type_bindings.insert(SQLSimpleIdentifier::new("end"), element_sql_type);
            let row_type: SQLType = SQLRowType::new(row_type_bindings).into();
            Ok(SQLCast::new(row_literal, row_type).into())
        }

        // Variant casts
        CastKind::ToVariant(_) => Ok(try_cast(expr, SQLBaseType::Json.into())),
        CastKind::FromVariant(variant_kind) => translate_from_variant(expr, variant_kind, span),

        // Interval/timestamp range casts
        CastKind::IntervalToTimestampRange => translate_interval_to_timestamp_range(expr),
        CastKind::TimestampToTimestampRange => translate_timestamp_to_timestamp_range(expr),
        CastKind::IntervalRangeToTimestampRange => {
            translate_interval_range_to_timestamp_range(expr)
        }
    }
}

/// Apply struct expansion: build a new struct with fields extracted/cast from source, plus nulls for new fields.
fn apply_struct_expansion(
    source_expr: SQLExpression,
    field_casts: &[(SimpleIdentifier, CastKind)],
    target_type: &Type,
    span: &impl Spannable,
) -> ExprTranslationResult {
    let target_struct = match target_type {
        Type::Struct(s) => s,
        _ => {
            return Err(TranslationError::msg(
                span,
                &format!(
                    "StructExpansion target type is not a struct: {:?}",
                    target_type
                ),
            )
            .single());
        }
    };

    let mut row_elements = Vec::new();
    let mut row_type_bindings = ordermap::OrderMap::new();

    for (field_name, cast_kind) in field_casts {
        let sql_name = SQLSimpleIdentifier::new(field_name.name());
        let field_type = target_struct.lookup(field_name).ok_or_else(|| {
            TranslationError::msg(
                span,
                &format!("Field '{}' must exist in target struct", field_name),
            )
            .single()
        })?;
        let sql_type = hamelin_type_to_sql_type(field_type, span)?;

        let field_expr = match cast_kind {
            CastKind::NullToType => {
                // Field doesn't exist in source - use CAST(NULL AS type)
                let null_expr: SQLExpression = SQLNullLiteral {}.into();
                SQLCast::new(null_expr, sql_type.clone()).into()
            }
            // ArrayElementCast with StructExpansion inside: build the transform
            // lambda with an anonymous ROW body to avoid Trino's PushCastIntoRow
            // optimizer stripping field names when it decomposes the outer
            // CAST(ROW(...) AS ROW(...)).
            CastKind::ArrayElementCast(inner_cast)
                if matches!(inner_cast.as_ref(), CastKind::StructExpansion(_)) =>
            {
                let field_ref: SQLExpression =
                    Dot::new(source_expr.clone(), sql_name.clone().into()).into();
                translate_array_element_cast_with_anonymous_struct(
                    field_ref, inner_cast, field_type, span,
                )?
            }
            _ => {
                // Field exists in source - extract and apply cast
                let field_ref: SQLExpression =
                    Dot::new(source_expr.clone(), sql_name.clone().into()).into();
                apply_cast_kind(field_ref, cast_kind, field_type, span)?
            }
        };

        row_elements.push(field_expr);
        row_type_bindings.insert(sql_name, sql_type);
    }

    // Build ROW(...) AS ROW("f1" t1, "f2" t2, ...)
    let row_literal: SQLExpression = SQLRowLiteral::new(row_elements).into();
    let row_type: SQLType = SQLRowType::new(row_type_bindings).into();
    Ok(SQLCast::new(row_literal, row_type).into())
}

/// Translate an ArrayElementCast where the inner cast is a StructExpansion,
/// using an anonymous ROW in the transform lambda body.
///
/// This produces `transform(arr, x -> ROW(field1, field2, ...))` instead of
/// `transform(arr, x -> CAST(ROW(...) AS ROW("a" T, "b" T)))`. The outer
/// `CAST(ROW(...) AS ROW("name" ARRAY(ROW("a" T, "b" T))))` from the parent
/// struct expansion will apply the correct field names via Trino's PushCastIntoRow.
fn translate_array_element_cast_with_anonymous_struct(
    inner_sql: SQLExpression,
    inner_cast_kind: &CastKind,
    target_type: &Type,
    span: &impl Spannable,
) -> ExprTranslationResult {
    let element_type = match target_type {
        Type::Array(arr) => &arr.element_type,
        _ => {
            return Err(TranslationError::msg(
                span,
                "ArrayElementCast target type is not an array",
            )
            .single());
        }
    };

    let field_casts = match inner_cast_kind {
        CastKind::StructExpansion(fc) => fc,
        _ => return apply_cast_kind(inner_sql, inner_cast_kind, target_type, span),
    };

    let lambda_param = SQLSimpleIdentifier::new("x");
    let param_ref: SQLExpression = SQLColumnReference::new(lambda_param.clone().into()).into();

    let element_cast =
        build_anonymous_struct_expansion(param_ref, field_casts, element_type, span)?;

    let lambda: SQLExpression = SQLLambda::new(vec![lambda_param], element_cast).into();

    Ok(FunctionCallApply::with_two("transform", inner_sql, lambda).into())
}

/// Build an anonymous ROW(...) for struct expansion inside a transform lambda.
///
/// This is like `apply_struct_expansion` but returns `ROW(field1, field2, ...)`
/// without wrapping in `CAST(... AS ROW("name" type, ...))`. The field naming
/// is left to the caller (the outer CAST ROW from the parent struct expansion).
///
/// This avoids a Trino bug where `PushCastIntoRow` strips field names from
/// `CAST(ROW(...) AS ROW(...))` inside transform lambdas when the outer
/// expression is also wrapped in `CAST(ROW(...) AS ROW(...))`.
fn build_anonymous_struct_expansion(
    source_expr: SQLExpression,
    field_casts: &[(SimpleIdentifier, CastKind)],
    target_type: &Type,
    span: &impl Spannable,
) -> ExprTranslationResult {
    let target_struct = match target_type {
        Type::Struct(s) => s,
        _ => {
            return Err(TranslationError::msg(
                span,
                &format!(
                    "StructExpansion target type is not a struct: {:?}",
                    target_type
                ),
            )
            .single());
        }
    };

    let mut row_elements = Vec::new();

    for (field_name, cast_kind) in field_casts {
        let sql_name = SQLSimpleIdentifier::new(field_name.name());
        let field_type = target_struct.lookup(field_name).ok_or_else(|| {
            TranslationError::msg(
                span,
                &format!("Field '{}' must exist in target struct", field_name),
            )
            .single()
        })?;

        let field_expr = match cast_kind {
            CastKind::NullToType => {
                let sql_type = hamelin_type_to_sql_type(field_type, span)?;
                let null_expr: SQLExpression = SQLNullLiteral {}.into();
                SQLCast::new(null_expr, sql_type).into()
            }
            _ => {
                let field_ref: SQLExpression =
                    Dot::new(source_expr.clone(), sql_name.clone().into()).into();
                apply_cast_kind(field_ref, cast_kind, field_type, span)?
            }
        };

        row_elements.push(field_expr);
    }

    Ok(SQLRowLiteral::new(row_elements).into())
}

/// Translate IntervalToTimestampRange cast.
///
/// Converts an interval to a timestamp range: ROW(now(), now() + interval)
fn translate_interval_to_timestamp_range(inner_sql: SQLExpression) -> ExprTranslationResult {
    // For simplicity, we'll use: ROW(now(), now() + interval)
    // The type checker should have validated this is a valid conversion
    let current_ts: SQLExpression = FunctionCallApply::with_no_arguments("now").into();

    // begin = current_timestamp, end = current_timestamp + interval
    let end_expr: SQLExpression =
        BinaryOperatorApply::new(Operator::Plus, current_ts.clone(), inner_sql).into();

    let row_literal: SQLExpression = SQLRowLiteral::new(vec![current_ts, end_expr]).into();

    let mut row_type_bindings = ordermap::OrderMap::new();
    row_type_bindings.insert(
        SQLSimpleIdentifier::new("begin"),
        SQLTimestampTzType::new(6).into(),
    );
    row_type_bindings.insert(
        SQLSimpleIdentifier::new("end"),
        SQLTimestampTzType::new(6).into(),
    );
    let row_type: SQLType = SQLRowType::new(row_type_bindings).into();

    Ok(SQLCast::new(row_literal, row_type).into())
}

/// Translate TimestampToTimestampRange cast.
///
/// Converts a timestamp to a point range: ROW(ts, ts)
fn translate_timestamp_to_timestamp_range(inner_sql: SQLExpression) -> ExprTranslationResult {
    // Both begin and end are the same timestamp (point range)
    let row_literal: SQLExpression = SQLRowLiteral::new(vec![inner_sql.clone(), inner_sql]).into();

    let mut row_type_bindings = ordermap::OrderMap::new();
    row_type_bindings.insert(
        SQLSimpleIdentifier::new("begin"),
        SQLTimestampTzType::new(6).into(),
    );
    row_type_bindings.insert(
        SQLSimpleIdentifier::new("end"),
        SQLTimestampTzType::new(6).into(),
    );
    let row_type: SQLType = SQLRowType::new(row_type_bindings).into();

    Ok(SQLCast::new(row_literal, row_type).into())
}

/// Translate IntervalRangeToTimestampRange cast.
///
/// Converts an interval range to a timestamp range:
/// ROW(now() + begin_interval, now() + end_interval)
fn translate_interval_range_to_timestamp_range(inner_sql: SQLExpression) -> ExprTranslationResult {
    let current_ts: SQLExpression = FunctionCallApply::with_no_arguments("now").into();

    // Extract begin and end intervals from the range
    let begin_interval: SQLExpression =
        Dot::new(inner_sql.clone(), SQLSimpleIdentifier::new("begin").into()).into();
    let end_interval: SQLExpression =
        Dot::new(inner_sql, SQLSimpleIdentifier::new("end").into()).into();

    // begin = current_timestamp + begin_interval
    let begin_ts: SQLExpression =
        BinaryOperatorApply::new(Operator::Plus, current_ts.clone(), begin_interval).into();

    // end = current_timestamp + end_interval
    let end_ts: SQLExpression =
        BinaryOperatorApply::new(Operator::Plus, current_ts, end_interval).into();

    let row_literal: SQLExpression = SQLRowLiteral::new(vec![begin_ts, end_ts]).into();

    let mut row_type_bindings = ordermap::OrderMap::new();
    row_type_bindings.insert(
        SQLSimpleIdentifier::new("begin"),
        SQLTimestampTzType::new(6).into(),
    );
    row_type_bindings.insert(
        SQLSimpleIdentifier::new("end"),
        SQLTimestampTzType::new(6).into(),
    );
    let row_type: SQLType = SQLRowType::new(row_type_bindings).into();

    Ok(SQLCast::new(row_literal, row_type).into())
}

/// Convert a VariantCastKind to the corresponding SQL type.
fn variant_cast_kind_to_sql_type(
    kind: &VariantCastKind,
    span: &impl Spannable,
) -> Result<SQLType, TranslationErrors> {
    match kind {
        VariantCastKind::Int => Ok(SQLBaseType::BigInt.into()),
        VariantCastKind::Double => Ok(SQLBaseType::Double.into()),
        VariantCastKind::Decimal { precision, scale } => {
            Ok(SQLDecimalType::new(*precision as u64, *scale as u64).into())
        }
        VariantCastKind::String => Ok(SQLBaseType::VarChar.into()),
        VariantCastKind::Boolean => Ok(SQLBaseType::Boolean.into()),
        VariantCastKind::Timestamp => Ok(SQLTimestampTzType::new(6).into()),
        VariantCastKind::Array(element_kind) => {
            let element_type = variant_cast_kind_to_sql_type(element_kind, span)?;
            Ok(SQLArrayType::new(element_type).into())
        }
        VariantCastKind::Struct(fields) => {
            let mut bindings = ordermap::OrderMap::new();
            for (name, field_kind) in fields {
                let sql_name = SQLSimpleIdentifier::new(name);
                let sql_type = variant_cast_kind_to_sql_type(field_kind, span)?;
                bindings.insert(sql_name, sql_type);
            }
            Ok(SQLRowType::new(bindings).into())
        }
        VariantCastKind::Map(key_kind, value_kind) => {
            let key_type = variant_cast_kind_to_sql_type(key_kind, span)?;
            let value_type = variant_cast_kind_to_sql_type(value_kind, span)?;
            Ok(SQLMapType::new(key_type, value_type).into())
        }
        // Unknown type doesn't have a meaningful SQL type - use JSON as fallback
        VariantCastKind::Unknown => Ok(SQLBaseType::Json.into()),
        // Variant maps to JSON type
        VariantCastKind::Variant => Ok(SQLBaseType::Json.into()),
    }
}

/// Translate a timestamp truncation expression.
///
/// For multiplier == 1, produces `date_trunc(unit, ts)`.
/// For multiplier > 1, produces arithmetic that snaps to an N-unit grid:
///   `date_trunc(parent_unit, ts) + INTERVAL '1' unit * (FLOOR(EXTRACT(unit FROM ts) / N) * N)`
fn translate_ts_trunc(ctx: &TranslationContext, ts_trunc: &TypedTsTrunc) -> ExprTranslationResult {
    let inner_expr = IRExpression::new(ts_trunc.expression.clone());
    let inner_sql = translate_expression(ctx, &inner_expr)?;

    if ts_trunc.multiplier <= 1 {
        let unit_str = trunc_unit_to_sql_str(&ts_trunc.unit);
        return Ok(FunctionCallApply::with_two(
            "date_trunc",
            StringLiteral::new(unit_str).into(),
            inner_sql,
        )
        .into());
    }

    translate_ts_trunc_with_multiplier(inner_sql, &ts_trunc.unit, ts_trunc.multiplier)
}

fn trunc_unit_to_sql_str(unit: &TruncUnit) -> &'static str {
    match unit {
        TruncUnit::Second => "second",
        TruncUnit::Minute => "minute",
        TruncUnit::Hour => "hour",
        TruncUnit::Day => "day",
        TruncUnit::Week => "week",
        TruncUnit::Month => "month",
        TruncUnit::Quarter => "quarter",
        TruncUnit::Year => "year",
    }
}

/// Build the SQL arithmetic for truncation with a multiplier > 1.
///
/// Sub-day and day units use epoch-modular arithmetic:
///   `from_unixtime(floor(to_unixtime(ts) / step) * step)`
/// where step = N * unit_seconds.
///
/// Calendar units (month) anchor to start-of-year with component arithmetic.
fn translate_ts_trunc_with_multiplier(
    inner_sql: SQLExpression,
    unit: &TruncUnit,
    multiplier: u32,
) -> ExprTranslationResult {
    let step_seconds: Option<i64> = match unit {
        TruncUnit::Second => Some(multiplier as i64),
        TruncUnit::Minute => Some(multiplier as i64 * 60),
        TruncUnit::Hour => Some(multiplier as i64 * 3600),
        TruncUnit::Day => Some(multiplier as i64 * 86400),
        TruncUnit::Month => {
            return translate_ts_trunc_month_multiplier(inner_sql, multiplier);
        }
        TruncUnit::Quarter => {
            return translate_ts_trunc_quarter_multiplier(inner_sql, multiplier);
        }
        TruncUnit::Week => {
            return translate_ts_trunc_week_multiplier(inner_sql, multiplier);
        }
        TruncUnit::Year => {
            return translate_ts_trunc_year_multiplier(inner_sql, multiplier);
        }
    };

    translate_ts_trunc_epoch_modular(inner_sql, step_seconds.unwrap_or(1))
}

/// Epoch-modular truncation: `from_unixtime(floor(to_unixtime(ts) / step) * step)`
fn translate_ts_trunc_epoch_modular(
    inner_sql: SQLExpression,
    step_seconds: i64,
) -> ExprTranslationResult {
    let epoch: SQLExpression = FunctionCallApply::with_one("to_unixtime", inner_sql).into();
    let step: SQLExpression = SQLIntegerLiteral::from_int(step_seconds).into();

    let divided = BinaryOperatorApply::new(Operator::Slash, epoch, step.clone());
    let floored: SQLExpression = FunctionCallApply::with_one("floor", divided.into()).into();
    let snapped = BinaryOperatorApply::new(Operator::Asterisk, floored, step);

    Ok(FunctionCallApply::with_one("from_unixtime", snapped.into()).into())
}

/// Month with multiplier: year-anchored calendar arithmetic.
fn translate_ts_trunc_month_multiplier(
    inner_sql: SQLExpression,
    multiplier: u32,
) -> ExprTranslationResult {
    // date_trunc('year', ts)
    let truncated_year = FunctionCallApply::with_two(
        "date_trunc",
        StringLiteral::new("year").into(),
        inner_sql.clone(),
    );

    // (EXTRACT(MONTH FROM ts) - 1) / N * N
    let extract_month: SQLExpression =
        ExtractFunction::new(ExtractField::Month, inner_sql.clone()).into();
    let one: SQLExpression = SQLIntegerLiteral::from_int(1).into();
    let month_zero_based = BinaryOperatorApply::new(Operator::Minus, extract_month, one);
    let n: SQLExpression = SQLIntegerLiteral::from_int(multiplier as i64).into();
    let divided = BinaryOperatorApply::new(Operator::Slash, month_zero_based.into(), n.clone());
    let floored: SQLExpression = FunctionCallApply::with_one("floor", divided.into()).into();
    let snapped = BinaryOperatorApply::new(Operator::Asterisk, floored, n);

    // INTERVAL '1' MONTH * snapped
    let one_month: SQLExpression = SQLIntervalLiteral::new("1", SQLIntervalUnit::Month).into();
    let offset = BinaryOperatorApply::new(
        Operator::Asterisk,
        one_month,
        SQLCast::new(snapped.into(), SQLBaseType::BigInt.into()).into(),
    );

    Ok(BinaryOperatorApply::new(Operator::Plus, truncated_year.into(), offset.into()).into())
}

/// Quarter with multiplier: treat as N*3 months from start of year.
fn translate_ts_trunc_quarter_multiplier(
    inner_sql: SQLExpression,
    multiplier: u32,
) -> ExprTranslationResult {
    translate_ts_trunc_month_multiplier(inner_sql, multiplier * 3)
}

/// Year with multiplier: calendar-based arithmetic.
///
/// Computes `date_trunc('year', ts) - INTERVAL '1' YEAR * (year - floor(year / N) * N)`
/// where year = EXTRACT(YEAR FROM ts). This uses Euclidean modulo (always non-negative)
/// to snap to N-year boundaries aligned to year 0 (matching the eval engine).
fn translate_ts_trunc_year_multiplier(
    inner_sql: SQLExpression,
    multiplier: u32,
) -> ExprTranslationResult {
    let truncated_year = FunctionCallApply::with_two(
        "date_trunc",
        StringLiteral::new("year").into(),
        inner_sql.clone(),
    );

    let year: SQLExpression = ExtractFunction::new(ExtractField::Year, inner_sql).into();
    // Cast year to DOUBLE for correct floor division (integer division truncates toward zero)
    let year_double: SQLExpression = SQLCast::new(year.clone(), SQLBaseType::Double.into()).into();
    let n: SQLExpression = SQLIntegerLiteral::from_int(multiplier as i64).into();

    // Use Euclidean modulo: year - floor(year / N) * N
    // This ensures the remainder is always non-negative, matching the eval engine's div_euclid
    let divided = BinaryOperatorApply::new(Operator::Slash, year_double, n.clone());
    let floored: SQLExpression = FunctionCallApply::with_one("floor", divided.into()).into();
    let floored_times_n = BinaryOperatorApply::new(Operator::Asterisk, floored, n);
    let remainder = BinaryOperatorApply::new(Operator::Minus, year, floored_times_n.into());

    let one_year: SQLExpression = SQLIntervalLiteral::new("1", SQLIntervalUnit::Year).into();
    let offset = BinaryOperatorApply::new(
        Operator::Asterisk,
        one_year,
        SQLCast::new(remainder.into(), SQLBaseType::BigInt.into()).into(),
    );

    Ok(BinaryOperatorApply::new(Operator::Minus, truncated_year.into(), offset.into()).into())
}

/// Week with multiplier: Monday-aligned epoch-based arithmetic.
///
/// First truncates to Monday via `date_trunc('week', ts)`, then snaps to N-week
/// boundaries using epoch seconds relative to Monday 1970-01-05 (the first Monday
/// of the Unix era), matching the eval engine's reference point.
fn translate_ts_trunc_week_multiplier(
    inner_sql: SQLExpression,
    multiplier: u32,
) -> ExprTranslationResult {
    let monday: SQLExpression =
        FunctionCallApply::with_two("date_trunc", StringLiteral::new("week").into(), inner_sql)
            .into();

    let epoch: SQLExpression = FunctionCallApply::with_one("to_unixtime", monday).into();

    // 1970-01-05 00:00 UTC = 4 days * 86400 = 345600 seconds from Unix epoch
    let ref_epoch: SQLExpression = SQLIntegerLiteral::from_int(345600).into();
    let step: SQLExpression = SQLIntegerLiteral::from_int(multiplier as i64 * 7 * 86400).into();

    let shifted = BinaryOperatorApply::new(Operator::Minus, epoch, ref_epoch.clone());
    let divided = BinaryOperatorApply::new(Operator::Slash, shifted.into(), step.clone());
    let floored: SQLExpression = FunctionCallApply::with_one("floor", divided.into()).into();
    let snapped = BinaryOperatorApply::new(Operator::Asterisk, floored, step);
    let result_epoch = BinaryOperatorApply::new(Operator::Plus, snapped.into(), ref_epoch);

    Ok(FunctionCallApply::with_one("from_unixtime", result_epoch.into()).into())
}

/// Translate a lambda expression.
fn translate_lambda(ctx: &TranslationContext, lambda: &TypedLambda) -> ExprTranslationResult {
    let params: Vec<SQLSimpleIdentifier> = lambda
        .parameters
        .iter()
        .map(|p| p.name.clone().into())
        .collect();

    // Translate the body
    let body_expr = IRExpression::new(lambda.body.clone());
    let body_sql = translate_expression(ctx, &body_expr)?;

    Ok(SQLLambda::new(params, body_sql).into())
}

/// Translate a function application expression.
///
/// This translates `TypedApply` by:
/// 1. Recursively translating all argument expressions to SQL
/// 2. Wrapping each in `ExpressionTranslation` (SQL + type)
/// 3. Calling the registry's translate method with the function def and bindings
fn translate_apply(
    ctx: &TranslationContext,
    apply: &TypedApply,
    resolved_type: &Type,
    span: &impl Spannable,
) -> ExprTranslationResult {
    // Translate each argument expression to SQL and wrap in ExpressionTranslation
    let translated_binding = apply
        .parameter_binding
        .clone()
        .try_map(|typed_expr| {
            let ir_expr = IRExpression::new(typed_expr.clone());
            let sql = translate_expression(ctx, &ir_expr)?;
            let typ = (*typed_expr.resolved_type).clone();
            let expr_span = typed_expr.ast.span.to_range();
            Ok(ExpressionTranslation {
                sql,
                typ,
                span: expr_span,
                special: None,
                nested_special: Vec::new(),
            })
        })
        .map_err(|e: TranslationErrors| e)?;

    // Call the registry's translate method
    ctx.registry
        .translate(
            apply.function_def.as_ref(),
            apply.function_def.name(),
            &ctx.fctx,
            translated_binding,
            resolved_type.clone(),
        )
        .map(|et| et.sql)
        .map_err(|e| {
            TranslationError::msg(span, &format!("Function translation failed: {}", e)).single()
        })
}

/// Translate a field lookup expression.
fn translate_field_lookup(
    ctx: &TranslationContext,
    field_lookup: &TypedFieldLookup,
    _span: &impl Spannable,
) -> ExprTranslationResult {
    let inner_expr = IRExpression::new(field_lookup.value.clone());
    let inner_sql = translate_expression(ctx, &inner_expr)?;

    match &field_lookup.access {
        FieldAccess::StructField(parsed_id) => match parsed_id {
            ParsedSimpleIdentifier::Valid(id) => {
                let sql_ident: SQLIdentifier = SQLSimpleIdentifier::new(id.as_str()).into();
                Ok(Dot::new(inner_sql, sql_ident).into())
            }
            ParsedSimpleIdentifier::Error(err) => Err(err.as_ref().clone().single()),
        },
        FieldAccess::TupleElement(index) => {
            // Trino anonymous ROWs use subscript notation: row[1], row[2], etc. (1-indexed)
            Ok(IndexLookup::new(
                inner_sql,
                SQLIntegerLiteral::from_int((index + 1) as i64).into(),
            )
            .to_sql_expression())
        }
        FieldAccess::VariantField(parsed_id) => match parsed_id {
            ParsedSimpleIdentifier::Valid(id) => {
                // Variant field access uses json_extract
                let path = format!("$[\"{}\"]", id.as_str());
                Ok(FunctionCallApply::with_two(
                    "json_extract",
                    inner_sql,
                    StringLiteral::new(&path).into(),
                )
                .into())
            }
            ParsedSimpleIdentifier::Error(err) => Err(err.as_ref().clone().single()),
        },
        FieldAccess::RangeBegin => {
            let sql_ident: SQLIdentifier = SQLSimpleIdentifier::new("begin").into();
            Ok(Dot::new(inner_sql, sql_ident).into())
        }
        FieldAccess::RangeEnd => {
            let sql_ident: SQLIdentifier = SQLSimpleIdentifier::new("end").into();
            Ok(Dot::new(inner_sql, sql_ident).into())
        }
        FieldAccess::BroadcastStructField(parsed_id) => match parsed_id {
            ParsedSimpleIdentifier::Valid(id) => {
                let param_ref: SQLExpression =
                    SQLColumnReference::new(SQLSimpleIdentifier::new("__x").into()).into();
                let sql_ident: SQLIdentifier = SQLSimpleIdentifier::new(id.as_str()).into();
                let body: SQLExpression = Dot::new(param_ref, sql_ident).into();
                Ok(transform_lambda(inner_sql, body))
            }
            ParsedSimpleIdentifier::Error(err) => Err(err.as_ref().clone().single()),
        },
        FieldAccess::BroadcastTupleElement(index) => {
            let param_ref: SQLExpression =
                SQLColumnReference::new(SQLSimpleIdentifier::new("__x").into()).into();
            // Trino anonymous ROWs use 1-indexed subscript access: row[1], row[2], ...
            let body: SQLExpression = IndexLookup::new(
                param_ref,
                SQLIntegerLiteral::from_int((*index + 1) as i64).into(),
            )
            .to_sql_expression();
            Ok(transform_lambda(inner_sql, body))
        }
        FieldAccess::BroadcastVariantField(parsed_id) => match parsed_id {
            ParsedSimpleIdentifier::Valid(id) => {
                let param_ref: SQLExpression =
                    SQLColumnReference::new(SQLSimpleIdentifier::new("__x").into()).into();
                let path = format!("$[\"{}\"]", id.as_str());
                let body: SQLExpression = FunctionCallApply::with_two(
                    "json_extract",
                    param_ref,
                    StringLiteral::new(&path).into(),
                )
                .into();
                Ok(transform_lambda(inner_sql, body))
            }
            ParsedSimpleIdentifier::Error(err) => Err(err.as_ref().clone().single()),
        },
    }
}

/// Build `transform(<arr>, __x -> <body>)` where `__x` is the lambda parameter
/// name referenced in the body.
fn transform_lambda(arr: SQLExpression, body: SQLExpression) -> SQLExpression {
    let param = SQLSimpleIdentifier::new("__x");
    let lambda: SQLExpression = SQLLambda::new(vec![param], body).into();
    FunctionCallApply::with_two("transform", arr, lambda).into()
}

/// Translate a variant index access expression.
fn translate_variant_index_access(
    ctx: &TranslationContext,
    variant_access: &TypedVariantIndexAccess,
    _span: &impl Spannable,
) -> ExprTranslationResult {
    let inner_expr = IRExpression::new(variant_access.value.clone());
    let inner_sql = translate_expression(ctx, &inner_expr)?;

    // Variant index access uses json_extract with array index
    let path = format!("$[{}]", variant_access.variant_index);
    Ok(
        FunctionCallApply::with_two("json_extract", inner_sql, StringLiteral::new(&path).into())
            .into(),
    )
}

/// Translate an array literal expression.
fn translate_array_literal(
    ctx: &TranslationContext,
    arr: &TypedArrayLiteral,
) -> ExprTranslationResult {
    let mut elements = Vec::with_capacity(arr.elements.len());
    for elem in &arr.elements {
        let ir_elem = IRExpression::new(elem.clone());
        elements.push(translate_expression(ctx, &ir_elem)?);
    }
    Ok(SQLArrayLiteral::new(elements).into())
}

/// Translate a tuple literal expression.
fn translate_tuple_literal(
    ctx: &TranslationContext,
    tuple: &TypedTupleLiteral,
) -> ExprTranslationResult {
    let mut values = Vec::with_capacity(tuple.elements.len());
    for elem in &tuple.elements {
        let ir_elem = IRExpression::new(elem.clone());
        values.push(translate_expression(ctx, &ir_elem)?);
    }
    Ok(SQLRowLiteral::new(values).into())
}

/// Translate a struct literal expression.
///
/// Struct literals in Trino require a CAST to provide field names:
/// `CAST(ROW(v1, v2) AS ROW(f1 type1, f2 type2))`
fn translate_struct_literal(
    ctx: &TranslationContext,
    strct: &TypedStructLiteral,
    resolved_type: &Type,
    span: &impl Spannable,
) -> ExprTranslationResult {
    let mut values = Vec::with_capacity(strct.fields.len());
    for (_, field_expr) in &strct.fields {
        let ir_elem = IRExpression::new(field_expr.clone());
        values.push(translate_expression(ctx, &ir_elem)?);
    }

    let row_literal: SQLExpression = SQLRowLiteral::new(values).into();
    let sql_type = hamelin_type_to_sql_type(resolved_type, span)?;

    Ok(SQLCast::new(row_literal, sql_type).into())
}

/// Convert a Hamelin Type to a SQL Type.
fn hamelin_type_to_sql_type(
    typ: &Type,
    span: &impl Spannable,
) -> Result<SQLType, TranslationErrors> {
    match typ {
        Type::Int => Ok(SQLBaseType::BigInt.into()),
        Type::Double => Ok(SQLBaseType::Double.into()),
        Type::String => Ok(SQLBaseType::VarChar.into()),
        Type::Boolean => Ok(SQLBaseType::Boolean.into()),
        Type::Timestamp => Ok(SQLTimestampTzType::new(6).into()),
        Type::Binary => Ok(SQLBaseType::VarBinary.into()),
        Type::Interval => Ok(SQLBaseType::IntervalDayToSecond.into()),
        Type::CalendarInterval => Ok(SQLBaseType::IntervalYearToMonth.into()),
        Type::Variant => Ok(SQLBaseType::Json.into()),
        Type::Decimal(decimal) => {
            Ok(SQLDecimalType::new(decimal.precision as u64, decimal.scale as u64).into())
        }
        Type::Array(arr) => {
            let elem_type = hamelin_type_to_sql_type(&arr.element_type, span)?;
            Ok(SQLArrayType::new(elem_type).into())
        }
        Type::Struct(s) => {
            let mut bindings = ordermap::OrderMap::new();
            for (name, field_type) in s.iter() {
                let sql_name = SQLSimpleIdentifier::new(name.name());
                let sql_type = hamelin_type_to_sql_type(field_type, span)?;
                bindings.insert(sql_name, sql_type);
            }
            Ok(SQLRowType::new(bindings).into())
        }
        Type::Map(map) => {
            let key_type = hamelin_type_to_sql_type(&map.key_type, span)?;
            let value_type = hamelin_type_to_sql_type(&map.value_type, span)?;
            Ok(SQLMapType::new(key_type, value_type).into())
        }
        Type::Range(range) | Type::RangeInclusive(range) => {
            // Range is represented as ROW(begin, end) in SQL
            let elem_type = hamelin_type_to_sql_type(&range.of, span)?;
            let bindings = ordermap::OrderMap::from_iter([
                (SQLSimpleIdentifier::new("begin"), elem_type.clone()),
                (SQLSimpleIdentifier::new("end"), elem_type),
            ]);
            Ok(SQLRowType::new(bindings).into())
        }
        Type::Tuple(tuple) => {
            let mut elements = Vec::new();
            for elem_type in tuple.elements.iter() {
                elements.push(hamelin_type_to_sql_type(elem_type, span)?);
            }
            Ok(SQLAnonRowType::new(elements).into())
        }
        Type::Unknown | Type::Rows | Type::Function(_) => Err(TranslationError::msg(
            span,
            &format!("Cannot convert type {} to SQL", typ),
        )
        .single()),
    }
}

/// Translate a leaf expression (literal).
///
/// Leaf expressions are literals in the raw AST. We access them via `expr.ast.kind`.
fn translate_leaf(_ctx: &TranslationContext, expr: &IRExpression) -> ExprTranslationResult {
    let typed_expr = expr.inner();

    match &typed_expr.ast.kind {
        ExpressionKind::IntLiteral(lit) => Ok(SQLIntegerLiteral::from_int(lit.int).into()),

        ExpressionKind::DecimalLiteral(lit) => {
            // Format the decimal with proper scale
            let formatted = format_decimal(lit.unscaled_value, lit.scale);
            Ok(SQLDecimalLiteral::new(&formatted).into())
        }

        ExpressionKind::ScientificLiteral(lit) => {
            Ok(SQLScientificLiteral::new(&format!("{:e}", lit.value)).into())
        }

        ExpressionKind::DoubleLiteral(lit) => {
            Ok(SQLScientificLiteral::new(&format!("{:e}", lit.value)).into())
        }

        ExpressionKind::BooleanLiteral(lit) => Ok(SQLBooleanLiteral::new(lit.value).into()),

        ExpressionKind::StringLiteral(lit) => Ok(SQLStringLiteral::new(&lit.value).into()),

        ExpressionKind::BinaryLiteral(lit) => {
            // Format as hex string
            let hex = lit
                .value
                .iter()
                .map(|b| format!("{:02x}", b))
                .collect::<String>();
            Ok(SQLBinaryLiteral::new(&format!("'{}'", hex)).into())
        }

        ExpressionKind::NullLiteral(_) => Ok(SQLNullLiteral::default().into()),

        ExpressionKind::IntervalLiteral(lit) => {
            Ok(translate_interval_literal(lit.value, &lit.unit))
        }

        // These are handled by other TypedExpressionKind variants, not Leaf
        ExpressionKind::ArrayLiteral(_)
        | ExpressionKind::TupleLiteral(_)
        | ExpressionKind::PairLiteral(_)
        | ExpressionKind::StructLiteral(_)
        | ExpressionKind::FieldReference(_)
        | ExpressionKind::UnaryPrefixOperator(_)
        | ExpressionKind::UnaryPostfixOperator(_)
        | ExpressionKind::BinaryOperator(_)
        | ExpressionKind::FunctionCall(_)
        | ExpressionKind::IndexAccess(_)
        | ExpressionKind::FieldLookup(_)
        | ExpressionKind::Cast(_)
        | ExpressionKind::TsTrunc(_)
        | ExpressionKind::Lambda(_) => Err(TranslationError::msg(
            expr,
            &format!(
                "Expected leaf literal, got {:?}",
                std::mem::discriminant(&typed_expr.ast.kind)
            ),
        )
        .single()),

        // RowsLiteral should have been lowered away
        ExpressionKind::RowsLiteral(_) => Err(TranslationError::msg(
            expr,
            "RowsLiteral should have been lowered before translation",
        )
        .single()),

        // UnboundRangeLiteral translates to CAST(ROW(NULL, NULL) AS ROW("begin" T, "end" T))
        ExpressionKind::UnboundRangeLiteral(_) => {
            // Get the range type from the resolved type
            let range_type = match typed_expr.resolved_type.as_ref() {
                Type::Range(range) | Type::RangeInclusive(range) => range,
                _ => {
                    return Err(TranslationError::msg(
                        expr,
                        "UnboundRangeLiteral should have Range type",
                    )
                    .single());
                }
            };

            // Get the SQL type for the element (handle Unknown specially)
            let elem_sql_type: SQLType = if *range_type.of == Type::Unknown {
                SQLBaseType::Unknown.into()
            } else {
                hamelin_type_to_sql_type(&range_type.of, expr)?
            };

            // Build ROW(NULL, NULL)
            let row_literal: SQLExpression = SQLRowLiteral::new(vec![
                SQLNullLiteral::default().into(),
                SQLNullLiteral::default().into(),
            ])
            .into();

            // Build ROW("begin" TYPE, "end" TYPE)
            let mut row_type_bindings = ordermap::OrderMap::new();
            row_type_bindings.insert(SQLSimpleIdentifier::new("begin"), elem_sql_type.clone());
            row_type_bindings.insert(SQLSimpleIdentifier::new("end"), elem_sql_type);
            let row_type: SQLType = SQLRowType::new(row_type_bindings).into();

            Ok(SQLCast::new(row_literal, row_type).into())
        }

        ExpressionKind::Error(err) => Err(err.error.as_ref().clone().single()),
    }
}

/// Format a decimal literal with the correct scale.
fn format_decimal(unscaled_value: i128, scale: u32) -> String {
    if scale == 0 {
        return format!("{}.0", unscaled_value);
    }

    let is_negative = unscaled_value < 0;
    let abs_value = unscaled_value.unsigned_abs();
    let str_value = abs_value.to_string();

    let scale_usize = scale as usize;
    if str_value.len() <= scale_usize {
        // Need leading zeros: 0.00123
        let leading_zeros = scale_usize - str_value.len();
        let sign = if is_negative { "-" } else { "" };
        format!("{}0.{}{}", sign, "0".repeat(leading_zeros), str_value)
    } else {
        // Insert decimal point
        let decimal_pos = str_value.len() - scale_usize;
        let sign = if is_negative { "-" } else { "" };
        format!(
            "{}{}.{}",
            sign,
            &str_value[..decimal_pos],
            &str_value[decimal_pos..]
        )
    }
}

/// Translate a Hamelin interval literal to the appropriate SQL expression.
///
/// Matches legacy behavior:
/// - Sub-second units (ms) use `parse_duration('Nunit')`
/// - Seconds through days use `parse_duration('Nunit')`
/// - Weeks use `7 * INTERVAL 'N' DAY`
/// - Months use `INTERVAL 'N' MONTH`
/// - Quarters use `3 * INTERVAL 'N' MONTH`
/// - Years use `INTERVAL 'N' YEAR`
fn translate_interval_literal(value: i64, unit: &IntervalUnit) -> SQLExpression {
    match unit {
        IntervalUnit::Millisecond => parse_duration(value, "ms"),
        IntervalUnit::Second => parse_duration(value, "s"),
        IntervalUnit::Minute => parse_duration(value, "m"),
        IntervalUnit::Hour => parse_duration(value, "h"),
        IntervalUnit::Day => parse_duration(value, "d"),
        IntervalUnit::Week => BinaryOperatorApply::new(
            Operator::Asterisk,
            SQLIntegerLiteral::new("7").into(),
            SQLIntervalLiteral::new(value, SQLIntervalUnit::Day).into(),
        )
        .into(),
        IntervalUnit::Month => SQLIntervalLiteral::new(value, SQLIntervalUnit::Month).into(),
        IntervalUnit::Quarter => BinaryOperatorApply::new(
            Operator::Asterisk,
            SQLIntegerLiteral::new("3").into(),
            SQLIntervalLiteral::new(value, SQLIntervalUnit::Month).into(),
        )
        .into(),
        IntervalUnit::Year => SQLIntervalLiteral::new(value, SQLIntervalUnit::Year).into(),
    }
}

fn parse_duration(value: i64, unit: &str) -> SQLExpression {
    FunctionCallApply::with_one(
        "parse_duration",
        SQLStringLiteral::new(&format!("{}{}", value, unit)).into(),
    )
    .into()
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_format_decimal() {
        assert_eq!(format_decimal(12345, 2), "123.45");
        assert_eq!(format_decimal(12345, 0), "12345.0");
        assert_eq!(format_decimal(123, 5), "0.00123");
        assert_eq!(format_decimal(-12345, 2), "-123.45");
        assert_eq!(format_decimal(0, 2), "0.00");
    }
}
