Fix escaping during check execution

docs/dqx/docs/dev/docs_authoring.mdx

@@ -215,7 +215,7 @@ leak the component markup into these elements.
 Any page, section, or subsection heading can be tagged. The following conventions are used for tagging
 feature documentation:
 
-* Untagged features represent generally-available functionality. If a version tag is missing, the feature has been available since before release version 0.9.0.
+* Untagged features represent generally available functionality. If a version tag is missing, the feature has been available since before release version 0.9.0.
 * If a feature page is tagged, all untagged subheadings share the same version and lifecycle stage.
 * If any subheading is tagged, that capability has it own version and lifecycle stage which differs from the page-level version and lifecycle stage.
 * Sections with structural content, explanations, or examples should not be tagged.
@@ -224,13 +224,10 @@ feature documentation:
 
 The following components are available:
 
-*`<FeatureLifecycleStage stage="experimental | beta | ga | deprecated" />` — a status badge linked
-  to the matching section of the [Feature lifecycle](/docs/reference/feature_lifecycle) reference.
-*`<AvailableSinceVersion productName="DQX" version="0.14.0" />` — "Available since DQX
-  v0.14.0", linked to that release's notes.
-*`<DeprecatedInVersion productName="DQX" version="0.16.0" replacement="the new_check function" />`
-  — "Deprecated in DQX v0.16.0", with an optional replacement named in the tooltip.
-*`<FeatureTags>` — the row container that places the tags on their own line under the heading.
+* `<FeatureLifecycleStage stage="experimental | beta | ga | deprecated" />` — a status badge linked to the matching section of the [Feature Lifecycle](/docs/reference/feature_lifecycle) reference.
+* `<AvailableSinceVersion version="0.14.0" />` — "Available since v0.14.0", linked to that release's notes.
+* `<DeprecatedInVersion version="0.16.0" replacement="the new_check function" />` — "Deprecated in v0.16.0", with an optional replacement named in the tooltip.
+* `<FeatureTags>` — the row container that places the tags on their own line under the heading.
 
 ## Content alignment and structure of folders
 

docs/dqx/docs/reference/feature_lifecycle.mdx

@@ -1,10 +1,10 @@
 ---
 sidebar_position: 500.5
-title: Feature lifecycle
-sidebar_label: Feature lifecycle
+title: Feature Lifecycle
+sidebar_label: Feature Lifecycle
 ---
 
-# Feature lifecycle
+# Feature Lifecycle
 
 DQX features move through release stages as they mature. When a feature is not yet generally
 available, its documentation page shows a status badge next to the title — for example, the

docs/dqx/docs/reference/quality_checks.mdx

@@ -4668,6 +4668,56 @@ When using custom message expressions:
   </TabItem>
 </Tabs>
 
+## SQL escaping for checked columns
+Most rules accept a `column` argument that can be a column name or a SQL expression. DQX automatically back-quotes column names that are not valid SQL identifiers (e.g. `Customer Name` or `Päivämäärä`) on a best-effort basis. Columns that cannot be resolved are reported as skipped rather than failing a run.
+
+Because column names and a SQL expression cannot always be differentiated, some ambiguous cases are not back-quoted automatically: 
+* A name that contains expression characters such as parentheses (e.g. `amount (usd)`)
+* An operator-free expression that is treated as a name (e.g. `col not null`). 
+
+When working with these cases, back-quote the column name when defining the check.
+
+<Tabs>
+<TabItem value="YAML" label="YAML">
+```yaml
+# Back-quote ambiguous column names when defining checks
+- name: amount_is_null_or_empty
+  check:
+    function: is_not_null_and_not_empty
+    arguments:
+      column: "`amount (usd)`"
+
+- name: col_not_null_is_null_or_empty
+  check:
+    function: is_not_null_and_not_empty
+    arguments:
+      column: "`col not null`"
+```
+</TabItem>
+<TabItem value="Python" label="Python">
+```python
+import yaml
+import pyspark.sql.functions as F
+from databricks.labs.dqx.rule import DQRowRule
+from databricks.labs.dqx import check_funcs
+
+# Back-quote ambiguous column names when defining checks
+checks = [
+  DQRowRule(
+    name="amount_is_null_or_empty",
+    check_func=check_funcs.is_not_null_and_not_empty,
+    column="`amount (usd)`",
+  ),
+  DQRowRule(
+    name="col_not_null_is_null_or_empty",
+    check_func=check_funcs.is_not_null_and_not_empty,
+    column="`col not null`",
+  ),
+]
+```
+</TabItem>
+</Tabs>
+
 ## Converting checks between formats
 
 In DQX, checks can be defined either as Python classes or YAML declarations. When using YAML, the files are first parsed into dictionaries and then transformed into DQX class instances under the hood. Since both formats share the same internal structure, they are interchangeable and can be safely converted between one another.

src/databricks/labs/dqx/check_funcs.py

@@ -24,6 +24,8 @@
     is_sql_query_safe,
     safe_filter_expr,
     normalize_col_str,
+    normalize_column_expr,
+    unquote_column_name,
     get_columns_as_strings,
     to_lowercase,
 )
@@ -4854,8 +4856,12 @@ def get_normalized_column_and_expr(column: str | Column) -> tuple[str, str, Colu
             - Spark Column expression corresponding to the input.
     """
     col_expr = _get_column_expr(column)
-    column_str = get_column_name_or_alias(col_expr)
-    col_str_norm = get_column_name_or_alias(col_expr, normalize=True)
+    if isinstance(column, str):
+        column_str = unquote_column_name(column)
+        col_str_norm = normalize_col_str(column_str)
+    else:
+        column_str = get_column_name_or_alias(col_expr)
+        col_str_norm = get_column_name_or_alias(col_expr, normalize=True)
 
     return col_str_norm, column_str, col_expr
 
@@ -4989,13 +4995,17 @@ def _get_column_expr(column: Column | str) -> Column:
     """
     Convert a column input (string or Column) into a Spark Column expression.
 
+    String inputs are normalized via *normalize_column_expr* so that column names requiring SQL identifier
+    escaping (e.g. column names with spaces, non-ASCII letters, etc.) are back-quoted and resolve correctly,
+    while genuine SQL expressions are passed through unchanged.
+
     Args:
         column: The input column, provided as either a string column name or a Spark Column expression.
 
     Returns:
         A Spark Column expression corresponding to the input.
     """
-    return F.expr(column) if isinstance(column, str) else column
+    return F.expr(normalize_column_expr(column)) if isinstance(column, str) else column
 
 
 def _handle_fk_composite_keys(

src/databricks/labs/dqx/datacontract/contract_rules_generator.py

@@ -33,6 +33,7 @@
 from databricks.labs.dqx.errors import InvalidPhysicalTypeError, ODCSContractError, ParameterError
 from databricks.labs.dqx.telemetry import telemetry_logger
 from databricks.labs.dqx.package_utils import missing_required_packages
+from databricks.labs.dqx.utils import normalize_column_expr
 
 # DQLLMEngine is referenced only as a type annotation. Eagerly importing it
 # requires installation of [llm] extras which may not be installed or wanted
@@ -745,6 +746,7 @@ def _generate_range_rules_from_options(
         has_float_limits = (minimum is not None and isinstance(minimum, float)) or (
             maximum is not None and isinstance(maximum, float)
         )
+        normalized_column = normalize_column_expr(column_path)
 
         if minimum is not None and maximum is not None:
             if has_float_limits:
@@ -753,7 +755,7 @@ def _generate_range_rules_from_options(
                         "check": {
                             "function": "sql_expression",
                             "arguments": {
-                                "expression": f"{column_path} >= {minimum} AND {column_path} <= {maximum}",
+                                "expression": f"{normalized_column} >= {minimum} AND {normalized_column} <= {maximum}",
                                 "columns": [column_path],
                             },
                         },
@@ -794,7 +796,7 @@ def _generate_range_rules_from_options(
                         "check": {
                             "function": "sql_expression",
                             "arguments": {
-                                "expression": f"{column_path} >= {minimum}",
+                                "expression": f"{normalized_column} >= {minimum}",
                                 "columns": [column_path],
                             },
                         },
@@ -834,7 +836,7 @@ def _generate_range_rules_from_options(
                         "check": {
                             "function": "sql_expression",
                             "arguments": {
-                                "expression": f"{column_path} <= {maximum}",
+                                "expression": f"{normalized_column} <= {maximum}",
                                 "columns": [column_path],
                             },
                         },
@@ -879,13 +881,15 @@ def _generate_string_length_rules_from_options(
         if min_length is None and max_length is None:
             return []
 
+        normalized_column = normalize_column_expr(column_path)
+
         if min_length is not None and max_length is not None and min_length == max_length:
             return [
                 {
                     "check": {
                         "function": "sql_expression",
                         "arguments": {
-                            "expression": f"LENGTH({column_path}) = {min_length}",
+                            "expression": f"LENGTH({normalized_column}) = {min_length}",
                             "columns": [column_path],
                         },
                     },
@@ -905,7 +909,7 @@ def _generate_string_length_rules_from_options(
                     "check": {
                         "function": "sql_expression",
                         "arguments": {
-                            "expression": f"LENGTH({column_path}) >= {min_length} AND LENGTH({column_path}) <= {max_length}",
+                            "expression": f"LENGTH({normalized_column}) >= {min_length} AND LENGTH({normalized_column}) <= {max_length}",
                             "columns": [column_path],
                         },
                     },
@@ -925,7 +929,7 @@ def _generate_string_length_rules_from_options(
                     "check": {
                         "function": "sql_expression",
                         "arguments": {
-                            "expression": f"LENGTH({column_path}) >= {min_length}",
+                            "expression": f"LENGTH({normalized_column}) >= {min_length}",
                             "columns": [column_path],
                         },
                     },
@@ -945,7 +949,7 @@ def _generate_string_length_rules_from_options(
                     "check": {
                         "function": "sql_expression",
                         "arguments": {
-                            "expression": f"LENGTH({column_path}) <= {max_length}",
+                            "expression": f"LENGTH({normalized_column}) <= {max_length}",
                             "columns": [column_path],
                         },
                     },

src/databricks/labs/dqx/manager.py

@@ -17,6 +17,7 @@
 from databricks.labs.dqx.utils import (
     get_column_name_or_alias,
     is_simple_column_expression,
+    normalize_column_expr,
     quote_column_name,
     is_sql_query_safe,
     safe_filter_expr,
@@ -127,11 +128,11 @@ def invalid_columns(self) -> list[str]:
         """
         invalid_cols = []
 
-        if self.check.column is not None and self._is_invalid_column(self.check.column):
+        if self.check.column is not None and self._is_invalid_check_column(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):
+                if self._is_invalid_check_column(column):
                     invalid_cols.append(self._display_column_name(column))
 
         return invalid_cols
@@ -328,21 +329,27 @@ def _get_invalid_cols_message(self) -> str:
 
         return invalid_cols_message
 
+    def _is_invalid_check_column(self, column: str | Column) -> bool:
+        """
+        Returns True if a check column cannot be resolved in the input DataFrame, otherwise False.
+
+        A string column is validated exactly as it will be executed: it is passed through
+        *normalize_column_expr* (which back-quotes names requiring SQL identifier escaping, e.g.
+        "Customer Name") first, so validation and execution agree and an unresolvable name is skipped
+        rather than aborting the run.
+        """
+        resolved: str | Column = normalize_column_expr(column) if isinstance(column, str) else column
+        return self._is_invalid_column(resolved)
+
     def _is_invalid_column(self, column: str | Column) -> bool:
         """
-        Returns True if the specified column is invalid (i.e., cannot be resolved in the input DataFrame),
-        otherwise False.
+        Returns True if the specified column or expression is invalid (i.e., cannot be resolved in the
+        input DataFrame), otherwise False.
         """
         try:
             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(
@@ -351,14 +358,3 @@ 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

@@ -78,6 +78,9 @@ def _validate_spark_column(value: Any) -> Any:
 COLUMN_NORMALIZE_EXPRESSION = re.compile("[^a-zA-Z0-9]+")
 COLUMN_PATTERN = re.compile(r"Column<'(.*?)(?: AS (\w+))?'>$", re.DOTALL)
 INVALID_COLUMN_NAME_PATTERN = re.compile(r"[\s,;{}\(\)\n\t=]+")
+VALID_UNQUOTED_IDENTIFIER_PATTERN = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
+SQL_EXPRESSION_STRUCTURE_PATTERN = re.compile(r"""[()\[\]{},;'"`*]""")
+SQL_EXPRESSION_OPERATOR_PATTERN = re.compile(r"[-+/%<>=!&|~^]\s|\s[-+/%<>=!&|~^]")
 _UNRESOLVED_PLACEHOLDER_PATTERN = re.compile(r"\{\{[^}]*\}\}")
 
 # Destructive SQL statement keywords rejected by `is_sql_query_safe`. SELECT is intentionally
@@ -308,6 +311,62 @@ def quote_column_name(name: str) -> str:
     return f"`{escaped}`"
 
 
+def unquote_column_name(name: str) -> str:
+    """
+    Removes surrounding back-quotes from a column name, reversing :func:`quote_column_name`.
+
+    A column reference that the user already back-quoted (e.g. "`Customer Name`") is unwrapped to its
+    plain form ("Customer Name") for use in display names and messages. Strings that are not a single
+    back-quoted identifier (plain names, dotted paths, SQL expressions) are returned unchanged.
+
+    Args:
+        name: Column reference provided as a string.
+
+    Returns:
+        The column name without surrounding back-quotes.
+    """
+    if len(name) >= 2 and name.startswith("`") and name.endswith("`"):
+        return name[1:-1].replace("``", "`")
+    return name
+
+
+def normalize_column_expr(column: str) -> str:
+    """
+    Prepares a column reference string for use with ``F.expr``, back-quoting names that require SQL
+    identifier escaping while leaving SQL expressions untouched.
+
+    Check functions accept a column as either a plain name or a SQL expression string. A plain name that
+    is not a valid bare identifier (spaces, dashes, non-ASCII letters, etc., e.g. "Customer Name",
+    "gross-margin" or "Päivämäärä") does not parse when passed to ``F.expr`` and must be back-quoted. A
+    SQL expression (e.g. "a + b", "substr(x, 1, 2)", "*") must be passed through as-is.
+
+    Name and expression cannot be told apart with certainty from a string alone. This function uses a
+    conservative heuristic. A string is treated as an expression, and returned unchanged, when it
+    contains characters that typically appear in expressions (parentheses, brackets, quotes, comma, star)
+    or arithmetic/comparison operators that are whitespace-separated (e.g. "a + b"). Everything else is
+    treated as a dotted column path: each segment is left alone if it is a valid bare identifier. Otherwise
+    it is back-quoted (e.g. "struct_col.field1" is unchanged, "Customer Name" becomes "`Customer Name`")
+
+    This heuristic intentionally excludes two ambiguous cases: names that contain expression characters
+    (e.g. "amount (usd)"), and operator-free SQL expressions such as "col IS NOT NULL" (which is treated
+    as a name). Callers with such columns should back-quote the name themselves, pass a Column expression,
+    or use the ``sql_expression`` check.
+
+    Args:
+        column: Column reference provided as a string (plain name, nested path, or SQL expression).
+
+    Returns:
+        A string safe to pass to ``F.expr``.
+    """
+    if SQL_EXPRESSION_STRUCTURE_PATTERN.search(column) or SQL_EXPRESSION_OPERATOR_PATTERN.search(column):
+        return column
+
+    return ".".join(
+        segment if VALID_UNQUOTED_IDENTIFIER_PATTERN.match(segment) else quote_column_name(segment)
+        for segment in column.split(".")
+    )
+
+
 def normalize_col_str(col_str: str) -> str:
     """
     Normalizes string to be compatible with metastore column names by applying the following transformations: