diff --git a/integration_tests/tests/test_column_anomalies.py b/integration_tests/tests/test_column_anomalies.py index af93e25c0..f78eb9cd8 100644 --- a/integration_tests/tests/test_column_anomalies.py +++ b/integration_tests/tests/test_column_anomalies.py @@ -1,6 +1,7 @@ from datetime import datetime, timedelta from typing import Any, Dict, List +import dateutil.parser from data_generator import DATE_FORMAT, generate_dates from dbt_project import DbtProject from parametrization import Parametrization @@ -720,3 +721,57 @@ def test_col_excl_detect_train_seven_day_bucket(test_id: str, dbt_project: DbtPr "Expected FAIL when exclude_detection_period_from_training=True " "(large bucket fix: detection period set to bucket size)" ) + + +# Regression test for https://github.com/elementary-data/elementary/issues/2370 +def test_column_anomalies_weekly_buckets_with_sunday_rows( + test_id: str, dbt_project: DbtProject +): + utc_today = datetime.utcnow().date() + last_sunday = utc_today - timedelta(days=utc_today.isoweekday() % 7 or 7) + sundays = [last_sunday - timedelta(weeks=week) for week in range(12)] + + # Nulls only on Sundays, with a different count each week, so a Sunday + # counted in the wrong week changes the weekly null_count. + expected_null_counts = {} + data: List[Dict[str, Any]] = [] + for week, sunday in enumerate(sundays): + null_count = 1 + week % 3 + expected_null_counts[sunday - timedelta(days=6)] = null_count + data += [ + {TIMESTAMP_COLUMN: sunday.strftime(DATE_FORMAT), "superhero": None} + for _ in range(null_count) + ] + data += [ + {TIMESTAMP_COLUMN: cur_date.strftime(DATE_FORMAT), "superhero": "Superman"} + for cur_date in generate_dates(base_date=last_sunday, days_back=12 * 7) + ] + + test_args = { + **DBT_TEST_ARGS, + "time_bucket": {"period": "week", "count": 1}, + "training_period": {"period": "day", "count": 12 * 7}, + } + test_result = dbt_project.test( + test_id, DBT_TEST_NAME, test_args, data=data, test_column="superhero" + ) + assert test_result["status"] == "pass" + + metrics = dbt_project.run_query( + f""" + select bucket_start, metric_value + from {{{{ ref("data_monitoring_metrics") }}}} + where metric_name = 'null_count' + and lower(full_table_name) like '%{test_id.lower()}' + """ + ) + null_counts = { + dateutil.parser.parse(metric["bucket_start"]).date(): metric["metric_value"] + for metric in metrics + if metric["bucket_start"] is not None + } + assert len(null_counts) == len(metrics), "Found metrics without a bucket" + assert null_counts + for bucket_start, null_count in null_counts.items(): + assert bucket_start.isoweekday() == 1, f"{bucket_start} is not a Monday" + assert null_count == expected_null_counts[bucket_start], bucket_start diff --git a/macros/edr/data_monitoring/monitors_query/get_start_bucket_in_data.sql b/macros/edr/data_monitoring/monitors_query/get_start_bucket_in_data.sql index 0c3cb23b6..79c6bd12e 100644 --- a/macros/edr/data_monitoring/monitors_query/get_start_bucket_in_data.sql +++ b/macros/edr/data_monitoring/monitors_query/get_start_bucket_in_data.sql @@ -1,4 +1,9 @@ {% macro get_start_bucket_in_data(timestamp_column, min_bucket_start, time_bucket) %} + {#- Count weeks as 7 days: where a week starts in datediff depends on the + adapter (Sunday on Postgres), while buckets start on min_bucket_start. -#} + {% if time_bucket.period | lower == "week" %} + {% set time_bucket = {"period": "day", "count": time_bucket.count * 7} %} + {% endif %} {% set bucket_start_datediff_expr %} floor({{ elementary.edr_datediff(min_bucket_start, elementary.edr_cast_as_timestamp(timestamp_column), time_bucket.period) }} / {{ time_bucket.count }}) * {{ time_bucket.count }} {% endset %}