From 5eca4e57eebbf6470f33f7d6a0d9351f870c5f53 Mon Sep 17 00:00:00 2001 From: Joost Boonzajer Flaes Date: Wed, 30 Sep 2026 16:04:43 +0300 Subject: [PATCH 1/9] feat: add include_current_bucket to dimension_anomalies Anomaly tests only score complete time buckets, so data that lands once per bucket (e.g. a monthly upload per country) is only checked after the bucket has ended. With include_current_bucket: true, the bucket that is still in progress is scored on every run as well. Dimension values without rows in the current bucket are not zero-filled, since they have not arrived yet. The current bucket is never treated as cached, so it is recalculated each run. Also align the backfill start to its bucket in get_metric_buckets_min_and_max. For sources and incremental models the backfill start was a plain date, so buckets that are not a whole number of days (months, weeks) were generated from a mid-bucket start on reruns, producing empty buckets and false failures. Co-Authored-By: Claude Opus 5.5 --- .../tests/test_anomaly_test_configuration.py | 1 + .../tests/test_dimension_anomalies.py | 135 ++++++++++++++++++ .../get_anomaly_scores_query.sql | 13 +- .../get_buckets_configuration.sql | 45 +++++- .../dimension_monitoring_query.sql | 14 +- .../get_anomalies_test_configuration.sql | 13 +- macros/edr/tests/test_dimension_anomalies.sql | 9 +- 7 files changed, 219 insertions(+), 11 deletions(-) diff --git a/integration_tests/tests/test_anomaly_test_configuration.py b/integration_tests/tests/test_anomaly_test_configuration.py index 89fcd3291..4a9b3f597 100644 --- a/integration_tests/tests/test_anomaly_test_configuration.py +++ b/integration_tests/tests/test_anomaly_test_configuration.py @@ -91,6 +91,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 51d7a05b0..ba7a4c40d 100644 --- a/integration_tests/tests/test_dimension_anomalies.py +++ b/integration_tests/tests/test_dimension_anomalies.py @@ -315,3 +315,138 @@ def test_anomaly_in_detection_period( ) assert test_result["status"] == expected_status + + +def _generate_current_bucket_data(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. + """ + utc_today = datetime.utcnow().date() + 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 +): + data = _generate_current_bucket_data( + ["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(datetime.utcnow().date().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(["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(["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_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_start = datetime.utcnow().date().replace(day=1) + month_starts = [current_month_start] + for _ in range(12): + previous = month_starts[-1] - timedelta(days=1) + month_starts.append(previous.replace(day=1)) + current_month, *previous_months = month_starts + + 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 test_monthly_buckets_stay_aligned_on_rerun(test_id: str, dbt_project: DbtProject): + """On a rerun the backfill window starts mid-month; buckets must still be calendar months.""" + current_month_start = datetime.utcnow().date().replace(day=1) + month_starts = [current_month_start] + for _ in range(12): + previous = month_starts[-1] - timedelta(days=1) + month_starts.append(previous.replace(day=1)) + + data: List[Dict[str, Any]] = [ + {TIMESTAMP_COLUMN: month_start.strftime(DATE_FORMAT), "country": country} + for month_start in month_starts[1:] + for country, rows in {"PL": 100, "MY": 20}.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}, + } + + first_result = dbt_project.test(test_id, DBT_TEST_NAME, test_args, data=data) + assert first_result["status"] == "pass" + second_result = dbt_project.test(test_id, DBT_TEST_NAME, test_args, data=data) + assert second_result["status"] == "pass" 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..62aea5913 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 and test_configuration.timestamp_column %} + {%- 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 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 cbbba17d1..e8e1f33a8 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,25 @@ {{ 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 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 -%} + {{ + return( + 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 +61,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 +72,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.edr_cast_as_timestamp( elementary.edr_datetime_to_sql( elementary.get_backfill_bucket_start( @@ -85,7 +115,7 @@ ), full_buckets_calc as ( select *, - floor({{ elementary.edr_datediff('days_back_start', 'detection_end', metric_properties.time_bucket.period) }} / {{ metric_properties.time_bucket.count }}) * {{ metric_properties.time_bucket.count }} as periods_until_max + (floor({{ elementary.edr_datediff('days_back_start', 'detection_end', metric_properties.time_bucket.period) }} / {{ metric_properties.time_bucket.count }}){% if include_current_bucket %} + 1{% endif %}) * {{ metric_properties.time_bucket.count }} as periods_until_max from bucket_times ) select @@ -98,14 +128,15 @@ {%- 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 }} and upper(full_table_name) = upper('{{ full_table_name }}') and metric_properties = {{ elementary.dict_to_quoted_json(metric_properties) }} @@ -128,11 +159,13 @@ min_bucket_start_candidates as ( select bucket_start from missing_bucket_starts union all - select {{ backfill_bucket_start }} as bucket_start + {# Align the backfill start to the start of its bucket. Otherwise buckets that are + not a whole number of days (e.g. months) are generated from a mid-bucket start. #} + select {{ elementary.get_start_bucket_in_data(backfill_bucket_start, trunc_min_bucket_start_expr, metric_properties.time_bucket) }} as bucket_start ) 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..796bd93ff 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,9 @@ min_bucket_start, max_bucket_end, metric_properties, - metric_name=none + metric_name=none, + include_current_bucket=false, + detection_end=none ) %} {% set metric_name = metric_name or "dimension" %} {% set full_table_name_str = elementary.edr_quote( @@ -105,6 +107,16 @@ 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 + {# The current bucket is still in progress, so a dimension without rows in it + has not arrived yet rather than dropped to zero. Only complete buckets are filled. #} + {% if include_current_bucket and detection_end %} + and edr_bucket_end + <= {{ + elementary.edr_cast_as_timestamp( + elementary.edr_datetime_to_sql(detection_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..36d47ce73 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,8 @@ training_period, exclude_final_results, min_value, - exclude_detection_period_from_training + exclude_detection_period_from_training, + include_current_bucket=none ) %} {%- set model_graph_node = elementary.get_model_graph_node(model_relation) %} @@ -126,6 +127,15 @@ model_graph_node, ) %} + {% 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 %} + {% set test_configuration = { "timestamp_column": metric_props.timestamp_column, "where_expression": metric_props.where_expression, @@ -145,6 +155,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..eb096117d 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,7 @@ 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, ) ) %} @@ -105,6 +107,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 +122,10 @@ min_bucket_start, max_bucket_end, metric_properties, + include_current_bucket=test_configuration.include_current_bucket, + detection_end=elementary.get_detection_end( + test_configuration.detection_delay + ), ) %} {{ elementary.debug_log( From 5be5f753270b20b1a718d9d87dc8161366fd27ac Mon Sep 17 00:00:00 2001 From: Joost Boonzajer Flaes Date: Wed, 30 Sep 2026 17:33:13 +0300 Subject: [PATCH 2/9] fix: harden include_current_bucket and fix CI failures - Cast the current bucket end back to a timestamp; BigQuery's edr_timeadd returns a date for week/month buckets. - Only treat metrics as cached when they were calculated after their bucket ended, so the last partial snapshot of a bucket is recalculated once it completes instead of being frozen as its final value. - Only resolve include_current_bucket for dimension_anomalies, so setting it in vars or on a model does not affect or break other anomaly tests. - Skip monthly tests on Redshift (no month buckets) and cover weekly buckets in the rerun alignment test. - Compute the current date once in the current bucket tests. Co-Authored-By: Claude Opus 5.5 --- .../tests/test_anomaly_test_configuration.py | 2 + .../tests/test_dimension_anomalies.py | 60 ++++++++++++------- .../get_anomaly_scores_query.sql | 2 +- .../get_buckets_configuration.sql | 14 +++-- .../get_anomalies_test_configuration.sql | 20 ++++--- macros/edr/tests/test_dimension_anomalies.sql | 1 + 6 files changed, 64 insertions(+), 35 deletions(-) diff --git a/integration_tests/tests/test_anomaly_test_configuration.py b/integration_tests/tests/test_anomaly_test_configuration.py index 4a9b3f597..6190e29b7 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), } diff --git a/integration_tests/tests/test_dimension_anomalies.py b/integration_tests/tests/test_dimension_anomalies.py index ba7a4c40d..398aed313 100644 --- a/integration_tests/tests/test_dimension_anomalies.py +++ b/integration_tests/tests/test_dimension_anomalies.py @@ -1,5 +1,5 @@ import json -from datetime import datetime, timedelta +from datetime import date, datetime, timedelta from typing import Any, Dict, List import pytest @@ -317,12 +317,11 @@ def test_anomaly_in_detection_period( assert test_result["status"] == expected_status -def _generate_current_bucket_data(today_superheroes: List[str]): +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. """ - utc_today = datetime.utcnow().date() 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)) @@ -338,8 +337,9 @@ def _generate_current_bucket_data(today_superheroes: List[str]): 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( - ["Superman", "Superman", "Superman", "Spiderman"] + utc_today, ["Superman", "Superman", "Superman", "Spiderman"] ) # By default only complete buckets are tested, so today's anomaly is not detected yet. @@ -354,7 +354,7 @@ def test_include_current_bucket_detects_anomaly_in_current_bucket( 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(datetime.utcnow().date().strftime("%Y-%m-%d")) + x["bucket_start"].startswith(utc_today.strftime("%Y-%m-%d")) for x in anomalous_points ) @@ -362,7 +362,9 @@ def test_include_current_bucket_detects_anomaly_in_current_bucket( def test_include_current_bucket_passes_on_normal_current_bucket( test_id: str, dbt_project: DbtProject ): - data = _generate_current_bucket_data(["Superman", "Spiderman"]) + 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" @@ -372,22 +374,19 @@ 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(["Spiderman"]) + 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" +# Redshift does not support monthly time buckets. +@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_start = datetime.utcnow().date().replace(day=1) - month_starts = [current_month_start] - for _ in range(12): - previous = month_starts[-1] - timedelta(days=1) - month_starts.append(previous.replace(day=1)) - current_month, *previous_months = month_starts + current_month, *previous_months = _get_previous_bucket_starts("month", 12) rows_per_upload = {"PL": 100, "MY": 20} data: List[Dict[str, Any]] = [ @@ -424,24 +423,39 @@ def test_include_current_bucket_monthly_snapshot_upload( assert set(x["dimension_value"] for x in anomalous_points) == {"MY"} -def test_monthly_buckets_stay_aligned_on_rerun(test_id: str, dbt_project: DbtProject): - """On a rerun the backfill window starts mid-month; buckets must still be calendar months.""" - current_month_start = datetime.utcnow().date().replace(day=1) - month_starts = [current_month_start] - for _ in range(12): - previous = month_starts[-1] - timedelta(days=1) - month_starts.append(previous.replace(day=1)) +def _get_previous_bucket_starts(period: str, count: int) -> List[date]: + """The start of the current week or month, followed by the starts of the `count` before it.""" + utc_today = datetime.utcnow().date() + if period == "week": + current_week_start = utc_today - timedelta(days=utc_today.weekday()) + return [current_week_start - timedelta(weeks=i) for i in range(count + 1)] + bucket_starts = [utc_today.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", ["week", "month"]) +def test_buckets_stay_aligned_on_rerun( + test_id: str, dbt_project: DbtProject, target: str, period: str +): + """On a rerun the backfill window starts mid-bucket; buckets must still be calendar weeks/months.""" + if target == "redshift" and period == "month": + pytest.skip("Redshift does not support monthly time buckets.") + # Uploads on the first day of each bucket, for the buckets before the current one. + bucket_starts = _get_previous_bucket_starts(period, 12)[1:] data: List[Dict[str, Any]] = [ - {TIMESTAMP_COLUMN: month_start.strftime(DATE_FORMAT), "country": country} - for month_start in month_starts[1:] + {TIMESTAMP_COLUMN: bucket_start.strftime(DATE_FORMAT), "country": country} + for bucket_start in bucket_starts for country, rows in {"PL": 100, "MY": 20}.items() for _ in range(rows) ] test_args = { "timestamp_column": TIMESTAMP_COLUMN, "dimensions": ["country"], - "time_bucket": {"period": "month", "count": 1}, + "time_bucket": {"period": period, "count": 1}, "training_period": {"period": "day", "count": 400}, "detection_period": {"period": "day", "count": 31}, } 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 62aea5913..0ab98d4fd 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 @@ -74,7 +74,7 @@ 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 and test_configuration.timestamp_column %} + {%- if test_configuration.include_current_bucket %} {%- set buckets_end_expr = elementary.get_current_bucket_end_expr( detection_end_expr, min_bucket_start_expr, 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 e8e1f33a8..2441df265 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 @@ -31,12 +31,15 @@ {%- 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_timeadd( - time_bucket.period, - elementary.edr_cast_as_int(periods_until_current_bucket_end), - min_bucket_start_expr, + 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, + ) ) ) }} @@ -138,6 +141,9 @@ 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 }} + {# Only metrics calculated after their bucket ended are complete. A metric of the bucket + that was still in progress (include_current_bucket) is recalculated once it ends. #} + and {{ elementary.edr_cast_as_timestamp('updated_at') }} >= {{ elementary.edr_cast_as_timestamp('bucket_end') }} and upper(full_table_name) = upper('{{ full_table_name }}') and metric_properties = {{ elementary.dict_to_quoted_json(metric_properties) }} {%- if metric_names %} 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 36d47ce73..a3c6e159f 100644 --- a/macros/edr/tests/test_configuration/get_anomalies_test_configuration.sql +++ b/macros/edr/tests/test_configuration/get_anomalies_test_configuration.sql @@ -26,7 +26,8 @@ exclude_final_results, min_value, exclude_detection_period_from_training, - include_current_bucket=none + include_current_bucket=none, + supports_current_bucket=false ) %} {%- set model_graph_node = elementary.get_model_graph_node(model_relation) %} @@ -127,13 +128,18 @@ model_graph_node, ) %} - {% 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" + {# 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 = { diff --git a/macros/edr/tests/test_dimension_anomalies.sql b/macros/edr/tests/test_dimension_anomalies.sql index eb096117d..3953a262b 100644 --- a/macros/edr/tests/test_dimension_anomalies.sql +++ b/macros/edr/tests/test_dimension_anomalies.sql @@ -86,6 +86,7 @@ min_value=min_value, exclude_detection_period_from_training=exclude_detection_period_from_training, include_current_bucket=include_current_bucket, + supports_current_bucket=true, ) ) %} From b6288cf5b69bc61173572cbb401ba485a25daec2 Mon Sep 17 00:00:00 2001 From: Joost Boonzajer Flaes Date: Thu, 1 Oct 2026 11:04:24 +0300 Subject: [PATCH 3/9] fix: keep testing the last complete bucket with include_current_bucket The detection window ends backfill_days before the latest scored bucket. With include_current_bucket that is the bucket still in progress, so the window moved forward by one bucket and the last complete bucket dropped out of it. Because backfill_days is raised to a single bucket for weekly (7 days) and monthly (30 days) buckets, a finished bucket was then usually never tested again: a dimension that skipped a whole week or month was zero-filled but not flagged. The window now ends at the last complete bucket, as it does without the option, and the current bucket is tested on top of it. Co-Authored-By: Claude Opus 5.5 --- .../tests/test_dimension_anomalies.py | 80 +++++++++++++++++-- .../tests/test_utils/get_anomaly_query.sql | 19 ++++- 2 files changed, 92 insertions(+), 7 deletions(-) diff --git a/integration_tests/tests/test_dimension_anomalies.py b/integration_tests/tests/test_dimension_anomalies.py index 398aed313..0e4d8c40f 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 pytest from data_generator import DATE_FORMAT, generate_dates @@ -423,13 +423,15 @@ def test_include_current_bucket_monthly_snapshot_upload( assert set(x["dimension_value"] for x in anomalous_points) == {"MY"} -def _get_previous_bucket_starts(period: str, count: int) -> List[date]: - """The start of the current week or month, followed by the starts of the `count` before it.""" - utc_today = datetime.utcnow().date() +def _get_previous_bucket_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 = utc_today - timedelta(days=utc_today.weekday()) + 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 = [utc_today.replace(day=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)) @@ -464,3 +466,69 @@ def test_buckets_stay_aligned_on_rerun( assert first_result["status"] == "pass" second_result = dbt_project.test(test_id, DBT_TEST_NAME, test_args, data=data) assert second_result["status"] == "pass" + + +@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_previous_bucket_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. + 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) diff --git a/macros/edr/tests/test_utils/get_anomaly_query.sql b/macros/edr/tests/test_utils/get_anomaly_query.sql index 00bc1d2a0..fd9a12561 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 ( From 6a8fb82d5aef819d4bf864cd3f9f7aa2e3fdef67 Mon Sep 17 00:00:00 2001 From: Joost Boonzajer Flaes Date: Thu, 1 Oct 2026 11:04:53 +0300 Subject: [PATCH 4/9] fix: do not score stale current-bucket metrics with include_current_bucket Metrics are appended to data_monitoring_metrics on every run and read back as the newest row per metric id. The current bucket is not zero-filled, so a dimension that had rows earlier in the bucket but has none now writes no new row, and the metric from the earlier run kept being scored until the bucket ended. Deleting a bad upload therefore did not clear the failure it caused. The bucket in progress is recalculated on every run, so its metrics now only come from the current run. Co-Authored-By: Claude Opus 5.5 --- .../tests/test_dimension_anomalies.py | 19 +++++++++++++++++++ .../get_anomaly_scores_query.sql | 5 +++++ 2 files changed, 24 insertions(+) diff --git a/integration_tests/tests/test_dimension_anomalies.py b/integration_tests/tests/test_dimension_anomalies.py index 0e4d8c40f..4633058b3 100644 --- a/integration_tests/tests/test_dimension_anomalies.py +++ b/integration_tests/tests/test_dimension_anomalies.py @@ -380,6 +380,25 @@ def test_include_current_bucket_ignores_dimension_not_arrived_yet( 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" + + # Redshift does not support monthly time buckets. @pytest.mark.skip_targets(["redshift"]) def test_include_current_bucket_monthly_snapshot_upload( 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 0ab98d4fd..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 @@ -140,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). From a6361919066f8efe2f1e81b423faa2b8e86e9fb7 Mon Sep 17 00:00:00 2001 From: Joost Boonzajer Flaes Date: Mon, 5 Oct 2026 17:05:00 +0300 Subject: [PATCH 5/9] fix: only apply the partial-bucket cache rule with include_current_bucket A stored metric only counted as cached if it was calculated after its bucket ended, and this applied to every anomaly test and collect_metrics. The rule compares the warehouse clock (updated_at) with dbt's run time, so clock skew or a current_timestamp that isn't really UTC (DuckDB in a non-UTC session) could make complete buckets look partial. Only tests with include_current_bucket store partial buckets, so the rule now only applies to them and the default path is unchanged. Co-Authored-By: Claude Opus 5.5 --- .../get_buckets_configuration.sql | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) 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 2441df265..26f2f19bf 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 @@ -141,9 +141,11 @@ 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 }} - {# Only metrics calculated after their bucket ended are complete. A metric of the bucket - that was still in progress (include_current_bucket) is recalculated once it ends. #} + {# 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 %} From b97056ac52b660d35661094115d45b8a54592cf9 Mon Sep 17 00:00:00 2001 From: Joost Boonzajer Flaes Date: Mon, 5 Oct 2026 17:05:00 +0300 Subject: [PATCH 6/9] refactor: pass a single zero-fill cutoff to dimension_monitoring_query include_current_bucket and detection_end had to be passed together, and passing the flag without detection_end silently zero-filled the current bucket. The caller now passes zero_fill_max_bucket_end, which is the detection end when include_current_bucket is on. Co-Authored-By: Claude Opus 5.5 --- .../monitors_query/dimension_monitoring_query.sql | 14 ++++++++------ macros/edr/tests/test_dimension_anomalies.sql | 9 ++++++--- 2 files changed, 14 insertions(+), 9 deletions(-) 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 796bd93ff..231c2997c 100644 --- a/macros/edr/data_monitoring/monitors_query/dimension_monitoring_query.sql +++ b/macros/edr/data_monitoring/monitors_query/dimension_monitoring_query.sql @@ -6,8 +6,7 @@ max_bucket_end, metric_properties, metric_name=none, - include_current_bucket=false, - detection_end=none + zero_fill_max_bucket_end=none ) %} {% set metric_name = metric_name or "dimension" %} {% set full_table_name_str = elementary.edr_quote( @@ -107,13 +106,16 @@ 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 - {# The current bucket is still in progress, so a dimension without rows in it - has not arrived yet rather than dropped to zero. Only complete buckets are filled. #} - {% if include_current_bucket and detection_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(detection_end) + elementary.edr_datetime_to_sql( + zero_fill_max_bucket_end + ) ) }} {% endif %} diff --git a/macros/edr/tests/test_dimension_anomalies.sql b/macros/edr/tests/test_dimension_anomalies.sql index 3953a262b..fef1d747c 100644 --- a/macros/edr/tests/test_dimension_anomalies.sql +++ b/macros/edr/tests/test_dimension_anomalies.sql @@ -123,9 +123,12 @@ min_bucket_start, max_bucket_end, metric_properties, - include_current_bucket=test_configuration.include_current_bucket, - detection_end=elementary.get_detection_end( - test_configuration.detection_delay + zero_fill_max_bucket_end=( + elementary.get_detection_end( + test_configuration.detection_delay + ) + if test_configuration.include_current_bucket + else none ), ) %} {{ From f78338511634ff5828b423f39eccf647d6a66c05 Mon Sep 17 00:00:00 2001 From: Joost Boonzajer Flaes Date: Mon, 5 Oct 2026 17:05:00 +0300 Subject: [PATCH 7/9] test: cover more include_current_bucket cases Adds tests for the missing timestamp_column error, 4-hour buckets on a source and on a table model, and a partial bucket that is recalculated after a gap longer than backfill_days. Also explains why the fixed dates of the previous-bucket test stay valid after they pass. Co-Authored-By: Claude Opus 5.5 --- .../tests/test_dimension_anomalies.py | 136 +++++++++++++++++- 1 file changed, 135 insertions(+), 1 deletion(-) diff --git a/integration_tests/tests/test_dimension_anomalies.py b/integration_tests/tests/test_dimension_anomalies.py index 4633058b3..490a4a911 100644 --- a/integration_tests/tests/test_dimension_anomalies.py +++ b/integration_tests/tests/test_dimension_anomalies.py @@ -2,6 +2,7 @@ from datetime import date, datetime, timedelta from typing import Any, Dict, List, Optional +import dateutil.parser import pytest from data_generator import DATE_FORMAT, generate_dates from dbt_project import DbtProject @@ -399,6 +400,79 @@ def test_include_current_bucket_ignores_deleted_rows( 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 + ) + + # Redshift does not support monthly time buckets. @pytest.mark.skip_targets(["redshift"]) def test_include_current_bucket_monthly_snapshot_upload( @@ -529,7 +603,10 @@ def test_include_current_bucket_tests_previous_bucket( } # 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. + # 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, @@ -551,3 +628,60 @@ def test_include_current_bucket_tests_previous_bucket( 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) + + +# Redshift does not support monthly time buckets. +@pytest.mark.skip_targets(["redshift"]) +def test_include_current_bucket_recalculates_partial_bucket_after_gap( + 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_previous_bucket_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] From 6cfc79ba025bdc072cf123985ee7a90458156ecc Mon Sep 17 00:00:00 2001 From: Joost Boonzajer Flaes Date: Mon, 5 Oct 2026 17:17:08 +0300 Subject: [PATCH 8/9] refactor: align the backfill start where it is defined Moves the bucket alignment from the incremental query into the backfill_bucket_start set, so the query line is unchanged from master. Co-Authored-By: Claude Opus 5.5 --- .../get_buckets_configuration.sql | 18 ++++++++++-------- 1 file changed, 10 insertions(+), 8 deletions(-) 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 26f2f19bf..703197427 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 @@ -85,12 +85,16 @@ ) %} {%- else %} {%- set buckets_end_expr = detection_end_expr %} {%- endif %} - {%- set backfill_bucket_start = elementary.edr_cast_as_timestamp( - elementary.edr_datetime_to_sql( - elementary.get_backfill_bucket_start( - detection_end, backfill_days + {%- set backfill_bucket_start = elementary.get_start_bucket_in_data( + elementary.edr_cast_as_timestamp( + elementary.edr_datetime_to_sql( + elementary.get_backfill_bucket_start( + detection_end, backfill_days + ) ) - ) + ), + trunc_min_bucket_start_expr, + metric_properties.time_bucket, ) %} {%- set full_table_name = elementary.relation_to_full_name(model_relation) %} {%- set force_metrics_backfill = elementary.get_config_var( @@ -167,9 +171,7 @@ min_bucket_start_candidates as ( select bucket_start from missing_bucket_starts union all - {# Align the backfill start to the start of its bucket. Otherwise buckets that are - not a whole number of days (e.g. months) are generated from a mid-bucket start. #} - select {{ elementary.get_start_bucket_in_data(backfill_bucket_start, trunc_min_bucket_start_expr, metric_properties.time_bucket) }} as bucket_start + select {{ backfill_bucket_start }} as bucket_start ) select min(bucket_start) as min_bucket_start, From 34470dc0e008d67daa51bb15b8e61f9b0e9f9e42 Mon Sep 17 00:00:00 2001 From: Joost Boonzajer Flaes Date: Tue, 6 Oct 2026 11:13:51 +0300 Subject: [PATCH 9/9] test: shorten a test name over the Postgres identifier limit The test name is also the seeded table name, and at 65 characters it was rejected by Postgres, which caps identifiers at 63. Co-Authored-By: Claude Opus 5.5 --- integration_tests/tests/test_dimension_anomalies.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/integration_tests/tests/test_dimension_anomalies.py b/integration_tests/tests/test_dimension_anomalies.py index 490a4a911..404b79326 100644 --- a/integration_tests/tests/test_dimension_anomalies.py +++ b/integration_tests/tests/test_dimension_anomalies.py @@ -632,7 +632,7 @@ def test_include_current_bucket_tests_previous_bucket( # Redshift does not support monthly time buckets. @pytest.mark.skip_targets(["redshift"]) -def test_include_current_bucket_recalculates_partial_bucket_after_gap( +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."""