Fix validation for columns requiring SQL identifier escaping

src/databricks/labs/dqx/manager.py

@@ -13,7 +13,7 @@
     DQRule,
 )
 from databricks.labs.dqx.schema.dq_result_schema import dq_result_item_schema
-from databricks.labs.dqx.utils import get_column_name_or_alias
+from databricks.labs.dqx.utils import get_column_name_or_alias, is_simple_column_expression, quote_column_name
 
 logger = logging.getLogger(__name__)
 
@@ -74,18 +74,32 @@ def filter_condition(self) -> Column:
     def invalid_columns(self) -> list[str]:
         """
         Returns list of invalid check columns in the input DataFrame.
+
+        Column names that contain characters requiring SQL identifier escaping (e.g. spaces, such as
+        "Customer Name") are back-quoted so they are unambiguous in the skip / error messages.
         """
         invalid_cols = []
 
         if self.check.column is not None and self._is_invalid_column(self.check.column):
-            invalid_cols.append(get_column_name_or_alias(self.check.column))
+            invalid_cols.append(self._display_column_name(self.check.column))
         elif self.check.columns is not None:  # either column or columns can be provided, but not both
             for column in self.check.columns:
                 if self._is_invalid_column(column):
-                    invalid_cols.append(get_column_name_or_alias(column))
+                    invalid_cols.append(self._display_column_name(column))
 
         return invalid_cols
 
+    @staticmethod
+    def _display_column_name(column: str | Column) -> str:
+        """
+        Returns the column name for use in skip / error messages, back-quoting names that require SQL
+        identifier escaping (e.g. "Customer Name" -> "`Customer Name`").
+        """
+        name = get_column_name_or_alias(column)
+        if is_simple_column_expression(name):
+            return name
+        return quote_column_name(name)
+
     @cached_property
     def has_invalid_columns(self) -> bool:
         """
@@ -260,6 +274,12 @@ def _is_invalid_column(self, column: str | Column) -> bool:
             col_expr = F.expr(column) if isinstance(column, str) else column
             _ = self.df.select(col_expr).schema  # perform logical plan validation without triggering computation
         except AnalysisException as e:
+            # The input string may be a SQL expression (e.g. "a + b") or a plain column name that may require
+            # SQL identifier escaping (spaces / non-ASCII / reserved chars, e.g. "Customer Name"). To validate
+            # the input string, we first attempt F.expr() and retry failing strings as a backtick-quoted identifiers.
+            # String expressions parse cleanly and only genuine single-identifier names pass fallback validation.
+            if isinstance(column, str) and not self._is_invalid_quoted_column(column):
+                return False
             # If column is not accessible or column expression cannot be evaluated, an AnalysisException is thrown.
             # Note: This does not cover all error conditions. Some issues only appear during a Spark action.
             logger.debug(
@@ -268,3 +288,14 @@ def _is_invalid_column(self, column: str | Column) -> bool:
             )
             return True
         return False
+
+    def _is_invalid_quoted_column(self, column: str) -> bool:
+        """
+        Returns True if the string cannot be resolved as a single backtick-quoted column identifier,
+        otherwise False.
+        """
+        try:
+            _ = self.df.select(F.expr(quote_column_name(column))).schema
+        except AnalysisException:
+            return True
+        return False

src/databricks/labs/dqx/utils.py

@@ -220,6 +220,24 @@ def normalize_bound_args(val: Any, allow_simple_expressions_only: bool = True) -
     return _normalize_leaf_value(val, allow_simple_expressions_only)
 
 
+def quote_column_name(name: str) -> str:
+    """
+    Wraps a column name in backticks so it can be used as a SQL identifier.
+
+    Column names containing spaces, non-ASCII characters, or other characters that require escaping
+    (e.g. "Customer Name", "Ääkkönen") are not valid bare SQL identifiers and must be back-quoted before
+    being parsed by ``F.expr``.
+
+    Args:
+        name: Column name to quote.
+
+    Returns:
+        The column name wrapped in backticks with embedded backticks escaped.
+    """
+    escaped = name.replace("`", "``")
+    return f"`{escaped}`"
+
+
 def normalize_col_str(col_str: str) -> str:
     """
     Normalizes string to be compatible with metastore column names by applying the following transformations:

tests/integration/test_apply_checks.py

@@ -10040,6 +10040,38 @@ def test_apply_checks_with_has_valid_schema_extra_columns_in_params(ws, spark):
     assert_df_equality(checked.sort("id"), expected.sort("id"), ignore_nullable=True)
 
 
+def test_apply_checks_with_has_valid_schema_special_char_columns_are_valid(ws, spark):
+    """Column names with spaces / non-ASCII characters (that require SQL identifier escaping
+    must be treated as valid and the check must run instead of being skipped."""
+    dq_engine = DQEngine(workspace_client=ws, extra_params=EXTRA_PARAMS)
+
+    schema = "id int, `Customer Name` string, `Ääkkönen` int"
+    test_df = spark.createDataFrame([[1, "Alice", 10], [2, "Bob", 20]], schema)
+
+    checks = [
+        DQDatasetRule(
+            name="has_valid_schema",
+            criticality="warn",
+            check_func=check_funcs.has_valid_schema,
+            check_func_kwargs={
+                "expected_schema": "id int, `Customer Name` string, `Ääkkönen` int",
+                "columns": ["id", "Customer Name", "Ääkkönen"],
+                "strict": False,
+            },
+        ),
+    ]
+    checked = dq_engine.apply_checks(test_df, checks)
+
+    expected = spark.createDataFrame(
+        [
+            [1, "Alice", 10, None, None],
+            [2, "Bob", 20, None, None],
+        ],
+        schema + REPORTING_COLUMNS,
+    )
+    assert_df_equality(checked.sort("id"), expected.sort("id"), ignore_nullable=True)
+
+
 def test_apply_checks_and_save_in_tables_for_patterns_missing_output_suffix(ws, spark):
     dq_engine = DQEngine(ws)
 

tests/integration/test_apply_checks_suppress_skipped.py

@@ -1,7 +1,7 @@
 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 databricks.labs.dqx.rule import DQDatasetRule, DQRowRule
 from tests.integration.conftest import (
     EXTRA_PARAMS,
     RUN_TIME,
@@ -57,3 +57,55 @@ def test_apply_checks_skipped_flag_set_on_skipped_entries(ws, spark):
     assert len(errors) == 1
     assert errors[0]["name"] == "missing_col_is_null"
     assert errors[0]["skipped"] is True
+
+
+def test_apply_checks_suppress_skipped_suppresses_missing_special_char_column(ws, spark):
+    """A missing special-char column is still skipped, and suppress_skipped removes the entry."""
+    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)
+    special_schema = "id int, `Customer Name` string"
+    test_df = spark.createDataFrame([[1, "Alice"], [2, "Bob"]], special_schema)
+
+    checks = [
+        DQDatasetRule(
+            name="has_valid_schema",
+            criticality="error",
+            check_func=check_funcs.has_valid_schema,
+            check_func_kwargs={
+                "expected_schema": "id int, `Customer Name` string, `Missing Column` string",
+                "columns": ["Customer Name", "Missing Column"],
+                "strict": False,
+            },
+        ),
+    ]
+
+    good_df, bad_df = dq_engine.apply_checks_and_split(test_df, checks)
+    assert bad_df.count() == 0
+    assert good_df.count() == 2
+
+
+def test_apply_checks_missing_special_char_column_is_skipped_and_back_quoted(ws, spark):
+    """A special-char column name that does not exist is still skipped, and is back-quoted in the message."""
+    dq_engine = DQEngine(workspace_client=ws, extra_params=EXTRA_PARAMS)
+    special_schema = "id int, `Customer Name` string"
+    test_df = spark.createDataFrame([[1, "Alice"]], special_schema)
+
+    checks = [
+        DQDatasetRule(
+            name="has_valid_schema",
+            criticality="error",
+            check_func=check_funcs.has_valid_schema,
+            check_func_kwargs={
+                "expected_schema": "id int, `Customer Name` string, `Missing Column` string",
+                "columns": ["Customer Name", "Missing Column"],
+                "strict": False,
+            },
+        ),
+    ]
+    checked = dq_engine.apply_checks(test_df, checks)
+
+    errors = checked.select("_errors").first()["_errors"]
+    assert len(errors) == 1
+    assert errors[0]["name"] == "has_valid_schema"
+    assert errors[0]["skipped"] is True
+    assert errors[0]["message"] == "Check evaluation skipped due to invalid check columns: ['`Missing Column`']"