hamelin_lib 0.22.1

//! Tests for invalid schema transformations and error cases
//!
//! These tests validate that the type checker correctly identifies and reports
//! errors in queries that reference non-existent fields, use incorrect types,
//! or perform invalid operations.

use std::sync::Arc;

use crate::err::TranslationErrors;
use crate::tree::ast::expression::Expression;
use crate::tree::ast::identifier::SimpleIdentifier;
use crate::tree::ast::pipeline::Pipeline;
use crate::tree::ast::query::Query;
use crate::tree::ast::ParseWithErrors;
use crate::tree::options::ExpressionTypeCheckOptions;
use crate::tree::tests::shared::{catalog_with_tables, type_check_with_catalog};
use crate::tree::typed_ast::command::TypedCommandKind;
use crate::tree::typed_ast::environment::FieldResolution;
use crate::type_check;
use crate::type_check_expression;
use crate::types::array::Array;
use crate::types::decimal_type::Decimal;
use crate::types::map::Map;
use crate::types::range::Range;
use crate::types::struct_type::Struct;
use crate::types::{Type, BOOLEAN, DOUBLE, INT, STRING};
use rstest::rstest;

#[rstest]
#[case::typeof_struct("typeof(value)", "{hamelin_type: string, sql_type: string}")]
#[case::integer("1", "int")]
#[case::string("'true'", "string")]
#[case::field("value", "int")]
#[case::arithmetic("value + 1", "int")]
#[case::boolean_array("[true, false]", "array(boolean)")]
fn test_where_requires_boolean(#[case] predicate: &str, #[case] actual_type: &str) {
    let input = format!("FROM events | WHERE {predicate}");
    let pipeline = Pipeline::parse_result(&input).unwrap();
    let catalog = catalog_with_tables(vec![("events", vec![("value", INT)])]);
    let result = type_check_with_catalog(pipeline, catalog);

    assert_eq!(result.errors.len(), 1, "{:?}", result.errors);
    let error = result.errors.iter().next().unwrap();
    assert_eq!(
        error.primary.message,
        format!("Expected type boolean, got {actual_type}")
    );
    let start = input.find(predicate).unwrap();
    assert_eq!(error.primary.interval, start..=input.len() - 1);
}

#[rstest]
#[case::literal("true")]
#[case::field("active")]
#[case::comparison("value > 0")]
#[case::logical("active AND value > 0")]
#[case::typeof_comparison("typeof(value).hamelin_type == 'int'")]
fn test_where_accepts_boolean(#[case] predicate: &str) {
    let input = format!("FROM events | WHERE {predicate}");
    let pipeline = Pipeline::parse_result(&input).unwrap();
    let catalog = catalog_with_tables(vec![("events", vec![("value", INT), ("active", BOOLEAN)])]);
    let result = type_check_with_catalog(pipeline, catalog);

    assert!(result.errors.is_empty(), "{:?}", result.errors);
}

#[test]
fn test_where_missing_field_does_not_add_type_error() {
    let pipeline = Pipeline::parse_result("SET value = 1 | WHERE missing").unwrap();
    let result = type_check(pipeline);

    assert_eq!(result.errors.len(), 1, "{:?}", result.errors);
    assert!(!result
        .errors
        .iter()
        .next()
        .unwrap()
        .primary
        .message
        .contains("Expected type"));
}

// ======================== Error Tests with SET ========================

#[rstest]
// SELECT errors - non-existent fields
#[case::select_nonexistent_field(
    vec![("name", STRING), ("age", INT)],
    "SELECT email",
    1
)]
#[case::select_multiple_nonexistent_fields(
    vec![("name", STRING)],
    "SELECT age, email, phone",
    1
)]
#[case::select_mix_valid_invalid(
    vec![("name", STRING), ("age", INT)],
    "SELECT name, email",
    1
)]
// WHERE errors - non-existent fields
#[case::where_nonexistent_field(
    vec![("name", STRING), ("age", INT)],
    "WHERE email == 'test@example.com'",
    1
)]
#[case::where_nonexistent_in_and(
    vec![("name", STRING)],
    "WHERE name == 'Alice' AND age > 18",
    1
)]
#[case::where_both_sides_nonexistent(
    vec![("name", STRING)],
    "WHERE email == phone",
    1
)]
// SET errors - non-existent fields
#[case::set_nonexistent_source(
    vec![("name", STRING)],
    "SET full_name = email",
    1
)]
#[case::set_nonexistent_in_expression(
    vec![("age", INT)],
    "SET result = age + missing_field",
    1
)]
#[case::set_multiple_missing(
    vec![("x", INT)],
    "SET result = x + y * z",
    1
)]
// AGG errors - non-existent fields
#[case::agg_nonexistent_field(
    vec![("amount", INT)],
    "AGG total = sum(missing_field)",
    1
)]
#[case::agg_nonexistent_group_by(
    vec![("amount", INT)],
    "AGG total = sum(amount) BY region",
    1
)]
#[case::agg_both_missing(
    vec![("price", INT)],
    "AGG total = sum(amount) BY region",
    1
)]
#[case::agg_bare_column_with_by(
    vec![("amount", DOUBLE), ("name", STRING)],
    "AGG amount BY name",
    1
)]
#[case::agg_bare_column_without_by(
    vec![("amount", DOUBLE)],
    "AGG amount",
    1
)]
// SORT errors - non-existent fields
#[case::sort_nonexistent_field(
    vec![("name", STRING)],
    "SORT age DESC",
    1
)]
#[case::sort_multiple_nonexistent(
    vec![("name", STRING)],
    "SORT age DESC, email ASC",
    1
)]
#[case::sort_mix_valid_invalid(
    vec![("name", STRING), ("age", INT)],
    "SORT name ASC, email DESC",
    1
)]
// DROP errors - non-existent fields
#[case::drop_nonexistent_field(
    vec![("name", STRING), ("age", INT)],
    "DROP email",
    1
)]
#[case::drop_multiple_nonexistent(
    vec![("name", STRING)],
    "DROP age, email",
    1
)]
#[case::drop_mix_valid_invalid(
    vec![("name", STRING), ("age", INT)],
    "DROP name, email",
    1
)]
// Function call errors
#[case::undefined_function(
    vec![("name", STRING)],
    "SELECT result = undefined_function(name)",
    1
)]
#[case::function_with_nonexistent_field(
    vec![("name", STRING)],
    "SELECT result = len(email)",
    1
)]
// Field access after DROP
#[case::dropped_field_in_select(
    vec![("name", STRING), ("age", INT), ("email", STRING)],
    "DROP email | SELECT name, email",
    1
)]
#[case::dropped_field_in_where(
    vec![("name", STRING), ("age", INT), ("active", BOOLEAN)],
    "DROP active | WHERE active == true",
    1
)]
// AGG schema transformation errors
#[case::non_grouped_field_after_agg(
    vec![("region", STRING), ("category", STRING), ("amount", INT)],
    "AGG total = sum(amount) BY region | SELECT region, category",
    1
)]
#[case::original_field_after_agg(
    vec![("region", STRING), ("amount", INT)],
    "AGG total = sum(amount) BY region | SELECT amount",
    1
)]
#[case::stddev_rejects_non_numeric(
    vec![("name", STRING)],
    "AGG bad_stddev = stddev(name)",
    1
)]
fn test_error_cases_with_set(
    #[case] input_fields: Vec<(&str, Type)>,
    #[case] command: &str,
    #[case] min_errors: usize,
) -> Result<(), TranslationErrors> {
    // Build a SET command that establishes the input schema
    let set_clause = input_fields
        .iter()
        .map(|(name, typ)| {
            let value = match typ {
                t if t == &STRING => "'dummy'",
                t if t == &INT => "0",
                t if t == &DOUBLE => "0.0",
                t if t == &BOOLEAN => "true",
                _ => "'unknown'",
            };
            format!("{} = {}", name, value)
        })
        .collect::<Vec<_>>()
        .join(", ");

    let query = format!("SET {} | {}", set_clause, command);
    let pipeline = Pipeline::parse_result(&query)?;

    let errors = type_check(pipeline).errors;
    assert!(errors.len() >= min_errors);
    Ok(())
}

// ======================== Error Tests with FROM ========================

#[rstest]
// FROM errors - non-existent tables
#[case::from_nonexistent_table(
    vec![],
    "FROM missing_table",
    1
)]
#[case::from_empty_catalog(
    vec![],
    "FROM users",
    1
)]
// Accessing replaced struct fields after struct literal assignment
#[case::access_replaced_field_after_struct_literal(
    vec![("events", vec![("id", INT)])],
    "FROM events | SET x.a = 1 | SET x = {y: 2} | SET z = x.a",
    1
)]
#[case::access_deep_replaced_field_after_struct_literal(
    vec![("events", vec![("id", INT)])],
    "FROM events | SET x.a.b = 1 | SET x = {y: 2} | SELECT x.a.b",
    1
)]
#[case::access_replaced_field_in_where(
    vec![("events", vec![("id", INT)])],
    "FROM events | SET x.a = 1 | SET x = {y: 2} | WHERE x.a > 0",
    1
)]
// Pipeline error propagation
#[case::error_in_first_command(
    vec![("users", vec![("name", STRING), ("age", INT)])],
    "FROM users | SELECT email",
    1
)]
#[case::error_in_middle_command(
    vec![("users", vec![("name", STRING), ("age", INT)])],
    "FROM users | WHERE email == 'test' | SELECT name",
    1
)]
#[case::error_in_last_command(
    vec![("users", vec![("name", STRING), ("age", INT)])],
    "FROM users | WHERE age > 18 | SELECT name, email",
    1
)]
#[case::multiple_errors_in_pipeline(
    vec![("users", vec![("name", STRING)])],
    "FROM users | WHERE age > 18 | SELECT name, email",
    2
)]
fn test_error_cases_with_from(
    #[case] tables: Vec<(&str, Vec<(&str, Type)>)>,
    #[case] query: &str,
    #[case] min_errors: usize,
) -> Result<(), TranslationErrors> {
    let pipeline = Pipeline::parse_result(query)?;

    let catalog = catalog_with_tables(tables);
    let errors = type_check_with_catalog(pipeline, catalog).errors;

    assert!(errors.len() >= min_errors);
    Ok(())
}

fn error_messages(errors: &TranslationErrors) -> Vec<&str> {
    errors
        .iter()
        .map(|error| error.primary.message.as_str())
        .collect()
}

#[test]
fn missing_source_suppresses_dependent_field_errors() -> Result<(), TranslationErrors> {
    let pipeline = Pipeline::parse_result("FROM missing | WHERE foo == 1 | SELECT bar, baz")?;
    let catalog = catalog_with_tables(vec![]);
    let errors = type_check_with_catalog(pipeline, catalog).errors;

    assert_eq!(error_messages(&errors), ["Dataset not found: test:missing"]);
    Ok(())
}

#[test]
fn complete_schema_keeps_independent_field_errors() -> Result<(), TranslationErrors> {
    let pipeline = Pipeline::parse_result("FROM users | WHERE age > 18 | SELECT email")?;
    let catalog = catalog_with_tables(vec![("users", vec![("name", STRING)])]);
    let errors = type_check_with_catalog(pipeline, catalog).errors;

    assert_eq!(
        error_messages(&errors),
        ["Field not found", "Field not found"]
    );
    Ok(())
}

#[test]
fn missing_join_source_does_not_hide_left_side_error() -> Result<(), TranslationErrors> {
    let pipeline =
        Pipeline::parse_result("FROM users | JOIN m = missing ON m.id == id | SELECT typo")?;
    let catalog = catalog_with_tables(vec![("users", vec![("id", INT)])]);
    let errors = type_check_with_catalog(pipeline, catalog).errors;

    assert_eq!(
        error_messages(&errors),
        ["Dataset not found: test:missing", "Field not found"]
    );
    Ok(())
}

#[test]
fn missing_join_alias_suppresses_dependent_errors() -> Result<(), TranslationErrors> {
    let pipeline =
        Pipeline::parse_result("FROM users | JOIN m = missing ON m.id == id | SELECT m.value")?;
    let catalog = catalog_with_tables(vec![("users", vec![("id", INT)])]);
    let errors = type_check_with_catalog(pipeline, catalog).errors;

    assert_eq!(error_messages(&errors), ["Dataset not found: test:missing"]);
    Ok(())
}

#[test]
fn missing_source_keeps_independent_expression_error() -> Result<(), TranslationErrors> {
    let pipeline = Pipeline::parse_result("FROM missing | SET x = foo, y = undefined_function(1)")?;
    let catalog = catalog_with_tables(vec![]);
    let errors = type_check_with_catalog(pipeline, catalog).errors;

    assert_eq!(
        error_messages(&errors),
        [
            "Dataset not found: test:missing",
            "no function by this name"
        ]
    );
    Ok(())
}

#[test]
fn missing_source_suppresses_dependent_shape_errors() -> Result<(), TranslationErrors> {
    let pipeline = Pipeline::parse_result("FROM missing | EXPLODE foo")?;
    let catalog = catalog_with_tables(vec![]);
    let errors = type_check_with_catalog(pipeline, catalog).errors;

    assert_eq!(error_messages(&errors), ["Dataset not found: test:missing"]);
    Ok(())
}

#[test]
fn missing_source_suppresses_dependent_unnest_shape_errors() -> Result<(), TranslationErrors> {
    let pipeline = Pipeline::parse_result("FROM missing | UNNEST foo | SELECT bar")?;
    let catalog = catalog_with_tables(vec![]);
    let errors = type_check_with_catalog(pipeline, catalog).errors;

    assert_eq!(error_messages(&errors), ["Dataset not found: test:missing"]);
    Ok(())
}

#[test]
fn invalid_unnest_expression_keeps_later_field_errors() -> Result<(), TranslationErrors> {
    let pipeline = Pipeline::parse_result("SET seed = 1 | UNNEST missing | SELECT other")?;
    let errors = type_check_with_catalog(pipeline, catalog_with_tables(vec![])).errors;

    assert_eq!(
        error_messages(&errors),
        ["Field not found", "Field not found"]
    );
    Ok(())
}

#[test]
fn drop_suppresses_a_dependent_lookup_and_closes_the_field() -> Result<(), TranslationErrors> {
    let pipeline = Pipeline::parse_result("FROM missing | DROP foo | SELECT foo")?;
    let catalog = catalog_with_tables(vec![]);
    let errors = type_check_with_catalog(pipeline, catalog).errors;

    assert_eq!(
        error_messages(&errors),
        ["Dataset not found: test:missing", "Field not found"]
    );
    Ok(())
}

#[test]
fn invalid_function_binding_becomes_a_new_dependency() -> Result<(), TranslationErrors> {
    let setup = Pipeline::parse_result("FROM missing | SET x = flatten(foo)")?;
    let setup_catalog = catalog_with_tables(vec![]);
    let setup_result = type_check_with_catalog(setup, setup_catalog);
    let setup_commands = setup_result.output.commands().unwrap();
    let TypedCommandKind::Set(set) = &setup_commands[1].kind else {
        panic!("expected SET command");
    };
    assert!(set.projections.assignments[0]
        .expression
        .upstream_error()
        .is_none());
    assert!(matches!(
        setup_result
            .output
            .environment()
            .resolve(&SimpleIdentifier::new("x").into()),
        FieldResolution::Found { dependencies, .. } if !dependencies.is_empty()
    ));

    let pipeline = Pipeline::parse_result("FROM missing | SET x = flatten(foo) | EXPLODE x")?;
    let catalog = catalog_with_tables(vec![]);
    let errors = type_check_with_catalog(pipeline, catalog).errors;

    assert_eq!(
        error_messages(&errors),
        [
            "Dataset not found: test:missing",
            "could not find a matching function definition"
        ]
    );
    Ok(())
}

#[rstest]
#[case::set("SET seed = 1 | SET x = missing | EXPLODE x | SELECT x.value")]
#[case::select("SET seed = 1 | SELECT x = missing | EXPLODE x | SELECT x.value")]
#[case::distinct("SET seed = 1 | DISTINCT x = missing | EXPLODE x")]
#[case::agg("SET seed = 1 | AGG x = sum(missing) | EXPLODE x")]
#[case::window(
    "SET timestamp = '2024-01-01' AS timestamp, seed = 1 | WINDOW x = sum(missing) | EXPLODE x"
)]
#[case::suppress(
    "SET timestamp = '2024-01-01' AS timestamp, seed = 1 | SUPPRESS 1m BY x = missing | EXPLODE x"
)]
fn projections_preserve_missing_field_dependencies(
    #[case] input: &str,
) -> Result<(), TranslationErrors> {
    let pipeline = Pipeline::parse_result(input)?;
    let errors = type_check_with_catalog(pipeline, catalog_with_tables(vec![])).errors;

    assert_eq!(error_messages(&errors), ["Field not found"]);
    Ok(())
}

#[test]
fn missing_function_suppresses_dependent_shape_errors() -> Result<(), TranslationErrors> {
    let pipeline = Pipeline::parse_result(
        "SET seed = 1 | SET x = unknown_func(1) | EXPLODE x | SELECT x.value",
    )?;
    let errors = type_check_with_catalog(pipeline, catalog_with_tables(vec![])).errors;

    assert_eq!(error_messages(&errors), ["no function by this name"]);
    Ok(())
}

#[test]
fn missing_function_keeps_independent_argument_error() -> Result<(), TranslationErrors> {
    let pipeline = Pipeline::parse_result("SET seed = 1 | SET x = unknown_func(missing)")?;
    let errors = type_check_with_catalog(pipeline, catalog_with_tables(vec![])).errors;

    assert_eq!(
        error_messages(&errors),
        ["Field not found", "no function by this name"]
    );
    Ok(())
}

#[test]
fn function_arity_error_stays_independent_of_argument_error() -> Result<(), TranslationErrors> {
    let pipeline = Pipeline::parse_result(
        "SET seed = 1 | SET x = len(missing, 1) | EXPLODE x | SELECT x.value",
    )?;
    let errors = type_check_with_catalog(pipeline, catalog_with_tables(vec![])).errors;

    assert_eq!(
        error_messages(&errors),
        [
            "Field not found",
            "could not find a matching function definition"
        ]
    );
    Ok(())
}

#[test]
fn unknown_named_argument_stays_independent_of_argument_error() -> Result<(), TranslationErrors> {
    let pipeline = Pipeline::parse_result("SET seed = 1 | SET x = lower(other = missing)")?;
    let errors = type_check_with_catalog(pipeline, catalog_with_tables(vec![])).errors;

    assert_eq!(
        error_messages(&errors),
        [
            "Field not found",
            "could not find a matching function definition"
        ]
    );
    Ok(())
}

#[rstest]
#[case::dependent_function(
    "SET seed = 1 | SET x = len(missing) | EXPLODE x",
    "Expected type array(any), got int"
)]
#[case::cast(
    "SET seed = 1 | SET x = missing AS int | EXPLODE x",
    "Expected type array(any), got int"
)]
fn valid_operations_with_known_result_types_keep_independent_shape_errors(
    #[case] input: &str,
    #[case] expected_shape_error: &str,
) -> Result<(), TranslationErrors> {
    let pipeline = Pipeline::parse_result(input)?;
    let errors = type_check_with_catalog(pipeline, catalog_with_tables(vec![])).errors;

    assert_eq!(errors.len(), 2, "{errors}");
    assert_eq!(error_messages(&errors).last(), Some(&expected_shape_error));
    Ok(())
}

#[test]
fn invalid_function_call_has_unknown_result_type() -> Result<(), TranslationErrors> {
    let pipeline = Pipeline::parse_result("SET seed = 1 | SET x = lower(1) | EXPLODE x")?;
    let errors = type_check_with_catalog(pipeline, catalog_with_tables(vec![])).errors;

    assert_eq!(
        error_messages(&errors),
        ["could not find a matching function definition"]
    );
    Ok(())
}

#[test]
fn nested_dependency_does_not_hide_sibling_field_error() -> Result<(), TranslationErrors> {
    let pipeline = Pipeline::parse_result(
        "SET s = { good: 1 } | SET s.bad = missing | SELECT good = s.good, other = s.other",
    )?;
    let errors = type_check_with_catalog(pipeline, catalog_with_tables(vec![])).errors;

    assert_eq!(
        error_messages(&errors),
        ["Field not found", "Field other not found in struct"]
    );
    Ok(())
}

#[test]
fn open_source_does_not_hide_known_struct_field_error() -> Result<(), TranslationErrors> {
    let pipeline = Pipeline::parse_result("FROM missing | SET s = { a: 1 } | SELECT value = s.b")?;
    let errors = type_check_with_catalog(pipeline, catalog_with_tables(vec![])).errors;

    assert_eq!(
        error_messages(&errors),
        [
            "Dataset not found: test:missing",
            "Field b not found in struct"
        ]
    );
    Ok(())
}

#[test]
fn nested_assignment_preserves_open_parent_scope() -> Result<(), TranslationErrors> {
    let pipeline = Pipeline::parse_result("FROM missing | SET x.a = 1 | SELECT value = x.b")?;
    let errors = type_check_with_catalog(pipeline, catalog_with_tables(vec![])).errors;

    assert_eq!(error_messages(&errors), ["Dataset not found: test:missing"]);
    Ok(())
}

#[test]
fn deep_assignment_preserves_each_open_parent_scope() -> Result<(), TranslationErrors> {
    let pipeline = Pipeline::parse_result(
        "FROM missing | SET x.a.b = 1 | SELECT first = x.other, second = x.a.other",
    )?;
    let errors = type_check_with_catalog(pipeline, catalog_with_tables(vec![])).errors;

    assert_eq!(error_messages(&errors), ["Dataset not found: test:missing"]);
    Ok(())
}

#[test]
fn nested_assignment_does_not_reopen_known_struct() -> Result<(), TranslationErrors> {
    let pipeline = Pipeline::parse_result(
        "FROM missing | SET s = { a: 1 } | SET s.b = 2 | SELECT value = s.c",
    )?;
    let errors = type_check_with_catalog(pipeline, catalog_with_tables(vec![])).errors;

    assert_eq!(
        error_messages(&errors),
        [
            "Dataset not found: test:missing",
            "Field c not found in struct"
        ]
    );
    Ok(())
}

#[test]
fn top_level_assignment_clears_descendant_open_scopes() -> Result<(), TranslationErrors> {
    let pipeline = Pipeline::parse_result(
        "FROM missing | SET b.x.y = 1 | SET b = { a: 1 } | SELECT value = b.x",
    )?;
    let errors = type_check_with_catalog(pipeline, catalog_with_tables(vec![])).errors;

    assert_eq!(
        error_messages(&errors),
        [
            "Dataset not found: test:missing",
            "Field x not found in struct"
        ]
    );
    Ok(())
}

#[test]
fn open_source_does_not_mask_scalar_def_fields() -> Result<(), TranslationErrors> {
    let query = Query::parse_result("DEF cfg = { a: 1 }; FROM missing | SELECT cfg.a")?;
    let errors = type_check_with_catalog(query, catalog_with_tables(vec![])).errors;

    assert_eq!(error_messages(&errors), ["Dataset not found: test:missing"]);
    Ok(())
}

#[test]
fn open_source_keeps_invalid_scalar_def_field_error() -> Result<(), TranslationErrors> {
    let query = Query::parse_result("DEF cfg = { a: 1 }; FROM missing | SELECT cfg.nope")?;
    let errors = type_check_with_catalog(query, catalog_with_tables(vec![])).errors;

    assert_eq!(
        error_messages(&errors),
        [
            "Dataset not found: test:missing",
            "Field nope not found in struct"
        ]
    );
    Ok(())
}

#[test]
fn join_preserves_right_side_schema_uncertainty() -> Result<(), TranslationErrors> {
    let query = Query::parse_result(
        "DEF b = FROM missing | SET x = 1; FROM users | JOIN b = b ON b.x == id | SELECT b.other",
    )?;
    let catalog = catalog_with_tables(vec![("users", vec![("id", INT)])]);
    let errors = type_check_with_catalog(query, catalog).errors;

    assert_eq!(error_messages(&errors), ["Dataset not found: test:missing"]);
    Ok(())
}

#[test]
fn aliased_from_preserves_nested_schema_uncertainty() -> Result<(), TranslationErrors> {
    let query = Query::parse_result(
        "DEF b = FROM missing | SET x = 1; FROM b = b | SELECT value = b.other",
    )?;
    let errors = type_check_with_catalog(query, catalog_with_tables(vec![])).errors;

    assert_eq!(error_messages(&errors), ["Dataset not found: test:missing"]);
    Ok(())
}

#[test]
fn aliased_from_prefixes_path_dependencies() -> Result<(), TranslationErrors> {
    let query = Query::parse_result(
        "DEF b = SET s = { good: 1 } | SET s.bad = missing; FROM b = b | EXPLODE b.s.bad",
    )?;
    let errors = type_check_with_catalog(query, catalog_with_tables(vec![])).errors;

    assert_eq!(error_messages(&errors), ["Field not found"]);
    Ok(())
}

#[rstest]
#[case::shadow("SET x = missing | SET x = 1 | EXPLODE x")]
#[case::drop("SET x = missing | DROP x | SELECT x")]
fn assignment_and_drop_clear_old_dependencies(
    #[case] input: &str,
) -> Result<(), TranslationErrors> {
    let pipeline = Pipeline::parse_result(input)?;
    let errors = type_check_with_catalog(pipeline, catalog_with_tables(vec![])).errors;

    assert_eq!(errors.len(), 2, "{errors}");
    Ok(())
}

#[rstest]
#[case::field_lookup("SET seed = 1 | SET x = missing.value | EXPLODE x", 1)]
#[case::index("SET seed = 1 | SET x = missing[0] | EXPLODE x", 1)]
#[case::index_argument("SET seed = 1 | SET x = missing[other] | EXPLODE x", 2)]
#[case::ts_trunc("SET seed = 1 | SET x = missing@m | EXPLODE x", 2)]
fn child_expressions_suppress_only_dependent_errors(
    #[case] input: &str,
    #[case] expected_errors: usize,
) -> Result<(), TranslationErrors> {
    let pipeline = Pipeline::parse_result(input)?;
    let errors = type_check_with_catalog(pipeline, catalog_with_tables(vec![])).errors;

    assert_eq!(errors.len(), expected_errors, "{errors}");
    Ok(())
}

#[test]
fn explode_declares_dependent_output_after_type_error() -> Result<(), TranslationErrors> {
    let pipeline = Pipeline::parse_result("SET seed = 1 | EXPLODE x = seed | SELECT x.value")?;
    let errors = type_check_with_catalog(pipeline, catalog_with_tables(vec![])).errors;

    assert_eq!(
        error_messages(&errors),
        ["Expected type array(any), got int"]
    );
    Ok(())
}

#[test]
fn explicit_projection_closes_an_open_schema() -> Result<(), TranslationErrors> {
    let pipeline = Pipeline::parse_result("FROM missing | SELECT one = 1 | SELECT typo")?;
    let catalog = catalog_with_tables(vec![]);
    let errors = type_check_with_catalog(pipeline, catalog).errors;

    assert_eq!(
        error_messages(&errors),
        ["Dataset not found: test:missing", "Field not found"]
    );
    Ok(())
}

#[test]
fn failed_def_preserves_schema_uncertainty() -> Result<(), TranslationErrors> {
    let query =
        Query::parse_result("DEF broken = FROM missing | SELECT foo; FROM broken | SELECT foo")?;
    let catalog = catalog_with_tables(vec![]);
    let errors = type_check_with_catalog(query, catalog).errors;

    assert_eq!(error_messages(&errors), ["Dataset not found: test:missing"]);
    Ok(())
}

#[rstest]
#[case::parse("FROM missing | PARSE foo '*' parsed | SELECT other")]
#[case::unnest("FROM missing | SET s = {a: 1} | UNNEST s | SELECT other")]
#[case::nest("FROM missing | NEST nested | SELECT nested.other")]
#[case::lambda("FROM missing | SET x = transform([1], n -> foo)")]
fn schema_rebuilds_preserve_uncertainty(#[case] input: &str) -> Result<(), TranslationErrors> {
    let query = Query::parse_result(input)?;
    let errors = type_check_with_catalog(query, catalog_with_tables(vec![])).errors;

    assert_eq!(error_messages(&errors), ["Dataset not found: test:missing"]);
    Ok(())
}

#[rstest]
#[case::missing_field("SELECT event.missing_field", "event.missing_field")]
#[case::valid_sibling("SELECT event.category, event.missing_field", "event.missing_field")]
#[case::nested_field("SELECT event.details.missing_field", "event.details.missing_field")]
#[case::quoted_field("SELECT event.`missing field`", "event.`missing field`")]
#[case::where_expression("WHERE event.missing_field == 'test'", "event.missing_field")]
fn test_missing_struct_field_error_span(
    #[case] command: &str,
    #[case] expected_highlight: &str,
) -> Result<(), TranslationErrors> {
    let query = format!("FROM signals | {}", command);
    let catalog = catalog_with_tables(vec![(
        "signals",
        vec![(
            "event",
            Struct::default()
                .with_str("category", STRING)
                .with_str("details", Struct::default().with_str("kind", STRING).into())
                .into(),
        )],
    )]);
    let pipeline = Pipeline::parse_result(&query)?;
    let errors = type_check_with_catalog(pipeline, catalog).errors;
    eprintln!("{}", errors.clone().contextualize(query.clone()));

    assert_eq!(errors.len(), 1);
    let error = &errors.0[0];
    assert!(error.primary.message.contains("not found in struct"));
    assert_eq!(&query[error.primary.interval.clone()], expected_highlight);
    Ok(())
}

// ======================== Type Error Resilience Tests ========================

#[rstest]
// Invalid binary operations
#[case::string_add_boolean("'hello' + true", "could not find a matching function")]
#[case::string_multiply_int("'hello' * 5", "could not find a matching function")]
#[case::boolean_add_int("true + 5", "could not find a matching function")]
#[case::string_subtract("'hello' - 'world'", "could not find a matching function")]
// Invalid function arguments
#[case::len_int("len(42)", "matching function definition")]
#[case::len_boolean("len(true)", "matching function definition")]
#[case::upper_int("upper(42)", "matching function definition")]
#[case::abs_string("abs('hello')", "matching function definition")]
#[case::round_string("round('3.14')", "matching function definition")]
// Unknown functions
#[case::unknown_function("unknown_func(42)", "no function by this name")]
#[case::undefined_function("not_a_function('test')", "no function by this name")]
// Mixed-type array errors
#[case::mixed_array_int_string("[1, 'two', 3]", "could not determine array element type")]
#[case::mixed_array_bool_int("[true, 1, false]", "could not determine array element type")]
// flatten rejects bad input types
#[case::flatten_non_array_of_array("flatten([1, 2, 3])", "matching function definition")]
#[case::flatten_non_array("flatten('not an array')", "matching function definition")]
// IN/NOT IN array type mismatch
#[case::in_array_type_mismatch("'hello' IN [1, 2]", "type")]
#[case::not_in_array_type_mismatch("'hello' NOT IN [1, 2]", "type")]
// IN/NOT IN tuple type mismatch
#[case::in_tuple_type_mismatch("'hello' IN (1, 2)", "type")]
#[case::in_tuple_mixed_types("'hello' IN (1, 'foo')", "type")]
// IN/NOT IN range type mismatch
#[case::in_range_type_mismatch("'hello' IN 1..2", "type")]
#[case::not_in_range_type_mismatch("'hello' NOT IN 1..2", "type")]
// Range construction errors
#[case::range_int_string("1..'hello'", "matching function definition")]
#[case::range_int_double("1..1.5", "matching function definition")]
#[case::range_int_interval("1..-1h", "matching function definition")]
// Inclusive range construction errors
#[case::range_inclusive_int_string("1..='hello'", "matching function definition")]
#[case::range_inclusive_int_interval("1..=-1h", "matching function definition")]
// if() type errors
#[case::if_non_bool_condition("if('hello', 0, 0)", "matching function definition")]
#[case::if_non_bool_condition_two_args("if('hello', 0)", "matching function definition")]
#[case::if_branch_type_mismatch("if(true, 'hello', 1)", "matching function definition")]
// case() type errors
#[case::case_branch_type_mismatch("case(true: 'hello', false: 0)", "type")]
#[case::case_non_bool_condition(
    "case('hello': 'hello', false: 'hello')",
    "matching function definition"
)]
#[case::case_all_non_bool(
    "case('hello': 'hello', 'hello': 'hello')",
    "matching function definition"
)]
// coalesce() type errors
#[case::coalesce_type_mismatch("coalesce(1, 'hello')", "type")]
// IP function type errors
#[case::cidr_contains_int_first_arg(
    "cidr_contains(123, '192.168.1.100')",
    "matching function definition"
)]
#[case::cidr_contains_int_second_arg(
    "cidr_contains('192.168.1.0/24', 456)",
    "matching function definition"
)]
#[case::is_ipv4_int_arg("is_ipv4(123)", "matching function definition")]
#[case::is_ipv6_int_arg("is_ipv6(456)", "matching function definition")]
fn test_type_errors(#[case] input: &str, #[case] expected_error: &str) {
    let result = type_check_expression(
        Expression::parse_result(input).unwrap(),
        ExpressionTypeCheckOptions::builder().build(),
    )
    .into_result();
    assert!(
        result.is_err(),
        "Expected error but got success for: {}",
        input
    );

    let err = result.unwrap_err();
    let err_string = err.to_string();

    // Print the error for debugging (shows ariadne formatted output)
    eprintln!("Error for '{}': {}", input, err_string);

    assert!(
        err_string.contains(expected_error),
        "Error '{}' does not contain expected fragment '{}'",
        err_string,
        expected_error
    );
}

// ======================== Duplicate Field Error Tests ========================

#[rstest]
// SELECT duplicate fields
#[case::select_duplicate_same_type(
    vec![("x", INT), ("y", INT)],
    "SELECT a = x, a = y",
    1
)]
#[case::select_duplicate_diff_type(
    vec![("x", INT), ("y", STRING)],
    "SELECT a = x, a = y",
    1
)]
#[case::select_nested_struct_conflict(
    vec![("x", INT), ("y", STRING)],
    "SET result = {x: x, y: y} | SELECT result, result.x = 'conflict'",
    1
)]
// SET duplicate fields
#[case::set_duplicate_same_type(
    vec![("x", INT), ("y", INT)],
    "SET a = x, a = y",
    1
)]
#[case::set_duplicate_diff_type(
    vec![("x", INT), ("y", STRING)],
    "SET a = x, a = y",
    1
)]
// WINDOW duplicate fields
#[case::window_duplicate_same_type(
    vec![("x", INT)],
    "WINDOW a = count(), a = sum(x)",
    1
)]
// AGG duplicate fields
#[case::agg_duplicate_same_type(
    vec![("x", INT), ("y", STRING)],
    "AGG a = count(), a = sum(x) BY y",
    1
)]
#[case::agg_duplicate_groupby_conflict(
    vec![("x", INT), ("y", STRING)],
    "AGG x = count() BY x",
    1
)]
// SET prefix clashes: scalar then child
#[case::set_prefix_scalar_then_child(
    vec![("x", INT), ("y", INT)],
    "SET a = x, a.b = y",
    1
)]
// SET prefix clashes: child then scalar
#[case::set_prefix_child_then_scalar(
    vec![("x", INT), ("y", INT)],
    "SET a.b = x, a = y",
    1
)]
// SET prefix clashes: deep nesting
#[case::set_prefix_deep_nesting(
    vec![("x", INT), ("y", INT)],
    "SET a.b = x, a.b.c = y",
    1
)]
// WINDOW prefix clash
#[case::window_prefix_clash(
    vec![("x", INT)],
    "WINDOW a = count(), a.b = sum(x)",
    1
)]
// SELECT prefix clash: scalar then child
#[case::select_prefix_scalar_then_child(
    vec![("x", INT), ("y", INT)],
    "SELECT a = x, a.b = y",
    1
)]
// PARSE duplicate fields
#[case::parse_duplicate_fields(
    vec![("msg", STRING)],
    "PARSE msg '*-*' a, a",
    1
)]
// Struct literal duplicate fields
#[case::struct_literal_duplicate_field(
    vec![("x", INT), ("y", INT)],
    "SET s = {a: x, a: y}",
    1
)]
fn test_duplicate_field_errors(
    #[case] input_fields: Vec<(&str, Type)>,
    #[case] pipeline: &str,
    #[case] min_errors: usize,
) -> Result<(), TranslationErrors> {
    // Build a SET command that establishes the input schema
    let set_clause = input_fields
        .iter()
        .map(|(name, typ)| {
            let value = match typ {
                t if t == &STRING => "'dummy'",
                t if t == &INT => "0",
                t if t == &DOUBLE => "0.0",
                t if t == &BOOLEAN => "true",
                _ => "'unknown'",
            };
            format!("{} = {}", name, value)
        })
        .collect::<Vec<_>>()
        .join(", ");

    let query = format!("SET {} | {}", set_clause, pipeline);
    let parsed = Pipeline::parse_result(&query)?;

    let errors = type_check(parsed).errors;
    assert!(
        errors.len() >= min_errors,
        "Expected at least {} errors, got {}",
        min_errors,
        errors.len()
    );
    Ok(())
}

// ======================== Misplaced Side-Effect Command Tests ========================

#[rstest]
// APPEND in non-final position
#[case::append_then_select(
    vec![("events", vec![("id", INT), ("name", STRING)])],
    "FROM events | APPEND output_table | SELECT id",
    "APPEND command must be the final command"
)]
#[case::append_then_where(
    vec![("events", vec![("id", INT), ("name", STRING)])],
    "FROM events | APPEND output_table | WHERE id > 0",
    "APPEND command must be the final command"
)]
#[case::append_in_middle(
    vec![("events", vec![("id", INT), ("name", STRING)])],
    "FROM events | APPEND output_table | SET x = 1 | SELECT id",
    "APPEND command must be the final command"
)]
fn test_misplaced_side_effect_commands(
    #[case] tables: Vec<(&str, Vec<(&str, Type)>)>,
    #[case] query: &str,
    #[case] expected_error: &str,
) -> Result<(), TranslationErrors> {
    let pipeline = Pipeline::parse_result(query)?;

    let catalog = catalog_with_tables(tables);
    let errors = type_check_with_catalog::<Pipeline>(Arc::new(pipeline), catalog).errors;

    // Print errors for debugging
    eprintln!("Errors: {}", errors);

    assert!(
        !errors.is_empty(),
        "Expected at least one error for misplaced side-effect command"
    );

    let error_string = errors.to_string();
    assert!(
        error_string.contains(expected_error),
        "Error '{}' does not contain expected fragment '{}'",
        error_string,
        expected_error
    );

    Ok(())
}

// ======================== AGG validation (valid cases) ========================

#[test]
fn agg_allows_scalar_def_outside_aggregate() {
    let input = "DEF multiplier = 2;\nSET amount = 1, region = 'r'\n| AGG total = sum(amount) * multiplier BY region";
    let (query, parse_errors) = Query::parse_with_errors(input);
    assert!(parse_errors.is_empty(), "{parse_errors:#?}");
    let result = type_check::<Query>(Arc::new(query));
    assert!(result.errors.is_empty(), "{}", result.errors);
}

#[test]
fn agg_allows_lambda_param_in_body_under_aggregate() {
    let input = "SET items = [1, 2], grp = 1\n| AGG result = array_agg(transform(items, x -> x + 1)) BY grp";
    let (query, parse_errors) = Query::parse_with_errors(input);
    assert!(parse_errors.is_empty(), "{parse_errors:#?}");
    let result = type_check::<Query>(Arc::new(query));
    assert!(result.errors.is_empty(), "{}", result.errors);
}

#[test]
fn agg_allows_outer_lambda_param_in_nested_lambda_body_under_aggregate() {
    let input = "SET items = [1, 2], grp = 1\n| AGG result = array_agg(transform(items, outer -> transform([outer], inner -> outer + inner))) BY grp";
    let (query, parse_errors) = Query::parse_with_errors(input);
    assert!(parse_errors.is_empty(), "{parse_errors:#?}");
    let result = type_check::<Query>(Arc::new(query));
    assert!(result.errors.is_empty(), "{}", result.errors);
}

// ======================== Expression Parse Error Tests ========================

#[rstest]
#[case::in_empty_tuple("'hello' IN ()")]
fn test_expression_parse_errors(#[case] input: &str) {
    assert!(
        Expression::parse_result(input).is_err(),
        "Expected parse error for: {}",
        input
    );
}

// ======================== Type Argument Tests ========================

#[rstest]
// Exact argument counts
#[case::array("array(int)", Array::new(INT).into())]
#[case::range("range(int)", Range::new(INT).into())]
#[case::range_inclusive("range_inclusive(int)", Type::RangeInclusive(Range::new(INT)))]
#[case::map("map(string, int)", Map::new(STRING, INT).into())]
#[case::decimal("decimal(10, 2)", Decimal::new(10, 2).unwrap().into())]
// Trailing commas
#[case::array_trailing_comma("array(int,)", Array::new(INT).into())]
#[case::range_trailing_comma("range(int,)", Range::new(INT).into())]
#[case::range_inclusive_trailing_comma(
    "range_inclusive(int,)",
    Type::RangeInclusive(Range::new(INT))
)]
#[case::map_trailing_comma("map(string, int,)", Map::new(STRING, INT).into())]
#[case::decimal_trailing_comma("decimal(10, 2,)", Decimal::new(10, 2).unwrap().into())]
// Case-insensitive type names
#[case::array_uppercase("ARRAY(int)", Array::new(INT).into())]
#[case::decimal_uppercase("DECIMAL(10, 2)", Decimal::new(10, 2).unwrap().into())]
// Nested parameterized types
#[case::nested(
    "map(string, array(range(int)))",
    Map::new(STRING, Array::new(Range::new(INT).into()).into()).into()
)]
fn test_type_arguments(#[case] typ: &str, #[case] expected_type: Type) {
    let input = format!("null AS {typ}");
    let parsed = Expression::parse_result(&input);
    if let Err(errors) = &parsed {
        eprintln!("{}", errors.clone().contextualize(input.clone()));
    }
    let typed = type_check_expression(
        parsed.unwrap(),
        ExpressionTypeCheckOptions::builder().build(),
    );
    eprintln!("{}", typed.errors.clone().contextualize(input.clone()));

    assert!(typed.errors.is_empty());
    assert_eq!(*typed.output.resolved_type, expected_type);
}

#[rstest]
// No arguments: rejected by the grammar
#[case::array_none("array()", &["invalid type expression", "no viable alternative at input 'array()'"])]
#[case::range_none("range()", &["invalid type expression", "no viable alternative at input 'range()'"])]
#[case::range_inclusive_none(
    "range_inclusive()",
    &["invalid type expression", "no viable alternative at input 'range_inclusive()'"]
)]
#[case::map_none("map()", &["invalid type expression", "no viable alternative at input 'map()'"])]
#[case::decimal_none("decimal()", &["invalid type expression", "no viable alternative at input 'decimal()'"])]
// Too few arguments
#[case::map_one("map(string)", &["map type takes 2 type arguments, got 1"])]
#[case::decimal_one("decimal(10)", &["decimal type takes 2 arguments (precision, scale), got 1"])]
// Too many arguments
#[case::array_two("array(int, string)", &["array type takes 1 type argument, got 2"])]
#[case::array_three("array(int, string, boolean)", &["array type takes 1 type argument, got 3"])]
#[case::range_two("range(int, string)", &["range type takes 1 type argument, got 2"])]
#[case::range_three("range(int, int, int)", &["range type takes 1 type argument, got 3"])]
#[case::range_inclusive_two(
    "range_inclusive(int, string)",
    &["range_inclusive type takes 1 type argument, got 2"]
)]
#[case::range_inclusive_three(
    "range_inclusive(int, int, int)",
    &["range_inclusive type takes 1 type argument, got 3"]
)]
#[case::map_three("map(string, int, boolean)", &["map type takes 2 type arguments, got 3"])]
#[case::map_four("map(string, int, boolean, double)", &["map type takes 2 type arguments, got 4"])]
#[case::decimal_three("decimal(10, 2, 5)", &["decimal type takes 2 arguments (precision, scale), got 3"])]
#[case::decimal_four("decimal(1, 2, 3, 4)", &["decimal type takes 2 arguments (precision, scale), got 4"])]
// Too many arguments with a trailing comma
#[case::array_two_trailing_comma("array(int, string,)", &["array type takes 1 type argument, got 2"])]
#[case::decimal_three_trailing_comma(
    "decimal(10, 2, 5,)",
    &["decimal type takes 2 arguments (precision, scale), got 3"]
)]
// Case-insensitive type names
#[case::array_uppercase_two("ARRAY(int, string)", &["array type takes 1 type argument, got 2"])]
// Nested types are validated too
#[case::nested_array_two("array(array(int, string))", &["array type takes 1 type argument, got 2"])]
#[case::nested_map_value_one("map(string, map(int))", &["map type takes 2 type arguments, got 1"])]
// Integer arguments where type arguments are expected
#[case::array_integer("array(10)", &["array type takes type arguments, not integers"])]
#[case::range_integer("range(1, 2)", &["range type takes type arguments, not integers"])]
#[case::range_inclusive_integer(
    "range_inclusive(1)",
    &["range_inclusive type takes type arguments, not integers"]
)]
#[case::map_integer("map(1, 2)", &["map type takes type arguments, not integers"])]
// Type arguments where integer arguments are expected
#[case::decimal_one_type(
    "decimal(int)",
    &["decimal type takes integer arguments (precision, scale), not types"]
)]
#[case::decimal_two_types(
    "decimal(int, int)",
    &["decimal type takes integer arguments (precision, scale), not types"]
)]
// Unknown type names
#[case::unknown_with_types("foo(int)", &["Unknown hamelin type: foo"])]
#[case::unknown_with_integers("foo(1)", &["Unknown hamelin type: foo"])]
fn test_type_argument_errors(#[case] typ: &str, #[case] expected_errors: &[&str]) {
    let input = format!("1 AS {typ}");
    let errors = Expression::parse_result(&input).expect_err("expected a parse error");
    eprintln!("{}", errors.clone().contextualize(input.clone()));

    assert_eq!(error_messages(&errors), expected_errors);
}

// ======================== Parse Error Pipeline Tests ========================
//
// When a query has parse errors (QueryKind::Error), the resulting TypedStatement
// should have a pipeline with kind: Err, not kind: Ok(ValidPipeline { commands: vec![] }).
// This maintains the invariant that Ok(ValidPipeline) always has at least one command.

#[rstest]
#[case::garbage_input("!@#$%")]
#[case::empty_input("")]
fn test_parse_error_query_produces_error_pipeline(#[case] input: &str) {
    let (query, _parse_errors) = Query::parse_with_errors(input);
    let typed = type_check::<Query>(Arc::new(query)).output;

    assert!(
        typed.pipeline.kind.is_err(),
        "Expected pipeline.kind to be Err for parse-error query '{}', but got Ok",
        input
    );
}