From c1958ef51a15be004ea42e4b293e9c150f183929 Mon Sep 17 00:00:00 2001 From: Adam Date: Tue, 29 Sep 2026 23:24:08 +0200 Subject: [PATCH 1/3] fix: count weekly buckets in days so Sunday rows land in their week On Postgres, datediff(week) starts weeks on Sunday while buckets run Monday to Monday, so Sunday rows were counted in the next bucket and the last one got no bucket at all, which crashed the anomaly test result sort. Fixes elementary-data/elementary#2370 --- .../tests/test_column_anomalies.py | 55 +++++++++++++++++++ .../get_start_bucket_in_data.sql | 5 ++ 2 files changed, 60 insertions(+) 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 %} From 9372a0d64cf0ee4801ce6e253706f70d62396a39 Mon Sep 17 00:00:00 2001 From: Elad Barkay Date: Wed, 7 Oct 2026 18:13:03 +0300 Subject: [PATCH 2/3] fix: count weeks in days for max_bucket_end too, and make the test adapter-agnostic get_metric_buckets_min_and_max had the same week-start mismatch: on Postgres and Spark, a Sunday run counted one week too many, so max_bucket_end was next Monday and the last bucket was unfinished. Both places now share get_datediff_time_bucket. The test assumed weeks start on Monday, which fails on SQL Server, BigQuery and Fabric. It now reads the week start from the buckets, checks that every full week is there, and computes each week's null count from the data. It pins run_started_at and covers a source (incremental path) and a model on a Sunday (regular path). Co-Authored-By: Claude Opus 5.5 --- .../tests/test_column_anomalies.py | 94 ++++++++++++++----- .../get_buckets_configuration.sql | 7 +- .../get_start_bucket_in_data.sql | 6 +- .../system_utils/get_datediff_time_bucket.sql | 9 ++ 4 files changed, 83 insertions(+), 33 deletions(-) create mode 100644 macros/edr/system/system_utils/get_datediff_time_bucket.sql diff --git a/integration_tests/tests/test_column_anomalies.py b/integration_tests/tests/test_column_anomalies.py index f78eb9cd8..c299c1d77 100644 --- a/integration_tests/tests/test_column_anomalies.py +++ b/integration_tests/tests/test_column_anomalies.py @@ -724,54 +724,96 @@ def test_col_excl_detect_train_seven_day_bucket(test_id: str, dbt_project: DbtPr # 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 +# Weeks start on Monday or on Sunday depending on the adapter, so the test reads +# the week start from the buckets, then checks every bucket against the data. +# A source takes the incremental path. A model takes the regular one, where a +# Sunday run used to add an unfinished bucket. The test name stays short: 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 ): - 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)] + training_days = 12 * 7 + run_date = run_started_at.date() + data_dates = generate_dates( + base_date=run_date - timedelta(days=1), days_back=training_days + 7 + ) # 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 += [ + sundays = [cur_date for cur_date in data_dates if cur_date.isoweekday() == 7] + sunday_null_counts = { + sunday: 1 + week % 3 for week, sunday in enumerate(sorted(sundays)) + } + data: List[Dict[str, Any]] = [ {TIMESTAMP_COLUMN: cur_date.strftime(DATE_FORMAT), "superhero": "Superman"} - for cur_date in generate_dates(base_date=last_sunday, days_back=12 * 7) + for cur_date in data_dates ] + for sunday, null_count in sunday_null_counts.items(): + data += [ + {TIMESTAMP_COLUMN: sunday.strftime(DATE_FORMAT), "superhero": None} + ] * null_count test_args = { **DBT_TEST_ARGS, "time_bucket": {"period": "week", "count": 1}, - "training_period": {"period": "day", "count": 12 * 7}, + "training_period": {"period": "day", "count": training_days}, } test_result = dbt_project.test( - test_id, DBT_TEST_NAME, test_args, data=data, test_column="superhero" + 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" + 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 '%{test_id.lower()}' + and lower(full_table_name) like '%{table_name}' """ ) + assert metrics, "Found no null_count metrics" + assert all( + metric["bucket_start"] is not None for metric in metrics + ), "Found metrics without a bucket" null_counts = { - dateutil.parser.parse(metric["bucket_start"]).date(): metric["metric_value"] + dateutil.parser.parse(str(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 + assert len(null_counts) == len(metrics), "Found duplicate buckets" + + # Every full week from the start of the training period to the run. + week_start_day = min(null_counts).isoweekday() + assert week_start_day in (1, 7), f"{min(null_counts)} is not a Monday or Sunday" + days_back_start = run_date - timedelta(days=training_days) + bucket_start = days_back_start - timedelta( + days=(days_back_start.isoweekday() - week_start_day) % 7 + ) + expected_null_counts = {} + while bucket_start + timedelta(weeks=1) <= run_date: + expected_null_counts[bucket_start] = sum( + null_count + for sunday, null_count in sunday_null_counts.items() + if bucket_start <= sunday < bucket_start + timedelta(weeks=1) + ) + bucket_start += timedelta(weeks=1) + assert null_counts == expected_null_counts 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..786e2d644 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 @@ -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 %} 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 79c6bd12e..c3e7f1e74 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,9 +1,5 @@ {% 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 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 %} diff --git a/macros/edr/system/system_utils/get_datediff_time_bucket.sql b/macros/edr/system/system_utils/get_datediff_time_bucket.sql new file mode 100644 index 000000000..5659fcc9b --- /dev/null +++ b/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 %} From a8df8e38275df61dd2144acaf82b64bfb4250376 Mon Sep 17 00:00:00 2001 From: Elad Barkay Date: Wed, 7 Oct 2026 19:03:48 +0300 Subject: [PATCH 3/3] test: shorten the weekly buckets test and move its checks into helpers Co-Authored-By: Claude Opus 5.5 --- .../tests/test_column_anomalies.py | 82 ++++++++++--------- 1 file changed, 44 insertions(+), 38 deletions(-) diff --git a/integration_tests/tests/test_column_anomalies.py b/integration_tests/tests/test_column_anomalies.py index c299c1d77..3cd9e64df 100644 --- a/integration_tests/tests/test_column_anomalies.py +++ b/integration_tests/tests/test_column_anomalies.py @@ -1,4 +1,4 @@ -from datetime import datetime, timedelta +from datetime import date, datetime, timedelta from typing import Any, Dict, List import dateutil.parser @@ -723,12 +723,9 @@ def test_col_excl_detect_train_seven_day_bucket(test_id: str, dbt_project: DbtPr ) -# Regression test for https://github.com/elementary-data/elementary/issues/2370 -# Weeks start on Monday or on Sunday depending on the adapter, so the test reads -# the week start from the buckets, then checks every bucket against the data. -# A source takes the incremental path. A model takes the regular one, where a -# Sunday run used to add an unfinished bucket. The test name stays short: the -# seeded table is named after it and Postgres allows 63 characters. +# 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", @@ -745,24 +742,21 @@ def test_weekly_buckets_sunday_rows( ): training_days = 12 * 7 run_date = run_started_at.date() - data_dates = generate_dates( + dates = generate_dates( base_date=run_date - timedelta(days=1), days_back=training_days + 7 ) - - # Nulls only on Sundays, with a different count each week, so a Sunday - # counted in the wrong week changes the weekly null_count. - sundays = [cur_date for cur_date in data_dates if cur_date.isoweekday() == 7] - sunday_null_counts = { - sunday: 1 + week % 3 for week, sunday in enumerate(sorted(sundays)) - } + # 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 data_dates + 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) ] - for sunday, null_count in sunday_null_counts.items(): - data += [ - {TIMESTAMP_COLUMN: sunday.strftime(DATE_FORMAT), "superhero": None} - ] * null_count test_args = { **DBT_TEST_ARGS, @@ -780,6 +774,17 @@ def test_weekly_buckets_sunday_rows( ) 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""" @@ -790,30 +795,31 @@ def test_weekly_buckets_sunday_rows( """ ) assert metrics, "Found no null_count metrics" - assert all( - metric["bucket_start"] is not None for metric in metrics - ), "Found metrics without a bucket" - null_counts = { + return { dateutil.parser.parse(str(metric["bucket_start"])).date(): metric[ "metric_value" ] for metric in metrics } - assert len(null_counts) == len(metrics), "Found duplicate buckets" - # Every full week from the start of the training period to the run. - week_start_day = min(null_counts).isoweekday() - assert week_start_day in (1, 7), f"{min(null_counts)} is not a Monday or Sunday" - days_back_start = run_date - timedelta(days=training_days) - bucket_start = days_back_start - timedelta( - days=(days_back_start.isoweekday() - week_start_day) % 7 + +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_null_counts = {} - while bucket_start + timedelta(weeks=1) <= run_date: - expected_null_counts[bucket_start] = sum( + 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_null_counts.items() - if bucket_start <= sunday < bucket_start + timedelta(weeks=1) + for sunday, null_count in sunday_nulls.items() + if week_start <= sunday < week_end ) - bucket_start += timedelta(weeks=1) - assert null_counts == expected_null_counts + week_start = week_end + return expected