Feature/validate upstream tables

[STEFANOVIVAS]

Changes

validate_upstream_table added to check_funcs.py — a new dataset-level check that compares an aggregate metric (row count by default, or any curated/built-in aggregate) computed on the checked DataFrame against the same aggregate computed on a reference (upstream) table or DataFrame, with optional abs_tolerance/rel_tolerance and independent row_filter/ref_row_filter on each side.

Linked issues

Resolves #1046

Tests

• manually tested
• added unit tests
• added integration tests
• added end-to-end tests
• added performance tests

Documentation and Demos

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

[CLAassistant]

All committers have signed the CLA.

[Copilot]

Pull request overview

This PR introduces a new dataset-level quality check, validate_upstream_table, to validate ingestion correctness by comparing an aggregate metric (row count by default, or other curated/built-in aggregates) computed on the checked DataFrame against the same metric computed on an upstream reference (DataFrame via ref_df_name or Unity Catalog table via ref_table), with optional absolute/relative tolerances and independent row filters.

Changes:

• Added validate_upstream_table to check_funcs.py, including parameter validation, aggregate evaluation on both sides, and tolerance/null-safe comparison logic.
• Added unit and integration tests covering parameter validation, tolerance behavior, row filtering, '*'/count(*) usage, and ref-table vs ref-DataFrame paths.
• Documented the new check in the quality checks reference, including YAML and Python usage examples.

Reviewed changes

Copilot reviewed 4 out of 4 changed files in this pull request and generated 1 comment.

File	Description
src/databricks/labs/dqx/check_funcs.py	Adds the new dataset-level upstream aggregate comparison check implementation.
tests/unit/test_dataset_checks.py	Adds rule-build-time parameter validation tests for the new check.
tests/integration/test_dataset_checks.py	Adds end-to-end Spark integration tests for the new check’s runtime behavior.
docs/dqx/docs/reference/quality_checks.mdx	Documents the new check and adds usage examples.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[ghanse]

Headed in the right direction. Left some comments. Do you also want to add a similar check for a custom SQL query?

[ghanse]

I think a naming convention like "matches_upstream_table" might be clearer. For example aggr_matches_upstream_table for this check.

[mwojtyczka]

Maybe keep it even more generic: aggr_matches_upstream_dataset since it can be a table or a dataframe

[STEFANOVIVAS]

In fact there isn’t even an obligation to be an upstream table, right? It’s just as just a validation between tables/dataframes.

[mwojtyczka]

@STEFANOVIVAS Correct, can you pls rename it to aggr_matches_dataset? this will also require update to docs and docstrings. We can have upstream table validation as one of the examples where this could be used.

[mwojtyczka]

Review of validate_upstream_table. Useful check and largely correct; main themes are: it reimplements existing aggregate machinery (_is_aggr_compare + _get_ref_df) rather than composing it, and a few edges around filter validation, the empty-upstream message, and the exact-equality default. Inline below.

Cleared: rel_tolerance has no div-by-zero (guarded by max()); the apply closure captures spark/ref_dfs correctly (standard dataset-check pattern); .limit(1) matches _is_aggr_compare; behavioral coverage lives in the integration tests (dataset checks can't be unit-tested without Spark).

[mwojtyczka]

Going in the right direction, left some comments

[mwojtyczka]

LGTM. Just one request to rename the function to aggr_matches_dataset to make it more generic and update docs and docs strings accordingly

[STEFANOVIVAS]

LGTM. Just one request to rename the function to aggr_matches_dataset to make it more generic and update docs and docs strings accordingly

Hi, @mwojtyczka! One thing to point out: Should we implement a group by clause in this check? _is_aggr_compare function already let us apply it in the input Dataframe. Could make sense because in many situations we need to make a per-group reconciliation (date, region,source_system,etc).

[mwojtyczka]

LGTM. Just one request to rename the function to aggr_matches_dataset to make it more generic and update docs and docs strings accordingly

Hi, @mwojtyczka! One thing to point out: Should we implement a group by clause in this check? _is_aggr_compare function already let us apply it in the input Dataframe. Could make sense because in many situations we need to make a per-group reconciliation (date, region,source_system,etc).

yes makes sense