hamelin_eval 0.10.13

use crate::eval::environment::Environment;
use crate::eval::error::{EvalError, EvalResult};
use crate::eval::timestamp::truncate_timestamp;
use crate::value::{Closure, DecimalValue, RangeValue, TimestampValue, Value};
use hamelin_lib::tree::ast::expression::{
    BinaryLiteral, BooleanLiteral, DecimalLiteral, DoubleLiteral, ExpressionKind, IntLiteral,
    IntervalLiteral, IntervalUnit, NullLiteral, RowsLiteral, ScientificLiteral, StringLiteral,
    UnboundRangeLiteral,
};
use hamelin_lib::tree::typed_ast::expression::{
    CastKind, FieldAccess, TypedApply, TypedArrayLiteral, TypedBroadcastApply, TypedCast,
    TypedExpression, TypedExpressionKind, TypedFieldLookup, TypedStructLiteral, TypedTsTrunc,
    TypedTupleLiteral, TypedVariantIndexAccess, VariantCastKind,
};
use hamelin_lib::types::Type;
use hamelin_lib::{f64_to_i64, parse_string_as_bool, parse_timestamp_to_utc};
use std::rc::Rc;

use chrono::{Duration, Utc};
use chronoutil::RelativeDuration;
use linear_map::LinearMap;
use ordermap::OrderMap;
use serde_json::Value as JsonValue;

use hamelin_lib::tree::ast::identifier::{ParsedSimpleIdentifier, SimpleIdentifier};

/// Evaluate a typed expression in the given environment
pub fn eval(expr: &TypedExpression, env: &Environment) -> EvalResult<Value> {
    // Type information is available from expr.resolved_type if needed
    eval_kind(&expr.kind, expr, env)
}

/// Invoke a closure with the given arguments.
///
/// This creates a new environment extending the closure's captured environment
/// with the lambda parameter bindings, then evaluates the lambda body.
pub fn invoke_closure(closure: &Closure, args: Vec<Value>) -> EvalResult<Value> {
    // Validate arity
    if args.len() != closure.lambda.parameters.len() {
        return Err(EvalError::execution(format!(
            "Lambda expects {} arguments, got {}",
            closure.lambda.parameters.len(),
            args.len()
        )));
    }

    // Create new environment extending the captured environment
    let mut invoke_env = closure.captured_env.clone();
    for (param, value) in closure.lambda.parameters.iter().zip(args) {
        invoke_env.bind(param.name.clone(), value);
    }

    // Evaluate the lambda body in the extended environment
    eval(&closure.lambda.body, &invoke_env)
}

/// Evaluate a typed expression kind
fn eval_kind(
    kind: &TypedExpressionKind,
    expr: &TypedExpression,
    env: &Environment,
) -> EvalResult<Value> {
    match kind {
        TypedExpressionKind::Leaf => eval_leaf(expr, env),
        TypedExpressionKind::FieldReference(_col_ref) => {
            // Column references are evaluated by looking up in the environment
            // This is the same as Leaf - the AST holds the column name
            eval_leaf(expr, env)
        }

        TypedExpressionKind::ArrayLiteral(array) => eval_array_literal(array, env),
        TypedExpressionKind::TupleLiteral(tuple) => eval_tuple_literal(tuple, env),
        TypedExpressionKind::StructLiteral(struct_lit) => eval_struct_literal(struct_lit, env),
        TypedExpressionKind::Apply(apply) => eval_apply(apply, env),
        TypedExpressionKind::BroadcastApply(broadcast) => eval_broadcast_apply(broadcast, env),
        TypedExpressionKind::VariantIndexAccess(via) => eval_variant_index_access(via, env),
        TypedExpressionKind::FieldLookup(lookup) => eval_field_lookup(lookup, env),
        TypedExpressionKind::Cast(cast) => eval_cast(cast, env),
        TypedExpressionKind::TsTrunc(ts_trunc) => eval_ts_trunc(ts_trunc, env),
        TypedExpressionKind::Error(_) => {
            Err(EvalError::execution("Cannot evaluate error expression"))
        }
        TypedExpressionKind::Lambda(lambda) => {
            // Typed lambdas capture the current environment for closure semantics
            Ok(Value::Closure(Closure {
                lambda: Rc::new(lambda.clone()),
                captured_env: env.clone(),
            }))
        }
    }
}

/// Evaluate a leaf expression (literals and column references)
fn eval_leaf(expr: &TypedExpression, env: &Environment) -> EvalResult<Value> {
    match &expr.ast.kind {
        ExpressionKind::IntLiteral(IntLiteral { int, .. }) => Ok(Value::Int(*int)),
        ExpressionKind::DecimalLiteral(DecimalLiteral {
            unscaled_value,
            scale,
            ..
        }) => Ok(Value::Decimal(DecimalValue {
            unscaled: *unscaled_value,
            scale: *scale as i32,
        })),
        ExpressionKind::ScientificLiteral(ScientificLiteral { value, .. }) => {
            Ok(Value::Double(*value))
        }
        ExpressionKind::DoubleLiteral(DoubleLiteral { value, .. }) => Ok(Value::Double(*value)),
        ExpressionKind::BooleanLiteral(BooleanLiteral { value, .. }) => Ok(Value::Boolean(*value)),
        ExpressionKind::StringLiteral(StringLiteral { value, .. }) => {
            Ok(Value::String(value.clone()))
        }
        ExpressionKind::BinaryLiteral(BinaryLiteral { value, .. }) => {
            Ok(Value::Binary(value.clone()))
        }
        ExpressionKind::NullLiteral(NullLiteral { .. }) => Ok(Value::Null),
        ExpressionKind::RowsLiteral(RowsLiteral { value, .. }) => Ok(Value::Rows(*value)),
        ExpressionKind::UnboundRangeLiteral(UnboundRangeLiteral { .. }) => {
            Ok(Value::Range(Box::new(RangeValue {
                lower: None,
                upper: None,
            })))
        }
        ExpressionKind::IntervalLiteral(IntervalLiteral { value, unit, .. }) => {
            // Convert interval literal to appropriate value type
            match unit {
                IntervalUnit::Month | IntervalUnit::Quarter | IntervalUnit::Year => {
                    // Calendar interval - store as number of months
                    let months = match unit {
                        IntervalUnit::Month => *value as i32,
                        IntervalUnit::Quarter => *value as i32 * 3,
                        IntervalUnit::Year => *value as i32 * 12,
                        _ => unreachable!(),
                    };

                    Ok(Value::CalendarInterval(months))
                }
                _ => {
                    // Regular interval - convert to microseconds
                    let microseconds = match unit {
                        IntervalUnit::Millisecond => *value * 1_000,
                        IntervalUnit::Second => *value * 1_000_000,
                        IntervalUnit::Minute => *value * 60 * 1_000_000,
                        IntervalUnit::Hour => *value * 60 * 60 * 1_000_000,
                        IntervalUnit::Day => *value * 24 * 60 * 60 * 1_000_000,
                        IntervalUnit::Week => *value * 7 * 24 * 60 * 60 * 1_000_000,
                        _ => unreachable!(),
                    };
                    Ok(Value::Interval(Duration::microseconds(microseconds)))
                }
            }
        }
        ExpressionKind::FieldReference(col_ref) => {
            // Look up the field in the environment
            let ast_simple_id = col_ref
                .field_name
                .valid_ref()
                .map_err(|_e| EvalError::execution("Invalid identifier"))?;
            // Convert AST identifier to SQL identifier for lookup
            let sql_simple_id = SimpleIdentifier::new(ast_simple_id.as_str());
            env.lookup(&sql_simple_id)
                .cloned()
                .ok_or_else(|| EvalError::VariableNotFound {
                    name: ast_simple_id.to_string(),
                })
        }
        _ => {
            // Other expression kinds shouldn't be Leaf
            Err(EvalError::execution(format!(
                "Unexpected leaf expression kind: {:?}",
                expr.ast.kind
            )))
        }
    }
}

/// Evaluate an array literal expression
fn eval_array_literal(array: &TypedArrayLiteral, env: &Environment) -> EvalResult<Value> {
    let mut values = Vec::with_capacity(array.elements.len());
    for element in &array.elements {
        values.push(eval(element, env)?);
    }
    Ok(Value::Array(values))
}

/// Evaluate a tuple literal expression
fn eval_tuple_literal(tuple: &TypedTupleLiteral, env: &Environment) -> EvalResult<Value> {
    let mut values = Vec::with_capacity(tuple.elements.len());
    for element in &tuple.elements {
        values.push(eval(element, env)?);
    }
    Ok(Value::Tuple(values))
}

/// Evaluate a struct literal expression
fn eval_struct_literal(struct_lit: &TypedStructLiteral, env: &Environment) -> EvalResult<Value> {
    let mut fields = OrderMap::new();
    for (name, expr) in &struct_lit.fields {
        let value = eval(expr, env)?;
        // Extract valid identifier, return error if invalid
        let simple_id = name
            .clone()
            .valid()
            .map_err(|_| EvalError::execution("Invalid field identifier in struct literal"))?;
        fields.insert(simple_id.into(), value);
    }
    Ok(Value::Struct(fields))
}

fn eval_ts_trunc(ts_trunc: &TypedTsTrunc, env: &Environment) -> EvalResult<Value> {
    let input_value = eval(&ts_trunc.expression, env)?;
    let timestamp = input_value.require_timestamp()?;
    let truncated = truncate_timestamp(&timestamp, &ts_trunc.unit, ts_trunc.multiplier)?;
    Ok(truncated.into())
}

/// Evaluate field lookup with access hints
fn eval_field_lookup(lookup: &TypedFieldLookup, env: &Environment) -> EvalResult<Value> {
    let value = eval(&lookup.value, env)?;

    match &lookup.access {
        FieldAccess::StructField(field_name) => eval_struct_field(&value, field_name),
        FieldAccess::TupleElement(index) => eval_tuple_element(&value, *index),
        FieldAccess::VariantField(field_name) => eval_variant_field(&value, field_name),
        FieldAccess::RangeBegin => match value {
            Value::Range(range) => Ok(range.lower.unwrap_or(Value::Null)),
            _ => Err(EvalError::execution(format!(
                "Expected range, got {}",
                value.type_name()
            ))),
        },
        FieldAccess::RangeEnd => match value {
            Value::Range(range) => Ok(range.upper.unwrap_or(Value::Null)),
            _ => Err(EvalError::execution(format!(
                "Expected range, got {}",
                value.type_name()
            ))),
        },
        FieldAccess::BroadcastStructField(field_name) => {
            eval_broadcast_field(&value, |element| eval_struct_field(element, field_name))
        }
        FieldAccess::BroadcastVariantField(field_name) => {
            eval_broadcast_field(&value, |element| eval_variant_field(element, field_name))
        }
        FieldAccess::BroadcastTupleElement(index) => {
            eval_broadcast_field(&value, |element| eval_tuple_element(element, *index))
        }
    }
}

/// Apply a per-element field accessor across an array value.
///
/// Null outer arrays propagate to null; null elements produce null output
/// elements (matching Arrow's semantics on `ListArray<StructArray>` where
/// the struct-level null bitmap makes its child columns return null at
/// those positions). Broadcasting never flattens (per SIM-4169), so when
/// the accessed value is itself an array the entry is preserved as-is
/// in the result — no concatenation.
fn eval_broadcast_field<F>(value: &Value, mut access: F) -> EvalResult<Value>
where
    F: FnMut(&Value) -> EvalResult<Value>,
{
    if value.is_null() {
        return Ok(Value::Null);
    }
    let elements = value.try_array()?;
    let mut results = Vec::with_capacity(elements.len());
    for element in elements {
        if element.is_null() {
            results.push(Value::Null);
        } else {
            results.push(access(element)?);
        }
    }
    Ok(Value::Array(results))
}

/// Extract a named field from a struct value.
fn eval_struct_field(value: &Value, field_name: &ParsedSimpleIdentifier) -> EvalResult<Value> {
    let fields = value.try_struct()?;
    let ast_simple_id = field_name
        .valid_ref()
        .map_err(|_| EvalError::execution("Invalid field identifier"))?;
    let sql_simple_id = SimpleIdentifier::new(ast_simple_id.as_str());
    fields
        .iter()
        .find(|(k, _)| **k == sql_simple_id)
        .map(|(_, v)| v.clone())
        .ok_or_else(|| EvalError::execution(format!("Field '{}' not found in struct", field_name)))
}

/// Extract a named field from a variant (JSON object) value.
fn eval_variant_field(value: &Value, field_name: &ParsedSimpleIdentifier) -> EvalResult<Value> {
    let variant = value.try_variant()?;
    match variant {
        serde_json::Value::Object(obj) => {
            let key: &str = field_name
                .valid_ref()
                .map_err(|e| EvalError::execution(format!("Invalid field name: {}", e)))?
                .as_str();
            let result_json = obj.get(key).cloned().unwrap_or(serde_json::Value::Null);
            Ok(Value::Variant(result_json))
        }
        _ => Err(EvalError::execution(format!(
            "Cannot access field '{}' on non-object variant",
            field_name
        ))),
    }
}

/// Extract an element from a tuple value by index.
fn eval_tuple_element(value: &Value, index: usize) -> EvalResult<Value> {
    let elements = value.try_tuple()?;
    if index >= elements.len() {
        return Err(EvalError::IndexOutOfBounds {
            index: index as i64,
            length: elements.len(),
        });
    }
    Ok(elements[index].clone())
}

/// Evaluate a function application (includes operators, function calls, and array/map indexing)
fn eval_apply(apply: &TypedApply, env: &Environment) -> EvalResult<Value> {
    // Transform the parameter binding from TypedExpression to Value by evaluating each expression
    let value_binding = apply
        .parameter_binding
        .clone()
        .try_map(|expr| eval(&expr, env))?;

    // Look up the function's eval implementation in the registry
    let type_id = apply.function_def.type_id();
    env.registry()
        .eval(type_id, value_binding)
        .ok_or_else(|| EvalError::NoEvalImplementation {
            function_name: apply.function_def.name().to_string(),
        })?
        .map_err(EvalError::from)
}

/// Evaluate a broadcast application - apply a function element-wise over an array
fn eval_broadcast_apply(broadcast: &TypedBroadcastApply, env: &Environment) -> EvalResult<Value> {
    // Evaluate all arguments first
    let value_binding = broadcast
        .parameter_binding
        .clone()
        .try_map(|expr| eval(&expr, env))?;

    // Get the array at the broadcast position
    let array_value = value_binding
        .get_by_index(broadcast.broadcast_position)
        .map_err(|e| EvalError::execution(format!("Broadcast position error: {}", e)))?;

    // Null array propagates to null result
    if matches!(array_value, Value::Null) {
        return Ok(Value::Null);
    }

    let elements = array_value.try_array()?;

    // Apply the function to each element
    let type_id = broadcast.function_def.type_id();
    let mut results = Vec::with_capacity(elements.len());

    for element in elements {
        // Create a new binding with this element substituted at the broadcast position
        let element_binding = value_binding
            .clone()
            .replace_by_index(broadcast.broadcast_position, element.clone())
            .map_err(|e| EvalError::execution(format!("Failed to substitute element: {}", e)))?;

        let result = env
            .registry()
            .eval(type_id, element_binding)
            .ok_or_else(|| EvalError::NoEvalImplementation {
                function_name: broadcast.function_def.name().to_string(),
            })?
            .map_err(EvalError::from)?;

        results.push(result);
    }

    Ok(Value::Array(results))
}

/// Evaluate a variant index access with compile-time constant index
fn eval_variant_index_access(
    via: &TypedVariantIndexAccess,
    env: &Environment,
) -> EvalResult<Value> {
    let container = eval(&via.value, env)?;

    match container {
        Value::Variant(variant) => {
            // For variant index access, treat as array indexing or object field access
            match variant {
                JsonValue::Array(arr) => {
                    // Direct array indexing
                    let result_json = if via.variant_index < arr.len() {
                        arr[via.variant_index].clone()
                    } else {
                        JsonValue::Null // Out of bounds returns JSON null
                    };
                    Ok(Value::Variant(result_json))
                }
                JsonValue::Object(obj) => {
                    // Look for field with numeric name (e.g., "_0", "_1")
                    let key = format!("_{}", via.variant_index);
                    let result_json = obj.get(&key).cloned().unwrap_or(JsonValue::Null);
                    Ok(Value::Variant(result_json))
                }
                _ => Err(EvalError::execution(format!(
                    "Cannot index into non-array/non-object variant with index {}",
                    via.variant_index
                ))),
            }
        }
        Value::Null => Ok(Value::Null),
        _ => Err(EvalError::execution(format!(
            "Expected variant value, got {:?}",
            container
        ))),
    }
}

/// Evaluate a cast expression
fn eval_cast(cast: &TypedCast, env: &Environment) -> EvalResult<Value> {
    let value = eval(&cast.value, env)?;

    // Use the precomputed cast kind from the type checker
    perform_cast(value, cast.cast_kind.clone(), &cast.target_type)
}

/// Perform a cast operation based on the cast kind
fn perform_cast(value: Value, cast_kind: CastKind, target_type: &Type) -> EvalResult<Value> {
    // Handle null - always becomes null regardless of cast kind
    if value.is_null() {
        return Ok(Value::Null);
    }

    match cast_kind {
        CastKind::Identity => Ok(value),
        CastKind::NullToType => Ok(Value::Null),

        // Numeric casts
        CastKind::IntToDouble => {
            if let Value::Int(i) = value {
                Ok(Value::Double(i as f64))
            } else {
                Err(EvalError::execution(format!(
                    "IntToDouble cast expected Int value, got {:?}",
                    value
                )))
            }
        }
        CastKind::IntToDecimal => {
            if let Value::Int(i) = value {
                let dec = DecimalValue {
                    unscaled: i as i128,
                    scale: 0,
                };
                cast_decimal_to_decimal(&dec, decimal_target_scale(target_type, dec.scale))
            } else {
                Err(EvalError::execution(format!(
                    "IntToDecimal cast expected Int value, got {:?}",
                    value
                )))
            }
        }
        CastKind::DoubleToInt => {
            if let Value::Double(d) = value {
                cast_double_to_int(d)
            } else {
                Err(EvalError::execution(format!(
                    "DoubleToInt cast expected Double value, got {:?}",
                    value
                )))
            }
        }
        CastKind::DoubleToDecimal => {
            if let Value::Double(d) = value {
                let intermediate = match cast_double_to_decimal(d)? {
                    Value::Decimal(dec) => dec,
                    other => return Ok(other),
                };
                cast_decimal_to_decimal(
                    &intermediate,
                    decimal_target_scale(target_type, intermediate.scale),
                )
            } else {
                Err(EvalError::execution(format!(
                    "DoubleToDecimal cast expected Double value, got {:?}",
                    value
                )))
            }
        }
        CastKind::DecimalToInt => {
            if let Value::Decimal(dec) = &value {
                cast_decimal_to_int(dec)
            } else {
                Err(EvalError::execution(format!(
                    "DecimalToInt cast expected Decimal value, got {:?}",
                    value
                )))
            }
        }
        CastKind::DecimalToDouble => {
            if let Value::Decimal(dec) = &value {
                cast_decimal_to_double(dec)
            } else {
                Err(EvalError::execution(format!(
                    "DecimalToDouble cast expected Decimal value, got {:?}",
                    value
                )))
            }
        }
        CastKind::DecimalToDecimal => {
            if let Value::Decimal(dec) = &value {
                let target_scale = match target_type {
                    Type::Decimal(d) => d.scale,
                    _ => dec.scale,
                };
                cast_decimal_to_decimal(dec, target_scale)
            } else {
                Err(EvalError::execution(format!(
                    "DecimalToDecimal cast expected Decimal value, got {:?}",
                    value
                )))
            }
        }

        // Boolean casts
        CastKind::IntToBoolean => {
            if let Value::Int(i) = value {
                Ok(Value::Boolean(i != 0))
            } else {
                Err(EvalError::execution(format!(
                    "IntToBoolean cast expected Int value, got {:?}",
                    value
                )))
            }
        }
        CastKind::BooleanToInt => {
            if let Value::Boolean(b) = value {
                Ok(Value::Int(if b { 1 } else { 0 }))
            } else {
                Err(EvalError::execution(format!(
                    "BooleanToInt cast expected Boolean value, got {:?}",
                    value
                )))
            }
        }
        CastKind::BooleanToDouble => {
            if let Value::Boolean(b) = value {
                Ok(Value::Double(if b { 1.0 } else { 0.0 }))
            } else {
                Err(EvalError::execution(format!(
                    "BooleanToDouble cast expected Boolean value, got {:?}",
                    value
                )))
            }
        }
        CastKind::BooleanToDecimal => {
            if let Value::Boolean(b) = value {
                let dec = DecimalValue {
                    unscaled: if b { 1 } else { 0 },
                    scale: 0,
                };
                cast_decimal_to_decimal(&dec, decimal_target_scale(target_type, dec.scale))
            } else {
                Err(EvalError::execution(format!(
                    "BooleanToDecimal cast expected Boolean value, got {:?}",
                    value
                )))
            }
        }
        CastKind::DoubleToBoolean => {
            if let Value::Double(d) = value {
                // IntToBoolean-style semantics extended to doubles: zero → false,
                // every other finite or non-finite value → true. Matches Arrow's
                // `cast_numeric_to_bool` and Trino's `CAST(DOUBLE AS BOOLEAN)`.
                Ok(Value::Boolean(d != 0.0))
            } else {
                Err(EvalError::execution(format!(
                    "DoubleToBoolean cast expected Double value, got {:?}",
                    value
                )))
            }
        }
        CastKind::DecimalToBoolean => {
            if let Value::Decimal(d) = value {
                Ok(Value::Boolean(d.unscaled != 0))
            } else {
                Err(EvalError::execution(format!(
                    "DecimalToBoolean cast expected Decimal value, got {:?}",
                    value
                )))
            }
        }

        // String casts - to string
        CastKind::ToStringFromInt
        | CastKind::ToStringFromDouble
        | CastKind::ToStringFromBoolean
        | CastKind::ToStringFromTimestamp
        | CastKind::ToStringFromBinary
        | CastKind::ToStringFromDecimal
        | CastKind::ToStringFromInterval
        | CastKind::ToStringFromCalendarInterval => cast_to_string(value),

        // String casts - from string
        CastKind::StringToInt => {
            if let Value::String(s) = &value {
                cast_string_to_int(s)
            } else {
                Err(EvalError::execution(format!(
                    "StringToInt cast expected String value, got {:?}",
                    value
                )))
            }
        }
        CastKind::StringToDouble => {
            if let Value::String(s) = &value {
                cast_string_to_double(s)
            } else {
                Err(EvalError::execution(format!(
                    "StringToDouble cast expected String value, got {:?}",
                    value
                )))
            }
        }
        CastKind::StringToBoolean => {
            if let Value::String(s) = &value {
                cast_string_to_boolean(s)
            } else {
                Err(EvalError::execution(format!(
                    "StringToBoolean cast expected String value, got {:?}",
                    value
                )))
            }
        }
        CastKind::StringToTimestamp => {
            if let Value::String(s) = &value {
                cast_string_to_timestamp(s)
            } else {
                Err(EvalError::execution(format!(
                    "StringToTimestamp cast expected String value, got {:?}",
                    value
                )))
            }
        }
        CastKind::StringToDecimal => {
            if let Value::String(s) = &value {
                let intermediate = match cast_string_to_decimal(s)? {
                    Value::Decimal(dec) => dec,
                    other => return Ok(other),
                };
                cast_decimal_to_decimal(
                    &intermediate,
                    decimal_target_scale(target_type, intermediate.scale),
                )
            } else {
                Err(EvalError::execution(format!(
                    "StringToDecimal cast expected String value, got {:?}",
                    value
                )))
            }
        }

        // Variant casts
        CastKind::ToVariant(_) => Ok(Value::Variant(to_json_value(value)?)),
        CastKind::FromVariant(ref kind) => {
            if let Value::Variant(v) = &value {
                cast_from_variant(v, kind)
            } else {
                Err(EvalError::execution(format!(
                    "FromVariant cast expected Variant value, got {:?}",
                    value
                )))
            }
        }

        // Collection casts
        CastKind::ArrayElementCast(element_cast_kind) => {
            if let Value::Array(arr) = &value {
                cast_array_with_kind(arr, element_cast_kind, target_type)
            } else {
                Err(EvalError::execution(format!(
                    "ArrayElementCast cast expected Array value, got {:?}",
                    value
                )))
            }
        }
        CastKind::TupleToStruct(field_casts) => {
            if let Value::Tuple(tuple) = &value {
                cast_tuple_to_struct_with_kinds(tuple, field_casts, target_type)
            } else {
                Err(EvalError::execution(format!(
                    "TupleToStruct cast expected Tuple value, got {:?}",
                    value
                )))
            }
        }

        // Range casts
        CastKind::RangeElementCast(element_cast_kind) => {
            if let Value::Range(range) = &value {
                cast_range_with_kind(range, element_cast_kind, target_type)
            } else {
                Err(EvalError::execution(format!(
                    "RangeElementCast cast expected Range value, got {:?}",
                    value
                )))
            }
        }
        CastKind::IntervalToTimestampRange => cast_interval_to_timestamp_range(value),
        CastKind::TimestampToTimestampRange => cast_timestamp_to_timestamp_range(value),
        CastKind::IntervalRangeToTimestampRange => cast_interval_range_to_timestamp_range(value),
        CastKind::StructExpansion(field_casts) => {
            if let Value::Struct(source_fields) = value {
                cast_struct_expansion(source_fields, field_casts, target_type)
            } else {
                Err(EvalError::execution(format!(
                    "StructExpansion cast expected Struct value, got {:?}",
                    value
                )))
            }
        }
    }
}

/// Cast a double to int. NaN, infinity, and out-of-range values produce null
/// (matching the documented `x AS T` contract: primitive casts never crash —
/// they null on failure).
fn cast_double_to_int(d: f64) -> EvalResult<Value> {
    Ok(f64_to_i64(d).map(Value::Int).unwrap_or(Value::Null))
}

/// Cast any value to string
fn cast_to_string(value: Value) -> EvalResult<Value> {
    let string_repr = match value {
        Value::Int(i) => i.to_string(),
        Value::Double(d) => d.to_string(),
        Value::Boolean(b) => {
            if b {
                "true".to_string()
            } else {
                "false".to_string()
            }
        }
        Value::Timestamp(t) => t.instant().to_rfc3339(),
        Value::Binary(bytes) => {
            let mut s = String::from("0x");
            for byte in bytes {
                s.push_str(&format!("{:02x}", byte));
            }
            s
        }
        Value::Decimal(d) => {
            let scale_factor = 10_i128.pow(d.scale as u32);
            let integer_part = d.unscaled / scale_factor;
            let fractional_part = (d.unscaled % scale_factor).abs();
            format!(
                "{}.{:0width$}",
                integer_part,
                fractional_part,
                width = d.scale as usize
            )
        }
        Value::String(s) => s,
        Value::Null => return Ok(Value::Null),
        Value::Interval(i) => format!("{}", i),
        Value::CalendarInterval(months) => {
            format!("{}", RelativeDuration::months(months).format_to_iso8601())
        }
        Value::Rows(r) => r.to_string(),
        Value::Array(_) => return Err(EvalError::execution("Cannot cast array to string")),
        Value::Tuple(_) => return Err(EvalError::execution("Cannot cast tuple to string")),
        Value::Struct(_) => return Err(EvalError::execution("Cannot cast struct to string")),
        Value::Map(_) => return Err(EvalError::execution("Cannot cast map to string")),
        Value::Range(_) => return Err(EvalError::execution("Cannot cast range to string")),
        Value::Variant(v) => v.to_string(),
        Value::Closure(_) => return Err(EvalError::execution("Cannot cast closure to string")),
        Value::Unknown => return Err(EvalError::execution("Cannot cast unknown to string")),
    };

    Ok(Value::String(string_repr))
}

/// Cast string to integer. Parse failure produces null, matching Trino's
/// TRY_CAST and the documented `x AS T` contract (casting.md).
fn cast_string_to_int(s: &str) -> EvalResult<Value> {
    Ok(s.parse::<i64>().map(Value::Int).unwrap_or(Value::Null))
}

/// Cast string to double. Parse failure produces null.
fn cast_string_to_double(s: &str) -> EvalResult<Value> {
    Ok(s.parse::<f64>().map(Value::Double).unwrap_or(Value::Null))
}

/// Cast string to boolean using the same acceptance set as Arrow's cast
/// kernel (and the variant-string → boolean path), so native and variant
/// casts agree: 'true'/'false'/'yes'/'no'/'on'/'off'/'1'/'0' and prefixes,
/// case-insensitive, with leading/trailing whitespace trimmed. Returns null
/// for inputs outside the accepted set.
fn cast_string_to_boolean(s: &str) -> EvalResult<Value> {
    Ok(parse_string_as_bool(s)
        .map(Value::Boolean)
        .unwrap_or(Value::Null))
}

/// Cast string to timestamp using the shared `parse_timestamp_to_utc`
/// format set, so native and variant casts agree on what strings parse:
/// RFC3339 first, then several timezone-aware and naive formats, then
/// date-only fallbacks. Returns null on parse failure.
fn cast_string_to_timestamp(s: &str) -> EvalResult<Value> {
    use crate::value::TimestampValue;
    Ok(parse_timestamp_to_utc(s)
        .map(|dt| TimestampValue::utc(dt).into())
        .unwrap_or(Value::Null))
}

/// Cast array elements with precomputed cast kind
fn cast_array_with_kind(
    arr: &[Value],
    element_cast_kind: Box<CastKind>,
    target_type: &Type,
) -> EvalResult<Value> {
    let mut result = Vec::with_capacity(arr.len());

    // Extract element type from target array type
    let element_type = if let Type::Array(array_type) = target_type {
        &*array_type.element_type
    } else {
        return Err(EvalError::execution(format!(
            "ArrayElementCast expected Array target type, got {:?}",
            target_type
        )));
    };

    for element in arr {
        result.push(perform_cast(
            element.clone(),
            (*element_cast_kind).clone(),
            element_type,
        )?);
    }
    Ok(Value::Array(result))
}

/// Cast range bounds with precomputed cast kind
fn cast_range_with_kind(
    range: &RangeValue,
    element_cast_kind: Box<CastKind>,
    target_type: &Type,
) -> EvalResult<Value> {
    // Extract element type from target range type
    let element_type =
        if let Type::Range(range_type) | Type::RangeInclusive(range_type) = target_type {
            &*range_type.of
        } else {
            return Err(EvalError::execution(format!(
                "RangeElementCast expected Range target type, got {:?}",
                target_type
            )));
        };

    let lower = range
        .lower
        .as_ref()
        .map(|v| perform_cast(v.clone(), (*element_cast_kind).clone(), element_type))
        .transpose()?;
    let upper = range
        .upper
        .as_ref()
        .map(|v| perform_cast(v.clone(), (*element_cast_kind).clone(), element_type))
        .transpose()?;

    Ok(Value::Range(Box::new(RangeValue { lower, upper })))
}

/// Convert value to JSON for variant storage
fn to_json_value(value: Value) -> EvalResult<serde_json::Value> {
    let json_val = match value {
        Value::Null => JsonValue::Null,
        Value::Boolean(b) => JsonValue::Bool(b),
        Value::Int(i) => JsonValue::Number(serde_json::Number::from(i)),
        Value::Double(d) => serde_json::Number::from_f64(d)
            .map(JsonValue::Number)
            .unwrap_or(JsonValue::Null),
        Value::String(s) => JsonValue::String(s),
        Value::Array(arr) => {
            let mut json_arr = Vec::with_capacity(arr.len());
            for val in arr {
                json_arr.push(to_json_value(val)?);
            }
            JsonValue::Array(json_arr)
        }
        Value::Variant(v) => v,
        Value::Binary(b) => {
            // Encode binary as hex string in JSON
            let mut hex_string = String::from("0x");
            for byte in b {
                hex_string.push_str(&format!("{:02x}", byte));
            }
            JsonValue::String(hex_string)
        }
        Value::Timestamp(t) => JsonValue::String(t.instant().to_rfc3339()),
        Value::Decimal(d) => {
            let scale_factor = 10_i128.pow(d.scale as u32);
            let integer_part = d.unscaled / scale_factor;
            let fractional_part = (d.unscaled % scale_factor).abs();
            let str_repr = format!(
                "{}.{:0width$}",
                integer_part,
                fractional_part,
                width = d.scale as usize
            );
            JsonValue::String(str_repr)
        }
        Value::Interval(_)
        | Value::CalendarInterval(_)
        | Value::Rows(_)
        | Value::Tuple(_)
        | Value::Struct(_)
        | Value::Map(_)
        | Value::Range(_)
        | Value::Closure(_)
        | Value::Unknown => {
            return Err(EvalError::execution(format!(
                "Cannot convert {:?} to JSON",
                value
            )));
        }
    };

    Ok(json_val)
}

/// Cast a variant value using a VariantCastKind descriptor
fn cast_from_variant(variant: &serde_json::Value, kind: &VariantCastKind) -> EvalResult<Value> {
    match variant {
        // JSON null in a variant is the null literal, not SQL null. When the
        // target is variant we preserve it as a variant value (matching
        // DataFusion's behavior in from_variant.rs); for any other target,
        // JSON null collapses to SQL null since no scalar type meaningfully
        // represents it.
        JsonValue::Null if !matches!(kind, VariantCastKind::Variant) => Ok(Value::Null),
        _ => match kind {
            VariantCastKind::Int => match variant {
                JsonValue::Number(n) => {
                    // Integer JSON numbers go direct; fractional numbers truncate toward zero
                    // to match native `double AS int` (DoubleToInt).
                    if let Some(i) = n.as_i64() {
                        Ok(Value::Int(i))
                    } else if let Some(f) = n.as_f64() {
                        Ok(f64_to_i64(f).map(Value::Int).unwrap_or(Value::Null))
                    } else {
                        Ok(Value::Null)
                    }
                }
                JsonValue::Bool(b) => Ok(Value::Int(if *b { 1 } else { 0 })),
                JsonValue::String(s) => {
                    // Double-cast: parse the string as int. Null on parse fail.
                    Ok(s.parse::<i64>().map(Value::Int).unwrap_or(Value::Null))
                }
                _ => Ok(Value::Null),
            },
            VariantCastKind::Double => match variant {
                // Boxed-type rule: never error from a variant cast. A JSON number
                // that won't fit in f64 (rare — only oversize integers) becomes
                // null, matching every sibling arm.
                JsonValue::Number(n) => Ok(n.as_f64().map(Value::Double).unwrap_or(Value::Null)),
                JsonValue::Bool(b) => Ok(Value::Double(if *b { 1.0 } else { 0.0 })),
                JsonValue::String(s) => {
                    Ok(s.parse::<f64>().map(Value::Double).unwrap_or(Value::Null))
                }
                _ => Ok(Value::Null),
            },
            VariantCastKind::Decimal { scale, .. } => {
                // Parse to a DecimalValue at whatever scale the source implies,
                // then rescale to the target so the result matches DataFusion
                // (e.g. parse_json('42') AS decimal(10, 2) → 42.00, not 42).
                let parsed = match variant {
                    JsonValue::Number(n) => n.to_string().parse::<DecimalValue>().ok(),
                    JsonValue::Bool(b) => Some(DecimalValue {
                        unscaled: if *b { 1 } else { 0 },
                        scale: 0,
                    }),
                    JsonValue::String(s) => s.parse::<DecimalValue>().ok(),
                    _ => None,
                };
                match parsed {
                    Some(d) => cast_decimal_to_decimal(&d, *scale),
                    None => Ok(Value::Null),
                }
            }
            VariantCastKind::String => match variant {
                JsonValue::String(s) => Ok(Value::String(s.clone())),
                JsonValue::Number(n) => Ok(Value::String(n.to_string())),
                JsonValue::Bool(b) => Ok(Value::String(b.to_string())),
                _ => Ok(Value::Null),
            },
            VariantCastKind::Boolean => match variant {
                JsonValue::Bool(b) => Ok(Value::Boolean(*b)),
                JsonValue::Number(n) => {
                    // Numeric → boolean: any nonzero value (int or double) is true,
                    // zero is false, matching IntToBoolean / DoubleToBoolean.
                    if let Some(f) = n.as_f64() {
                        Ok(Value::Boolean(f != 0.0))
                    } else {
                        Ok(Value::Null)
                    }
                }
                JsonValue::String(s) => Ok(parse_string_as_bool(s)
                    .map(Value::Boolean)
                    .unwrap_or(Value::Null)),
                _ => Ok(Value::Null),
            },
            VariantCastKind::Timestamp => match variant {
                // Variant-string → timestamp: delegate to Hamelin's standard string→timestamp cast.
                JsonValue::String(s) => cast_string_to_timestamp(s).or(Ok(Value::Null)),
                _ => Ok(Value::Null),
            },
            VariantCastKind::Array(element_kind) => match variant {
                JsonValue::Array(json_array) => {
                    let mut result = Vec::with_capacity(json_array.len());
                    for json_element in json_array {
                        result.push(cast_from_variant(json_element, element_kind)?);
                    }
                    Ok(Value::Array(result))
                }
                // Shape mismatch (non-array runtime type) → null per the boxed-type
                // rule; casts never error at runtime.
                _ => Ok(Value::Null),
            },
            VariantCastKind::Struct(field_kinds) => match variant {
                JsonValue::Object(obj) => {
                    let mut result = OrderMap::new();
                    for (field_name, field_kind) in field_kinds {
                        let json_value = obj.get(field_name).unwrap_or(&JsonValue::Null);
                        let casted_value = cast_from_variant(json_value, field_kind)?;
                        result.insert(SimpleIdentifier::new(field_name), casted_value);
                    }
                    Ok(Value::Struct(result))
                }
                // Shape mismatch (non-object runtime type) → null per the boxed-type
                // rule; casts never error at runtime.
                _ => Ok(Value::Null),
            },
            VariantCastKind::Map(key_kind, value_kind) => match variant {
                JsonValue::Object(obj) => {
                    let mut result = LinearMap::new();
                    for (key_str, json_value) in obj {
                        // JSON object keys are always strings, cast to target key type
                        let casted_key =
                            cast_from_variant(&JsonValue::String(key_str.clone()), key_kind)?;
                        let casted_value = cast_from_variant(json_value, value_kind)?;
                        result.insert(casted_key, casted_value);
                    }
                    Ok(Value::Map(result))
                }
                // Shape mismatch (non-object runtime type) → null per the boxed-type
                // rule; casts never error at runtime.
                _ => Ok(Value::Null),
            },
            // Unknown type - return null (this handles empty arrays cast to variant then back)
            VariantCastKind::Unknown => Ok(Value::Null),
            // Variant to variant - return as-is
            VariantCastKind::Variant => Ok(Value::Variant(variant.clone())),
        },
    }
}

/// Cast tuple to struct with precomputed cast kinds for each field
fn cast_tuple_to_struct_with_kinds(
    tuple: &[Value],
    field_casts: Vec<(String, CastKind)>,
    target_type: &Type,
) -> EvalResult<Value> {
    let mut result = OrderMap::new();

    // Extract struct type from target
    let struct_type = if let Type::Struct(s) = target_type {
        s
    } else {
        return Err(EvalError::execution(format!(
            "TupleToStruct expected Struct target type, got {:?}",
            target_type
        )));
    };

    // Map tuple elements to struct fields using precomputed cast kinds
    for (i, (field_name, cast_kind)) in field_casts.iter().enumerate() {
        let simple_field_name = SimpleIdentifier::new(field_name);

        if i < tuple.len() {
            // Get the field's type from the struct
            let field_type = struct_type.lookup(&simple_field_name).ok_or_else(|| {
                EvalError::execution(format!("Field cast for non-existent field: {}", field_name))
            })?;

            // Cast tuple element to the field's type
            let casted_value = perform_cast(tuple[i].clone(), cast_kind.clone(), field_type)?;
            result.insert(simple_field_name, casted_value);
        } else {
            // Not enough tuple elements - set to NULL
            result.insert(simple_field_name, Value::Null);
        }
    }

    Ok(Value::Struct(result))
}

/// Expand a struct by adding null fields and casting existing fields as needed.
fn cast_struct_expansion(
    source_fields: OrderMap<SimpleIdentifier, Value>,
    field_casts: Vec<(SimpleIdentifier, CastKind)>,
    target_type: &Type,
) -> EvalResult<Value> {
    let mut result = OrderMap::new();

    // Get target struct type for field types
    let target_struct = match target_type {
        Type::Struct(s) => s,
        _ => {
            return Err(EvalError::execution(format!(
                "StructExpansion expected Struct target type, got {:?}",
                target_type
            )));
        }
    };

    for (field_name, cast_kind) in field_casts {
        let field_type = target_struct.lookup(&field_name).ok_or_else(|| {
            EvalError::execution(format!("Field {} not in target struct", field_name))
        })?;

        let value = match &cast_kind {
            CastKind::NullToType => Value::Null,
            _ => {
                // Field exists in source - get and cast it
                let source_value = source_fields
                    .get(&field_name)
                    .cloned()
                    .unwrap_or(Value::Null);
                perform_cast(source_value, cast_kind, field_type)?
            }
        };
        result.insert(field_name, value);
    }

    Ok(Value::Struct(result))
}

/// Cast double to decimal with appropriate scale
fn cast_double_to_decimal(d: f64) -> EvalResult<Value> {
    if d.is_infinite() || d.is_nan() {
        return Ok(Value::Null);
    }

    // Use scale of 6 for doubles to preserve reasonable precision
    let scale = 6;
    let scale_factor = 10_f64.powi(scale);
    let unscaled = (d * scale_factor).round() as i128;

    Ok(Value::Decimal(DecimalValue { unscaled, scale }))
}

/// Cast decimal to integer by truncating decimal part. Overflow produces null.
fn cast_decimal_to_int(dec: &DecimalValue) -> EvalResult<Value> {
    let scale_factor = 10_i128.pow(dec.scale as u32);
    let integer_part = dec.unscaled / scale_factor;

    if integer_part < i64::MIN as i128 || integer_part > i64::MAX as i128 {
        return Ok(Value::Null);
    }

    Ok(Value::Int(integer_part as i64))
}

/// Pick the scale to apply when rescaling a freshly-constructed DecimalValue
/// to a target Decimal type. Falls back to the source's natural scale when the
/// target isn't a Decimal (defensive — the cast kind should make that unreachable).
fn decimal_target_scale(target_type: &Type, fallback: i32) -> i32 {
    match target_type {
        Type::Decimal(d) => d.scale,
        _ => fallback,
    }
}

/// Cast decimal to a different precision/scale. Overflow when widening scale
/// produces null (matches the `x AS T` null-on-failure contract).
fn cast_decimal_to_decimal(dec: &DecimalValue, target_scale: i32) -> EvalResult<Value> {
    let scale_diff = target_scale - dec.scale;
    let unscaled = if scale_diff > 0 {
        // Increasing scale — multiply to add trailing zeros
        let Some(unscaled) = dec.unscaled.checked_mul(10_i128.pow(scale_diff as u32)) else {
            return Ok(Value::Null);
        };
        unscaled
    } else if scale_diff < 0 {
        // Decreasing scale — divide (truncate toward zero)
        let divisor = 10_i128.pow((-scale_diff) as u32);
        dec.unscaled / divisor
    } else {
        dec.unscaled
    };

    Ok(Value::Decimal(DecimalValue {
        unscaled,
        scale: target_scale,
    }))
}

/// Cast decimal to double
fn cast_decimal_to_double(dec: &DecimalValue) -> EvalResult<Value> {
    let scale_factor = 10_f64.powi(dec.scale);
    let double_value = dec.unscaled as f64 / scale_factor;

    Ok(Value::Double(double_value))
}

/// Cast string to decimal by parsing decimal representation. Parse failure
/// or non-finite values produce null, matching the documented `x AS T`
/// contract.
fn cast_string_to_decimal(s: &str) -> EvalResult<Value> {
    // Parse as f64 first to handle various formats
    let Ok(parsed_double) = s.parse::<f64>() else {
        return Ok(Value::Null);
    };

    if parsed_double.is_infinite() || parsed_double.is_nan() {
        return Ok(Value::Null);
    }

    // Determine scale based on decimal places in the string
    let scale = if let Some(dot_pos) = s.find('.') {
        let decimal_part = &s[dot_pos + 1..];
        // Remove any trailing zeros and non-numeric characters
        let cleaned = decimal_part
            .chars()
            .take_while(|c| c.is_ascii_digit())
            .collect::<String>();
        cleaned.len() as i32
    } else {
        0
    };

    // Convert to unscaled integer representation
    let scale_factor = 10_f64.powi(scale);
    let unscaled = (parsed_double * scale_factor).round() as i128;

    Ok(Value::Decimal(DecimalValue { unscaled, scale }))
}

/// Cast interval to Range<Timestamp>
/// Positive interval: [now(), now() + interval)
/// Negative interval: [now() - interval, now())
fn cast_interval_to_timestamp_range(value: Value) -> EvalResult<Value> {
    let now = Utc::now();

    match value {
        Value::Interval(duration) => {
            let (lower, upper) = if duration < Duration::zero() {
                // Negative interval: [now() - interval, now())
                let lower_ts = now + duration; // duration is negative, so this subtracts
                (
                    Some(TimestampValue::utc(lower_ts).into()),
                    Some(TimestampValue::utc(now).into()),
                )
            } else {
                // Positive interval: [now(), now() + interval)
                let upper_ts = now + duration;
                (
                    Some(TimestampValue::utc(now).into()),
                    Some(TimestampValue::utc(upper_ts).into()),
                )
            };

            Ok(Value::Range(Box::new(RangeValue { lower, upper })))
        }
        Value::CalendarInterval(_) => {
            // For calendar intervals, we can't directly add them without a concrete base timestamp
            // For now, treat them as zero-duration ranges at now()
            Ok(Value::Range(Box::new(RangeValue {
                lower: Some(TimestampValue::utc(now).into()),
                upper: Some(TimestampValue::utc(now).into()),
            })))
        }
        _ => Err(EvalError::execution(format!(
            "IntervalToTimestampRange cast expected Interval value, got {:?}",
            value
        ))),
    }
}

/// Cast timestamp to Range<Timestamp>
/// Always creates [timestamp, now())
fn cast_timestamp_to_timestamp_range(value: Value) -> EvalResult<Value> {
    let now = Utc::now();

    match value {
        Value::Timestamp(ts) => Ok(Value::Range(Box::new(RangeValue {
            lower: Some(ts.into()),
            upper: Some(TimestampValue::utc(now).into()),
        }))),
        _ => Err(EvalError::execution(format!(
            "TimestampToTimestampRange cast expected Timestamp value, got {:?}",
            value
        ))),
    }
}

/// Cast Range<Interval> to Range<Timestamp>
/// Converts each bound by treating intervals relative to now()
fn cast_interval_range_to_timestamp_range(value: Value) -> EvalResult<Value> {
    let now = Utc::now();

    match value {
        Value::Range(range) => {
            let lower = match range.lower {
                Some(Value::Interval(duration)) => {
                    let ts = if duration < Duration::zero() {
                        now + duration // duration is negative, so this subtracts
                    } else {
                        now + duration
                    };
                    Some(TimestampValue::utc(ts).into())
                }
                Some(Value::CalendarInterval(_)) => {
                    // For calendar intervals, treat as now() for simplicity
                    Some(TimestampValue::utc(now).into())
                }
                None => None,
                _ => {
                    return Err(EvalError::execution(format!(
                        "IntervalRangeToTimestampRange expected Interval lower bound, got {:?}",
                        range.lower
                    )));
                }
            };

            let upper = match range.upper {
                Some(Value::Interval(duration)) => {
                    let ts = if duration < Duration::zero() {
                        now + duration // duration is negative, so this subtracts
                    } else {
                        now + duration
                    };
                    Some(TimestampValue::utc(ts).into())
                }
                Some(Value::CalendarInterval(_)) => {
                    // For calendar intervals, treat as now() for simplicity
                    Some(TimestampValue::utc(now).into())
                }
                None => None,
                _ => {
                    return Err(EvalError::execution(format!(
                        "IntervalRangeToTimestampRange expected Interval upper bound, got {:?}",
                        range.upper
                    )));
                }
            };

            Ok(Value::Range(Box::new(RangeValue { lower, upper })))
        }
        _ => Err(EvalError::execution(format!(
            "IntervalRangeToTimestampRange cast expected Range value, got {:?}",
            value
        ))),
    }
}