hamelin_analysis 0.9.4

//! Tests for error cases where incremental refresh is not supported

use rstest::rstest;
use std::collections::HashMap;

use crate::incremental::{compute_incremental_ranges_for_pipeline, TimeRange};
use hamelin_lib::tree::{
    ast::identifier::Identifier,
    builder::{call, eq, field_ref, hours, pipeline, string, PipelineBuilder},
};

use super::helpers::{build_pipeline, stale_ranges, timestamp_field};

#[rstest]
#[case::drop_timestamp(
    pipeline()
        .from(|f| f.table_reference("events"))
        .drop(|d| d.field("timestamp")),
    stale_ranges("events", "2024-01-01 14:30:00", "2024-01-01 16:45:00"),
    "TimestampLineageError"
)]
#[case::select_without_timestamp(
    pipeline()
        .from(|f| f.table_reference("events"))
        .select(|s| s.field("user").field("severity")),
    stale_ranges("events", "2024-01-01 14:30:00", "2024-01-01 16:45:00"),
    "TimestampLineageError"
)]
#[case::agg_without_timestamp_groupby(
    pipeline()
        .from(|f| f.table_reference("events"))
        .agg(|a| a
            .named_aggregate("count", call("count"))
            .named_group("user_id", field_ref("user.id"))
        ),
    stale_ranges("events", "2024-01-01 14:30:00", "2024-01-01 16:45:00"),
    "AggWithoutTimestampGroupBy"
)]
#[case::parse_timestamp_field(
    pipeline()
        .from(|f| f.table_reference("events"))
        .parse(|p| p
            .pattern(string(r"(\d+)"))
            .identifier("timestamp")
            .source(field_ref("timestamp"))
        ),
    stale_ranges("events", "2024-01-01 14:30:00", "2024-01-01 16:45:00"),
    "TimestampLineageError"
)]
#[case::explode_timestamp_field(
    pipeline()
        .from(|f| f.table_reference("events"))
        .explode(|e| e.named_field("timestamp", field_ref("tags"))),
    stale_ranges("events", "2024-01-01 14:30:00", "2024-01-01 16:45:00"),
    "TimestampLineageError"
)]
#[case::nest_into_timestamp(
    pipeline()
        .from(|f| f.table_reference("events"))
        .nest("timestamp"),
    stale_ranges("events", "2024-01-01 14:30:00", "2024-01-01 16:45:00"),
    "TimestampLineageError"
)]
#[case::join_not_supported(
    pipeline()
        .from(|f| f.table_reference("events"))
        .join("users", eq(field_ref("events.user_id"), field_ref("users.id"))),
    stale_ranges("events", "2024-01-01 14:30:00", "2024-01-01 16:45:00"),
    "CommandNotSupported"
)]
// TODO: Re-enable when we have DML builder support
// #[case::append_not_supported(
//     pipeline()
//         .from(|f| f.table_reference("events"))
//         .append("other_events", |a| a),
//     stale_ranges("events", "2024-01-01 14:30:00", "2024-01-01 16:45:00"),
//     "CommandNotSupported"
// )]
#[case::lookup_not_supported(
    pipeline()
        .from(|f| f.table_reference("events"))
        .lookup("users", |l| l.on(eq(field_ref("events.user_id"), field_ref("users.id")))),
    stale_ranges("events", "2024-01-01 14:30:00", "2024-01-01 16:45:00"),
    "CommandNotSupported"
)]
#[case::non_deterministic_in_where(
    pipeline()
        .from(|f| f.table_reference("events"))
        .where_cmd(eq(field_ref("timestamp"), call("now"))),
    stale_ranges("events", "2024-01-01 14:30:00", "2024-01-01 16:45:00"),
    "NonDeterministicFunction"
)]
#[case::non_deterministic_in_set(
    pipeline()
        .from(|f| f.table_reference("events"))
        .set_cmd(|l| l.named_field("current_time", call("now"))),
    stale_ranges("events", "2024-01-01 14:30:00", "2024-01-01 16:45:00"),
    "NonDeterministicFunction"
)]
#[case::non_deterministic_in_select(
    pipeline()
        .from(|f| f.table_reference("events"))
        .select(|s| s.field("timestamp").named_field("today", call("today"))),
    stale_ranges("events", "2024-01-01 14:30:00", "2024-01-01 16:45:00"),
    "NonDeterministicFunction"
)]
#[case::non_deterministic_in_window(
    pipeline()
        .from(|f| f.table_reference("events"))
        .window(|w| w.named_field("current", call("yesterday"))),
    stale_ranges("events", "2024-01-01 14:30:00", "2024-01-01 16:45:00"),
    "NonDeterministicFunction"
)]
#[case::non_deterministic_in_agg(
    pipeline()
        .from(|f| f.table_reference("events"))
        .agg(|a| a
            .named_aggregate("count", call("count"))
            .named_aggregate("as_of", call("tomorrow"))
            .group_by("timestamp")
        ),
    stale_ranges("events", "2024-01-01 14:30:00", "2024-01-01 16:45:00"),
    "NonDeterministicFunction"
)]
#[case::non_deterministic_in_within_positive_interval(
    pipeline()
        .from(|f| f.table_reference("events"))
        .within(hours(1)),
    stale_ranges("events", "2024-01-01 14:30:00", "2024-01-01 16:45:00"),
    "NonDeterministicFunction"
)]
#[case::non_deterministic_in_within_negative_interval(
    pipeline()
        .from(|f| f.table_reference("events"))
        .within(hours(-1)),
    stale_ranges("events", "2024-01-01 14:30:00", "2024-01-01 16:45:00"),
    "NonDeterministicFunction"
)]
fn test_error_cases(
    #[case] pipeline_builder: PipelineBuilder,
    #[case] stale_ranges_map: HashMap<Identifier, Option<TimeRange>>,
    #[case] expected_error_variant: &str,
) {
    let pipeline = build_pipeline(pipeline_builder);
    let result = compute_incremental_ranges_for_pipeline(
        &pipeline,
        stale_ranges_map,
        &timestamp_field(),
        None,
        false,
    );

    assert!(result.is_err(), "Expected error, but got success");
    let error = result.unwrap_err();

    // Check that the error matches the expected variant
    let error_string = format!("{:?}", error);
    assert!(
        error_string.contains(expected_error_variant),
        "Expected error variant '{}', but got: {:?}",
        expected_error_variant,
        error
    );
}