hamelin_analysis 0.8.0

Analysis utilities for Hamelin query language
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119

//! Tests for strategy detection (which incremental strategies are supported)

use rstest::rstest;

use crate::incremental::{detect_supported_strategies_for_pipeline, IncrementalStrategyKind};
use hamelin_lib::tree::builder::{
    at_hour, call, eq, field_ref, hours, pipeline, string, PipelineBuilder,
};

use super::helpers::build_pipeline;

#[rstest]
#[case::simple_filter(
    pipeline()
        .from(|f| f.table_reference("events"))
        .where_cmd(eq(field_ref("severity"), string("high"))),
    vec![IncrementalStrategyKind::CascadedAppend, IncrementalStrategyKind::TimeRangeRefresh]
)]
#[case::set_and_select(
    pipeline()
        .from(|f| f.table_reference("events"))
        .set_cmd(|l| l.named_field("risk_score", field_ref("severity")))
        .select(|s| s.field("timestamp").field("user").field("risk_score")),
    vec![IncrementalStrategyKind::CascadedAppend, IncrementalStrategyKind::TimeRangeRefresh]
)]
#[case::drop_non_timestamp(
    pipeline()
        .from(|f| f.table_reference("events"))
        .drop(|d| d.field("extra_field")),
    vec![IncrementalStrategyKind::CascadedAppend, IncrementalStrategyKind::TimeRangeRefresh]
)]
#[case::window_disqualifies_cascaded_append(
    pipeline()
        .from(|f| f.table_reference("events"))
        .window(|w| w
            .named_field("count", call("count"))
            .within(hours(-1))
        ),
    vec![IncrementalStrategyKind::TimeRangeRefresh]
)]
#[case::agg_disqualifies_cascaded_append(
    pipeline()
        .from(|f| f.table_reference("events"))
        .agg(|a| a
            .named_aggregate("count", call("count"))
            .named_group("timestamp", at_hour(field_ref("timestamp")))
        ),
    vec![IncrementalStrategyKind::TimeRangeRefresh]
)]
#[case::parse_disqualifies_cascaded_append(
    pipeline()
        .from(|f| f.table_reference("events"))
        .parse(|p| p
            .pattern("*")
            .identifier("entire_event")
        ),
    vec![IncrementalStrategyKind::TimeRangeRefresh]
)]
#[case::explode_allows_both_strategies(
    pipeline()
        .from(|f| f.table_reference("events"))
        .explode(|e| e.named_field("tag", field_ref("tags"))),
    vec![IncrementalStrategyKind::CascadedAppend, IncrementalStrategyKind::TimeRangeRefresh]
)]
// TODO: Fix UNNEST - may need schema or builder adjustments
// #[case::unnest_allows_both_strategies(
//     pipeline()
//         .from(|f| f.table_reference("events"))
//         .unnest(field_ref("tags")),
//     vec![IncrementalStrategyKind::CascadedAppend, IncrementalStrategyKind::TimeRangeRefresh]
// )]
#[case::nest_disqualifies_cascaded_append(
    pipeline()
        .from(|f| f.table_reference("events"))
        .nest("user_info"),
    vec![IncrementalStrategyKind::TimeRangeRefresh]
)]
#[case::non_deterministic_disqualifies_all_strategies_in_where(
    pipeline()
        .from(|f| f.table_reference("events"))
        .where_cmd(eq(field_ref("timestamp"), call("now"))),
    vec![]
)]
#[case::non_deterministic_disqualifies_all_strategies_in_set(
    pipeline()
        .from(|f| f.table_reference("events"))
        .set_cmd(|l| l.named_field("current_time", call("today"))),
    vec![]
)]
#[case::non_deterministic_disqualifies_all_strategies_in_window(
    pipeline()
        .from(|f| f.table_reference("events"))
        .window(|w| w.named_field("current", call("yesterday"))),
    vec![]
)]
#[case::non_deterministic_disqualifies_all_strategies_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")
        ),
    vec![]
)]
#[case::non_deterministic_disqualifies_all_strategies_in_within(
    pipeline()
        .from(|f| f.table_reference("events"))
        .within(hours(1)),
    vec![]
)]
fn test_strategy_detection(
    #[case] pipeline_builder: PipelineBuilder,
    #[case] expected_strategies: Vec<IncrementalStrategyKind>,
) {
    let pipeline = build_pipeline(pipeline_builder);
    let result = detect_supported_strategies_for_pipeline(pipeline.into(), false);
    assert_eq!(result.supported, expected_strategies);
}