feat: add suppress_skipped option and skipped flag for skipped checks

src/databricks/labs/dqx/config.py

@@ -215,6 +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
 
 
 @dataclass

src/databricks/labs/dqx/engine.py

@@ -95,6 +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.observer = observer
         if self.observer:
@@ -476,6 +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,
                 rule_fingerprint=check.rule_fingerprint,
                 rule_set_fingerprint=rule_set_fingerprint,
             )

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
     rule_fingerprint: str | None = None
     rule_set_fingerprint: str | None = None
 
@@ -127,8 +128,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)
@@ -142,7 +146,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)
@@ -158,6 +162,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"),
             F.lit(self.rule_fingerprint).alias("rule_fingerprint"),
             F.lit(self.rule_set_fingerprint).alias("rule_set_fingerprint"),
         ).cast(dq_result_item_schema)

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),
         StructField("rule_fingerprint", StringType(), nullable=True),
         StructField("rule_set_fingerprint", StringType(), nullable=True),
     ]

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),
             ],
@@ -9751,6 +9753,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",
@@ -9761,6 +9764,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",
@@ -9772,6 +9776,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",
@@ -9783,6 +9788,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",
@@ -9793,6 +9799,7 @@ def test_apply_checks_skip_checks_with_missing_columns(ws, spark):
                         "run_time": RUN_TIME,
                         "run_id": RUN_ID,
                         "user_metadata": {},
+                        "skipped": True,
                     },
                 ],
                 [
@@ -9805,6 +9812,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,
                     },
                 ],
             ]
@@ -9929,6 +9937,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",
@@ -9939,6 +9948,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",
@@ -9950,6 +9960,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",
@@ -9961,6 +9972,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",
@@ -9971,6 +9983,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,
                     },
                 ],
                 [
@@ -9983,10 +9996,59 @@ 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,
                     },
                 ],
             ]
         ],
         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

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(

tests/unit/test_engine.py

@@ -64,6 +64,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()