Add support for summary metrics in Lakeflow Declarative Pipelines / DLT

AGENTS.md

@@ -115,7 +115,7 @@ tests/
 - **Cover all changes with tests.** New check functions and rule logic → unit tests. Workspace interactions → integration tests. Bug fixes → regression tests.
 - **Unit tests** (`tests/unit/`) run without Spark or a live workspace and must stay fast.
 - **Integration tests** (`tests/integration/`) require a real workspace and spark session; do not add workspace API calls to unit tests.
-- Test **behaviour, not implementation details**: assert on outputs and observable state, not on private methods or internal data structures.
+- Test **behaviour, not implementation details**: assert on outputs and observable state, not on private methods or internal data structures. Do not call or access private/protected members (e.g. `obj._helper`) from tests — exercise the public API instead. If something is only reachable via a private member, that is a design smell: make it public or extract a shared helper, do not reach past the boundary (and never silence the resulting `protected-access` lint — see Critical Rule 6).
 - Use **dependency injection to enable testing**: construct dependencies with `create_autospec` rather than patching internal module state.
 - Use **pytest fixtures** (`conftest.py`) to share setup and teardown logic across tests. Unit-level fixtures live in `tests/unit/conftest.py`; integration-level fixtures in `tests/integration/conftest.py`. Do not duplicate fixture logic inline in individual tests.
 - For workspace resource creation and cleanup in integration tests, use the pytester `factory` helper — see [## Testing](#testing) for the established patterns.
@@ -162,6 +162,8 @@ Use `ConfigSerializer` — it preserves nested types. `dataclasses.asdict()` los
 
 Fix the code instead of adding `# pylint: disable`, `# type: ignore`, `# noqa`, or per-file ignores. Use project-wide exceptions in `pyproject.toml` only when there is no viable fix (e.g., third-party API compatibility).
 
+In particular, never disable `protected-access` (pylint `W0212`) — inline, per-file, or globally in `pyproject.toml` — to let tests or callers reach private/protected members (`_name`). That is a hack that masks a design smell and is technical debt. Instead, exercise the public API, or if a member genuinely needs outside access, make it public or extract a shared helper.
+
 ---
 
 ## Security Requirements

demos/dqx_dlt_demo.py

@@ -5,7 +5,12 @@
 
 # MAGIC
 # MAGIC %md
-# MAGIC ## Create Lakeflow Pipeline (formerly Delta Live Tables - DLT)
+# MAGIC ## DQX in a Lakeflow Pipeline (formerly Delta Live Tables - DLT)
+# MAGIC
+# MAGIC This demo applies DQX checks and reports issues as additional columns (`_errors` / `_warnings`),
+# MAGIC persisting the checked data as a `silver` table. Summary metrics are then computed as a materialized
+# MAGIC view over that table. For a pipeline that quarantines invalid records into a separate table, see
+# MAGIC `dqx_dlt_demo_quarantine.py`.
 # MAGIC
 # MAGIC Create new ETL Pipeline to execute this notebook (see [here](https://docs.databricks.com/aws/en/getting-started/data-pipeline-get-started)):
 # MAGIC 1. Upload the notebook to a Databricks Workspace
@@ -23,27 +28,21 @@
 
 # COMMAND ----------
 
-# MAGIC %md
-# MAGIC ## Define Lakeflow Pipeline
-
-# COMMAND ----------
-
+import yaml
 from databricks.labs.dqx.engine import DQEngine
+from databricks.labs.dqx.metrics_observer import DQMetricsObserver
 from databricks.sdk import WorkspaceClient
 
+# compute_summary_metrics requires an observer on the engine; it reads any custom_metrics from it.
+dq_engine = DQEngine(WorkspaceClient(), observer=DQMetricsObserver())
+
 # COMMAND ----------
 
-@dlt.view
-def bronze():
-  df = spark.readStream.format("delta") \
-    .load("/databricks-datasets/delta-sharing/samples/nyctaxi_2019")
-  return df
+# MAGIC %md
+# MAGIC ## Define Data Quality checks
 
 # COMMAND ----------
 
-# Define Data Quality checks
-import yaml
-
 # Define checks in YAML format. They can also be defined using classes or loaded from a file or table.
 checks = yaml.safe_load("""
 - check:
@@ -109,26 +108,31 @@ def bronze():
 
 # COMMAND ----------
 
-dq_engine = DQEngine(WorkspaceClient())
+# MAGIC %md
+# MAGIC ## Define Lakeflow Pipeline (bronze -> silver -> metrics)
+
+# COMMAND ----------
 
-# Read data from Bronze and apply checks
+# Bronze: raw input as a streaming view.
 @dlt.view
-def bronze_dq_check():
-  df = dlt.read_stream("bronze")
-  return dq_engine.apply_checks_by_metadata(df, checks)
+def bronze():
+  return spark.readStream.format("delta") \
+    .load("/databricks-datasets/delta-sharing/samples/nyctaxi_2019")
 
 # COMMAND ----------
 
-# # get rows without errors or warnings, and drop auxiliary columns
+# Silver: apply checks and report issues as additional columns (_errors / _warnings).
+# Persist as a table so it can be used downstream (including by the metrics view below).
 @dlt.table
 def silver():
-  df = dlt.read_stream("bronze_dq_check")
-  return dq_engine.get_valid(df)
+  df = dlt.read_stream("bronze")
+  return dq_engine.apply_checks_by_metadata(df, checks)
 
 # COMMAND ----------
 
-# get only rows with errors or warnings
+# Summary Metrics: materialized view computed by aggregation over the silver table. 
+# One row per metric (input / error / warning / valid row counts and per-check breakdown).
 @dlt.table
-def quarantine():
-  df = dlt.read_stream("bronze_dq_check")
-  return dq_engine.get_invalid(df)
+def dq_summary_metrics():
+  df = dlt.read("silver")
+  return dq_engine.compute_summary_metrics(df, checks=checks)

demos/dqx_dlt_demo_quarantine.py

@@ -0,0 +1,154 @@
+# Databricks notebook source
+import dlt
+
+# COMMAND ----------
+
+# MAGIC
+# MAGIC %md
+# MAGIC ## DQX in a Lakeflow Pipeline (formerly Delta Live Tables - DLT) — data quality with quarantine pattern
+# MAGIC
+# MAGIC This demo applies DQX checks and **splits** the data into a valid `silver` table and a `quarantine`
+# MAGIC table. It also persists the checked layer (`bronze_dq_check`) so summary metrics can be computed as
+# MAGIC a materialized view over all rows. For a simpler pipeline that reports issues as columns without
+# MAGIC quarantining, see `dqx_dlt_demo.py`.
+# MAGIC
+# MAGIC Create new ETL Pipeline to execute this notebook (see [here](https://docs.databricks.com/aws/en/getting-started/data-pipeline-get-started)):
+# MAGIC 1. Upload the notebook to a Databricks Workspace
+# MAGIC 2. Go to `Workflows` tab > `Create` > `ETL Pipeline` > `Add existing assets` > select the source code path and root directory
+# MAGIC 3. Add DQX library as a [dependency](https://docs.databricks.com/aws/en/dlt/dlt-multi-file-editor#environment) to the pipeline: Go to `Settings` > `Edit environment` > Add `databricks‑labs‑dqx` as dependency
+# MAGIC 4. Run the pipeline
+# MAGIC
+# MAGIC
+# MAGIC As an alternative to setting the environment as described above, you can also [install](https://docs.databricks.com/aws/en/dlt/external-dependencies) DQX directly in the notebook. Put the below commands as first cells in the notebook:
+# MAGIC
+# MAGIC %`pip install databricks-labs-dqx`
+# MAGIC
+# MAGIC `dbutils.library.restartPython()`
+# MAGIC
+
+# COMMAND ----------
+
+import yaml
+from databricks.labs.dqx.engine import DQEngine
+from databricks.labs.dqx.metrics_observer import DQMetricsObserver
+from databricks.sdk import WorkspaceClient
+
+# compute_summary_metrics requires an observer on the engine; it reads any custom_metrics from it.
+dq_engine = DQEngine(WorkspaceClient(), observer=DQMetricsObserver())
+
+# COMMAND ----------
+
+# MAGIC %md
+# MAGIC ## Define Data Quality checks
+
+# COMMAND ----------
+
+# Define checks in YAML format. They can also be defined using classes or loaded from a file or a table.
+checks = yaml.safe_load("""
+- check:
+    function: is_not_null
+    arguments:
+      column: vendor_id
+  name: vendor_id_is_null
+  criticality: error
+- check:
+    function: is_not_null_and_not_empty
+    arguments:
+      column: vendor_id
+      trim_strings: true
+  name: vendor_id_is_null_or_empty
+  criticality: error
+
+- check:
+    function: is_not_null
+    arguments:
+      column: pickup_datetime
+  name: pickup_datetime_is_null
+  criticality: error
+- check:
+    function: is_not_in_future
+    arguments:
+      column: pickup_datetime
+  name: pickup_datetime_isnt_in_range
+  criticality: warn
+
+- check:
+    function: is_not_in_future
+    arguments:
+      column: pickup_datetime
+  name: pickup_datetime_not_in_future
+  criticality: warn
+- check:
+    function: is_not_in_future
+    arguments:
+      column: dropoff_datetime
+  name: dropoff_datetime_not_in_future
+  criticality: warn
+- check:
+    function: is_not_null
+    arguments:
+      column: passenger_count
+  name: passenger_count_is_null
+  criticality: error
+- check:
+    function: is_in_range
+    arguments:
+      column: passenger_count
+      min_limit: 0
+      max_limit: 6
+  name: passenger_incorrect_count
+  criticality: warn
+- check:
+    function: is_not_null
+    arguments:
+      column: trip_distance
+  name: trip_distance_is_null
+  criticality: error
+""")
+
+# COMMAND ----------
+
+# MAGIC %md
+# MAGIC ## Define Lakeflow Pipeline (bronze -> silver + quarantine -> metrics)
+
+# COMMAND ----------
+
+# Bronze: raw input as a streaming view.
+@dlt.view
+def bronze():
+  return spark.readStream.format("delta") \
+    .load("/databricks-datasets/delta-sharing/samples/nyctaxi_2019")
+
+# COMMAND ----------
+
+# Apply checks and persist the checked data as a table so the summary-metrics materialized view
+# below can read the result columns (_errors/_warnings) after the pipeline writes them.
+@dlt.table
+def bronze_dq_check():
+  df = dlt.read_stream("bronze")
+  return dq_engine.apply_checks_by_metadata(df, checks)
+
+# COMMAND ----------
+
+# Silver: rows without errors or warnings, with the auxiliary result columns dropped.
+@dlt.table
+def silver():
+  df = dlt.read_stream("bronze_dq_check")
+  return dq_engine.get_valid(df)
+
+# COMMAND ----------
+
+# Quarantine: only rows with errors or warnings.
+@dlt.table
+def quarantine():
+  df = dlt.read_stream("bronze_dq_check")
+  return dq_engine.get_invalid(df)
+
+# COMMAND ----------
+
+# Summary Metrics: materialized view computed by aggregation over the checked table. 
+# One row per metric (input / error / warning / valid row counts and per-check breakdown).
+@dlt.table
+def dq_summary_metrics():
+  df = dlt.read("bronze_dq_check")
+  return dq_engine.compute_summary_metrics(df, checks=checks)

docs/dqx/docs/demos.mdx

@@ -18,7 +18,8 @@ Import the following notebooks in the Databricks workspace to try DQX out:
 * [DQX Demo Notebook for Data Contract Integration (ODCS)](https://github.com/databrickslabs/dqx/blob/v0.15.0/demos/dqx_demo_datacontract_odcs.py) - demonstrates how to generate DQX quality rules from ODCS (Open Data Contract Standard) data contracts, including predefined rules from schema constraints, explicit custom rules, and contract metadata tracking.
 * [DQX Demo Notebook for Spark Structured Streaming (Native End-to-End Approach)](https://github.com/databrickslabs/dqx/blob/v0.15.0/demos/dqx_streaming_demo_native.py) - demonstrates how to use DQX as a library with Spark Structured Streaming, using the built-in end-to-end method to handle both reading and writing.
 * [DQX Demo Notebook for Spark Structured Streaming (DIY Approach)](https://github.com/databrickslabs/dqx/blob/v0.15.0/demos/dqx_streaming_demo_diy.py) - demonstrates how to use DQX as a library with Spark Structured Streaming, while handling reading and writing on your own outside DQX using Spark API.
-* [DQX Demo Notebook for Lakeflow Pipelines (formerly DLT)](https://github.com/databrickslabs/dqx/blob/v0.15.0/demos/dqx_dlt_demo.py) - demonstrates how to use DQX as a library with Lakeflow Pipelines.
+* [DQX Demo Notebook for Lakeflow Pipelines (formerly DLT)](https://github.com/databrickslabs/dqx/blob/v0.15.0/demos/dqx_dlt_demo.py) - demonstrates how to use DQX in a Lakeflow Pipeline, reporting issues as additional columns and computing data quality summary metrics.
+* [DQX Demo Notebook for Lakeflow Pipelines with Quarantine](https://github.com/databrickslabs/dqx/blob/v0.15.0/demos/dqx_dlt_demo_quarantine.py) - demonstrates how to use DQX in a Lakeflow Pipeline to split valid and invalid records into valid and quarantine tables and compute summary metrics.
 * [DQX usage with Declarative Automation Bundles](https://github.com/databrickslabs/dqx/blob/v0.15.0/demos/dqx_demo_asset_bundle/README.md) - demonstrates how to use DQX as a library with Declarative Automation Bundles (formerly Databricks Asset Bundles).
 * [DQX usage with dbt](https://github.com/databrickslabs/dqx/blob/v0.15.0/demos/dqx_demo_dbt/README.md) - demonstrates how to use DQX as a library with dbt projects.
 * [DQX Demo on Primary-Key detection](https://github.com/databrickslabs/dqx/blob/v0.15.0/demos/dqx_demo_llm_pk_detection.py) - demonstrates how to detect primary keys and generate uniqueness rules.

docs/dqx/docs/guide/quality_checks_apply.mdx

@@ -1019,6 +1019,53 @@ To enable summary metrics programmatically, create and pass a `DQMetricsObserver
   </TabItem>
 </Tabs>
 
+### Enabling summary metrics in Lakeflow Pipelines
+
+Inside a Lakeflow Pipeline (Spark Declarative Pipeline / DLT), do the following:
+
+1. Persist the checked data as a table (so the result columns are available downstream).
+2. Add a materialized view that calls `compute_summary_metrics` on that table, passing the same `checks` you applied.
+
+<Tabs>
+  <TabItem value="Python" label="Python" default>
+    ```python
+    import dlt
+    from databricks.labs.dqx.engine import DQEngine
+    from databricks.labs.dqx.metrics_observer import DQMetricsObserver
+    from databricks.sdk import WorkspaceClient
+
+    # compute_summary_metrics requires an observer on the engine; it reads any custom_metrics from it.
+    dq_engine = DQEngine(WorkspaceClient(), observer=DQMetricsObserver())
+
+    @dlt.view
+    def bronze():
+      return spark.readStream.table("catalog.schema.input")
+
+    # 1. Apply checks and persist the checked data as a table.
+    @dlt.table
+    def silver():
+      df = dlt.read_stream("bronze")
+      return dq_engine.apply_checks_by_metadata(df, checks)
+
+    # 2. Compute summary metrics as a materialized view over the checked table.
+    @dlt.table
+    def dq_summary_metrics():
+      df = dlt.read("silver")
+      return dq_engine.compute_summary_metrics(df, checks=checks)
+    ```
+  </TabItem>
+</Tabs>
+
+This works for both batch and streaming pipelines. See the [Lakeflow pipeline demo](https://github.com/databrickslabs/dqx/blob/v0.15.0/demos/dqx_dlt_demo.py) for a complete example, or the [quarantine variant](https://github.com/databrickslabs/dqx/blob/v0.15.0/demos/dqx_dlt_demo_quarantine.py) that splits valid and invalid records into separate tables.
+
+<Admonition type="info" title="Summary Metrics: Snapshot vs. history">
+The example is using a materialized view to compute the summary metrics. The view is recomputed on each pipeline update, so it holds a **current snapshot** of the metrics over all rows in the checked table. Because it aggregates the whole table, `input_row_count` is the cumulative total, not a per-run count.
+
+To keep a **history** of metrics over time you need an append target: for **streaming** pipelines append per-window metrics from a windowed streaming table (recommended), and for **batch** pipelines append `compute_summary_metrics` output from a companion job. See the [Summary Metrics guide](/docs/guide/summary_metrics) for examples.
+</Admonition>
+
+**Why not the observer directly?** The observer (`observe()`) and the streaming metrics listener do not work inside a declarative pipeline: the pipeline runtime — not your code — triggers the write action, so the Spark Observation is never populated (`observation.get` stalls) and the streaming listener receives no events. DQX detects the pipeline runtime and automatically skips wiring `observe()` there. `compute_summary_metrics` computes the same metrics as an aggregation instead, but still reads from the engine's observer — so **an observer must be configured** on the engine (calling it without one raises `InvalidParameterError`), and if you need **custom metrics**, configure it with them (`DQMetricsObserver(custom_metrics=[...])`). See the [Summary Metrics guide](/docs/guide/summary_metrics) for details.
+
 ### Enabling summary metrics in DQX workflows
 
 Summary metrics can also be enabled in DQX workflows. Metrics are configured: