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 datetime import date, 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,105 @@ 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:
+# Sunday rows must land in their week, whichever day the adapter starts weeks on.
+# Short name: the seeded table is named after it, and Postgres allows 63 characters.
+@Parametrization.autodetect_parameters()
+@Parametrization.case(
+    name="source_wednesday_run",
+    as_model=False,
+    run_started_at=datetime(2026, 9, 16, 12),
+)
+@Parametrization.case(
+    name="model_sunday_run",
+    as_model=True,
+    run_started_at=datetime(2026, 9, 13, 12),
+)
+def test_weekly_buckets_sunday_rows(
+    test_id: str, dbt_project: DbtProject, as_model: bool, run_started_at: datetime
+):
+    training_days = 12 * 7
+    run_date = run_started_at.date()
+    dates = generate_dates(
+        base_date=run_date - timedelta(days=1), days_back=training_days + 7
+    )
+    # A different null count each Sunday, so a Sunday in the wrong week shows.
+    sundays = sorted(cur_date for cur_date in dates if cur_date.isoweekday() == 7)
+    sunday_nulls = {sunday: 1 + week % 3 for week, sunday in enumerate(sundays)}
+    data: List[Dict[str, Any]] = [
+        {TIMESTAMP_COLUMN: cur_date.strftime(DATE_FORMAT), "superhero": "Superman"}
+        for cur_date in dates
+    ]
+    data += [
+        {TIMESTAMP_COLUMN: sunday.strftime(DATE_FORMAT), "superhero": None}
+        for sunday, null_count in sunday_nulls.items()
+        for _ in range(null_count)
+    ]
+
+    test_args = {
+        **DBT_TEST_ARGS,
+        "time_bucket": {"period": "week", "count": 1},
+        "training_period": {"period": "day", "count": training_days},
+    }
+    test_result = dbt_project.test(
+        test_id,
+        DBT_TEST_NAME,
+        test_args,
+        data=data,
+        test_column="superhero",
+        as_model=as_model,
+        test_vars={"custom_run_started_at": run_started_at.isoformat()},
+    )
+    assert test_result["status"] == "pass"
+
+    null_counts = _read_weekly_null_counts(dbt_project, test_id)
+    week_start_day = min(null_counts).isoweekday()
+    assert week_start_day in (1, 7), f"{min(null_counts)} is not a Monday or Sunday"
+    training_start = run_date - timedelta(days=training_days)
+    assert null_counts == _expected_weekly_null_counts(
+        sunday_nulls, training_start, run_date, week_start_day
+    )
+
+
+def _read_weekly_null_counts(dbt_project: DbtProject, test_id: str) -> Dict[date, Any]:
+    """The stored null_count of each bucket, by bucket start date."""
+    table_name = test_id.replace("[", "_").replace("]", "_").lower()
+    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 '%{table_name}'
+        """
+    )
+    assert metrics, "Found no null_count metrics"
+    return {
+        dateutil.parser.parse(str(metric["bucket_start"])).date(): metric[
+            "metric_value"
+        ]
+        for metric in metrics
+    }
+
+
+def _expected_weekly_null_counts(
+    sunday_nulls: Dict[date, int],
+    training_start: date,
+    run_date: date,
+    week_start_day: int,
+) -> Dict[date, int]:
+    """Every full week from the week of training_start to run_date, with its nulls."""
+    week_start = training_start - timedelta(
+        days=(training_start.isoweekday() - week_start_day) % 7
+    )
+    expected = {}
+    while week_start + timedelta(weeks=1) <= run_date:
+        week_end = week_start + timedelta(weeks=1)
+        expected[week_start] = sum(
+            null_count
+            for sunday, null_count in sunday_nulls.items()
+            if week_start <= sunday < week_end
+        )
+        week_start = week_end
+    return expected

macros/edr/data_monitoring/data_monitors_configuration/get_buckets_configuration.sql

@@ -59,6 +59,9 @@
             )
         )
     ) %}
+    {%- set datediff_time_bucket = elementary.get_datediff_time_bucket(
+        metric_properties.time_bucket
+    ) %}
     {%- set full_table_name = elementary.relation_to_full_name(model_relation) %}
     {%- set force_metrics_backfill = elementary.get_config_var(
         "force_metrics_backfill"
@@ -85,12 +88,12 @@
         ),
         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', datediff_time_bucket.period) }} / {{ datediff_time_bucket.count }}) * {{ datediff_time_bucket.count }} as periods_until_max
             from bucket_times
         )
         select
              days_back_start as min_bucket_start,
-             {{ elementary.edr_timeadd(metric_properties.time_bucket.period, 'periods_until_max', 'days_back_start') }} {# Add full buckets to last_max_bucket_end #}
+             {{ elementary.edr_timeadd(datediff_time_bucket.period, 'periods_until_max', 'days_back_start') }} {# Add full buckets to last_max_bucket_end #}
         as max_bucket_end
         from full_buckets_calc
     {%- endset %}

macros/edr/data_monitoring/monitors_query/get_start_bucket_in_data.sql

@@ -1,4 +1,5 @@
 {% macro get_start_bucket_in_data(timestamp_column, min_bucket_start, time_bucket) %}
+    {% set time_bucket = elementary.get_datediff_time_bucket(time_bucket) %}
     {% 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 %}

macros/edr/system/system_utils/get_datediff_time_bucket.sql

@@ -0,0 +1,9 @@
+{#- The time bucket to count with edr_datediff. Weeks are counted as 7 days: where
+    a week starts in datediff depends on the adapter (Sunday on Postgres and Spark),
+    and may differ from date_trunc and from the first bucket start. -#}
+{% macro get_datediff_time_bucket(time_bucket) %}
+    {% if time_bucket.period | lower == "week" %}
+        {% do return({"period": "day", "count": time_bucket.count * 7}) %}
+    {% endif %}
+    {% do return(time_bucket) %}
+{% endmacro %}