From 081029ff8418722a4463cab4901582c9f3de7c45 Mon Sep 17 00:00:00 2001 From: Roshan1299 Date: Tue, 10 Mar 2026 02:53:49 -0600 Subject: [PATCH 01/12] feat(schema): add skipped boolean field to dq_result_item_schema --- src/databricks/labs/dqx/schema/dq_result_schema.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/databricks/labs/dqx/schema/dq_result_schema.py b/src/databricks/labs/dqx/schema/dq_result_schema.py index 112fe0d7f..17380694e 100644 --- a/src/databricks/labs/dqx/schema/dq_result_schema.py +++ b/src/databricks/labs/dqx/schema/dq_result_schema.py @@ -1,4 +1,4 @@ -from pyspark.sql.types import StructType, StructField, ArrayType, StringType, TimestampType, MapType +from pyspark.sql.types import StructType, StructField, ArrayType, BooleanType, StringType, TimestampType, MapType dq_result_item_schema = StructType( [ @@ -10,6 +10,7 @@ StructField("run_time", TimestampType(), nullable=True), StructField("run_id", StringType(), nullable=True), StructField("user_metadata", MapType(StringType(), StringType()), nullable=True), + StructField("skipped", BooleanType(), nullable=True), ] ) From bb1fe7777999c807d334f9eea139df88f291ae3e Mon Sep 17 00:00:00 2001 From: Roshan1299 Date: Tue, 10 Mar 2026 02:53:58 -0600 Subject: [PATCH 02/12] feat(config): add skip_quietly option to ExtraParams --- src/databricks/labs/dqx/config.py | 1 + 1 file changed, 1 insertion(+) diff --git a/src/databricks/labs/dqx/config.py b/src/databricks/labs/dqx/config.py index bb336bee2..1c41d45ab 100644 --- a/src/databricks/labs/dqx/config.py +++ b/src/databricks/labs/dqx/config.py @@ -212,6 +212,7 @@ class ExtraParams: user_metadata: dict[str, str] = field(default_factory=dict) run_time_overwrite: str | None = None run_id_overwrite: str | None = None + skip_quietly: bool = False @dataclass From d82937c85fc8e9a5efa887f870bc2db13b09a44d Mon Sep 17 00:00:00 2001 From: Roshan1299 Date: Tue, 10 Mar 2026 02:54:12 -0600 Subject: [PATCH 03/12] feat(engine): propagate skip_quietly from ExtraParams to DQRuleManager --- src/databricks/labs/dqx/engine.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/databricks/labs/dqx/engine.py b/src/databricks/labs/dqx/engine.py index 792fa5328..48ffddc60 100644 --- a/src/databricks/labs/dqx/engine.py +++ b/src/databricks/labs/dqx/engine.py @@ -91,6 +91,7 @@ def __init__( datetime.fromisoformat(extra_params.run_time_overwrite) if extra_params.run_time_overwrite else None ) self.engine_user_metadata = extra_params.user_metadata + self.skip_quietly = extra_params.skip_quietly self.observer = observer if self.observer: @@ -467,6 +468,7 @@ def _create_results_array( engine_user_metadata=self.engine_user_metadata, run_time_overwrite=self.run_time_overwrite, ref_dfs=ref_dfs, + skip_quietly=self.skip_quietly, ) log_telemetry(self.ws, "check", check.check_func.__name__) result = manager.process() From 3d680c7d3657ec415f66d01c47ad0998294aaf70 Mon Sep 17 00:00:00 2001 From: Roshan1299 Date: Tue, 10 Mar 2026 02:54:21 -0600 Subject: [PATCH 04/12] feat(manager): implement skip_quietly suppression and skipped flag in check results --- src/databricks/labs/dqx/manager.py | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/src/databricks/labs/dqx/manager.py b/src/databricks/labs/dqx/manager.py index 9d08d7dd6..2c92e80eb 100644 --- a/src/databricks/labs/dqx/manager.py +++ b/src/databricks/labs/dqx/manager.py @@ -49,6 +49,7 @@ class DQRuleManager: run_time_overwrite: datetime | None run_id: str ref_dfs: dict[str, DataFrame] | None = None + skip_quietly: bool = False @cached_property def user_metadata(self) -> dict[str, str]: @@ -125,8 +126,11 @@ def process(self) -> DQCheckResult: """ invalid_cols_message = self._get_invalid_cols_message() if invalid_cols_message: - # overwrite message but preserve all other fields in the result - result_struct = self._build_result_struct(condition=F.lit(invalid_cols_message)) + if self.skip_quietly: + # return null condition so this check produces no entry in _errors/_warnings + return DQCheckResult(condition=F.lit(None), check_df=self.df) + # overwrite message but preserve all other fields in the result, marking the check as skipped + result_struct = self._build_result_struct(condition=F.lit(invalid_cols_message), skipped=True) return DQCheckResult(condition=result_struct, check_df=self.df) executor = DQRuleExecutorFactory.create(self.check) @@ -140,7 +144,7 @@ def _wrap_result(self, raw_result: DQCheckResult) -> DQCheckResult: condition=check_result, check_df=raw_result.check_df, info_column_name=raw_result.info_column_name ) - def _build_result_struct(self, condition: Column) -> Column: + def _build_result_struct(self, condition: Column, skipped: bool = False) -> Column: # Use current_timestamp() to make sure streaming gets per-micro-batch timestamps, # or use literal run time if explicitly overridden run_time_expr = F.current_timestamp() if self.run_time_overwrite is None else F.lit(self.run_time_overwrite) @@ -156,6 +160,7 @@ def _build_result_struct(self, condition: Column) -> Column: F.create_map(*[item for kv in self.user_metadata.items() for item in (F.lit(kv[0]), F.lit(kv[1]))]).alias( "user_metadata" ), + F.lit(True if skipped else None).alias("skipped"), ).cast(dq_result_item_schema) def _get_invalid_cols_message(self) -> str: From a0cf8bd0bc328d757bc992007be24d45b8ee887d Mon Sep 17 00:00:00 2001 From: Roshan1299 Date: Tue, 10 Mar 2026 02:54:29 -0600 Subject: [PATCH 05/12] test: add tests for skip_quietly and skipped flag, update existing skip expectations --- tests/integration/test_apply_checks.py | 60 ++++++++++++++++++++++++++ tests/unit/test_config.py | 6 +++ tests/unit/test_engine.py | 19 ++++++++ 3 files changed, 85 insertions(+) diff --git a/tests/integration/test_apply_checks.py b/tests/integration/test_apply_checks.py index 3c6065e3c..99d87d165 100755 --- a/tests/integration/test_apply_checks.py +++ b/tests/integration/test_apply_checks.py @@ -9316,6 +9316,7 @@ def test_apply_checks_skip_checks_with_missing_columns(ws, spark): "run_time": RUN_TIME, "run_id": RUN_ID, "user_metadata": {}, + "skipped": True, }, { "name": "missing_col_is_null", @@ -9326,6 +9327,7 @@ def test_apply_checks_skip_checks_with_missing_columns(ws, spark): "run_time": RUN_TIME, "run_id": RUN_ID, "user_metadata": {}, + "skipped": True, }, { "name": "missing_col_sql_expression", @@ -9337,6 +9339,7 @@ def test_apply_checks_skip_checks_with_missing_columns(ws, spark): "run_time": RUN_TIME, "run_id": RUN_ID, "user_metadata": {}, + "skipped": True, }, { "name": "missing_col_is_unique", @@ -9348,6 +9351,7 @@ def test_apply_checks_skip_checks_with_missing_columns(ws, spark): "run_time": RUN_TIME, "run_id": RUN_ID, "user_metadata": {}, + "skipped": True, }, { "name": "invalid_col_sql_expression", @@ -9358,6 +9362,7 @@ def test_apply_checks_skip_checks_with_missing_columns(ws, spark): "run_time": RUN_TIME, "run_id": RUN_ID, "user_metadata": {}, + "skipped": True, }, ], [ @@ -9370,6 +9375,7 @@ def test_apply_checks_skip_checks_with_missing_columns(ws, spark): "run_time": RUN_TIME, "run_id": RUN_ID, "user_metadata": {"tag1": "value1", "tag2": "value2"}, + "skipped": True, }, ], ] @@ -9494,6 +9500,7 @@ def test_apply_checks_by_metadata_skip_checks_with_missing_columns(ws, spark): "run_time": RUN_TIME, "run_id": RUN_ID, "user_metadata": {}, + "skipped": True, }, { "name": "missing_col_is_null", @@ -9504,6 +9511,7 @@ def test_apply_checks_by_metadata_skip_checks_with_missing_columns(ws, spark): "run_time": RUN_TIME, "run_id": RUN_ID, "user_metadata": {}, + "skipped": True, }, { "name": "missing_col_sql_expression", @@ -9515,6 +9523,7 @@ def test_apply_checks_by_metadata_skip_checks_with_missing_columns(ws, spark): "run_time": RUN_TIME, "run_id": RUN_ID, "user_metadata": {}, + "skipped": True, }, { "name": "missing_col_is_unique", @@ -9526,6 +9535,7 @@ def test_apply_checks_by_metadata_skip_checks_with_missing_columns(ws, spark): "run_time": RUN_TIME, "run_id": RUN_ID, "user_metadata": {}, + "skipped": True, }, { "name": "invalid_col_sql_expression", @@ -9536,6 +9546,7 @@ def test_apply_checks_by_metadata_skip_checks_with_missing_columns(ws, spark): "run_time": RUN_TIME, "run_id": RUN_ID, "user_metadata": {}, + "skipped": True, }, ], [ @@ -9548,6 +9559,7 @@ def test_apply_checks_by_metadata_skip_checks_with_missing_columns(ws, spark): "run_time": RUN_TIME, "run_id": RUN_ID, "user_metadata": {"tag1": "value1", "tag2": "value2"}, + "skipped": True, }, ], ] @@ -9555,3 +9567,51 @@ def test_apply_checks_by_metadata_skip_checks_with_missing_columns(ws, spark): SCHEMA + complex_cols_schema + REPORTING_COLUMNS, ) assert_df_equality(checked, expected, ignore_nullable=True) + + +def test_apply_checks_skip_quietly_suppresses_skipped_entries(ws, spark): + extra_params = ExtraParams(run_time_overwrite=RUN_TIME.isoformat(), run_id_overwrite=RUN_ID, skip_quietly=True) + dq_engine = DQEngine(workspace_client=ws, extra_params=extra_params) + test_df = spark.createDataFrame([[1, None, 3]], SCHEMA) + + checks = [ + DQRowRule(name="b_is_null", criticality="error", check_func=check_funcs.is_not_null, column="b"), + DQRowRule( + name="missing_col_is_null", + criticality="error", + check_func=check_funcs.is_not_null, + column="missing_col", + ), + ] + + good_df, bad_df = dq_engine.apply_checks_and_split(test_df, checks) + + # row is in bad_df only because of the real failure (b is null), not the skipped check + assert bad_df.count() == 1 + errors = bad_df.select("_errors").first()["_errors"] + assert len(errors) == 1 + assert errors[0]["name"] == "b_is_null" + + # good_df is empty since the real check failed + assert good_df.count() == 0 + + +def test_apply_checks_skipped_flag_set_on_skipped_entries(ws, spark): + dq_engine = DQEngine(workspace_client=ws, extra_params=EXTRA_PARAMS) + test_df = spark.createDataFrame([[1, 2, 3]], SCHEMA) + + checks = [ + DQRowRule( + name="missing_col_is_null", + criticality="error", + check_func=check_funcs.is_not_null, + column="missing_col", + ), + ] + + checked = dq_engine.apply_checks(test_df, checks) + + errors = checked.select("_errors").first()["_errors"] + assert len(errors) == 1 + assert errors[0]["name"] == "missing_col_is_null" + assert errors[0]["skipped"] is True diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py index d21f1255a..14d51d827 100644 --- a/tests/unit/test_config.py +++ b/tests/unit/test_config.py @@ -271,6 +271,7 @@ def test_extra_params_defaults(): assert not config.user_metadata assert config.run_time_overwrite is None assert config.run_id_overwrite is None + assert config.skip_quietly is False def test_extra_params_custom_values(): @@ -286,6 +287,11 @@ def test_extra_params_custom_values(): assert config.run_id_overwrite == "custom_run_id" +def test_extra_params_skip_quietly(): + config = ExtraParams(skip_quietly=True) + assert config.skip_quietly is True + + # Test WorkspaceConfig.as_dict() def test_workspace_config_as_dict(): config = WorkspaceConfig( diff --git a/tests/unit/test_engine.py b/tests/unit/test_engine.py index 1d187607e..3ff48110c 100644 --- a/tests/unit/test_engine.py +++ b/tests/unit/test_engine.py @@ -58,6 +58,25 @@ def test_engine_core_creation_with_observer(): assert engine_core.run_id == observer.id +def test_engine_core_skip_quietly_default(): + spark_mock = create_autospec(SparkSession) + ws = create_autospec(WorkspaceClient) + + engine_core = DQEngineCore(spark=spark_mock, workspace_client=ws) + + assert engine_core.skip_quietly is False + + +def test_engine_core_skip_quietly_enabled(): + spark_mock = create_autospec(SparkSession) + ws = create_autospec(WorkspaceClient) + extra_params = ExtraParams(skip_quietly=True) + + engine_core = DQEngineCore(spark=spark_mock, workspace_client=ws, extra_params=extra_params) + + assert engine_core.skip_quietly is True + + def test_engine_creation_no_workspace_connection(mock_workspace_client, mock_spark): mock_workspace_client.clusters.select_spark_version.side_effect = DatabricksError() From 9b53187349801546acaf4bb6e9602a22b5ba8c81 Mon Sep 17 00:00:00 2001 From: Roshan1299 Date: Thu, 19 Mar 2026 14:45:43 -0600 Subject: [PATCH 06/12] test: fix test_define_user_metadata_and_extract_dq_results for skipped field Signed-off-by: Roshan1299 --- tests/integration/test_apply_checks.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/tests/integration/test_apply_checks.py b/tests/integration/test_apply_checks.py index 2def79281..f188753f8 100755 --- a/tests/integration/test_apply_checks.py +++ b/tests/integration/test_apply_checks.py @@ -7193,6 +7193,7 @@ def test_define_user_metadata_and_extract_dq_results(ws, spark): RUN_TIME, RUN_ID, user_metadata, + None, get_rule_fingerprint_from_checks(versioning_rules_checks, "a_is_null_or_empty", "error"), get_rule_set_fingerprint_from_checks(versioning_rules_checks), ], @@ -7205,6 +7206,7 @@ def test_define_user_metadata_and_extract_dq_results(ws, spark): RUN_TIME, RUN_ID, user_metadata, + None, get_rule_fingerprint_from_checks(versioning_rules_checks, "a_is_null", "error"), get_rule_set_fingerprint_from_checks(versioning_rules_checks), ], From 0ea9507b6f63e769f02448564d522fa2cf963f39 Mon Sep 17 00:00:00 2001 From: Roshan1299 Date: Fri, 20 Mar 2026 03:03:50 -0600 Subject: [PATCH 07/12] fix(schema): move skipped field to end of result schema and cast null condition Signed-off-by: Roshan1299 --- src/databricks/labs/dqx/manager.py | 8 +-- .../labs/dqx/schema/dq_result_schema.py | 2 +- tests/integration/test_apply_checks.py | 54 ++----------------- 3 files changed, 9 insertions(+), 55 deletions(-) diff --git a/src/databricks/labs/dqx/manager.py b/src/databricks/labs/dqx/manager.py index 4c38741e7..aa9a3f6e6 100644 --- a/src/databricks/labs/dqx/manager.py +++ b/src/databricks/labs/dqx/manager.py @@ -49,7 +49,7 @@ class DQRuleManager: run_time_overwrite: datetime | None run_id: str ref_dfs: dict[str, DataFrame] | None = None - skip_quietly: bool = False + suppress_skipped: bool = False rule_fingerprint: str | None = None rule_set_fingerprint: str | None = None @@ -128,9 +128,9 @@ def process(self) -> DQCheckResult: """ invalid_cols_message = self._get_invalid_cols_message() if invalid_cols_message: - if self.skip_quietly: + if self.suppress_skipped: # return null condition so this check produces no entry in _errors/_warnings - return DQCheckResult(condition=F.lit(None), check_df=self.df) + return DQCheckResult(condition=F.lit(None).cast(dq_result_item_schema), check_df=self.df) # overwrite message but preserve all other fields in the result, marking the check as skipped result_struct = self._build_result_struct(condition=F.lit(invalid_cols_message), skipped=True) return DQCheckResult(condition=result_struct, check_df=self.df) @@ -162,9 +162,9 @@ def _build_result_struct(self, condition: Column, skipped: bool = False) -> Colu F.create_map(*[item for kv in self.user_metadata.items() for item in (F.lit(kv[0]), F.lit(kv[1]))]).alias( "user_metadata" ), - F.lit(True if skipped else None).alias("skipped"), F.lit(self.rule_fingerprint).alias("rule_fingerprint"), F.lit(self.rule_set_fingerprint).alias("rule_set_fingerprint"), + F.lit(skipped or None).alias("skipped"), ).cast(dq_result_item_schema) def _get_invalid_cols_message(self) -> str: diff --git a/src/databricks/labs/dqx/schema/dq_result_schema.py b/src/databricks/labs/dqx/schema/dq_result_schema.py index 3ec749de7..9aeea76c0 100644 --- a/src/databricks/labs/dqx/schema/dq_result_schema.py +++ b/src/databricks/labs/dqx/schema/dq_result_schema.py @@ -10,9 +10,9 @@ StructField("run_time", TimestampType(), nullable=True), StructField("run_id", StringType(), nullable=True), StructField("user_metadata", MapType(StringType(), StringType()), nullable=True), - StructField("skipped", BooleanType(), nullable=True), StructField("rule_fingerprint", StringType(), nullable=True), StructField("rule_set_fingerprint", StringType(), nullable=True), + StructField("skipped", BooleanType(), nullable=True), ] ) diff --git a/tests/integration/test_apply_checks.py b/tests/integration/test_apply_checks.py index f188753f8..ec9fb7026 100755 --- a/tests/integration/test_apply_checks.py +++ b/tests/integration/test_apply_checks.py @@ -7193,9 +7193,9 @@ def test_define_user_metadata_and_extract_dq_results(ws, spark): RUN_TIME, RUN_ID, user_metadata, - None, get_rule_fingerprint_from_checks(versioning_rules_checks, "a_is_null_or_empty", "error"), get_rule_set_fingerprint_from_checks(versioning_rules_checks), + None, ], [ "a_is_null", @@ -7206,9 +7206,9 @@ def test_define_user_metadata_and_extract_dq_results(ws, spark): RUN_TIME, RUN_ID, user_metadata, - None, get_rule_fingerprint_from_checks(versioning_rules_checks, "a_is_null", "error"), get_rule_set_fingerprint_from_checks(versioning_rules_checks), + None, ], ], dq_result_schema.elementType, @@ -7227,6 +7227,7 @@ def test_define_user_metadata_and_extract_dq_results(ws, spark): user_metadata, get_rule_fingerprint_from_checks(versioning_rules_checks, "a_is_null_or_empty", "warn"), get_rule_set_fingerprint_from_checks(versioning_rules_checks), + None, ], [ "a_is_null", @@ -7239,6 +7240,7 @@ def test_define_user_metadata_and_extract_dq_results(ws, spark): user_metadata, get_rule_fingerprint_from_checks(versioning_rules_checks, "a_is_null", "warn"), get_rule_set_fingerprint_from_checks(versioning_rules_checks), + None, ], ], dq_result_schema.elementType, @@ -10004,51 +10006,3 @@ def test_apply_checks_by_metadata_skip_checks_with_missing_columns(ws, spark): SCHEMA + complex_cols_schema + REPORTING_COLUMNS, ) assert_df_equality(checked, expected, ignore_nullable=True) - - -def test_apply_checks_skip_quietly_suppresses_skipped_entries(ws, spark): - extra_params = ExtraParams(run_time_overwrite=RUN_TIME.isoformat(), run_id_overwrite=RUN_ID, skip_quietly=True) - dq_engine = DQEngine(workspace_client=ws, extra_params=extra_params) - test_df = spark.createDataFrame([[1, None, 3]], SCHEMA) - - checks = [ - DQRowRule(name="b_is_null", criticality="error", check_func=check_funcs.is_not_null, column="b"), - DQRowRule( - name="missing_col_is_null", - criticality="error", - check_func=check_funcs.is_not_null, - column="missing_col", - ), - ] - - good_df, bad_df = dq_engine.apply_checks_and_split(test_df, checks) - - # row is in bad_df only because of the real failure (b is null), not the skipped check - assert bad_df.count() == 1 - errors = bad_df.select("_errors").first()["_errors"] - assert len(errors) == 1 - assert errors[0]["name"] == "b_is_null" - - # good_df is empty since the real check failed - assert good_df.count() == 0 - - -def test_apply_checks_skipped_flag_set_on_skipped_entries(ws, spark): - dq_engine = DQEngine(workspace_client=ws, extra_params=EXTRA_PARAMS) - test_df = spark.createDataFrame([[1, 2, 3]], SCHEMA) - - checks = [ - DQRowRule( - name="missing_col_is_null", - criticality="error", - check_func=check_funcs.is_not_null, - column="missing_col", - ), - ] - - checked = dq_engine.apply_checks(test_df, checks) - - errors = checked.select("_errors").first()["_errors"] - assert len(errors) == 1 - assert errors[0]["name"] == "missing_col_is_null" - assert errors[0]["skipped"] is True From f310a88062d362da7859b89ebe421c370ac685ce Mon Sep 17 00:00:00 2001 From: Roshan1299 Date: Fri, 20 Mar 2026 03:04:12 -0600 Subject: [PATCH 08/12] fix(config): rename skip_quietly to suppress_skipped Signed-off-by: Roshan1299 --- src/databricks/labs/dqx/config.py | 2 +- src/databricks/labs/dqx/engine.py | 4 ++-- tests/unit/test_config.py | 8 ++++---- tests/unit/test_engine.py | 10 +++++----- 4 files changed, 12 insertions(+), 12 deletions(-) diff --git a/src/databricks/labs/dqx/config.py b/src/databricks/labs/dqx/config.py index 812a3b34c..e889189ca 100644 --- a/src/databricks/labs/dqx/config.py +++ b/src/databricks/labs/dqx/config.py @@ -215,7 +215,7 @@ class ExtraParams: user_metadata: dict[str, str] = field(default_factory=dict) run_time_overwrite: str | None = None run_id_overwrite: str | None = None - skip_quietly: bool = False + suppress_skipped: bool = False @dataclass diff --git a/src/databricks/labs/dqx/engine.py b/src/databricks/labs/dqx/engine.py index b4e2ed716..1982e3e4e 100644 --- a/src/databricks/labs/dqx/engine.py +++ b/src/databricks/labs/dqx/engine.py @@ -95,7 +95,7 @@ def __init__( datetime.fromisoformat(extra_params.run_time_overwrite) if extra_params.run_time_overwrite else None ) self.engine_user_metadata = extra_params.user_metadata - self.skip_quietly = extra_params.skip_quietly + self.suppress_skipped = extra_params.suppress_skipped self.observer = observer if self.observer: @@ -477,7 +477,7 @@ def _create_results_array( engine_user_metadata=self.engine_user_metadata, run_time_overwrite=self.run_time_overwrite, ref_dfs=ref_dfs, - skip_quietly=self.skip_quietly, + suppress_skipped=self.suppress_skipped, rule_fingerprint=check.rule_fingerprint, rule_set_fingerprint=rule_set_fingerprint, ) diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py index 40251e320..868a04775 100644 --- a/tests/unit/test_config.py +++ b/tests/unit/test_config.py @@ -271,7 +271,7 @@ def test_extra_params_defaults(): assert not config.user_metadata assert config.run_time_overwrite is None assert config.run_id_overwrite is None - assert config.skip_quietly is False + assert config.suppress_skipped is False def test_extra_params_custom_values(): @@ -287,9 +287,9 @@ def test_extra_params_custom_values(): assert config.run_id_overwrite == "custom_run_id" -def test_extra_params_skip_quietly(): - config = ExtraParams(skip_quietly=True) - assert config.skip_quietly is True +def test_extra_params_suppress_skipped(): + config = ExtraParams(suppress_skipped=True) + assert config.suppress_skipped is True # Test WorkspaceConfig.as_dict() diff --git a/tests/unit/test_engine.py b/tests/unit/test_engine.py index c411d8519..55c105940 100644 --- a/tests/unit/test_engine.py +++ b/tests/unit/test_engine.py @@ -64,23 +64,23 @@ def test_engine_core_creation_with_observer(): assert engine_core.run_id == observer.id -def test_engine_core_skip_quietly_default(): +def test_engine_core_suppress_skipped_default(): spark_mock = create_autospec(SparkSession) ws = create_autospec(WorkspaceClient) engine_core = DQEngineCore(spark=spark_mock, workspace_client=ws) - assert engine_core.skip_quietly is False + assert engine_core.suppress_skipped is False -def test_engine_core_skip_quietly_enabled(): +def test_engine_core_suppress_skipped_enabled(): spark_mock = create_autospec(SparkSession) ws = create_autospec(WorkspaceClient) - extra_params = ExtraParams(skip_quietly=True) + extra_params = ExtraParams(suppress_skipped=True) engine_core = DQEngineCore(spark=spark_mock, workspace_client=ws, extra_params=extra_params) - assert engine_core.skip_quietly is True + assert engine_core.suppress_skipped is True def test_engine_creation_no_workspace_connection(mock_workspace_client, mock_spark): From e6bed258e3450f017dbfd92868e6c3835cf5f958 Mon Sep 17 00:00:00 2001 From: Roshan1299 Date: Fri, 20 Mar 2026 03:04:29 -0600 Subject: [PATCH 09/12] docs: add suppress_skipped documentation, unit tests, and changelog entry Signed-off-by: Roshan1299 --- CHANGELOG.md | 5 ++ .../docs/guide/additional_configuration.mdx | 42 +++++++++++++ docs/dqx/docs/guide/quality_checks_apply.mdx | 2 +- .../test_apply_checks_suppress_skipped.py | 59 +++++++++++++++++++ tests/unit/test_manager.py | 56 ++++++++++++++++++ 5 files changed, 163 insertions(+), 1 deletion(-) create mode 100644 tests/integration/test_apply_checks_suppress_skipped.py create mode 100644 tests/unit/test_manager.py diff --git a/CHANGELOG.md b/CHANGELOG.md index f02ed8c8c..8af7f5805 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,10 @@ # Version changelog +## 0.14.0 + +* Added `suppress_skipped` option to `ExtraParams` ([#953](https://github.com/databrickslabs/dqx/issues/953)). When `suppress_skipped=True`, checks skipped due to missing columns or invalid filters produce no entry in `_errors` or `_warnings`, so skipped rows are excluded entirely from the bad DataFrame rather than recorded with a skip message. +* Added `skipped` field to DQ result schema ([#953](https://github.com/databrickslabs/dqx/issues/953)). When a check is skipped due to an unresolvable column, filter, or SQL expression, the result struct now includes `skipped=True`, enabling downstream consumers to distinguish skipped checks from failed ones without parsing the message string. + ## 0.13.0 * New DQX Data Quality Dashboard ([#1019](https://github.com/databrickslabs/dqx/issues/1019)). The data quality dashboard has been significantly enhanced to provide a centralized view of data quality metrics across all tables, allowing users to monitor and track data quality issues with greater ease. The dashboard now consists of three tabs - Data Quality Summary, Data Quality by Table (Time Series), and Data Quality by Table (Full Snapshot) - each catering to different monitoring scenarios, and offers customizable parameters for reporting column names and filtering tables with data quality issues. Additionally, the installation process for the dashboard has been simplified, with options to import it directly to a Workspace or deploy it automatically using the Databricks CLI. diff --git a/docs/dqx/docs/guide/additional_configuration.mdx b/docs/dqx/docs/guide/additional_configuration.mdx index 25e16acf8..c72811b2a 100644 --- a/docs/dqx/docs/guide/additional_configuration.mdx +++ b/docs/dqx/docs/guide/additional_configuration.mdx @@ -129,3 +129,45 @@ You can customize the names of these result columns by specifying extra paramete The `_dq_info` reporting column is an **array of structs**: one element per dataset-level check that produces info (e.g. one per `has_no_row_anomalies`). Each element has a shared schema; for row anomaly detection the populated field is `anomaly`, with fields such as `score`, `severity_percentile`, `is_anomaly`, `threshold`, `model`, `contributions`, and `confidence_std`. For the full field list and types, see [Schema of the info column (_dq_info)](/docs/guide/row_anomaly_detection#schema-of-the-info-column-_dq_info) in the Row Anomaly Detection guide. + +## Suppressing skipped check entries + +By default, when a check is skipped (e.g. because a referenced column or filter cannot be resolved in the input DataFrame), DQX records an entry in `_errors` or `_warnings` with `skipped=True` and a message describing why the check was skipped. + +You can suppress these entries entirely by setting `suppress_skipped=True` in `ExtraParams`. When enabled, skipped checks produce no entry in `_errors` or `_warnings`, so the row is not included in the bad DataFrame. + + + + ```python + from databricks.sdk import WorkspaceClient + from databricks.labs.dqx.engine import DQEngine + from databricks.labs.dqx.config import ExtraParams + + # suppress entries for checks skipped due to missing columns or invalid filters + extra_parameters = ExtraParams(suppress_skipped=True) + + ws = WorkspaceClient() + dq_engine = DQEngine(ws, extra_params=extra_parameters) + ``` + + + You can set the following fields in the [configuration file](/docs/installation/#configuration-file) to suppress skipped check entries when using DQX workflows: + ```yaml + extra_params: + suppress_skipped: true + ``` + + + +## Identifying skipped checks in results + +When a check is skipped and `suppress_skipped` is not enabled, the result struct includes a `skipped` boolean field set to `True`. This allows downstream consumers to distinguish skipped checks from actual data quality failures without parsing the message string. + +```python +checked_df = dq_engine.apply_checks(df, checks) + +# filter to only skipped entries in _errors +from pyspark.sql import functions as F + +skipped = checked_df.select(F.explode("_errors").alias("e")).filter("e.skipped = true") +``` diff --git a/docs/dqx/docs/guide/quality_checks_apply.mdx b/docs/dqx/docs/guide/quality_checks_apply.mdx index e4321b4c0..1b0b742ef 100644 --- a/docs/dqx/docs/guide/quality_checks_apply.mdx +++ b/docs/dqx/docs/guide/quality_checks_apply.mdx @@ -36,7 +36,7 @@ Checks can be applied to the input data by one of the following methods of the ` You can see the full list of `DQEngine` methods [here](/docs/reference/engine/#dqx-engine-methods). -The engine ensures that the specified `column`, `columns`, `filter`, or sql 'expression' fields can be resolved in the input DataFrame. If any of these fields are invalid, the check evaluation is skipped, and the results include the check failure with a message identifying the invalid fields. +The engine ensures that the specified `column`, `columns`, `filter`, or sql 'expression' fields can be resolved in the input DataFrame. If any of these fields are invalid, the check evaluation is skipped, and the results include the check failure with a message identifying the invalid fields and `skipped=True` in the result struct. You can suppress these entries entirely or identify them downstream — see [Suppressing skipped check entries](/docs/guide/additional_configuration#suppressing-skipped-check-entries). The engine will raise an error if you try to apply checks with invalid definition (e.g. wrong syntax). In addition, you can also perform a standalone syntax validation of the checks as described [here](/docs/guide/quality_checks_definition#validating-syntax-of-quality-checks). diff --git a/tests/integration/test_apply_checks_suppress_skipped.py b/tests/integration/test_apply_checks_suppress_skipped.py new file mode 100644 index 000000000..88351f71c --- /dev/null +++ b/tests/integration/test_apply_checks_suppress_skipped.py @@ -0,0 +1,59 @@ +from databricks.labs.dqx import check_funcs +from databricks.labs.dqx.config import ExtraParams +from databricks.labs.dqx.engine import DQEngine +from databricks.labs.dqx.rule import DQRowRule +from tests.integration.conftest import ( + EXTRA_PARAMS, + RUN_TIME, + RUN_ID, +) + +SCHEMA = "a: int, b: int, c: int" + + +def test_apply_checks_suppress_skipped_suppresses_skipped_entries(ws, spark): + extra_params = ExtraParams(run_time_overwrite=RUN_TIME.isoformat(), run_id_overwrite=RUN_ID, suppress_skipped=True) + dq_engine = DQEngine(workspace_client=ws, extra_params=extra_params) + test_df = spark.createDataFrame([[1, None, 3]], SCHEMA) + + checks = [ + DQRowRule(name="b_is_null", criticality="error", check_func=check_funcs.is_not_null, column="b"), + DQRowRule( + name="missing_col_is_null", + criticality="error", + check_func=check_funcs.is_not_null, + column="missing_col", + ), + ] + + good_df, bad_df = dq_engine.apply_checks_and_split(test_df, checks) + + # row is in bad_df only because of the real failure (b is null), not the skipped check + assert bad_df.count() == 1 + errors = bad_df.select("_errors").first()["_errors"] + assert len(errors) == 1 + assert errors[0]["name"] == "b_is_null" + + # good_df is empty since the real check failed + assert good_df.count() == 0 + + +def test_apply_checks_skipped_flag_set_on_skipped_entries(ws, spark): + dq_engine = DQEngine(workspace_client=ws, extra_params=EXTRA_PARAMS) + test_df = spark.createDataFrame([[1, 2, 3]], SCHEMA) + + checks = [ + DQRowRule( + name="missing_col_is_null", + criticality="error", + check_func=check_funcs.is_not_null, + column="missing_col", + ), + ] + + checked = dq_engine.apply_checks(test_df, checks) + + errors = checked.select("_errors").first()["_errors"] + assert len(errors) == 1 + assert errors[0]["name"] == "missing_col_is_null" + assert errors[0]["skipped"] is True diff --git a/tests/unit/test_manager.py b/tests/unit/test_manager.py new file mode 100644 index 000000000..6af4e2566 --- /dev/null +++ b/tests/unit/test_manager.py @@ -0,0 +1,56 @@ +from unittest.mock import create_autospec, PropertyMock + +from pyspark.errors import AnalysisException +from pyspark.sql import DataFrame, SparkSession +from pyspark.sql.column import Column + +from databricks.labs.dqx import check_funcs +from databricks.labs.dqx.executor import DQCheckResult +from databricks.labs.dqx.manager import DQRuleManager +from databricks.labs.dqx.rule import DQRowRule + + +def _make_manager_with_missing_column(df: DataFrame, spark: SparkSession, *, suppress_skipped: bool) -> DQRuleManager: + return DQRuleManager( + check=DQRowRule(check_func=check_funcs.is_not_null, column="missing_col"), + df=df, + spark=spark, + engine_user_metadata={}, + run_time_overwrite=None, + run_id="test-run", + suppress_skipped=suppress_skipped, + ) + + +def test_rule_manager_suppress_skipped_returns_null_condition_for_invalid_column(): + """When suppress_skipped=True and a column cannot be resolved, process() returns a null-cast Column + so the check produces no entry in _errors/_warnings.""" + df_mock = create_autospec(DataFrame) + spark_mock = create_autospec(SparkSession) + type(df_mock.select.return_value).schema = PropertyMock( + side_effect=AnalysisException("Column 'missing_col' not found") + ) + + manager = _make_manager_with_missing_column(df_mock, spark_mock, suppress_skipped=True) + result = manager.process() + + assert isinstance(result, DQCheckResult) + assert isinstance(result.condition, Column) + assert result.check_df is df_mock + + +def test_rule_manager_suppress_skipped_false_returns_struct_for_invalid_column(): + """When suppress_skipped=False and a column cannot be resolved, process() returns a result struct + (not null) so the skipped check is recorded in _errors/_warnings.""" + df_mock = create_autospec(DataFrame) + spark_mock = create_autospec(SparkSession) + type(df_mock.select.return_value).schema = PropertyMock( + side_effect=AnalysisException("Column 'missing_col' not found") + ) + + manager = _make_manager_with_missing_column(df_mock, spark_mock, suppress_skipped=False) + result = manager.process() + + assert isinstance(result, DQCheckResult) + assert isinstance(result.condition, Column) + assert result.check_df is df_mock From 84bce10c93d0fc9d41eb07458de049695ba93d73 Mon Sep 17 00:00:00 2001 From: Roshan1299 Date: Fri, 20 Mar 2026 14:16:54 -0600 Subject: [PATCH 10/12] chore: remove changelog entry, generated during release Signed-off-by: Roshan1299 --- CHANGELOG.md | 5 ----- 1 file changed, 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8af7f5805..f02ed8c8c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,10 +1,5 @@ # Version changelog -## 0.14.0 - -* Added `suppress_skipped` option to `ExtraParams` ([#953](https://github.com/databrickslabs/dqx/issues/953)). When `suppress_skipped=True`, checks skipped due to missing columns or invalid filters produce no entry in `_errors` or `_warnings`, so skipped rows are excluded entirely from the bad DataFrame rather than recorded with a skip message. -* Added `skipped` field to DQ result schema ([#953](https://github.com/databrickslabs/dqx/issues/953)). When a check is skipped due to an unresolvable column, filter, or SQL expression, the result struct now includes `skipped=True`, enabling downstream consumers to distinguish skipped checks from failed ones without parsing the message string. - ## 0.13.0 * New DQX Data Quality Dashboard ([#1019](https://github.com/databrickslabs/dqx/issues/1019)). The data quality dashboard has been significantly enhanced to provide a centralized view of data quality metrics across all tables, allowing users to monitor and track data quality issues with greater ease. The dashboard now consists of three tabs - Data Quality Summary, Data Quality by Table (Time Series), and Data Quality by Table (Full Snapshot) - each catering to different monitoring scenarios, and offers customizable parameters for reporting column names and filtering tables with data quality issues. Additionally, the installation process for the dashboard has been simplified, with options to import it directly to a Workspace or deploy it automatically using the Databricks CLI. From e6d772d3a5d75c6377840031f07201779a9b91e1 Mon Sep 17 00:00:00 2001 From: Marcin Wojtyczka Date: Mon, 23 Mar 2026 19:46:30 +0100 Subject: [PATCH 11/12] Apply suggestion from @mwojtyczka --- docs/dqx/docs/guide/additional_configuration.mdx | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/dqx/docs/guide/additional_configuration.mdx b/docs/dqx/docs/guide/additional_configuration.mdx index c72811b2a..69701e6fb 100644 --- a/docs/dqx/docs/guide/additional_configuration.mdx +++ b/docs/dqx/docs/guide/additional_configuration.mdx @@ -169,5 +169,5 @@ checked_df = dq_engine.apply_checks(df, checks) # filter to only skipped entries in _errors from pyspark.sql import functions as F -skipped = checked_df.select(F.explode("_errors").alias("e")).filter("e.skipped = true") +skipped = checked_df.select(F.explode("_errors").alias("e")).filter(F.col("e.skipped") == True) ``` From 1234d01ab1aee5fce60d8088906e075e9dff4864 Mon Sep 17 00:00:00 2001 From: Marcin Wojtyczka Date: Mon, 23 Mar 2026 21:36:17 +0100 Subject: [PATCH 12/12] fix tests --- tests/integration/test_apply_checks.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/tests/integration/test_apply_checks.py b/tests/integration/test_apply_checks.py index 56819a4d4..e3ed75034 100755 --- a/tests/integration/test_apply_checks.py +++ b/tests/integration/test_apply_checks.py @@ -9647,6 +9647,7 @@ def test_apply_checks_with_has_valid_schema_extra_columns_in_params(ws, spark): "run_time": RUN_TIME, "run_id": RUN_ID, "user_metadata": {}, + "skipped": True, } expected_skip_permissive = { @@ -9658,6 +9659,7 @@ def test_apply_checks_with_has_valid_schema_extra_columns_in_params(ws, spark): "run_time": RUN_TIME, "run_id": RUN_ID, "user_metadata": {}, + "skipped": True, } expected = spark.createDataFrame(