fix: count weekly buckets in days so Sunday rows land in their week

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

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 %}