Skip to content
Merged
Show file tree
Hide file tree
Changes from 9 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
31 changes: 30 additions & 1 deletion docs/dqx/docs/reference/quality_checks.mdx
Original file line number Diff line number Diff line change
Expand Up @@ -1654,7 +1654,7 @@ You can also define your own custom dataset-level checks (see [Creating custom c
| `sql_query` | Checks whether the condition column produced by a SQL query is satisfied. The check supports two modes: **Row-level validation** (when `merge_columns` is provided) - query results are joined back to the input DataFrame to mark specific rows; **Dataset-level validation** (when `merge_columns` is None or empty) - the check result applies to all rows (or filtered rows if `row_filter` is used), making it ideal for aggregate validations with custom metrics. The query must return a boolean condition column (True = fail, False = pass). For row-level checks: if merge columns aren't unique, multiple query rows can attach to a single input row, potentially causing false positives. Performance tip: for complex queries, writing a custom dataset-level rule is usually more performant than `sql_query` check. | `query`: query string, must return condition column (and merge columns if provided); `input_placeholder`: name to be used in the sql query as `{{ input_placeholder }}` to refer to the input DataFrame, optional reference DataFrames are referred by the name provided in the dictionary of reference DataFrames (e.g. `{{ ref_df_key }}`, dictionary of DataFrames can be passed when applying checks); `merge_columns`: (optional) list of columns used for merging with the input DataFrame which must exist in the input DataFrame and be present in output of the sql query; when not provided (None or empty list), the check result applies to all rows in the dataset (dataset-level validation); `condition_column`: name of the column indicating a violation (False = pass, True = fail); `msg`: (optional) message to output; `name`: (optional) name of the resulting check (it can be overwritten by `name` specified at the check level); `negate`: if the condition should be negated |
| `compare_datasets` | Compares two DataFrames at both row and column levels, providing detailed information about differences, including new or missing rows and column-level changes. Only columns present in both the source and reference DataFrames are compared. Use with caution if `check_missing_records` is enabled, as this may increase the number of rows in the output beyond the original input DataFrame. The comparison does not support Map types (any column comparison on map type is skipped automatically). Comparing datasets is valuable for validating data during migrations, detecting drift, performing regression testing, or verifying synchronization between source and target systems. | `columns`: columns to use for row matching with the reference DataFrame (can be a list of string column names or column expressions, but only simple column expressions are allowed such as 'F.col("col1")'), if not having primary keys or wanting to match against all columns you can pass 'df.columns'; `ref_columns`: list of columns in the reference DataFrame or Table to row match against the source DataFrame (can be a list of string column names or column expressions, but only simple column expressions are allowed such as 'F.col("col1")'), if not having primary keys or wanting to match against all columns you can pass 'ref_df.columns'; note that `columns` are matched with `ref_columns` by position, so the order of the provided columns in both lists must be exactly aligned; `exclude_columns`: (optional) list of columns to exclude from the value comparison but not from row matching (can be a list of string column names or column expressions, but only simple column expressions are allowed such as 'F.col("col1")'); the `exclude_columns` field does not alter the list of columns used to determine row matches (columns), it only controls which columns are skipped during the value comparison; `ref_df_name`: (optional) name of the reference DataFrame (dictionary of DataFrames can be passed when applying checks); `ref_table`: (optional) fully qualified reference table name; either `ref_df_name` or `ref_table` must be provided but never both; the number of passed `columns` and `ref_columns` must match and keys are checks in the given order; `check_missing_records`: perform a FULL OUTER JOIN to identify records that are missing from source or reference DataFrames, default is False; use with caution as this may increase the number of rows in the output, as unmatched rows from both sides are included; `null_safe_row_matching`: (optional) treat NULLs as equal when matching rows using `columns` and `ref_columns` (default: True); `null_safe_column_value_matching`: (optional) treat NULLs as equal when comparing column values (default: True) |
| `is_data_fresh_per_time_window` | Freshness check that validates whether at least X records arrive within every Y-minute time window. | `column`: timestamp column (can be a string column name or a column expression); `window_minutes`: time window in minutes to check for data arrival; `min_records_per_window`: minimum number of records expected per time window; `lookback_windows`: (optional) number of time windows to look back from `curr_timestamp`, it filters records to include only those within the specified number of time windows from `curr_timestamp` (if no lookback is provided, the check is applied to the entire dataset); `curr_timestamp`: (optional) current timestamp column (if not provided, current_timestamp() function is used) |
| `has_valid_schema` | Schema check that validates whether the DataFrame schema matches an expected schema. In non-strict mode, validates that all expected columns exist with compatible types (allows extra columns). In strict mode, validates exact schema match (same columns, same order, same types) for all columns by default or for all columns specified in `columns`. This check is applied at the dataset level and reports schema violations for all rows in the DataFrame when incompatibilities are detected. | `expected_schema`: (optional) expected schema as a DDL string (e.g., "id INT, name STRING") or StructType object; `ref_df_name`: (optional) name of the reference DataFrame to load the schema from (dictionary of DataFrames can be passed when applying checks); `ref_table`: (optional) fully qualified reference table name to load the schema from (e.g. "catalog.schema.table"); exactly one of `expected_schema`, `ref_df_name`, or `ref_table` must be provided; `columns`: (optional) list of columns to validate (if not provided, all columns are considered); `strict`: (optional) whether to perform strict schema validation (default: False) - False: validates that all expected columns exist with compatible types, True: validates exact schema match |
| `has_valid_schema` | Schema check that validates whether the DataFrame schema matches an expected schema. In non-strict mode, validates that all expected columns exist with compatible types (allows extra columns). In strict mode, validates exact schema match (same columns, same order, same types) for all columns by default or for all columns specified in `columns`. This check is applied at the dataset level and reports schema violations for all rows in the DataFrame when incompatibilities are detected. All columns in the `ignore_columns` list will be ignored even if the column is present in the `columns` list. | `expected_schema`: (optional) expected schema as a DDL string (e.g., "id INT, name STRING") or StructType object; `ref_df_name`: (optional) name of the reference DataFrame to load the schema from (dictionary of DataFrames can be passed when applying checks); `ref_table`: (optional) fully qualified reference table name to load the schema from (e.g. "catalog.schema.table"); exactly one of `expected_schema`, `ref_df_name`, or `ref_table` must be provided; `columns`: (optional) list of columns to validate (if not provided, all columns are considered); `strict`: (optional) whether to perform strict schema validation (default: False) - False: validates that all expected columns exist with compatible types, True: validates exact schema match; `ignore_columns`: (optional) list of columns to ignore during validation (if not provided, all columns are considered); |
| `has_no_outliers` | Checks whether the values in the input column contain any outliers. This function implements a median absolute deviation (MAD) algorithm to find outliers. | `column`: column of type numeric to check (can be a string column name or a column expression); |

**Compare datasets check**
Expand Down Expand Up @@ -2001,6 +2001,16 @@ Complex data types are supported as well.
- id
- name

# has_valid_schema check with specific ignored columns
- criticality: warn
check:
function: has_valid_schema
arguments:
expected_schema: "id INT, name STRING, age INT, contact_info STRUCT<email: STRING, phone: STRING, address: STRING>"
ignore_columns:
- last_update_date
- last_updated_by

# has_valid_schema check using reference table
- criticality: error
check:
Expand Down Expand Up @@ -2456,6 +2466,25 @@ checks = [
},
),

# has_valid_schema check with specific ignored columns, expected schema defined using StructType
DQDatasetRule(
criticality="warn",
check_func=check_funcs.has_valid_schema,
check_func_kwargs={
"expected_schema": StructType([
StructField("id", IntegerType(), True),
StructField("name", StringType(), True),
StructField("age", IntegerType(), True),
StructField("contact_info", StructType([
StructField("email", StringType(), True),
StructField("phone", StringType(), True),
StructField("address", StringType(), True),
]), True)
]),
"ignore_columns": ["last_update_date", "last_updated_by"],
},
),

# has_valid_schema check using reference table
DQDatasetRule(
criticality="error",
Expand Down
16 changes: 15 additions & 1 deletion src/databricks/labs/dqx/check_funcs.py
Original file line number Diff line number Diff line change
Expand Up @@ -1931,6 +1931,7 @@ def has_valid_schema(
ref_table: str | None = None,
columns: list[str | Column] | None = None,
strict: bool = False,
ignore_columns: list[str] | None = None,
) -> tuple[Column, Callable]:
"""
Build a schema compatibility check condition and closure for dataset-level validation.
Expand All @@ -1940,6 +1941,8 @@ def has_valid_schema(
In strict mode, validates that the schema matches exactly (same columns, same order, same types)
for the columns specified in columns or for all columns if columns is not specified.

All columns in the `ignore_columns` list will be ignored even if the column is present in the `columns` list.

Args:
expected_schema: Expected schema as a DDL string (e.g., "id INT, name STRING") or StructType object.
ref_df_name: Name of the reference DataFrame (used when passing DataFrames directly).
Expand All @@ -1948,6 +1951,8 @@ def has_valid_schema(
strict: Whether to perform strict schema validation (default: False).
- False: Validates that all expected columns exist with compatible types (allows extra columns)
- True: Validates exact schema match (same columns, same order, same types)
ignore_columns: Optional list of column names in the checked DataFrame schema to
ignore for validation.

Returns:
A tuple of:
Expand Down Expand Up @@ -1976,6 +1981,12 @@ def has_valid_schema(
if columns:
column_names = [get_column_name_or_alias(col) if not isinstance(col, str) else col for col in columns]

ignore_column_names: list[str] | None = None
if ignore_columns:
ignore_column_names = [
get_column_name_or_alias(col) if not isinstance(col, str) else col for col in ignore_columns
]

expected_schema = _get_schema(expected_schema or types.StructType(), column_names)

unique_str = uuid.uuid4().hex # make sure any column added to the dataframe is unique
Expand Down Expand Up @@ -2003,7 +2014,10 @@ def apply(df: DataFrame, spark: SparkSession, ref_dfs: dict[str, DataFrame]) ->
else:
_expected_schema = expected_schema

actual_schema = df.select(*columns).schema if columns else df.schema
selected_column_names = column_names if column_names else df.columns
if ignore_column_names:
selected_column_names = [col for col in selected_column_names if col not in ignore_column_names]

Copilot AI Jan 22, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The list comprehension on line 2019 performs O(n*m) lookups where n is the number of selected columns and m is the number of ignore columns. Convert ignore_column_names to a set before the list comprehension for O(n) performance: ignore_set = set(ignore_column_names) then use if col not in ignore_set.

Suggested change
selected_column_names = [col for col in selected_column_names if col not in ignore_column_names]
ignore_set = set(ignore_column_names)
selected_column_names = [col for col in selected_column_names if col not in ignore_set]

Copilot uses AI. Check for mistakes.

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

good feedback

actual_schema = df.select(*selected_column_names).schema

if strict:
errors = _get_strict_schema_comparison(actual_schema, _expected_schema)
Expand Down
32 changes: 31 additions & 1 deletion src/databricks/labs/dqx/engine.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
import logging
from concurrent import futures
from collections.abc import Callable
from dataclasses import replace
from datetime import datetime
from functools import cached_property
from typing import Any
Expand All @@ -14,6 +15,7 @@
from pyspark.sql.streaming import StreamingQuery

from databricks.labs.dqx.base import DQEngineBase, DQEngineCoreBase
from databricks.labs.dqx import check_funcs
from databricks.labs.dqx.checks_resolver import resolve_custom_check_functions_from_path
from databricks.labs.dqx.checks_serializer import deserialize_checks
from databricks.labs.dqx.config_serializer import ConfigSerializer
Expand Down Expand Up @@ -345,6 +347,29 @@ def _all_are_dq_rules(checks: list[DQRule]) -> bool:
"""Check if all elements in the checks list are instances of DQRule."""
return all(isinstance(check, DQRule) for check in checks)

def _preselect_schema_validation_columns(self, df: DataFrame, check: DQRule) -> DQRule:
Comment thread
ghanse marked this conversation as resolved.
Outdated
"""
Determines the selected columns for schema validation checks. This is required when using `has_valid_schema` to
ignore columns added during quality checking. This includes:
* DQX result columns (e.g. '_warnings' and '_errors')
* Internal columns used for dataset-level checks

Args:
df: Input DataFrame
check: DQRule to be modified
"""

# Default columns to all columns of the current DataFrame if not explicitly set
if check.check_func_kwargs.get("columns"):
return check

Comment thread
ghanse marked this conversation as resolved.
if check.check_func_args and len(check.check_func_args) >= 3:

@mwojtyczka mwojtyczka Jan 22, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I would remove this condition to make it more generic. We will have to use this function for anomaly detection as well, which requires a list of columns.

There are a couple of issues with using check_func_args here:

  • makes an assumption of the underlying implementation. Hard to maintain.
  • makes an assumption that the function is executed within the context of has_valid_schema. So there is potential for misuse.

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I used inspect here to handle if the schema is passed as an ordinal argument.

return check

rule_kwargs = check.check_func_kwargs.copy()
rule_kwargs["columns"] = [col for col in df.columns if col not in set(self._result_column_names.values())]
return replace(check, check_func_kwargs=rule_kwargs)

def _append_empty_checks(self, df: DataFrame) -> DataFrame:
"""Append empty checks at the end of DataFrame.

Expand Down Expand Up @@ -389,8 +414,13 @@ def _create_results_array(
current_df = df

for check in checks:
normalized_check = (

@mwojtyczka mwojtyczka Jan 22, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think we should make it more generic, since more checks will require this in the future. This will also allow the creation of custom functions that operate on the full schema.

@register_rule("dataset")
@full_schema_rule  # -> register in FULL_SCHEMA_CHECK_FUNC_REGISTRY
def has_valid_schema(
...

@register_rule("dataset")
@full_schema_rule 
def has_no_anomalies(
...

# any custom check function
@register_rule("dataset")
@full_schema_rule 
def custom_func(
normalized_check = (
    self._preselect_schema_validation_columns(df, check)
    if check.check_func in FULL_SCHEMA_CHECK_FUNC_REGISTRY
    else check
)

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Updated

self._preselect_schema_validation_columns(df, check)
if check.check_func is check_funcs.has_valid_schema
else check
)
manager = DQRuleManager(
check=check,
check=normalized_check,
df=current_df,
spark=self.spark,
run_id=self.run_id,
Expand Down
Loading
Loading