diff --git a/integration_tests/tests/test_anomaly_test_configuration.py b/integration_tests/tests/test_anomaly_test_configuration.py index 38d5b25ab..126c909a0 100644 --- a/integration_tests/tests/test_anomaly_test_configuration.py +++ b/integration_tests/tests/test_anomaly_test_configuration.py @@ -62,6 +62,8 @@ class ParamValues(Generic[T]): ), "exclude_final_results": ParamValues(*(["1 = 1"] * 3)), "min_value": ParamValues(0.5, 1.0, 2.0), + # Only dimension_anomalies supports it, so it is ignored here (and must not raise). + "include_current_bucket": ParamValues(True, True, True), } @@ -92,6 +94,7 @@ def get_value(key: str): "exclude_final_results": get_value("exclude_final_results"), "min_value": get_value("min_value"), "exclude_detection_period_from_training": None, + "include_current_bucket": None, } diff --git a/integration_tests/tests/test_dimension_anomalies.py b/integration_tests/tests/test_dimension_anomalies.py index 0ded83187..fc2247c39 100644 --- a/integration_tests/tests/test_dimension_anomalies.py +++ b/integration_tests/tests/test_dimension_anomalies.py @@ -1,6 +1,6 @@ import json from datetime import date, datetime, timedelta -from typing import Any, Dict, List +from typing import Any, Dict, List, Optional import dateutil.parser import pytest @@ -436,3 +436,340 @@ def test_buckets_stay_aligned_on_rerun( ) == 1 ) + + +def _generate_current_bucket_data(utc_today: date, today_superheroes: List[str]): + """One Superman and one Spiderman per day until yesterday, and today_superheroes today. + + Today is the bucket that is still in progress when the test runs. + """ + data: List[Dict[str, Any]] = [ + {TIMESTAMP_COLUMN: cur_date.strftime(DATE_FORMAT), "superhero": superhero} + for cur_date in generate_dates(base_date=utc_today - timedelta(1)) + for superhero in ["Superman", "Spiderman"] + ] + data += [ + {TIMESTAMP_COLUMN: utc_today.strftime(DATE_FORMAT), "superhero": superhero} + for superhero in today_superheroes + ] + return data + + +def test_include_current_bucket_detects_anomaly_in_current_bucket( + test_id: str, dbt_project: DbtProject +): + utc_today = datetime.utcnow().date() + data = _generate_current_bucket_data( + utc_today, ["Superman", "Superman", "Superman", "Spiderman"] + ) + + # By default only complete buckets are tested, so today's anomaly is not detected yet. + test_result = dbt_project.test(test_id, DBT_TEST_NAME, DBT_TEST_ARGS, data=data) + assert test_result["status"] == "pass" + + test_args = {**DBT_TEST_ARGS, "include_current_bucket": True} + test_result = dbt_project.test(test_id, DBT_TEST_NAME, test_args, data=data) + assert test_result["status"] == "fail" + + anomaly_test_points = get_latest_anomaly_test_points(dbt_project, test_id) + anomalous_points = [x for x in anomaly_test_points if x["is_anomalous"]] + assert set(x["dimension_value"] for x in anomalous_points) == {"Superman"} + assert all( + x["bucket_start"].startswith(utc_today.strftime("%Y-%m-%d")) + for x in anomalous_points + ) + + +def test_include_current_bucket_passes_on_normal_current_bucket( + test_id: str, dbt_project: DbtProject +): + data = _generate_current_bucket_data( + datetime.utcnow().date(), ["Superman", "Spiderman"] + ) + test_args = {**DBT_TEST_ARGS, "include_current_bucket": True} + test_result = dbt_project.test(test_id, DBT_TEST_NAME, test_args, data=data) + assert test_result["status"] == "pass" + + +def test_include_current_bucket_ignores_dimension_not_arrived_yet( + test_id: str, dbt_project: DbtProject +): + """A dimension without rows in the current bucket has not arrived yet, so it is not a drop to zero.""" + data = _generate_current_bucket_data(datetime.utcnow().date(), ["Spiderman"]) + test_args = {**DBT_TEST_ARGS, "include_current_bucket": True} + test_result = dbt_project.test(test_id, DBT_TEST_NAME, test_args, data=data) + assert test_result["status"] == "pass" + + +def test_include_current_bucket_ignores_deleted_rows( + test_id: str, dbt_project: DbtProject +): + """Rows deleted from the current bucket are not scored from the metrics of an earlier run.""" + utc_today = datetime.utcnow().date() + test_args = {**DBT_TEST_ARGS, "include_current_bucket": True} + + data = _generate_current_bucket_data( + utc_today, ["Superman", "Superman", "Superman", "Spiderman"] + ) + test_result = dbt_project.test(test_id, DBT_TEST_NAME, test_args, data=data) + assert test_result["status"] == "fail" + + # Today's Superman rows were a bad upload and got deleted, so Superman has not arrived yet. + data = _generate_current_bucket_data(utc_today, ["Spiderman"]) + test_result = dbt_project.test(test_id, DBT_TEST_NAME, test_args, data=data) + assert test_result["status"] == "pass" + + +def test_include_current_bucket_requires_timestamp_column( + test_id: str, dbt_project: DbtProject +): + data = _generate_current_bucket_data(datetime.utcnow().date(), ["Superman"]) + test_args = {"dimensions": ["superhero"], "include_current_bucket": True} + test_result = dbt_project.test(test_id, DBT_TEST_NAME, test_args, data=data) + assert test_result["status"] == "error" + + +@pytest.mark.parametrize("as_model", [False, True], ids=["source", "model"]) +def test_include_current_bucket_with_multi_hour_buckets( + test_id: str, dbt_project: DbtProject, as_model: bool +): + """4-hour buckets, so the current bucket end is calculated with count > 1. + + A source takes the incremental path. A table model recalculates all buckets, like force_metrics_backfill. + """ + test_id = test_id.replace("[", "_").replace("]", "_") + # In the middle of a bucket, so the current bucket cannot end while the test runs. + run_at = datetime.utcnow().replace( + hour=10, minute=0, second=0, microsecond=0 + ) - timedelta(days=1) + current_bucket_start = run_at.replace(hour=8) + data: List[Dict[str, Any]] = [ + {TIMESTAMP_COLUMN: bucket_start.strftime(DATE_FORMAT), "superhero": superhero} + for bucket_start in generate_dates( + base_date=current_bucket_start - timedelta(hours=4), + step=timedelta(hours=4), + days_back=7, + ) + for superhero in ["Superman", "Spiderman"] + ] + data += [ + { + TIMESTAMP_COLUMN: current_bucket_start.strftime(DATE_FORMAT), + "superhero": superhero, + } + for superhero in ["Superman", "Superman", "Superman", "Spiderman"] + ] + test_args = {**DBT_TEST_ARGS, "time_bucket": {"period": "hour", "count": 4}} + test_vars = {"custom_run_started_at": run_at.isoformat()} + + test_result = dbt_project.test( + test_id, + DBT_TEST_NAME, + test_args, + data=data, + as_model=as_model, + test_vars=test_vars, + ) + assert test_result["status"] == "pass" + + test_args["include_current_bucket"] = True + test_result = dbt_project.test( + test_id, + DBT_TEST_NAME, + test_args, + data=data, + as_model=as_model, + test_vars=test_vars, + ) + assert test_result["status"] == "fail" + + anomaly_test_points = get_latest_anomaly_test_points(dbt_project, test_id) + anomalous_points = [x for x in anomaly_test_points if x["is_anomalous"]] + assert set(x["dimension_value"] for x in anomalous_points) == {"Superman"} + assert all( + x["bucket_start"][:13].replace("T", " ") + == current_bucket_start.strftime("%Y-%m-%d %H") + for x in anomalous_points + ) + + +@pytest.mark.skip_targets(["redshift"]) +def test_include_current_bucket_monthly_snapshot_upload( + test_id: str, dbt_project: DbtProject +): + """Monthly uploads per country: a short upload is detected in the month it lands.""" + current_month, *previous_months = _get_week_or_month_starts("month", 12) + + rows_per_upload = {"PL": 100, "MY": 20} + data: List[Dict[str, Any]] = [ + {TIMESTAMP_COLUMN: month_start.strftime(DATE_FORMAT), "country": country} + for month_start in previous_months + for country, rows in rows_per_upload.items() + for _ in range(rows) + ] + # This month PL delivered a normal upload and MY a short one. + data += [ + {TIMESTAMP_COLUMN: current_month.strftime(DATE_FORMAT), "country": country} + for country, rows in {"PL": 100, "MY": 10}.items() + for _ in range(rows) + ] + + test_args = { + "timestamp_column": TIMESTAMP_COLUMN, + "dimensions": ["country"], + "time_bucket": {"period": "month", "count": 1}, + "training_period": {"period": "day", "count": 400}, + "detection_period": {"period": "day", "count": 31}, + } + + # By default this month is only tested once it has ended. + test_result = dbt_project.test(test_id, DBT_TEST_NAME, test_args, data=data) + assert test_result["status"] == "pass" + + test_args["include_current_bucket"] = True + test_result = dbt_project.test(test_id, DBT_TEST_NAME, test_args, data=data) + assert test_result["status"] == "fail" + + anomaly_test_points = get_latest_anomaly_test_points(dbt_project, test_id) + anomalous_points = [x for x in anomaly_test_points if x["is_anomalous"]] + assert set(x["dimension_value"] for x in anomalous_points) == {"MY"} + + +def _get_week_or_month_starts( + period: str, count: int, base_date: Optional[date] = None +) -> List[date]: + """The start of the week or month of base_date (default today), followed by the starts of the `count` before it.""" + base_date = base_date or datetime.utcnow().date() + if period == "week": + current_week_start = base_date - timedelta(days=base_date.weekday()) + return [current_week_start - timedelta(weeks=i) for i in range(count + 1)] + bucket_starts = [base_date.replace(day=1)] + for _ in range(count): + previous = bucket_starts[-1] - timedelta(days=1) + bucket_starts.append(previous.replace(day=1)) + return bucket_starts + + +@pytest.mark.parametrize( + "period,first_run_at,run_at", + [ + ("week", "2026-10-07T12:00:00", "2026-10-14T12:00:00"), + # October has more days than the 30 backfill_days of monthly buckets. + ("month", "2026-09-10T12:00:00", "2026-10-15T12:00:00"), + ], + ids=["week", "month"], +) +def test_include_current_bucket_tests_previous_bucket( + test_id: str, + dbt_project: DbtProject, + target: str, + period: str, + first_run_at: str, + run_at: str, +): + """MY skipped the previous bucket. Testing the current bucket must not stop testing the previous one.""" + if target == "redshift" and period == "month": + pytest.skip("Redshift does not support monthly time buckets.") + test_id = test_id.replace("[", "_").replace("]", "_") + current_bucket, previous_bucket, *older_buckets = _get_week_or_month_starts( + period, 12, base_date=datetime.fromisoformat(run_at).date() + ) + rows_per_bucket = {bucket: {"PL": 100, "MY": 20} for bucket in older_buckets} + rows_per_bucket[previous_bucket] = {"PL": 100} + rows_per_bucket[current_bucket] = {"PL": 100, "MY": 20} + data: List[Dict[str, Any]] = [ + {TIMESTAMP_COLUMN: bucket_start.strftime(DATE_FORMAT), "country": country} + for bucket_start, rows_per_country in rows_per_bucket.items() + for country, rows in rows_per_country.items() + for _ in range(rows) + ] + test_args = { + "timestamp_column": TIMESTAMP_COLUMN, + "dimensions": ["country"], + "time_bucket": {"period": period, "count": 1}, + "training_period": {"period": "day", "count": 400}, + "include_current_bucket": True, + } + + # A run during the previous bucket stores MY in the metrics history, so the next run + # fills in a zero for MY once the previous bucket is complete. The dates stay valid once + # they have passed: the first run's snapshot then counts as complete (updated_at is the + # real clock), but the previous bucket is inside the second run's backfill window, so + # it is recalculated anyway. + first_result = dbt_project.test( + test_id, + DBT_TEST_NAME, + test_args, + data=data, + test_vars={"custom_run_started_at": first_run_at}, + ) + assert first_result["status"] == "pass" + test_result = dbt_project.test( + test_id, + DBT_TEST_NAME, + test_args, + test_vars={"custom_run_started_at": run_at}, + ) + assert test_result["status"] == "fail" + + # Only checks that the bucket is complete, since some adapters start weeks on Sunday. + anomaly_test_points = get_latest_anomaly_test_points(dbt_project, test_id) + anomalous_points = [x for x in anomaly_test_points if x["is_anomalous"]] + assert set(x["dimension_value"] for x in anomalous_points) == {"MY"} + assert all(x["bucket_end"][:10] <= run_at[:10] for x in anomalous_points) + + +@pytest.mark.skip_targets(["redshift"]) +def test_include_current_bucket_recalculates_partial_bucket( + test_id: str, dbt_project: DbtProject +): + """A partial snapshot is recalculated once its bucket has ended, even when that is outside backfill_days.""" + current_month, *previous_months = _get_week_or_month_starts("month", 12) + next_month = (current_month + timedelta(days=32)).replace(day=1) + rows_per_bucket = {month: {"PL": 100, "MY": 20} for month in previous_months} + rows_per_bucket[current_month] = {"PL": 100} # MY never arrives this month + rows_per_bucket[next_month] = {"PL": 100, "MY": 20} + data: List[Dict[str, Any]] = [ + {TIMESTAMP_COLUMN: bucket_start.strftime(DATE_FORMAT), "country": country} + for bucket_start, rows_per_country in rows_per_bucket.items() + for country, rows in rows_per_country.items() + for _ in range(rows) + ] + test_args = { + "timestamp_column": TIMESTAMP_COLUMN, + "dimensions": ["country"], + "time_bucket": {"period": "month", "count": 1}, + "training_period": {"period": "day", "count": 400}, + "include_current_bucket": True, + } + + # Runs now, so it stores a partial snapshot of this month before the month ends. + dbt_project.test(test_id, DBT_TEST_NAME, test_args, data=data) + # The next run is two months later, when this month is outside the backfill window. + later_run_at = (next_month + timedelta(days=32)).replace(day=10) + dbt_project.test( + test_id, + DBT_TEST_NAME, + test_args, + test_vars={"custom_run_started_at": f"{later_run_at.isoformat()}T12:00:00"}, + ) + + results = dbt_project.run_query( + f""" + with my_metrics as ( + select + bucket_start, + metric_value, + row_number() over (partition by id order by updated_at desc) as row_num + from {{{{ ref("data_monitoring_metrics") }}}} + where dimension_value = 'MY' and lower(full_table_name) like '%{test_id.lower()}' + ) + select bucket_start, metric_value from my_metrics where row_num = 1 + """ + ) + current_month_values = [ + result["metric_value"] + for result in results + if dateutil.parser.parse(result["bucket_start"]).date() == current_month + ] + assert current_month_values == [0] diff --git a/macros/edr/data_monitoring/anomaly_detection/get_anomaly_scores_query.sql b/macros/edr/data_monitoring/anomaly_detection/get_anomaly_scores_query.sql index 3e76e3dab..412a309f9 100644 --- a/macros/edr/data_monitoring/anomaly_detection/get_anomaly_scores_query.sql +++ b/macros/edr/data_monitoring/anomaly_detection/get_anomaly_scores_query.sql @@ -73,6 +73,15 @@ {%- set min_bucket_start_expr = elementary.get_trunc_min_bucket_start_expr( detection_end, metric_properties, test_configuration.days_back ) %} + {# With include_current_bucket, the bucket that is still in progress is scored as well #} + {%- if test_configuration.include_current_bucket %} + {%- set buckets_end_expr = elementary.get_current_bucket_end_expr( + detection_end_expr, + min_bucket_start_expr, + metric_properties.time_bucket, + ) %} + {%- else %} {%- set buckets_end_expr = detection_end_expr %} + {%- endif %} {# Calculate detection period start for exclusion logic. backfill_days defines the window of recent data to test for anomalies on each run. @@ -103,9 +112,9 @@ with buckets as ( select edr_bucket_start, edr_bucket_end from ({{ elementary.complete_buckets_cte(metric_properties, min_bucket_start_expr, - detection_end_expr) }}) results + buckets_end_expr) }}) results where edr_bucket_start >= {{ min_bucket_start_expr }} - and edr_bucket_end <= {{ detection_end_expr }} + and edr_bucket_end <= {{ buckets_end_expr }} ), {% else %} with @@ -131,6 +140,11 @@ -- We use bucket_end because non-timestamp tests have only bucket_end field. where bucket_end > {{ min_bucket_start_expr }} + {% if test_configuration.include_current_bucket %} + -- The bucket that is still in progress is recalculated on every run, so its metrics only + -- come from this run. The ones stored by earlier runs can be stale. + and bucket_end <= {{ detection_end_expr }} + {% endif %} {% if test_configuration.timestamp_column %} -- For timestamped tests, verify that the bucket we got from the history is actually -- a valid one (this check is important for buckets that are not aligned with a day). diff --git a/macros/edr/data_monitoring/data_monitors_configuration/get_buckets_configuration.sql b/macros/edr/data_monitoring/data_monitors_configuration/get_buckets_configuration.sql index c5d89158c..3c4bd16f5 100644 --- a/macros/edr/data_monitoring/data_monitors_configuration/get_buckets_configuration.sql +++ b/macros/edr/data_monitoring/data_monitors_configuration/get_buckets_configuration.sql @@ -23,6 +23,29 @@ {{ return(trunc_min_bucket_start_expr) }} {% endmacro %} +{# The end of the bucket that contains detection_end, i.e. the bucket that is still in progress. + min_bucket_start_expr must be aligned to the time bucket (see get_trunc_min_bucket_start_expr). #} +{% macro get_current_bucket_end_expr( + detection_end_expr, min_bucket_start_expr, time_bucket +) %} + {%- set time_bucket = elementary.get_datediff_time_bucket(time_bucket) -%} + {%- set periods_until_current_bucket_end -%} + (floor({{ elementary.edr_datediff(min_bucket_start_expr, detection_end_expr, time_bucket.period) }} / {{ time_bucket.count }}) + 1) * {{ time_bucket.count }} + {%- endset -%} + {# edr_timeadd returns a date for week/month on some adapters (e.g. BigQuery), so cast back #} + {{ + return( + elementary.edr_cast_as_timestamp( + elementary.edr_timeadd( + time_bucket.period, + elementary.edr_cast_as_int(periods_until_current_bucket_end), + min_bucket_start_expr, + ) + ) + ) + }} +{% endmacro %} + {# This macro cant be used without truncating to full buckets #} {% macro get_backfill_bucket_start(detection_end, backfill_days) %} {% do return( @@ -42,7 +65,8 @@ column_name=none, metric_properties=none, unit_test=false, - unit_test_relation=none + unit_test_relation=none, + include_current_bucket=false ) %} {%- set detection_end = elementary.get_detection_end(detection_delay) %} @@ -52,6 +76,16 @@ {%- set trunc_min_bucket_start_expr = elementary.get_trunc_min_bucket_start_expr( detection_end, metric_properties, days_back ) %} + {# By default only complete buckets are monitored. With include_current_bucket, + the bucket that is still in progress is monitored as well. #} + {%- if include_current_bucket %} + {%- set buckets_end_expr = elementary.get_current_bucket_end_expr( + detection_end_expr, + trunc_min_bucket_start_expr, + metric_properties.time_bucket, + ) %} + {%- else %} {%- set buckets_end_expr = detection_end_expr %} + {%- endif %} {%- set backfill_bucket_start = elementary.get_start_bucket_in_data( elementary.edr_cast_as_timestamp( elementary.edr_datetime_to_sql( @@ -92,7 +126,7 @@ ), full_buckets_calc as ( select *, - floor({{ elementary.edr_datediff('days_back_start', 'detection_end', datediff_time_bucket.period) }} / {{ datediff_time_bucket.count }}) * {{ datediff_time_bucket.count }} as periods_until_max + (floor({{ elementary.edr_datediff('days_back_start', 'detection_end', datediff_time_bucket.period) }} / {{ datediff_time_bucket.count }}){% if include_current_bucket %} + 1{% endif %}) * {{ datediff_time_bucket.count }} as periods_until_max from bucket_times ) select @@ -105,15 +139,21 @@ {%- set incremental_bucket_times_query %} with all_buckets as ( select edr_bucket_start as bucket_start, edr_bucket_end as bucket_end - from ({{ elementary.complete_buckets_cte(metric_properties, trunc_min_bucket_start_expr, detection_end_expr) }}) results + from ({{ elementary.complete_buckets_cte(metric_properties, trunc_min_bucket_start_expr, buckets_end_expr) }}) results where edr_bucket_start >= {{ trunc_min_bucket_start_expr }} - and edr_bucket_end <= {{ detection_end_expr }} + and edr_bucket_end <= {{ buckets_end_expr }} ), buckets_with_existing_metrics as ( select distinct bucket_start, bucket_end from {{ data_monitoring_metrics_relation }} where bucket_start >= {{ trunc_min_bucket_start_expr }} + {# The current bucket is never treated as cached, so it is recalculated on every run #} and bucket_end <= {{ detection_end_expr }} + {# With include_current_bucket, only metrics calculated after their bucket ended are complete. + A metric of the bucket that was still in progress is recalculated once it ends. #} + {%- if include_current_bucket %} + and {{ elementary.edr_cast_as_timestamp('updated_at') }} >= {{ elementary.edr_cast_as_timestamp('bucket_end') }} + {%- endif %} and upper(full_table_name) = upper('{{ full_table_name }}') and metric_properties = {{ elementary.dict_to_quoted_json(metric_properties) }} {%- if metric_names %} @@ -139,7 +179,7 @@ ) select min(bucket_start) as min_bucket_start, - {{ detection_end_expr }} as max_bucket_end + {{ buckets_end_expr }} as max_bucket_end from min_bucket_start_candidates {% endset %} diff --git a/macros/edr/data_monitoring/monitors_query/dimension_monitoring_query.sql b/macros/edr/data_monitoring/monitors_query/dimension_monitoring_query.sql index ceb80546a..231c2997c 100644 --- a/macros/edr/data_monitoring/monitors_query/dimension_monitoring_query.sql +++ b/macros/edr/data_monitoring/monitors_query/dimension_monitoring_query.sql @@ -5,7 +5,8 @@ min_bucket_start, max_bucket_end, metric_properties, - metric_name=none + metric_name=none, + zero_fill_max_bucket_end=none ) %} {% set metric_name = metric_name or "dimension" %} {% set full_table_name_str = elementary.edr_quote( @@ -105,6 +106,19 @@ buckets.joiner = training_set_dimensions.joiner {# This makes sure we dont create empty buckets for dimensions before their first appearance #} and edr_bucket_end >= dimension_min_bucket_end + {# With include_current_bucket, the bucket in progress ends after zero_fill_max_bucket_end. + A dimension without rows in it has not arrived yet rather than dropped to zero, + so only complete buckets are filled. #} + {% if zero_fill_max_bucket_end %} + and edr_bucket_end + <= {{ + elementary.edr_cast_as_timestamp( + elementary.edr_datetime_to_sql( + zero_fill_max_bucket_end + ) + ) + }} + {% endif %} ) where dimension_value is not null ), diff --git a/macros/edr/tests/test_configuration/get_anomalies_test_configuration.sql b/macros/edr/tests/test_configuration/get_anomalies_test_configuration.sql index e94fb564e..a3c6e159f 100644 --- a/macros/edr/tests/test_configuration/get_anomalies_test_configuration.sql +++ b/macros/edr/tests/test_configuration/get_anomalies_test_configuration.sql @@ -25,7 +25,9 @@ training_period, exclude_final_results, min_value, - exclude_detection_period_from_training + exclude_detection_period_from_training, + include_current_bucket=none, + supports_current_bucket=false ) %} {%- set model_graph_node = elementary.get_model_graph_node(model_relation) %} @@ -126,6 +128,20 @@ model_graph_node, ) %} + {# Only resolved for tests that support it, so setting it in vars or on a model + does not affect (or break) the other anomaly tests #} + {% if supports_current_bucket %} + {% set include_current_bucket = elementary.get_test_argument( + "include_current_bucket", include_current_bucket, model_graph_node + ) %} + {% if include_current_bucket and not metric_props.timestamp_column %} + {% do exceptions.raise_compiler_error( + "include_current_bucket requires a timestamp_column" + ) %} + {% endif %} + {% else %} {% set include_current_bucket = none %} + {% endif %} + {% set test_configuration = { "timestamp_column": metric_props.timestamp_column, "where_expression": metric_props.where_expression, @@ -145,6 +161,7 @@ "anomaly_exclude_metrics": anomaly_exclude_metrics, "exclude_final_results": exclude_final_results, "exclude_detection_period_from_training": exclude_detection_period_from_training, + "include_current_bucket": include_current_bucket, } %} {%- set test_configuration = elementary.undefined_dict_keys_to_none( test_configuration diff --git a/macros/edr/tests/test_dimension_anomalies.sql b/macros/edr/tests/test_dimension_anomalies.sql index 00177425a..fef1d747c 100644 --- a/macros/edr/tests/test_dimension_anomalies.sql +++ b/macros/edr/tests/test_dimension_anomalies.sql @@ -19,7 +19,8 @@ training_period, exclude_final_results, min_value=none, - exclude_detection_period_from_training=none + exclude_detection_period_from_training=none, + include_current_bucket=none ) %} {{ config(tags=["elementary-tests"]) }} {%- if execute and elementary.is_test_command() and elementary.is_elementary_enabled() %} @@ -84,6 +85,8 @@ exclude_final_results=exclude_final_results, min_value=min_value, exclude_detection_period_from_training=exclude_detection_period_from_training, + include_current_bucket=include_current_bucket, + supports_current_bucket=true, ) ) %} @@ -105,6 +108,7 @@ days_back=test_configuration.days_back, detection_delay=test_configuration.detection_delay, metric_properties=metric_properties, + include_current_bucket=test_configuration.include_current_bucket, ) ) %} @@ -119,6 +123,13 @@ min_bucket_start, max_bucket_end, metric_properties, + zero_fill_max_bucket_end=( + elementary.get_detection_end( + test_configuration.detection_delay + ) + if test_configuration.include_current_bucket + else none + ), ) %} {{ elementary.debug_log( diff --git a/macros/edr/tests/test_utils/get_anomaly_query.sql b/macros/edr/tests/test_utils/get_anomaly_query.sql index be42c4029..c59668656 100644 --- a/macros/edr/tests/test_utils/get_anomaly_query.sql +++ b/macros/edr/tests/test_utils/get_anomaly_query.sql @@ -97,6 +97,23 @@ }} {%- endif %} {%- set backfill_period = "'-" ~ test_configuration.backfill_days ~ "'" %} + {# The detection window ends at the last complete bucket. With include_current_bucket, the bucket + that is still in progress is tested on top of that window instead of shifting it forward. #} + {%- set max_bucket_end_expr = ( + "max(bucket_end) over (partition by test_execution_id)" + ) %} + {%- if test_configuration.include_current_bucket %} + {%- set detection_end_expr = elementary.edr_cast_as_timestamp( + elementary.edr_datetime_to_sql( + elementary.get_detection_end( + test_configuration.detection_delay + ) + ) + ) %} + {%- set max_bucket_end_expr -%} + coalesce(max(case when bucket_end <= {{ detection_end_expr }} then bucket_end end) over (partition by test_execution_id), {{ detection_end_expr }}) + {%- endset -%} + {%- endif %} {%- set anomaly_query -%} with anomaly_scores as ( @@ -126,7 +143,7 @@ dimension, dimension_value, {{ elementary.anomaly_detection_description() }}, - max(bucket_end) over (partition by test_execution_id) as max_bucket_end + {{ max_bucket_end_expr }} as max_bucket_end from {{ elementary.get_elementary_test_table(elementary.get_elementary_test_table_name(), 'anomaly_scores') }} ), anomaly_scores_with_is_anomalous as (