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::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"));
}
#[rstest]
#[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
)]
#[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
)]
#[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
)]
#[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
)]
#[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
)]
#[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
)]
#[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
)]
#[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
)]
#[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> {
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(())
}
#[rstest]
#[case::from_nonexistent_table(
vec![],
"FROM missing_table",
1
)]
#[case::from_empty_catalog(
vec![],
"FROM users",
1
)]
#[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
)]
#[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(())
}
#[rstest]
#[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")]
#[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")]
#[case::unknown_function("unknown_func(42)", "no function by this name")]
#[case::undefined_function("not_a_function('test')", "no function by this name")]
#[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")]
#[case::flatten_non_array_of_array("flatten([1, 2, 3])", "matching function definition")]
#[case::flatten_non_array("flatten('not an array')", "matching function definition")]
#[case::in_array_type_mismatch("'hello' IN [1, 2]", "type")]
#[case::not_in_array_type_mismatch("'hello' NOT IN [1, 2]", "type")]
#[case::in_tuple_type_mismatch("'hello' IN (1, 2)", "type")]
#[case::in_tuple_mixed_types("'hello' IN (1, 'foo')", "type")]
#[case::in_range_type_mismatch("'hello' IN 1..2", "type")]
#[case::not_in_range_type_mismatch("'hello' NOT IN 1..2", "type")]
#[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")]
#[case::range_inclusive_int_string("1..='hello'", "matching function definition")]
#[case::range_inclusive_int_interval("1..=-1h", "matching function definition")]
#[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::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"
)]
#[case::coalesce_type_mismatch("coalesce(1, 'hello')", "type")]
#[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();
eprintln!("Error for '{}': {}", input, err_string);
assert!(
err_string.contains(expected_error),
"Error '{}' does not contain expected fragment '{}'",
err_string,
expected_error
);
}
#[rstest]
#[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
)]
#[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
)]
#[case::window_duplicate_same_type(
vec![("x", INT)],
"WINDOW a = count(), a = sum(x)",
1
)]
#[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
)]
#[case::set_prefix_scalar_then_child(
vec![("x", INT), ("y", INT)],
"SET a = x, a.b = y",
1
)]
#[case::set_prefix_child_then_scalar(
vec![("x", INT), ("y", INT)],
"SET a.b = x, a = y",
1
)]
#[case::set_prefix_deep_nesting(
vec![("x", INT), ("y", INT)],
"SET a.b = x, a.b.c = y",
1
)]
#[case::window_prefix_clash(
vec![("x", INT)],
"WINDOW a = count(), a.b = sum(x)",
1
)]
#[case::select_prefix_scalar_then_child(
vec![("x", INT), ("y", INT)],
"SELECT a = x, a.b = y",
1
)]
#[case::parse_duplicate_fields(
vec![("msg", STRING)],
"PARSE msg '*-*' a, a",
1
)]
#[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> {
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(())
}
#[rstest]
#[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;
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(())
}
#[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);
}
#[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
);
}
#[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
);
}