Extend summary metrics with per-check-name breakdowns

src/databricks/labs/dqx/engine.py

@@ -165,6 +165,10 @@ def apply_checks(
             ref_dfs,
             rule_set_fingerprint=rule_set_fingerprint,
         )
+
+        if self.observer:
+            self.observer.set_check_names([check.name for check in checks])
+
         observed_result = self._observe_metrics(result_df)
 
         if isinstance(observed_result, tuple):

src/databricks/labs/dqx/metrics_observer.py

@@ -1,4 +1,5 @@
-from dataclasses import dataclass
+import re
+from dataclasses import dataclass, field
 from datetime import datetime
 from functools import cached_property
 from typing import Any
@@ -8,6 +9,23 @@
 import pyspark.sql.functions as F
 
 
+def _sanitize_metric_alias(name: str) -> str:
+    """Sanitize a check name to produce a valid Spark SQL column alias.
+
+    Replaces any non-alphanumeric character (except underscore) with an underscore
+    and collapses consecutive underscores.
+
+    Args:
+        name: The raw check name.
+
+    Returns:
+        A sanitized string safe for use as a Spark SQL alias.
+    """
+    sanitized = re.sub(r"[^a-zA-Z0-9_]", "_", name)
+    sanitized = re.sub(r"_+", "_", sanitized)
+    return sanitized.strip("_")
+
+
 OBSERVATION_TABLE_SCHEMA = (
     "run_id string, run_name string, input_location string, output_location string, quarantine_location string, "
     "checks_location string, rule_set_fingerprint string, metric_name string, metric_value string, run_time timestamp, "
@@ -70,6 +88,7 @@ class DQMetricsObserver:
 
     _error_column_name: str = "_errors"
     _warning_column_name: str = "_warnings"
+    _check_names: list[str] = field(default_factory=list)
 
     @cached_property
     def id(self) -> str:
@@ -81,24 +100,57 @@ def id(self) -> str:
         """
         return self.id_overwrite or str(uuid4())
 
-    @cached_property
+    @property
     def metrics(self) -> list[str]:
         """
         Gets the observer metrics as Spark SQL expressions.
 
         Returns:
-            A list of Spark SQL expressions defining the observer metrics (both default and custom).
+            A list of Spark SQL expressions defining the observer metrics (both default, per-check, and custom).
         """
         default_metrics = [
             "count(1) as input_row_count",
             f"count(case when {self._error_column_name} is not null then 1 end) as error_row_count",
             f"count(case when {self._warning_column_name} is not null then 1 end) as warning_row_count",
             f"count(case when {self._error_column_name} is null and {self._warning_column_name} is null then 1 end) as valid_row_count",
         ]
+        default_metrics.extend(self._build_per_check_metrics())
         if self.custom_metrics:
             default_metrics.extend(self.custom_metrics)
         return default_metrics
 
+    def set_check_names(self, check_names: list[str]) -> None:
+        """
+        Sets the check names used to generate per-check summary metrics.
+
+        Args:
+            check_names: List of check names from the applied quality rules.
+        """
+        self._check_names = check_names
+
+    def _build_per_check_metrics(self) -> list[str]:
+        """
+        Builds per-check metric SQL expressions for each registered check name.
+
+        For each check name, generates a count of rows where that check appears in the errors or warnings array.
+
+        Returns:
+            A list of Spark SQL expressions for per-check metrics.
+        """
+        per_check_metrics: list[str] = []
+        for check_name in self._check_names:
+            safe_alias = _sanitize_metric_alias(check_name)
+            escaped_name = check_name.replace("'", "\\'")
+            per_check_metrics.append(
+                f"count(case when exists({self._error_column_name}, x -> x.name = '{escaped_name}') then 1 end) "
+                f"as {safe_alias}_error_count"
+            )
+            per_check_metrics.append(
+                f"count(case when exists({self._warning_column_name}, x -> x.name = '{escaped_name}') then 1 end) "
+                f"as {safe_alias}_warning_count"
+            )
+        return per_check_metrics
+
     @property
     def observation(self) -> Observation:
         """

tests/unit/test_observer.py

@@ -100,3 +100,91 @@ def test_dq_observer_observation_property():
     observation = observer.observation
     assert observation is not None
     assert isinstance(observation, Observation | SparkConnectObservation)
+
+
+def test_dq_observer_per_check_metrics_without_check_names():
+    """Test that no per-check metrics are generated when no check names are set."""
+    observer = DQMetricsObserver()
+    expected_default_metrics = [
+        "count(1) as input_row_count",
+        "count(case when _errors is not null then 1 end) as error_row_count",
+        "count(case when _warnings is not null then 1 end) as warning_row_count",
+        "count(case when _errors is null and _warnings is null then 1 end) as valid_row_count",
+    ]
+    assert observer.metrics == expected_default_metrics
+
+
+def test_dq_observer_per_check_metrics_with_check_names():
+    """Test that per-check metrics are generated after setting check names."""
+    observer = DQMetricsObserver()
+    observer.set_check_names(["id_is_not_null", "name_is_not_empty"])
+
+    metrics = observer.metrics
+    assert "count(1) as input_row_count" in metrics
+
+    assert (
+        "count(case when exists(_errors, x -> x.name = 'id_is_not_null') then 1 end) " "as id_is_not_null_error_count"
+    ) in metrics
+    assert (
+        "count(case when exists(_warnings, x -> x.name = 'id_is_not_null') then 1 end) "
+        "as id_is_not_null_warning_count"
+    ) in metrics
+    assert (
+        "count(case when exists(_errors, x -> x.name = 'name_is_not_empty') then 1 end) "
+        "as name_is_not_empty_error_count"
+    ) in metrics
+    assert (
+        "count(case when exists(_warnings, x -> x.name = 'name_is_not_empty') then 1 end) "
+        "as name_is_not_empty_warning_count"
+    ) in metrics
+
+
+def test_dq_observer_per_check_metrics_ordering():
+    """Test that per-check metrics appear between default and custom metrics."""
+    custom_metrics = ["avg(age) as avg_age"]
+    observer = DQMetricsObserver(custom_metrics=custom_metrics)
+    observer.set_check_names(["my_check"])
+
+    metrics = observer.metrics
+    # Default metrics come first (4 items), then per-check (2 per check), then custom
+    assert metrics[0] == "count(1) as input_row_count"
+    assert "my_check_error_count" in metrics[4]
+    assert "my_check_warning_count" in metrics[5]
+    assert metrics[6] == "avg(age) as avg_age"
+
+
+def test_dq_observer_per_check_metrics_with_custom_column_names():
+    """Test that per-check metrics use custom error/warning column names."""
+    observer = DQMetricsObserver()
+    observer.set_column_names(error_column_name="dq_errors", warning_column_name="dq_warnings")
+    observer.set_check_names(["my_check"])
+
+    metrics = observer.metrics
+    assert (
+        "count(case when exists(dq_errors, x -> x.name = 'my_check') then 1 end) " "as my_check_error_count"
+    ) in metrics
+    assert (
+        "count(case when exists(dq_warnings, x -> x.name = 'my_check') then 1 end) " "as my_check_warning_count"
+    ) in metrics
+
+
+def test_dq_observer_per_check_metrics_sanitizes_special_characters():
+    """Test that check names with special characters are sanitized in metric aliases."""
+    observer = DQMetricsObserver()
+    observer.set_check_names(["check-with-dashes", "check.with.dots"])
+
+    metrics = observer.metrics
+    # Aliases should be sanitized: dashes and dots become underscores
+    assert any("check_with_dashes_error_count" in m for m in metrics)
+    assert any("check_with_dots_error_count" in m for m in metrics)
+
+
+def test_dq_observer_set_check_names_replaces_previous():
+    """Test that calling set_check_names replaces previously set names."""
+    observer = DQMetricsObserver()
+    observer.set_check_names(["first_check"])
+    assert any("first_check_error_count" in m for m in observer.metrics)
+
+    observer.set_check_names(["second_check"])
+    assert not any("first_check_error_count" in m for m in observer.metrics)
+    assert any("second_check_error_count" in m for m in observer.metrics)