From 20d0342c3a1d4e47481f6768525cbf3b5f4dd7ab Mon Sep 17 00:00:00 2001 From: Joost Boonzajer Flaes Date: Mon, 5 Oct 2026 17:08:02 +0300 Subject: [PATCH 1/4] fix: align the backfill start to the time bucket on reruns On reruns of anomaly tests on sources and incremental models, the backfill window started at a plain date rather than at the start of a time bucket. For weekly, monthly and multi-day buckets, the buckets were then generated from that mid-bucket start. With monthly buckets, rows fell into no bucket and the resulting zeros were scored as a drop. The backfill start is now aligned to the start of its bucket. Co-Authored-By: Claude Opus 5.5 --- .../tests/test_dimension_anomalies.py | 65 ++++++++++++++++++- .../get_buckets_configuration.sql | 4 +- 2 files changed, 67 insertions(+), 2 deletions(-) diff --git a/integration_tests/tests/test_dimension_anomalies.py b/integration_tests/tests/test_dimension_anomalies.py index 51d7a05b0..609eb8001 100644 --- a/integration_tests/tests/test_dimension_anomalies.py +++ b/integration_tests/tests/test_dimension_anomalies.py @@ -1,7 +1,8 @@ import json -from datetime import datetime, timedelta +from datetime import date, datetime, timedelta from typing import Any, Dict, List +import dateutil.parser import pytest from data_generator import DATE_FORMAT, generate_dates from dbt_project import DbtProject @@ -315,3 +316,65 @@ def test_anomaly_in_detection_period( ) assert test_result["status"] == expected_status + + +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.") + test_id = test_id.replace("[", "_").replace("]", "_") + # 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: 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": period, "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" + + # Misaligned weekly buckets still hold one upload each, so check the stored bucket starts too. + stored_bucket_starts = { + dateutil.parser.parse(result["bucket_start"]).date() + for result in dbt_project.run_query( + f""" + select distinct bucket_start + from {{{{ ref("data_monitoring_metrics") }}}} + where lower(full_table_name) like '%{test_id.lower()}' + """ + ) + } + if period == "week": + assert ( + len({bucket_start.weekday() for bucket_start in stored_bucket_starts}) == 1 + ) + else: + assert {bucket_start.day for bucket_start in stored_bucket_starts} == {1} 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..f44a99126 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 @@ -128,7 +128,9 @@ 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, From 3d8a7d955776d2c77c5f61d12a0cd07b93c2c2db Mon Sep 17 00:00:00 2001 From: Joost Boonzajer Flaes Date: Mon, 5 Oct 2026 17:17:08 +0300 Subject: [PATCH 2/4] 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 f44a99126..b082bade1 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 @@ -52,12 +52,16 @@ {%- set trunc_min_bucket_start_expr = elementary.get_trunc_min_bucket_start_expr( detection_end, metric_properties, days_back ) %} - {%- 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( @@ -128,9 +132,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 6f31f6cb2382a5b400048f956c24b8dae8aea7d6 Mon Sep 17 00:00:00 2001 From: Joost Boonzajer Flaes Date: Wed, 7 Oct 2026 14:34:47 +0300 Subject: [PATCH 3/4] test: keep the rerun backfill start mid-bucket and cover multi-day buckets With a fixed 31-day detection period, the old backfill start landed on a bucket boundary on some dates (every Thursday for weeks, the 1st after a 31-day month), so the test could pass without the fix. Pick the backfill length so its start is never on a boundary, and add a 3-day bucket case for the floor(.../count) * count path of get_start_bucket_in_data. Co-Authored-By: Claude Opus 5.5 --- .../tests/test_dimension_anomalies.py | 85 +++++++++++++++---- 1 file changed, 69 insertions(+), 16 deletions(-) diff --git a/integration_tests/tests/test_dimension_anomalies.py b/integration_tests/tests/test_dimension_anomalies.py index 609eb8001..38c117ebb 100644 --- a/integration_tests/tests/test_dimension_anomalies.py +++ b/integration_tests/tests/test_dimension_anomalies.py @@ -318,12 +318,25 @@ def test_anomaly_in_detection_period( assert test_result["status"] == expected_status -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.""" +TRAINING_DAYS = 400 + + +def _get_previous_bucket_starts(time_bucket: Dict[str, Any], count: int) -> List[date]: + """The start of the current bucket, followed by the starts of the `count` before it.""" utc_today = datetime.utcnow().date() - if period == "week": + if time_bucket["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)] + if time_bucket["period"] == "day": + # Multi-day buckets are counted from the start of the training period. + days = time_bucket["count"] + grid_start = utc_today - timedelta(days=TRAINING_DAYS) + current_bucket_start = grid_start + timedelta( + days=(utc_today - grid_start).days // days * days + ) + return [ + current_bucket_start - timedelta(days=days * 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) @@ -331,16 +344,46 @@ def _get_previous_bucket_starts(period: str, count: int) -> List[date]: return bucket_starts -@pytest.mark.parametrize("period", ["week", "month"]) +def _get_mid_bucket_backfill_days(time_bucket: Dict[str, Any]) -> int: + """A backfill length of about a month whose start is not on a bucket boundary. + + Before the fix, a rerun built its buckets from the backfill start, so the test only + catches a regression when that start falls inside a bucket. Weeks start on Sunday or + Monday depending on the warehouse, so both are avoided. + """ + utc_today = datetime.utcnow().date() + backfill_days = 31 + while True: + backfill_start = utc_today - timedelta(days=backfill_days) + if time_bucket["period"] == "week": + on_boundary = backfill_start.weekday() in (0, 6) + elif time_bucket["period"] == "day": + on_boundary = (TRAINING_DAYS - backfill_days) % time_bucket["count"] == 0 + else: + on_boundary = backfill_start.day == 1 + if not on_boundary: + return backfill_days + backfill_days += 1 + + +@pytest.mark.parametrize( + "time_bucket", + [ + {"period": "week", "count": 1}, + {"period": "month", "count": 1}, + {"period": "day", "count": 3}, + ], + ids=["week", "month", "3_days"], +) def test_buckets_stay_aligned_on_rerun( - test_id: str, dbt_project: DbtProject, target: str, period: str + test_id: str, dbt_project: DbtProject, target: str, time_bucket: Dict[str, Any] ): - """On a rerun the backfill window starts mid-bucket; buckets must still be calendar weeks/months.""" - if target == "redshift" and period == "month": + """On a rerun the backfill window starts mid-bucket; buckets must stay on the same grid.""" + if target == "redshift" and time_bucket["period"] == "month": pytest.skip("Redshift does not support monthly time buckets.") test_id = test_id.replace("[", "_").replace("]", "_") # Uploads on the first day of each bucket, for the buckets before the current one. - bucket_starts = _get_previous_bucket_starts(period, 12)[1:] + bucket_starts = _get_previous_bucket_starts(time_bucket, 12)[1:] data: List[Dict[str, Any]] = [ {TIMESTAMP_COLUMN: bucket_start.strftime(DATE_FORMAT), "country": country} @@ -351,9 +394,12 @@ def test_buckets_stay_aligned_on_rerun( test_args = { "timestamp_column": TIMESTAMP_COLUMN, "dimensions": ["country"], - "time_bucket": {"period": period, "count": 1}, - "training_period": {"period": "day", "count": 400}, - "detection_period": {"period": "day", "count": 31}, + "time_bucket": time_bucket, + "training_period": {"period": "day", "count": TRAINING_DAYS}, + "detection_period": { + "period": "day", + "count": _get_mid_bucket_backfill_days(time_bucket), + }, } first_result = dbt_project.test(test_id, DBT_TEST_NAME, test_args, data=data) @@ -361,7 +407,7 @@ def test_buckets_stay_aligned_on_rerun( second_result = dbt_project.test(test_id, DBT_TEST_NAME, test_args, data=data) assert second_result["status"] == "pass" - # Misaligned weekly buckets still hold one upload each, so check the stored bucket starts too. + # Misaligned buckets can still hold one upload each, so check the stored bucket starts too. stored_bucket_starts = { dateutil.parser.parse(result["bucket_start"]).date() for result in dbt_project.run_query( @@ -372,9 +418,16 @@ def test_buckets_stay_aligned_on_rerun( """ ) } - if period == "week": + if time_bucket["period"] == "month": + assert {bucket_start.day for bucket_start in stored_bucket_starts} == {1} + else: + bucket_days = 7 if time_bucket["period"] == "week" else time_bucket["count"] assert ( - len({bucket_start.weekday() for bucket_start in stored_bucket_starts}) == 1 + len( + { + bucket_start.toordinal() % bucket_days + for bucket_start in stored_bucket_starts + } + ) + == 1 ) - else: - assert {bucket_start.day for bucket_start in stored_bucket_starts} == {1} From aa05c853fb28c091b3850f8a4c6ec08d6ebc8f4e Mon Sep 17 00:00:00 2001 From: Joost Boonzajer Flaes Date: Sun, 11 Oct 2026 10:22:50 +0300 Subject: [PATCH 4/4] test: upload in the latest Monday's week so Sunday-start warehouses pass on Sundays Co-Authored-By: Claude Opus 5.5 --- integration_tests/tests/test_dimension_anomalies.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/integration_tests/tests/test_dimension_anomalies.py b/integration_tests/tests/test_dimension_anomalies.py index 38c117ebb..0ded83187 100644 --- a/integration_tests/tests/test_dimension_anomalies.py +++ b/integration_tests/tests/test_dimension_anomalies.py @@ -384,6 +384,11 @@ def test_buckets_stay_aligned_on_rerun( test_id = test_id.replace("[", "_").replace("]", "_") # Uploads on the first day of each bucket, for the buckets before the current one. bucket_starts = _get_previous_bucket_starts(time_bucket, 12)[1:] + if time_bucket["period"] == "week": + # Uploads are on Mondays. On a Sunday, warehouses whose weeks start on Sunday + # already have the week of the latest Monday complete, so upload there too. + # Where weeks start on Monday, that upload falls in the current, unscored week. + bucket_starts = _get_previous_bucket_starts(time_bucket, 12) data: List[Dict[str, Any]] = [ {TIMESTAMP_COLUMN: bucket_start.strftime(DATE_FORMAT), "country": country}