feat: add suppress_skipped option and skipped flag for skipped checks

docs/dqx/docs/guide/additional_configuration.mdx

@@ -129,3 +129,45 @@ You can customize the names of these result columns by specifying extra paramete
 <Admonition type="info" title="Info column usage">
 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.
 </Admonition>
+
+## 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.
+
+<Tabs>
+  <TabItem value="Python" label="Python" default>
+    ```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)
+    ```
+  </TabItem>
+  <TabItem value="Workflows" label="Workflows">
+    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
+    ```
+  </TabItem>
+</Tabs>
+
+## 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")
+```

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).
 

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
+    suppress_skipped: 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.suppress_skipped = extra_params.suppress_skipped
 
         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,
+                suppress_skipped=self.suppress_skipped,
                 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
+    suppress_skipped: 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.suppress_skipped:
+                # return null condition so this check produces no entry in _errors/_warnings
+                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)
 
         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)
@@ -160,6 +164,7 @@ def _build_result_struct(self, condition: Column) -> Column:
             ),
             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:

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(
     [
@@ -12,6 +12,7 @@
         StructField("user_metadata", MapType(StringType(), StringType()), nullable=True),
         StructField("rule_fingerprint", StringType(), nullable=True),
         StructField("rule_set_fingerprint", StringType(), nullable=True),
+        StructField("skipped", BooleanType(), nullable=True),
     ]
 )
 

tests/integration/test_apply_checks.py

@@ -7234,6 +7234,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", "error"),
                 get_rule_set_fingerprint_from_checks(versioning_rules_checks),
+                None,
             ],
             [
                 "a_is_null",
@@ -7246,6 +7247,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", "error"),
                 get_rule_set_fingerprint_from_checks(versioning_rules_checks),
+                None,
             ],
         ],
         dq_result_schema.elementType,
@@ -7264,6 +7266,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",
@@ -7276,6 +7279,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,
@@ -9790,6 +9794,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",
@@ -9800,6 +9805,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",
@@ -9811,6 +9817,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",
@@ -9822,6 +9829,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",
@@ -9832,6 +9840,7 @@ def test_apply_checks_skip_checks_with_missing_columns(ws, spark):
                         "run_time": RUN_TIME,
                         "run_id": RUN_ID,
                         "user_metadata": {},
+                        "skipped": True,
                     },
                 ],
                 [
@@ -9844,6 +9853,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,
                     },
                 ],
             ]
@@ -9968,6 +9978,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",
@@ -9978,6 +9989,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",
@@ -9989,6 +10001,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",
@@ -10000,6 +10013,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",
@@ -10010,6 +10024,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,
                     },
                 ],
                 [
@@ -10022,6 +10037,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,
                     },
                 ],
             ]

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

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.suppress_skipped 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_suppress_skipped():
+    config = ExtraParams(suppress_skipped=True)
+    assert config.suppress_skipped 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_suppress_skipped_default():
+    spark_mock = create_autospec(SparkSession)
+    ws = create_autospec(WorkspaceClient)
+
+    engine_core = DQEngineCore(spark=spark_mock, workspace_client=ws)
+
+    assert engine_core.suppress_skipped is False
+
+
+def test_engine_core_suppress_skipped_enabled():
+    spark_mock = create_autospec(SparkSession)
+    ws = create_autospec(WorkspaceClient)
+    extra_params = ExtraParams(suppress_skipped=True)
+
+    engine_core = DQEngineCore(spark=spark_mock, workspace_client=ws, extra_params=extra_params)
+
+    assert engine_core.suppress_skipped is True
+
+
 def test_engine_creation_no_workspace_connection(mock_workspace_client, mock_spark):
     mock_workspace_client.clusters.select_spark_version.side_effect = DatabricksError()