hamelin_eval 0.10.13

use crate::value::Value;
use hamelin_lib::tree::ast::identifier::SimpleIdentifier;
use hamelin_lib::tree::builder::{
    array, call, cast, days, decimal, decimal_from_parts, field_ref, null, tuple,
};
use hamelin_lib::types::array::Array;
use hamelin_lib::types::range::Range;
use hamelin_lib::types::struct_type::Struct;
use hamelin_lib::types::{decimal_type, Type, BOOLEAN, DOUBLE, INT, STRING, TIMESTAMP, VARIANT};
use ordermap::OrderMap;

use super::test_helpers::test_context;

const DECIMAL: Type = Type::Decimal(decimal_type::Decimal {
    precision: 18,
    scale: 6,
});

#[test]
fn test_identity_casts() {
    let ctx = test_context();

    // Same type casts should pass through unchanged
    assert_eq!(ctx.eval_expr(&cast(42, INT)), Value::Int(42));
    assert_eq!(ctx.eval_expr(&cast(3.14, DOUBLE)), Value::Double(3.14));
    assert_eq!(
        ctx.eval_expr(&cast("hello", STRING)),
        Value::String("hello".to_string())
    );
    assert_eq!(ctx.eval_expr(&cast(true, BOOLEAN)), Value::Boolean(true));
}

#[test]
fn test_null_casts() {
    let ctx = test_context();

    // NULL should cast to any type and remain NULL
    assert_eq!(ctx.eval_expr(&cast(null(), INT)), Value::Null);
    assert_eq!(ctx.eval_expr(&cast(null(), STRING)), Value::Null);
    assert_eq!(ctx.eval_expr(&cast(null(), BOOLEAN)), Value::Null);
    assert_eq!(ctx.eval_expr(&cast(null(), DOUBLE)), Value::Null);
}

#[test]
fn test_numeric_casts() {
    let ctx = test_context();

    // Int to Double
    assert_eq!(ctx.eval_expr(&cast(42, DOUBLE)), Value::Double(42.0));
    assert_eq!(ctx.eval_expr(&cast(-17, DOUBLE)), Value::Double(-17.0));
    assert_eq!(ctx.eval_expr(&cast(0, DOUBLE)), Value::Double(0.0));

    // Double to Int (truncation)
    assert_eq!(ctx.eval_expr(&cast(3.14, INT)), Value::Int(3));
    assert_eq!(ctx.eval_expr(&cast(3.99, INT)), Value::Int(3));
    assert_eq!(ctx.eval_expr(&cast(-2.7, INT)), Value::Int(-2));
    assert_eq!(ctx.eval_expr(&cast(0.0, INT)), Value::Int(0));
}

#[test]
fn test_double_overflow_to_int_is_null() {
    // Out-of-range or non-finite doubles null when cast to int (matches the
    // `x AS T` null-on-failure contract).
    let ctx = test_context();

    assert_eq!(ctx.eval_expr(&cast(1e20, INT)), Value::Null); // Way beyond i64::MAX
    assert_eq!(ctx.eval_expr(&cast(-1e20, INT)), Value::Null); // Way beyond i64::MIN

    // Test infinite values (these would need special double literals)
    // For now, we'll skip these tests as they require more complex expression building
}

#[test]
fn test_string_casts() {
    let ctx = test_context();

    // To String conversions
    assert_eq!(
        ctx.eval_expr(&cast(42, STRING)),
        Value::String("42".to_string())
    );
    assert_eq!(
        ctx.eval_expr(&cast(3.14, STRING)),
        Value::String("3.14".to_string())
    );
    assert_eq!(
        ctx.eval_expr(&cast(true, STRING)),
        Value::String("true".to_string())
    );
    assert_eq!(
        ctx.eval_expr(&cast(false, STRING)),
        Value::String("false".to_string())
    );

    // From String conversions
    assert_eq!(ctx.eval_expr(&cast("42", INT)), Value::Int(42));
    assert_eq!(ctx.eval_expr(&cast("3.14", DOUBLE)), Value::Double(3.14));
    assert_eq!(ctx.eval_expr(&cast("true", BOOLEAN)), Value::Boolean(true));
    assert_eq!(
        ctx.eval_expr(&cast("false", BOOLEAN)),
        Value::Boolean(false)
    );
    assert_eq!(ctx.eval_expr(&cast("TRUE", BOOLEAN)), Value::Boolean(true)); // Case insensitive
    assert_eq!(
        ctx.eval_expr(&cast("False", BOOLEAN)),
        Value::Boolean(false)
    ); // Case insensitive

    // Wider Arrow-compatible acceptance set: yes/no, on/off, 1/0, prefixes,
    // case-insensitive, leading/trailing whitespace trimmed.
    assert_eq!(ctx.eval_expr(&cast("yes", BOOLEAN)), Value::Boolean(true));
    assert_eq!(ctx.eval_expr(&cast("no", BOOLEAN)), Value::Boolean(false));
    assert_eq!(ctx.eval_expr(&cast("on", BOOLEAN)), Value::Boolean(true));
    assert_eq!(ctx.eval_expr(&cast("off", BOOLEAN)), Value::Boolean(false));
    assert_eq!(ctx.eval_expr(&cast("1", BOOLEAN)), Value::Boolean(true));
    assert_eq!(ctx.eval_expr(&cast("0", BOOLEAN)), Value::Boolean(false));
    assert_eq!(ctx.eval_expr(&cast("YES", BOOLEAN)), Value::Boolean(true));
    assert_eq!(
        ctx.eval_expr(&cast("  true  ", BOOLEAN)),
        Value::Boolean(true)
    );
}

#[test]
fn test_string_parse_failure_is_null() {
    // `x AS T` for primitive targets nulls on parse failure rather than
    // erroring (matches Trino TRY_CAST and the casting.md contract). The
    // wrapping pipeline never crashes on bad input rows.
    let ctx = test_context();

    // Invalid integer strings
    assert_eq!(ctx.eval_expr(&cast("not_a_number", INT)), Value::Null);
    assert_eq!(ctx.eval_expr(&cast("3.14", INT)), Value::Null);
    assert_eq!(ctx.eval_expr(&cast("", INT)), Value::Null);
    // The VPC flow regression case from the production crash.
    assert_eq!(ctx.eval_expr(&cast("-", INT)), Value::Null);

    // Invalid double strings
    assert_eq!(ctx.eval_expr(&cast("not_a_number", DOUBLE)), Value::Null);
    assert_eq!(ctx.eval_expr(&cast("3.14.15", DOUBLE)), Value::Null);

    // Invalid boolean strings — anything outside Arrow's accepted set
    // ('true'/'false'/'yes'/'no'/'on'/'off'/'1'/'0' and prefixes, case
    // insensitive, trimmed) nulls. 'yes'/'1'/'0' DO parse under the wide
    // set (verified by test_boolean_casts below).
    assert_eq!(ctx.eval_expr(&cast("not_a_bool", BOOLEAN)), Value::Null);
    assert_eq!(ctx.eval_expr(&cast("maybe", BOOLEAN)), Value::Null);
}

#[test]
fn test_boolean_casts() {
    let ctx = test_context();

    // To Boolean
    assert_eq!(ctx.eval_expr(&cast(0, BOOLEAN)), Value::Boolean(false));
    assert_eq!(ctx.eval_expr(&cast(1, BOOLEAN)), Value::Boolean(true));
    assert_eq!(ctx.eval_expr(&cast(-1, BOOLEAN)), Value::Boolean(true));
    assert_eq!(ctx.eval_expr(&cast(42, BOOLEAN)), Value::Boolean(true));

    // From Boolean
    assert_eq!(ctx.eval_expr(&cast(true, INT)), Value::Int(1));
    assert_eq!(ctx.eval_expr(&cast(false, INT)), Value::Int(0));
    assert_eq!(
        ctx.eval_expr(&cast(true, STRING)),
        Value::String("true".to_string())
    );
    assert_eq!(
        ctx.eval_expr(&cast(false, STRING)),
        Value::String("false".to_string())
    );
}

#[test]
fn test_array_element_casting() {
    let ctx = test_context();

    // Array element type conversion
    let int_array = array().element(1).element(2).element(3);
    let result = ctx.eval_expr(&cast(int_array, Type::Array(Array::new(DOUBLE))));
    assert_eq!(
        result,
        Value::Array(vec![
            Value::Double(1.0),
            Value::Double(2.0),
            Value::Double(3.0),
        ])
    );

    let double_array = array().element(1.1).element(2.7).element(3.9);
    let result = ctx.eval_expr(&cast(double_array, Type::Array(Array::new(INT))));
    assert_eq!(
        result,
        Value::Array(vec![Value::Int(1), Value::Int(2), Value::Int(3)])
    );

    let string_array = array().element("1").element("2").element("3");
    let result = ctx.eval_expr(&cast(string_array, Type::Array(Array::new(INT))));
    assert_eq!(
        result,
        Value::Array(vec![Value::Int(1), Value::Int(2), Value::Int(3)])
    );
}

#[test]
fn test_tuple_to_struct_casting() {
    let ctx = test_context();

    // Create a struct type with two fields: name (String) and age (Int)
    let mut fields = OrderMap::new();
    fields.insert(SimpleIdentifier::new("name"), STRING);
    fields.insert(SimpleIdentifier::new("age"), INT);
    let struct_type = Type::Struct(Struct::new(fields));

    // Test tuple with matching elements
    let tuple_expr = tuple().element("Alice").element(30);
    println!("Testing tuple cast: {:?} -> {:?}", tuple_expr, struct_type);

    // First test if the tuple evaluates correctly
    let tuple_result = ctx.try_eval_expr(&tuple_expr);
    println!("Tuple evaluation result: {:?}", tuple_result);

    // Now test the cast
    let cast_result = ctx.try_eval_expr(&cast(tuple_expr, struct_type.clone()));
    println!("Cast result: {:?}", cast_result);

    if let Ok(Value::Struct(struct_map)) = cast_result {
        assert_eq!(
            struct_map.get(&SimpleIdentifier::new("name")),
            Some(&Value::String("Alice".to_string()))
        );
        assert_eq!(
            struct_map.get(&SimpleIdentifier::new("age")),
            Some(&Value::Int(30))
        );
    } else {
        panic!("Expected struct result, got: {:?}", cast_result);
    }

    // Test tuple with matching length but different types (type conversion)
    let type_convert_tuple = tuple().element(42).element("30"); // Int and String
    let result = ctx.try_eval_expr(&cast(type_convert_tuple, struct_type.clone()));
    println!("Type conversion tuple cast result: {:?}", result);

    if let Ok(Value::Struct(struct_map)) = result {
        assert_eq!(
            struct_map.get(&SimpleIdentifier::new("name")),
            Some(&Value::String("42".to_string()))
        );
        assert_eq!(
            struct_map.get(&SimpleIdentifier::new("age")),
            Some(&Value::Int(30))
        );
    } else {
        panic!(
            "Expected struct result for type conversion tuple, got: {:?}",
            result
        );
    }

    // Test that tuple with wrong length fails type checking (should not compile)
    let short_tuple = tuple().element("Bob");
    let short_result = ctx.try_eval_expr(&cast(short_tuple, struct_type));
    // This should fail at type checking phase, not evaluation
    assert!(
        short_result.is_err(),
        "Expected type checking error for mismatched tuple/struct lengths"
    );
}

#[test]
fn test_decimal_casts() {
    let ctx = test_context();

    // Int to Decimal — rescale to the target's scale (DECIMAL is scale 6 here)
    // so 42 AS DECIMAL(18, 6) yields 42.000000.
    assert_eq!(
        ctx.eval_expr(&cast(42, DECIMAL)),
        Value::Decimal(crate::value::DecimalValue {
            unscaled: 42_000_000,
            scale: 6
        })
    );

    // Double to Decimal
    let result = ctx.eval_expr(&cast(3.14, DECIMAL));
    if let Value::Decimal(dec) = result {
        assert_eq!(dec.scale, 6);
        // 3.14 * 10^6 = 3140000
        assert_eq!(dec.unscaled, 3140000);
    } else {
        panic!("Expected decimal result");
    }

    // Boolean to Decimal — true rescales to target.
    assert_eq!(
        ctx.eval_expr(&cast(true, DECIMAL)),
        Value::Decimal(crate::value::DecimalValue {
            unscaled: 1_000_000,
            scale: 6
        })
    );
    assert_eq!(
        ctx.eval_expr(&cast(false, DECIMAL)),
        Value::Decimal(crate::value::DecimalValue {
            unscaled: 0,
            scale: 6
        })
    );

    // Decimal to Int (truncation) - now using decimal literal builder
    let decimal_literal = decimal_from_parts(31415, 5, 4); // 3.1415 with precision 5, scale 4
    let result = ctx.eval_expr(&cast(decimal_literal, INT));
    assert_eq!(result, Value::Int(3)); // Should truncate to 3

    // Decimal to Double conversion
    let decimal_literal = decimal_from_parts(31415, 5, 4); // 3.1415
    let result = ctx.eval_expr(&cast(decimal_literal, DOUBLE));
    assert_eq!(result, Value::Double(3.1415));

    // String to Decimal — rescale parsed value to the target's scale.
    assert_eq!(
        ctx.eval_expr(&cast("3.14", DECIMAL)),
        Value::Decimal(crate::value::DecimalValue {
            unscaled: 3_140_000,
            scale: 6
        })
    );

    assert_eq!(
        ctx.eval_expr(&cast("42", DECIMAL)),
        Value::Decimal(crate::value::DecimalValue {
            unscaled: 42_000_000,
            scale: 6
        })
    );

    // Decimal to String — DECIMAL is scale 6, so the parsed "3.14159" is
    // rescaled to "3.141590" (trailing zero added).
    let decimal_expr = cast("3.14159", DECIMAL);
    let result = ctx.eval_expr(&cast(decimal_expr, STRING));
    assert_eq!(result, Value::String("3.141590".to_string()));

    // Test decimal convenience function with string parsing
    let decimal_literal = decimal("123.456").expect("Valid decimal string");
    let result = ctx.eval_expr(&decimal_literal);
    assert_eq!(
        result,
        Value::Decimal(crate::value::DecimalValue {
            unscaled: 123456,
            scale: 3
        })
    );

    // Test decimal to string conversion with literal
    let decimal_literal = decimal("99.99").expect("Valid decimal string");
    let result = ctx.eval_expr(&cast(decimal_literal, STRING));
    assert_eq!(result, Value::String("99.99".to_string()));
}

#[test]
fn test_decimal_literal_builder() {
    let ctx = test_context();

    // Test decimal literal creation with string parsing
    let decimal_literal = decimal("42.75").expect("Valid decimal string");
    let result = ctx.eval_expr(&decimal_literal);
    assert_eq!(
        result,
        Value::Decimal(crate::value::DecimalValue {
            unscaled: 4275,
            scale: 2
        })
    );

    // Test decimal literal creation from parts
    let decimal_literal = decimal_from_parts(12345, 5, 3); // 12.345
    let result = ctx.eval_expr(&decimal_literal);
    assert_eq!(
        result,
        Value::Decimal(crate::value::DecimalValue {
            unscaled: 12345,
            scale: 3
        })
    );
}

#[test]
fn test_edge_cases() {
    let ctx = test_context();

    // Empty array casting
    let empty_array = array();
    assert_eq!(
        ctx.eval_expr(&cast(empty_array, Type::Array(Array::new(STRING)))),
        Value::Array(vec![])
    );

    // Nested null values in arrays
    let array_with_nulls = array().element(1).element(null()).element(3);
    let result = ctx.eval_expr(&cast(array_with_nulls, Type::Array(Array::new(STRING))));
    assert_eq!(
        result,
        Value::Array(vec![
            Value::String("1".to_string()),
            Value::Null,
            Value::String("3".to_string()),
        ])
    );
}

#[test]
fn test_interval_to_timestamp_range() {
    let ctx = test_context();

    // Cast interval to Range<Timestamp>
    // Positive interval: [now(), now() + interval)
    let interval_expr = days(7);
    let result = ctx.eval_expr(&cast(interval_expr, Type::Range(Range::new(TIMESTAMP))));

    // Check that we got a range with two timestamps
    if let Value::Range(range) = result {
        assert!(range.lower.is_some(), "Lower bound should exist");
        assert!(range.upper.is_some(), "Upper bound should exist");

        // Check that both bounds are timestamps
        assert!(
            matches!(range.lower.as_ref().unwrap(), Value::Timestamp(_)),
            "Lower bound should be a timestamp"
        );
        assert!(
            matches!(range.upper.as_ref().unwrap(), Value::Timestamp(_)),
            "Upper bound should be a timestamp"
        );

        // For a positive interval, lower should be <= upper
        if let (Some(Value::Timestamp(lower_ts)), Some(Value::Timestamp(upper_ts))) =
            (range.lower.as_ref(), range.upper.as_ref())
        {
            assert!(
                lower_ts <= upper_ts,
                "For positive interval, lower should be <= upper"
            );
        }
    } else {
        panic!("Expected Range value, got: {:?}", result);
    }
}

#[test]
fn test_timestamp_to_timestamp_range() {
    let ctx = test_context();

    // Cast timestamp to Range<Timestamp>
    // Should create [timestamp, now())
    let timestamp_expr = call("now");
    let result = ctx.eval_expr(&cast(timestamp_expr, Type::Range(Range::new(TIMESTAMP))));

    // Check that we got a range with two timestamps
    if let Value::Range(range) = result {
        assert!(range.lower.is_some(), "Lower bound should exist");
        assert!(range.upper.is_some(), "Upper bound should exist");

        // Check that both bounds are timestamps
        assert!(
            matches!(range.lower.as_ref().unwrap(), Value::Timestamp(_)),
            "Lower bound should be a timestamp"
        );
        assert!(
            matches!(range.upper.as_ref().unwrap(), Value::Timestamp(_)),
            "Upper bound should be a timestamp"
        );
    } else {
        panic!("Expected Range value, got: {:?}", result);
    }
}

#[test]
fn test_interval_range_to_timestamp_range() {
    let ctx = test_context();

    // Cast Range<Interval> to Range<Timestamp>
    // Using Rust range syntax: days(1)..days(7)
    let interval_range = days(1)..days(7);
    let result = ctx.eval_expr(&cast(interval_range, Type::Range(Range::new(TIMESTAMP))));

    // Check that we got a range with two timestamps
    if let Value::Range(range) = result {
        assert!(range.lower.is_some(), "Lower bound should exist");
        assert!(range.upper.is_some(), "Upper bound should exist");

        // Check that both bounds are timestamps
        assert!(
            matches!(range.lower.as_ref().unwrap(), Value::Timestamp(_)),
            "Lower bound should be a timestamp"
        );
        assert!(
            matches!(range.upper.as_ref().unwrap(), Value::Timestamp(_)),
            "Upper bound should be a timestamp"
        );
    } else {
        panic!("Expected Range value, got: {:?}", result);
    }
}

#[test]
fn test_variant_to_decimal_honors_target_scale() {
    // Variant→decimal must rescale to the target's precision/scale, matching
    // DataFusion's from_variant.rs: `parse_json('42') AS decimal(10, 2)` →
    // DecimalValue { unscaled: 4200, scale: 2 }, not unscaled: 42, scale: 0.
    let mut ctx = test_context();
    ctx.set("v_int", Value::Variant(serde_json::json!(42)), VARIANT);
    ctx.set("v_str", Value::Variant(serde_json::json!("3.14")), VARIANT);
    ctx.set("v_bool", Value::Variant(serde_json::json!(true)), VARIANT);

    let dec_10_2 = Type::Decimal(decimal_type::Decimal {
        precision: 10,
        scale: 2,
    });

    assert_eq!(
        ctx.eval_expr(&cast(field_ref("v_int"), dec_10_2.clone())),
        Value::Decimal(crate::value::DecimalValue {
            unscaled: 4200,
            scale: 2,
        })
    );
    assert_eq!(
        ctx.eval_expr(&cast(field_ref("v_str"), dec_10_2.clone())),
        Value::Decimal(crate::value::DecimalValue {
            unscaled: 314,
            scale: 2,
        })
    );
    assert_eq!(
        ctx.eval_expr(&cast(field_ref("v_bool"), dec_10_2)),
        Value::Decimal(crate::value::DecimalValue {
            unscaled: 100,
            scale: 2,
        })
    );
}

#[test]
fn test_variant_null_field_preserved_when_target_is_variant() {
    // JSON null inside a variant is the null literal — distinct from SQL null.
    // When casting a variant struct to a target struct whose field type is
    // variant, the JSON null in that field must round-trip as
    // Value::Variant(JsonValue::Null), matching DataFusion's from_variant.rs
    // behavior. For non-variant target field types, JSON null collapses to
    // SQL null.
    let mut ctx = test_context();
    ctx.set(
        "v",
        Value::Variant(serde_json::json!({"a": null, "b": null})),
        VARIANT,
    );

    let mut variant_field_type = OrderMap::new();
    variant_field_type.insert(SimpleIdentifier::new("a"), VARIANT);
    let target_variant = Type::Struct(Struct::new(variant_field_type));

    let result = ctx.eval_expr(&cast(field_ref("v"), target_variant));
    let mut expected_variant = OrderMap::new();
    expected_variant.insert(
        SimpleIdentifier::new("a"),
        Value::Variant(serde_json::Value::Null),
    );
    assert_eq!(result, Value::Struct(expected_variant));

    let mut int_field_type = OrderMap::new();
    int_field_type.insert(SimpleIdentifier::new("b"), INT);
    let target_int = Type::Struct(Struct::new(int_field_type));

    let result = ctx.eval_expr(&cast(field_ref("v"), target_int));
    let mut expected_int = OrderMap::new();
    expected_int.insert(SimpleIdentifier::new("b"), Value::Null);
    assert_eq!(result, Value::Struct(expected_int));
}