diff --git a/integration_tests/tests/test_dimension_anomalies.py b/integration_tests/tests/test_dimension_anomalies.py index 51d7a05b0..0ded83187 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,123 @@ def test_anomaly_in_detection_period( ) assert test_result["status"] == expected_status + + +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 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) + bucket_starts.append(previous.replace(day=1)) + return bucket_starts + + +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, time_bucket: Dict[str, Any] +): + """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(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} + 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": 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) + 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 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( + f""" + select distinct bucket_start + from {{{{ ref("data_monitoring_metrics") }}}} + where lower(full_table_name) like '%{test_id.lower()}' + """ + ) + } + 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.toordinal() % bucket_days + 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 786e2d644..c5d89158c 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 datediff_time_bucket = elementary.get_datediff_time_bucket( metric_properties.time_bucket