fix: cast numeric monitors to Nullable on ClickHouse (#1094)

integration_tests/tests/test_column_anomalies.py

@@ -153,6 +153,73 @@ 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,
+        test_vars={"force_metrics_backfill": True},
+    )
+    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()
+    # The metrics table is append-only, so read only the latest row per metric.
+    metrics = dbt_project.run_query(
+        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
+            from {{{{ ref("data_monitoring_metrics") }}}}
+            where lower(full_table_name) like '%{table_name}'
+            and lower(column_name) = 'amount'
+        )
+        select bucket_end, metric_name, metric_value from metrics_ordered
+        where row_num = 1
+        """
+    )
+    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()
 @Parametrization.case(
     name="true_positive",

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