From 96731053195438d599b0e8ad3bab888e577d82d8 Mon Sep 17 00:00:00 2001 From: zerafachris <22851907+zerafachris@users.noreply.github.com> Date: Fri, 30 Oct 2026 08:07:13 +0100 Subject: [PATCH 1/4] fix: cast numeric monitors to Nullable on ClickHouse (#1094) On ClickHouse, max/min/average/sum/zero_count cast the column to Float32, which is non-Nullable, so a single NULL in the monitored window raised Code 349 instead of being ignored. Route these casts through a dispatched helper whose ClickHouse variant casts to Nullable(Float32), as the stddev and variance monitors already do. --- .../tests/test_column_anomalies.py | 28 +++++++++++++++++++ .../monitors/column_numeric_monitors.sql | 24 ++++++++++++---- 2 files changed, 47 insertions(+), 5 deletions(-) diff --git a/integration_tests/tests/test_column_anomalies.py b/integration_tests/tests/test_column_anomalies.py index 3cd9e64df..8a472702d 100644 --- a/integration_tests/tests/test_column_anomalies.py +++ b/integration_tests/tests/test_column_anomalies.py @@ -153,6 +153,34 @@ def test_column_anomalies_with_timestamp_as_sql_expression( assert test_result["status"] == "pass" +def test_anomalyless_numeric_column_anomalies_with_nulls( + test_id: str, dbt_project: DbtProject +): + utc_today = datetime.utcnow().date() + data: List[Dict[str, Any]] = [ + { + TIMESTAMP_COLUMN: cur_date.strftime(DATE_FORMAT), + "amount": amount, + } + for cur_date in generate_dates(base_date=utc_today - timedelta(1)) + for amount in [0, 5, 10, None] + ] + test_args = { + "timestamp_column": TIMESTAMP_COLUMN, + "column_anomalies": ["max", "min", "average", "sum", "zero_count"], + } + test_results = dbt_project.test( + test_id, + DBT_TEST_NAME, + test_args, + data=data, + test_column="amount", + multiple_results=True, + ) + assert test_results + assert all(res["status"] == "pass" for res in test_results) + + @Parametrization.autodetect_parameters() @Parametrization.case( name="true_positive", diff --git a/macros/edr/data_monitoring/monitors/column_numeric_monitors.sql b/macros/edr/data_monitoring/monitors/column_numeric_monitors.sql index 93a4db267..4629cd1ae 100644 --- a/macros/edr/data_monitoring/monitors/column_numeric_monitors.sql +++ b/macros/edr/data_monitoring/monitors/column_numeric_monitors.sql @@ -1,13 +1,27 @@ +{#- ClickHouse columns are non-Nullable by default, so casting a NULL to a + plain Float32 raises an error instead of yielding NULL. -#} +{% macro edr_cast_numeric_monitor_column(column_name) -%} + {{ adapter.dispatch("edr_cast_numeric_monitor_column", "elementary")(column_name) }} +{%- endmacro %} + +{% macro default__edr_cast_numeric_monitor_column(column_name) -%} + cast({{ column_name }} as {{ elementary.edr_type_float() }}) +{%- endmacro %} + +{% macro clickhouse__edr_cast_numeric_monitor_column(column_name) -%} + cast({{ column_name }} as Nullable({{ elementary.edr_type_float() }})) +{%- endmacro %} + {% macro max(column_name) -%} - max(cast({{ column_name }} as {{ elementary.edr_type_float() }})) + max({{ elementary.edr_cast_numeric_monitor_column(column_name) }}) {%- endmacro %} {% macro min(column_name) -%} - min(cast({{ column_name }} as {{ elementary.edr_type_float() }})) + min({{ elementary.edr_cast_numeric_monitor_column(column_name) }}) {%- endmacro %} {% macro average(column_name) -%} - avg(cast({{ column_name }} as {{ elementary.edr_type_float() }})) + avg({{ elementary.edr_cast_numeric_monitor_column(column_name) }}) {%- endmacro %} {% macro zero_count(column_name) %} @@ -16,7 +30,7 @@ case when {{ column_name }} is null then 1 - when cast({{ column_name }} as {{ elementary.edr_type_float() }}) = 0 + when {{ elementary.edr_cast_numeric_monitor_column(column_name) }} = 0 then 1 else 0 end @@ -92,7 +106,7 @@ {%- endmacro %} {% macro sum(column_name) -%} - sum(cast({{ column_name }} as {{ elementary.edr_type_float() }})) + sum({{ elementary.edr_cast_numeric_monitor_column(column_name) }}) {%- endmacro %} {#- edr_normalize_stddev – post-process a stddev column reference so that From 451a2b211a9719d5090c80eb1767da045d859a02 Mon Sep 17 00:00:00 2001 From: zerafachris <22851907+zerafachris@users.noreply.github.com> Date: Fri, 30 Oct 2026 14:02:14 +0100 Subject: [PATCH 2/4] test: assert stored numeric metric values in the NULL column test Read the max/min/average/sum/zero_count rows from data_monitoring_metrics so a NULL read as 0 (average 3.75 instead of 5) fails the test even when the anomaly checks still pass. Co-Authored-By: Claude Opus 5.5 --- .../tests/test_column_anomalies.py | 21 +++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/integration_tests/tests/test_column_anomalies.py b/integration_tests/tests/test_column_anomalies.py index 8a472702d..ccaa9cffb 100644 --- a/integration_tests/tests/test_column_anomalies.py +++ b/integration_tests/tests/test_column_anomalies.py @@ -180,6 +180,27 @@ def test_anomalyless_numeric_column_anomalies_with_nulls( assert test_results assert all(res["status"] == "pass" for res in test_results) + # NULLs must be skipped, not read as 0 (that would make average 3.75). + # zero_count counts NULLs as zeros by design. + table_name = test_id.replace("[", "_").replace("]", "_").lower() + metrics = dbt_project.run_query( + f""" + select metric_name, metric_value + from {{{{ ref("data_monitoring_metrics") }}}} + where lower(full_table_name) like '%{table_name}' + and lower(column_name) = 'amount' + """ + ) + expected = {"max": 10, "min": 0, "average": 5, "sum": 15, "zero_count": 2} + values_by_metric: Dict[str, set] = {} + for metric in metrics: + values_by_metric.setdefault(metric["metric_name"], set()).add( + float(metric["metric_value"]) + ) + assert values_by_metric == { + name: {float(value)} for name, value in expected.items() + } + @Parametrization.autodetect_parameters() @Parametrization.case( From 395e74678dc182a177718d4434d2b019e62ac239 Mon Sep 17 00:00:00 2001 From: zerafachris <22851907+zerafachris@users.noreply.github.com> Date: Sat, 31 Oct 2026 14:46:12 +0100 Subject: [PATCH 3/4] test: read only the latest metric rows and force a backfill in the NULL column test --- .../tests/test_column_anomalies.py | 17 +++++++++++++---- 1 file changed, 13 insertions(+), 4 deletions(-) diff --git a/integration_tests/tests/test_column_anomalies.py b/integration_tests/tests/test_column_anomalies.py index ccaa9cffb..b774fcb5c 100644 --- a/integration_tests/tests/test_column_anomalies.py +++ b/integration_tests/tests/test_column_anomalies.py @@ -176,6 +176,7 @@ def test_anomalyless_numeric_column_anomalies_with_nulls( data=data, test_column="amount", multiple_results=True, + test_vars={"force_metrics_backfill": True}, ) assert test_results assert all(res["status"] == "pass" for res in test_results) @@ -183,12 +184,20 @@ def test_anomalyless_numeric_column_anomalies_with_nulls( # NULLs must be skipped, not read as 0 (that would make average 3.75). # zero_count counts NULLs as zeros by design. table_name = test_id.replace("[", "_").replace("]", "_").lower() + # The metrics table is append-only, so read only the latest row per metric. metrics = dbt_project.run_query( f""" - select metric_name, metric_value - from {{{{ ref("data_monitoring_metrics") }}}} - where lower(full_table_name) like '%{table_name}' - and lower(column_name) = 'amount' + with metrics_ordered as ( + select + metric_name, + metric_value, + row_number() over (partition by id order by updated_at desc) as row_num + from {{{{ ref("data_monitoring_metrics") }}}} + where lower(full_table_name) like '%{table_name}' + and lower(column_name) = 'amount' + ) + select metric_name, metric_value from metrics_ordered + where row_num = 1 """ ) expected = {"max": 10, "min": 0, "average": 5, "sum": 15, "zero_count": 2} From 346f0a81d92b1b84d622563658bc9169386de997 Mon Sep 17 00:00:00 2001 From: zerafachris <22851907+zerafachris@users.noreply.github.com> Date: Sun, 1 Nov 2026 07:39:28 +0100 Subject: [PATCH 4/4] test: check the NULL column metrics per bucket Co-Authored-By: Claude Opus 5.5 --- .../tests/test_column_anomalies.py | 27 ++++++++++++------- 1 file changed, 18 insertions(+), 9 deletions(-) diff --git a/integration_tests/tests/test_column_anomalies.py b/integration_tests/tests/test_column_anomalies.py index b774fcb5c..7b3ede260 100644 --- a/integration_tests/tests/test_column_anomalies.py +++ b/integration_tests/tests/test_column_anomalies.py @@ -189,6 +189,7 @@ def test_anomalyless_numeric_column_anomalies_with_nulls( f""" with metrics_ordered as ( select + bucket_end, metric_name, metric_value, row_number() over (partition by id order by updated_at desc) as row_num @@ -196,19 +197,27 @@ def test_anomalyless_numeric_column_anomalies_with_nulls( where lower(full_table_name) like '%{table_name}' and lower(column_name) = 'amount' ) - select metric_name, metric_value from metrics_ordered + select bucket_end, metric_name, metric_value from metrics_ordered where row_num = 1 """ ) - expected = {"max": 10, "min": 0, "average": 5, "sum": 15, "zero_count": 2} - values_by_metric: Dict[str, set] = {} - for metric in metrics: - values_by_metric.setdefault(metric["metric_name"], set()).add( - float(metric["metric_value"]) - ) - assert values_by_metric == { - name: {float(value)} for name, value in expected.items() + expected = { + name: float(value) + for name, value in { + "max": 10, + "min": 0, + "average": 5, + "sum": 15, + "zero_count": 2, + }.items() } + values_by_bucket: Dict[str, Dict[str, float]] = {} + for metric in metrics: + values_by_bucket.setdefault(str(metric["bucket_end"]), {})[ + metric["metric_name"] + ] = float(metric["metric_value"]) + assert values_by_bucket + assert all(values == expected for values in values_by_bucket.values()) @Parametrization.autodetect_parameters()