Add support for summary metrics in Lakeflow Declarative Pipelines / DLT by mwojtyczka · Pull Request #1301 · databrickslabs/dqx

mwojtyczka · 2026-07-03T14:36:12Z

Changes

Adds aggregation-based data quality summary metrics that work inside Spark Declarative Pipelines (SDP / Lakeflow / DLT), where the existing observe()-based observer and the streaming metrics listener cannot be used because the pipeline runtime — not the caller — owns the write action.

New DQEngine.compute_summary_metrics(...) — computes the same metrics as the observer path (row counts, per-check breakdown, and observer custom metrics) but as a lazy aggregation over the result columns, returning a DataFrame with the OBSERVATION_TABLE_SCHEMA. This can back a materialized view over the checked table inside a pipeline. It uses the engine's DQMetricsObserver (including its custom_metrics) and raises InvalidParameterError when no observer is configured.
DQEngineCore._observe_metrics now detects a pipeline context (via is_dlt_pipeline) and skips wiring observe() there, returning the DataFrame unchanged. This keeps return engine.apply_checks*(df, checks) usable directly as a @dlt.table definition even with an observer configured; metrics are computed downstream by compute_summary_metrics.
DQMetricsObserver.build_metrics_df_from_aggregation — reshapes the one-row wide aggregation into the long OBSERVATION_TABLE_SCHEMA without triggering an action; robust to dotted custom-metric aliases.
Docs updated (summary_metrics.mdx, quality_checks_apply.mdx, demos.mdx) and two DLT demos added (report-as-columns and quarantine variants).
Updated demos to show how to use summary metrics for annotation and quarantine scenarios
AGENTS.md: strengthened the testing guidance on not accessing private/protected members from tests

Quarantine scenario:

Annotation scenario:

Linked issues

Resolves #1291
Resolves #947

Tests

Documentation and Demos

added/updated demos
added/updated docs
added/updated agent skills

Add DQEngine.compute_summary_metrics, which computes the same summary metrics as the observer/listener path but as a lazy aggregation over the result columns, returning a DataFrame with the OBSERVATION_TABLE_SCHEMA. Unlike observe()/streaming-listener metrics (which require the caller to own the write action), this can back a materialized view inside a Spark Declarative Pipeline, where the pipeline runtime owns the write. DQEngineCore._observe_metrics now detects an SDP/Lakeflow/DLT context via is_dlt_pipeline and skips wiring observe() there (returning the DataFrame unchanged) so apply_checks* can be returned directly from a @dlt.table definition even with an observer configured. compute_summary_metrics uses the engine's observer (incl. its custom_metrics) and raises InvalidParameterError when none is configured. Includes two DLT demos (report-as-columns and quarantine), docs updates, and unit/integration/e2e tests. Co-authored-by: Isaac

codecov · 2026-07-03T14:43:16Z

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 92.22%. Comparing base (67138ec) to head (d534894).
⚠️ Report is 1 commits behind head on main.

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #1301      +/-   ##
==========================================
- Coverage   92.56%   92.22%   -0.35%     
==========================================
  Files         102      102              
  Lines       10429    10461      +32     
==========================================
- Hits         9654     9648       -6     
- Misses        775      813      +38

Flag	Coverage Δ
anomaly	`18.86% <0.00%> (-35.17%)`	⬇️
anomaly-serverless	`20.92% <0.00%> (-33.11%)`	⬇️
integration	`49.23% <75.67%> (-1.02%)`	⬇️
integration-serverless	`50.87% <75.67%> (+0.25%)`	⬆️
unit	`58.87% <32.43%> (+0.03%)`	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

github-actions · 2026-07-03T15:20:13Z

✅ 805/805 passed, 42 skipped, 5h39m46s total

_{Running from acceptance #5170}

github-actions · 2026-07-03T15:21:40Z

✅ 178/178 passed, 2 skipped, 6h48m51s total

_{Running from anomaly #1284}

test_run_dqx_row_anomaly_detection_demo is the slowest e2e demo (trains two IsolationForest models and scores with contributions + AI explanations) and can exceed the previous 30-minute SDK job-wait on a cold serverless start, causing spurious TimeoutError failures. Raise both waits to 45 minutes, still well within the e2e CI job's 2h wrapper. Co-authored-by: Isaac

Note in the Snapshot vs. history admonition that a materialized view is single-query/single-pipeline and cannot be shared: unionByName the compute_summary_metrics results inside one MV to centralize several tables in a pipeline, and use the companion-job append into a plain Delta table (safe concurrent appends) to centralize across pipelines. Co-authored-by: Isaac

… run overwrites Add an integration test exercising build_metrics_df_from_aggregation's two previously-uncovered branches: the current_timestamp() run_time fallback (no run_time_overwrite) and the user_metadata map. These are the only new lines the integration suite missed for compute_summary_metrics. Also document that, as for DQX result materialized views, run_time_overwrite and run_id_overwrite must be set as static values in ExtraParams for the summary-metrics materialized view to be incrementally refreshable in Lakeflow (otherwise run_time defaults to current_timestamp() and run_id to a random UUID, so every update looks new). Co-authored-by: Isaac

ghanse

Overall LGTM, left a few comments regarding documentation and usage patterns.

ghanse · 2026-07-07T16:52:29Z

-  df = dlt.read_stream("bronze_dq_check")
-  return dq_engine.get_invalid(df)
+def dq_summary_metrics():
+  df = dlt.read("silver")


This defines an MV on the entire silver table with all of its historical data and does not scope the metrics to the latest run or micro-batch. Should we show a foreachBatch sink that reads data with the streaming APIs instead? This seems more scalable.

You are right, although foreachBatch is more complex and is for streaming pipelines only. I guess we should document all 3 approaches:

MV — simple, fully managed, broad runtime support, gives current-state snapshot (scalable when incrementalizable). Default for a current-quality view and the simplest starting point.

foreach_batch sink — recommended for large/streaming/incremental tables and when you want per-run/per-batch history; needs a streaming source, a recent runtime, and an idempotent write.

windowed streaming table — per-time-window history.

As for the demo, I'm not sure, maybe we can have a separate one with foreach_batch? the MV one will definitely be simpler.

ghanse · 2026-07-07T16:53:01Z

+# 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")


Same applies here.

ghanse · 2026-07-07T17:00:29Z

+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.


An MV will recompute metrics over the entire checked table. There are 2 possible challenges with this:

It might not be scalable for large tables where data is checked incrementally

MV will not have per-run or per-micro-batch quality results, the metrics will be recomputed every time

Should we instruct users to use a foreachBatch sink in these scenarios?

ghanse · 2026-07-07T17:19:45Z

+**Streaming pipelines (recommended):** append per-window metrics from a windowed streaming table. Add a watermark on a timestamp column and aggregate the observer's metric expressions per time window, so each window's metrics are appended as it completes:
+
+```python
+import pyspark.sql.functions as F
+from databricks.labs.dqx.metrics_observer import DQMetricsObserver
+
+observer = DQMetricsObserver()  # pass check names to get_metrics(check_names) to also include the per-check breakdown
+
+@dlt.table
+def dq_summary_metrics():
+    df = dlt.read_stream("silver").withWatermark("event_time", "10 minutes")
+    metric_exprs = [F.expr(m) for m in observer.get_metrics()]
+    return df.groupBy(F.window("event_time", "1 hour")).agg(*metric_exprs)
+```
+
+This needs a timestamp column to window on (an event or ingestion time) and a watermark; each window's metrics are emitted only once the watermark passes it, and very late rows are dropped. The output carries a `window` column plus the metric columns (it is not the long `OBSERVATION_TABLE_SCHEMA` shape) — so it is **not** schema-compatible with the companion-job output below, and the two cannot be appended into the same metrics table. Pick one history strategy per table.


ForeachBatch sink may be useful to avoid recomputing the results for large tables.

Reconcile the "Snapshot vs. history" note with the "Incremental materialized views" admonition: the metrics MV refreshes incrementally when the query is deterministic (static run_time_overwrite / run_id_overwrite) and the pipeline engine can incrementalize the aggregation, otherwise it fully recomputes — and either way it is a cumulative snapshot, not per-run. Add the same caveat to both DLT demos, pointing at the windowed streaming-table option for per-run / per-window metrics on large or incrementally-checked tables. Co-authored-by: Isaac

mwojtyczka requested a review from a team as a code owner July 3, 2026 14:36

mwojtyczka requested review from nehamilak-db and removed request for a team July 3, 2026 14:36

mwojtyczka changed the title ~~Add aggregation-based summary metrics for Spark Declarative Pipelines~~ Add Summary metrics for Spark Declarative Pipelines / DLT Jul 3, 2026

mwojtyczka changed the title ~~Add Summary metrics for Spark Declarative Pipelines / DLT~~ Add support for summary metrics in Spark Declarative Pipelines / DLT Jul 3, 2026

mwojtyczka changed the title ~~Add support for summary metrics in Spark Declarative Pipelines / DLT~~ Add support for summary metrics in Lakeflow Declarative Pipelines / DLT Jul 3, 2026

mwojtyczka requested a review from ghanse July 3, 2026 14:37

mwojtyczka temporarily deployed to tool July 3, 2026 14:37 — with GitHub Actions Inactive

mwojtyczka had a problem deploying to tool July 3, 2026 14:37 — with GitHub Actions Failure

mwojtyczka temporarily deployed to tool July 3, 2026 14:37 — with GitHub Actions Inactive

This was referenced Jul 3, 2026

[FEATURE]: DQMetricsObserve is incompatible with Spark Declarative Pipelines (SDP) #1291

Open

[BUG]: Materialize metrics in data streaming lakeflow pipelines #947

Open

mwojtyczka changed the title ~~Add support for summary metrics in Lakeflow Declarative Pipelines / DLT~~ Add support for summary metrics to Lakeflow Declarative Pipelines / DLT Jul 3, 2026

mwojtyczka changed the title ~~Add support for summary metrics to Lakeflow Declarative Pipelines / DLT~~ Add support for summary metrics in Lakeflow Declarative Pipelines / DLT Jul 3, 2026

mwojtyczka had a problem deploying to tool July 3, 2026 16:02 — with GitHub Actions Failure

mwojtyczka had a problem deploying to tool July 3, 2026 16:08 — with GitHub Actions Failure

mwojtyczka had a problem deploying to tool July 3, 2026 16:52 — with GitHub Actions Failure

mwojtyczka temporarily deployed to tool July 3, 2026 16:52 — with GitHub Actions Inactive

mwojtyczka had a problem deploying to tool July 3, 2026 16:52 — with GitHub Actions Failure

mwojtyczka temporarily deployed to tool July 3, 2026 17:17 — with GitHub Actions Inactive

mwojtyczka temporarily deployed to tool July 3, 2026 18:52 — with GitHub Actions Inactive

ghanse reviewed Jul 7, 2026

View reviewed changes

ghanse added the under-review This PR is currently being reviewed by one of DQX maintainers. label Jul 7, 2026

mwojtyczka temporarily deployed to tool July 12, 2026 22:38 — with GitHub Actions Inactive

mwojtyczka had a problem deploying to tool July 12, 2026 22:38 — with GitHub Actions Failure

mwojtyczka temporarily deployed to tool July 12, 2026 22:38 — with GitHub Actions Inactive

Merge branch 'main' into feat/metrics-sdp-support

2965c1b

mwojtyczka had a problem deploying to tool July 16, 2026 21:57 — with GitHub Actions Failure

Merge branch 'main' into feat/metrics-sdp-support

d534894

mwojtyczka temporarily deployed to tool July 17, 2026 10:36 — with GitHub Actions Inactive

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Add support for summary metrics in Lakeflow Declarative Pipelines / DLT#1301

Add support for summary metrics in Lakeflow Declarative Pipelines / DLT#1301
mwojtyczka wants to merge 7 commits into
mainfrom
feat/metrics-sdp-support

mwojtyczka commented Jul 3, 2026 •

edited

Loading

Uh oh!

codecov Bot commented Jul 3, 2026 •

edited

Loading

Uh oh!

github-actions Bot commented Jul 3, 2026 •

edited

Loading

Uh oh!

github-actions Bot commented Jul 3, 2026 •

edited

Loading

Uh oh!

ghanse left a comment

Uh oh!

ghanse Jul 7, 2026

Uh oh!

mwojtyczka Jul 12, 2026 •

edited

Loading

Uh oh!

ghanse Jul 7, 2026

Uh oh!

ghanse Jul 7, 2026

Uh oh!

ghanse Jul 7, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Uh oh!

Conversation

mwojtyczka commented Jul 3, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Changes

Linked issues

Tests

Documentation and Demos

Uh oh!

codecov Bot commented Jul 3, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

github-actions Bot commented Jul 3, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

github-actions Bot commented Jul 3, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

ghanse left a comment

Choose a reason for hiding this comment

Uh oh!

ghanse Jul 7, 2026

Choose a reason for hiding this comment

Uh oh!

mwojtyczka Jul 12, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ghanse Jul 7, 2026

Choose a reason for hiding this comment

Uh oh!

ghanse Jul 7, 2026

Choose a reason for hiding this comment

Uh oh!

ghanse Jul 7, 2026

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

mwojtyczka commented Jul 3, 2026 •

edited

Loading

codecov Bot commented Jul 3, 2026 •

edited

Loading

github-actions Bot commented Jul 3, 2026 •

edited

Loading

github-actions Bot commented Jul 3, 2026 •

edited

Loading

mwojtyczka Jul 12, 2026 •

edited

Loading