Skip to content
Merged
Show file tree
Hide file tree
Changes from 15 commits
Commits
Show all changes
22 commits
Select commit Hold shift + click to select a range
5bb8480
feat: Make merge_columns optional in sql_query check
vb-dbrks Nov 24, 2025
43812c8
fmt & tests: sql check complexity and reuse dataset level check. Adde…
vb-dbrks Nov 24, 2025
24abf0d
improve: tighten sql_query validation + share test helper
vb-dbrks Nov 25, 2025
55383ea
fix: test assertion when totals do not match
vb-dbrks Nov 25, 2025
2f43afa
Support Custom Folder Installation for CLI Commands (#942)
ghanse Nov 24, 2025
1694a72
fix: updated test assertion failure scenario totals mismatch / datase…
vb-dbrks Nov 25, 2025
270f1ab
Merge branch 'main' into 938-feature-sql_query-should-have-merge-colu…
vb-dbrks Nov 30, 2025
25c0bce
Merge branch 'main' into 938-feature-sql_query-should-have-merge-colu…
mwojtyczka Dec 7, 2025
5f4976c
Apply suggestion from @mwojtyczka
mwojtyczka Dec 7, 2025
03a1b1b
Apply suggestion from @mwojtyczka
mwojtyczka Dec 7, 2025
49326a3
Apply suggestion from @mwojtyczka
mwojtyczka Dec 7, 2025
7d98e2a
Apply suggestion from @mwojtyczka
mwojtyczka Dec 7, 2025
623c2df
Apply suggestion from @mwojtyczka
mwojtyczka Dec 7, 2025
b2062d4
Apply suggestion from @mwojtyczka
mwojtyczka Dec 7, 2025
c935591
Merge branch 'main' into 938-feature-sql_query-should-have-merge-colu…
mwojtyczka Dec 7, 2025
b943d5d
fmt
mwojtyczka Dec 7, 2025
f707a23
Update src/databricks/labs/dqx/check_funcs.py
mwojtyczka Dec 7, 2025
24cdd3f
Update src/databricks/labs/dqx/check_funcs.py
mwojtyczka Dec 7, 2025
ded5fee
fmt
mwojtyczka Dec 7, 2025
0f18b32
Merge remote-tracking branch 'origin/938-feature-sql_query-should-hav…
mwojtyczka Dec 7, 2025
e50db33
Add pytest-benchmark performance baseline
mwojtyczka Dec 7, 2025
cddee6b
fixed tests
mwojtyczka Dec 7, 2025
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
69 changes: 66 additions & 3 deletions demos/dqx_demo_library.py
Original file line number Diff line number Diff line change
Expand Up @@ -943,11 +943,15 @@ def not_ends_with(column: str, suffix: str) -> Column:
# COMMAND ----------

# MAGIC %md
# MAGIC #### Using `sql_query` check
# MAGIC #### Using `sql_query` check - Row-level validation
# MAGIC
# MAGIC The `sql_query` check supports two modes:
# MAGIC - **Row-level validation** (with `merge_columns`): Query results are joined back to mark specific rows
# MAGIC - **Dataset-level validation** (without `merge_columns`): Check result applies to all rows

# COMMAND ----------

# using DQX classes
# Row-level validation example: Check each sensor against its threshold
from databricks.labs.dqx.rule import DQDatasetRule
from databricks.labs.dqx.check_funcs import sql_query

Expand All @@ -973,7 +977,7 @@ def not_ends_with(column: str, suffix: str) -> Column:
check_func=sql_query,
check_func_kwargs={
"query": query,
"merge_columns": ["sensor_id"],
"merge_columns": ["sensor_id"], # Results joined back by sensor_id
"condition_column": "condition", # the check fails if this column evaluates to True
"msg": "one of the sensor reading is greater than limit",
"name": "sensor_reading_check",
Expand All @@ -990,6 +994,41 @@ def not_ends_with(column: str, suffix: str) -> Column:

# COMMAND ----------

# MAGIC %md
# MAGIC #### Using `sql_query` check - Dataset-level validation
# MAGIC
# MAGIC When `merge_columns` is not provided, the check applies to all rows (all pass or all fail together).
# MAGIC This is useful for dataset-level aggregate validations.

# COMMAND ----------

# Dataset-level validation example: Check total sensor count
dataset_query = """
SELECT COUNT(DISTINCT sensor_id) < 1 AS condition
FROM {{ sensor }}
"""

checks = [
DQDatasetRule(
criticality="warn",
check_func=sql_query,
check_func_kwargs={
"query": dataset_query,
# No merge_columns = dataset-level check (all rows get same result)
"condition_column": "condition",
"msg": "Dataset has no sensors",
"name": "dataset_has_sensors",
"input_placeholder": "sensor",
},
),
]

ref_dfs = {"sensor_specs": sensor_specs_df}
valid_and_quarantine_df = dq_engine.apply_checks(sensor_df, checks, ref_dfs=ref_dfs)
display(valid_and_quarantine_df)

# COMMAND ----------

# using YAML declarative approach
checks = yaml.safe_load(
"""
Expand Down Expand Up @@ -1028,6 +1067,30 @@ def not_ends_with(column: str, suffix: str) -> Column:

# COMMAND ----------

# YAML example for dataset-level validation (without merge_columns)
checks_dataset_level = yaml.safe_load(
"""
- criticality: warn
check:
function: sql_query
arguments:
# No merge_columns = dataset-level validation
condition_column: condition
msg: Dataset has no sensors
name: dataset_has_sensors
input_placeholder: sensor
query: |
SELECT COUNT(DISTINCT sensor_id) < 1 AS condition
FROM {{ sensor }}
"""
)

ref_dfs = {"sensor_specs": sensor_specs_df}
valid_and_quarantine_df = dq_engine.apply_checks_by_metadata(sensor_df, checks_dataset_level, ref_dfs=ref_dfs)
display(valid_and_quarantine_df)

# COMMAND ----------

# MAGIC %md
# MAGIC #### Defining custom python dataset-level check

Expand Down
32 changes: 29 additions & 3 deletions docs/dqx/docs/reference/quality_checks.mdx
Original file line number Diff line number Diff line change
Expand Up @@ -1393,7 +1393,7 @@ You can also define your own custom dataset-level checks (see [Creating custom c
| `is_aggr_equal` | Checks whether the aggregated values over group of rows or all rows are equal to the provided limit. | `column`: column to check (can be a string column name or a column expression), optional for 'count' aggregation; `limit`: limit as number, column name or sql expression (string literals must be single quoted, e.g. 'string_value'); `aggr_type`: aggregation function (default: "count"), supports 20 curated functions (count, sum, avg, stddev, percentile, etc.) plus any Databricks built-in aggregate; `group_by`: (optional) list of columns or column expressions to group the rows for aggregation (no grouping by default); `row_filter`: (optional) SQL expression to filter rows before aggregation; `aggr_params`: (optional) dict of parameters for aggregates requiring them |
| `is_aggr_not_equal` | Checks whether the aggregated values over group of rows or all rows are not equal to the provided limit. | `column`: column to check (can be a string column name or a column expression), optional for 'count' aggregation; `limit`: limit as number, column name or sql expression (string literals must be single quoted, e.g. 'string_value'); `aggr_type`: aggregation function (default: "count"), supports 20 curated functions (count, sum, avg, stddev, percentile, etc.) plus any Databricks built-in aggregate; `group_by`: (optional) list of columns or column expressions to group the rows for aggregation (no grouping by default); `row_filter`: (optional) SQL expression to filter rows before aggregation; `aggr_params`: (optional) dict of parameters for aggregates requiring them |
| `foreign_key` (aka is_in_list) | Checks whether input column or columns can be found in the reference DataFrame or Table (foreign key check). It supports foreign key check on single and composite keys. This check can be used to validate whether values in the input column(s) exist in a predefined list of allowed values (stored in the reference DataFrame or Table). It serves as a scalable alternative to `is_in_list` row-level checks, when working with large lists. | `columns`: columns to check (can be a list of string column names or column expressions); `ref_columns`: columns to check for existence in the reference DataFrame or Table (can be a list string column name or a column expression); `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; negate: if True the condition is negated (i.e. the check fails when the foreign key values exist in the reference DataFrame/Table), if False the check fails when the foreign key values do not exist in the reference |
| `sql_query` | Checks whether the condition column produced by a SQL query is satisfied. The check expects the query to return a boolean condition column indicating whether a record meets the requirement (True = fail, False = pass), and one or more merge columns so that results can be joined back to the input DataFrame to preserve all original records. Important considerations: if merge columns aren't unique, multiple query rows can attach to a single input row, potentially causing false positives. Performance tip: since the check must join back to the input DataFrame to retain original records, writing a custom dataset-level rule is usually more performant than `sql_query` check. | `query`: query string, must return all merge columns and condition column; `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`: 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; `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 |
| `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`: expected schema as a DDL string (e.g., "id INT, name STRING") or StructType object; `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 |
Expand Down Expand Up @@ -1649,7 +1649,7 @@ Complex data types are supported as well.
ref_df_name: ref_df_key
negate: true

# sql_query check
# sql_query check with merge_columns (row-level validation)
- criticality: error
check:
function: sql_query
Expand All @@ -1665,6 +1665,18 @@ Complex data types are supported as well.
name: sql_query_violation # optional
negate: false # optional, default False

Comment thread
mwojtyczka marked this conversation as resolved.
# sql_query check without merge_columns (dataset-level validation)
- criticality: error
check:
function: sql_query
arguments:
# sql query for dataset-level check (must return 1 record)
query: SELECT COUNT(*) = 0 AS condition FROM {{ input_view }}
input_placeholder: input_view # name to be used in the sql query as `{{ input_view }}` to refer to the input DataFrame
condition_column: condition # the check fails if this column evaluates to True
msg: dataset has no records # optional
name: dataset_is_empty # optional

# compare_datasets check
- criticality: error
check:
Expand Down Expand Up @@ -2021,7 +2033,7 @@ checks = [
"negate": True
},

# sql_query check
# sql_query check with merge_columns (row-level validation)
DQDatasetRule(
criticality="error",
check_func=sql_query,
Expand All @@ -2037,6 +2049,20 @@ checks = [
},
),

Comment thread
mwojtyczka marked this conversation as resolved.
# sql_query check without merge_columns (dataset-level validation)
DQDatasetRule(
criticality="error",
check_func=sql_query,
check_func_kwargs={
# sql query for dataset-level check (must return 1 record)
"query": "SELECT COUNT(*) = 0 AS condition FROM {{ input_view }}",
"input_placeholder": "input_view", # name to be used in the sql query as `{{ input_placeholder }}` to refer to the input DataFrame
"condition_column": "condition", # the check fails if this column evaluates to True
"msg": "dataset has no records", # optional
"name": "dataset_is_empty", # optional
},
),

# compare_datasets check
DQDatasetRule(
criticality="error",
Expand Down
Loading
Loading