Allow custom check messages

docs/dqx/docs/reference/quality_checks.mdx

@@ -4095,6 +4095,92 @@ Using DQX classes:
 When using dataset-level checks, the top-level `filter` condition is pushed down as `row_filter` to the check function and applied before aggregation, ensuring that the check operates only on the relevant subset of rows rather than on the aggregated results.
 </Admonition>
 
+## Customizing check messages
+
+Users can override the default failure message of any `DQRule` by specifying a custom message expression. Set `message_expr` to either a Spark SQL expression string or a Spark `Column` expression that returns a string-valued message when data fails a check.
+If a check cannot be evaluated (for example, because it references invalid columns or uses an invalid SQL expression), the results report the default skip message instead of the custom message.
+
+<Admonition type="tip" title="Using custom messages">
+When using custom message expressions:
+
+* Messages defined as SQL expressions are only validated when checks are run (e.g. with `apply_checks_by_metadata(...)`).
+* Wrap column references with `coalesce` to avoid null messages. In Spark SQL, `concat(..., null)` returns `null`.
+* Pass literal string values with quotes (e.g. `'Email must not be null'`) to ensure they are not evaluated as SQL expressions.
+* Avoid long messages or referencing unbounded string columns. Message text is **truncated at 500 characters**.
+* Avoid unquoted SQL keywords that may be executed naively (e.g. `DELETE` or `TRUNCATE`).
+</Admonition>
+
+<Tabs>
+  <TabItem value="Python" label="Python" default>
+    ```python
+    import pyspark.sql.functions as F
+    from databricks.labs.dqx import check_funcs
+    from databricks.labs.dqx.rule import DQRowRule
+
+    # static message: "Email must not be null"
+    checks = [
+      DQRowRule(
+        name="email_not_null",
+        criticality="error",
+        check_func=check_funcs.is_not_null,
+        column="email",
+        message_expr="'Email must not be null'",
+      )
+    ]
+
+    # dynamic message using a SQL expression string: "age_positive: age <value> is not valid"
+    checks = [
+      DQRowRule(
+        name="age_positive",
+        criticality="error",
+        check_func=check_funcs.is_not_less_than,
+        column="age",
+        check_func_kwargs={"limit": 0},
+        message_expr="concat('age_positive: age ', coalesce(cast(age as string), 'null'), ' is not valid')",
+      )
+    ]
+
+    # dynamic message using a Spark Column expression: "age_positive: age <value> is not valid"
+    checks = [
+      DQRowRule(
+        name="age_positive",
+        criticality="error",
+        check_func=check_funcs.is_not_less_than,
+        column="age",
+        check_func_kwargs={"limit": 0},
+        message_expr=F.concat(
+            F.lit("age_positive: age "),
+            F.coalesce(F.col("age").cast("string"), F.lit("null")),
+            F.lit(" is not valid"),
+        ),
+      )
+    ]
+    ```
+  </TabItem>
+  <TabItem value="YAML" label="YAML">
+    ```yaml
+    # static message: "Email must not be null"
+    - name: email_not_null
+      criticality: error
+      message_expr: "'Email must not be null'"
+      check:
+        function: is_not_null
+        arguments:
+          column: email
+
+    # dynamic message using a SQL expression string: "age_positive: age <value> is not valid"
+    - name: age_positive
+      criticality: error
+      message_expr: "concat('age_positive: age ', coalesce(cast(age as string), 'null'), ' is not valid')"
+      check:
+        function: is_not_less_than
+        arguments:
+          column: age
+          limit: 0
+    ```
+  </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/checks_serializer.py

@@ -267,6 +267,7 @@ def deserialize(self, checks: list[dict]) -> list[DQRule]:
             criticality = check_def.get("criticality", "error")
             filter_str = check_def.get("filter")
             user_metadata = check_def.get("user_metadata")
+            message_expr = check_def.get("message_expr")
 
             # Exclude `column` and `columns` from check_func_kwargs
             # as these are always included in the check function call
@@ -282,6 +283,7 @@ def deserialize(self, checks: list[dict]) -> list[DQRule]:
                     filter=filter_str,
                     check_func_kwargs=check_func_kwargs,
                     user_metadata=user_metadata,
+                    message_expr=message_expr,
                 ).get_rules()
             else:
                 rule_type = CHECK_FUNC_REGISTRY.get(func_name)
@@ -296,6 +298,7 @@ def deserialize(self, checks: list[dict]) -> list[DQRule]:
                             criticality=criticality,
                             filter=filter_str,
                             user_metadata=user_metadata,
+                            message_expr=message_expr,
                         )
                     )
                 else:  # default to row-level rule
@@ -309,6 +312,7 @@ def deserialize(self, checks: list[dict]) -> list[DQRule]:
                             criticality=criticality,
                             filter=filter_str,
                             user_metadata=user_metadata,
+                            message_expr=message_expr,
                         )
                     )
 

src/databricks/labs/dqx/manager.py

@@ -100,6 +100,15 @@ def has_invalid_filter(self) -> bool:
         """
         return self._is_invalid_column(self.filter_condition)
 
+    @cached_property
+    def has_invalid_custom_message(self) -> bool:
+        """
+        Returns a boolean indicating whether the custom message expression is invalid in the input DataFrame.
+        """
+        if self.check.message_expr is None:
+            return False
+        return self._is_invalid_column(self.check.message_expr)
+
     @cached_property
     def invalid_sql_expression(self) -> str | None:
         """
@@ -151,9 +160,11 @@ def _build_result_struct(self, condition: Column, skipped: bool = False) -> Colu
         # 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)
 
+        message_col = self._build_message_col(condition, skipped=skipped)
+
         return F.struct(
             F.lit(self.check.name).alias("name"),
-            condition.alias("message"),
+            message_col.alias("message"),
             self.check.columns_as_string_expr.alias("columns"),
             F.lit(self.check.filter or None).cast("string").alias("filter"),
             F.lit(self.check.check_func.__name__).alias("function"),
@@ -167,6 +178,35 @@ def _build_result_struct(self, condition: Column, skipped: bool = False) -> Colu
             F.lit(skipped or None).alias("skipped"),
         ).cast(dq_result_item_schema)
 
+    def _build_message_col(self, condition: Column, skipped: bool = False) -> Column:
+        """
+        Builds the message column, using the default message or the user-supplied ``message_expr`` from the
+        rule definition. The expression is evaluated as-is. Accepts either a Spark SQL expression string or a
+        Spark Column.
+
+        Args:
+            condition: Default DQX condition message returned by evaluating the DQX check function
+            skipped: Whether the check was skipped (default False)
+
+        Returns:
+            The custom DQX condition message if ``message_expr`` is set on the rule and the check was not skipped,
+            otherwise the default DQX condition message.
+        """
+        if skipped:
+            return condition
+
+        if self.check.message_expr is None:
+            return condition
+
+        _max_message_length = 500
+        message_expr = F.substr(
+            F.expr(self.check.message_expr) if isinstance(self.check.message_expr, str) else self.check.message_expr,
+            F.lit(1),
+            F.lit(_max_message_length),
+        )
+
+        return F.when(condition.isNotNull(), message_expr).otherwise(F.lit(None).cast("string"))
+
     def _get_invalid_cols_message(self) -> str:
         """
         Returns invalid columns message containing info about invalid columns to check should be applied to or filter.
@@ -187,6 +227,18 @@ def _get_invalid_cols_message(self) -> str:
                 f"Check evaluation skipped due to invalid check filter: '{self.check.filter}'"
             )
 
+        if self.has_invalid_custom_message:
+            if isinstance(self.check.message_expr, str):
+                custom_message_detail = f": '{self.check.message_expr}'"
+            else:
+                custom_message_detail = ""
+            logger.warning(
+                f"Skipping check '{self.check.name}' due to invalid custom message expression{custom_message_detail}"
+            )
+            invalid_cols_message_parts.append(
+                f"Check evaluation skipped due to invalid custom message expression{custom_message_detail}"
+            )
+
         if self.invalid_sql_expression:
             logger.warning(
                 f"Skipping check '{self.check.name}' due to invalid sql expression: '{self.invalid_sql_expression}'"

src/databricks/labs/dqx/rule.py

@@ -12,7 +12,7 @@
 from pyspark.sql import Column
 import pyspark.sql.functions as F
 from databricks.labs.dqx.utils import get_column_name_or_alias, normalize_bound_args
-from databricks.labs.dqx.errors import InvalidCheckError
+from databricks.labs.dqx.errors import InvalidCheckError, InvalidParameterError
 
 logger = logging.getLogger(__name__)
 
@@ -165,6 +165,12 @@ class DQRule(abc.ABC, DQRuleTypeMixin, SingleColumnMixin, MultipleColumnsMixin):
     * *check_func_args* (optional) - Positional arguments for the check function (excluding *column*).
     * *check_func_kwargs* (optional) - Keyword arguments for the check function (excluding *column*).
     * *user_metadata* (optional) - User-defined key-value pairs added to metadata generated by the check.
+    * *message_expr* (optional) - User-defined expression used as the check failure message. Accepts either
+        a Spark SQL expression string or a Spark *Column* expression. The expression is evaluated as-is.
+        Any column references, casts, or rule-identifying literals must be supplied directly by the caller
+        (e.g., ``F.concat(F.lit('age_positive: value '), F.col('age').cast('string'))`` or
+        ``"concat('age_positive: value ', cast(age as string))"``). The same message is shared across all
+        rules generated from a ``DQForEachColRule``.
     """
 
     check_func: Callable
@@ -176,6 +182,7 @@ class DQRule(abc.ABC, DQRuleTypeMixin, SingleColumnMixin, MultipleColumnsMixin):
     check_func_args: list[Any] = field(default_factory=list)
     check_func_kwargs: dict[str, Any] = field(default_factory=dict)
     user_metadata: dict[str, str] | None = None
+    message_expr: str | Column | None = None
 
     def __post_init__(self):
         self._validate_rule_type(self.check_func)
@@ -184,6 +191,8 @@ def __post_init__(self):
         self._validate_attributes()
         check_condition = self.get_check_condition()
         self._initialize_name_if_missing(check_condition)
+        if isinstance(self.message_expr, str):
+            self._validate_message_expression(self.message_expr)
 
     @abc.abstractmethod
     def get_check_condition(self) -> Column:
@@ -259,6 +268,12 @@ def to_dict(self) -> dict:
 
         if self.user_metadata:
             metadata["user_metadata"] = self.user_metadata
+        # Only string expressions can be round-tripped through metadata; Column objects are
+        # in-process Spark expressions with no canonical YAML/JSON representation.
+        if isinstance(self.message_expr, str):
+            metadata["message_expr"] = self.message_expr
+        elif self.message_expr is not None:
+            logger.warning("Message expressions of type 'Column' cannot be serialized; falling back to default message")
         return metadata
 
     def _initialize_column_if_missing(self):
@@ -341,6 +356,24 @@ def _is_optional_argument(self, signature: inspect.Signature, arg_name: str):
             return None  # Argument not present
         return param.default is not inspect.Parameter.empty
 
+    @staticmethod
+    def _validate_message_expression(message_expr: str) -> None:
+        """
+        Checks that the message expression is a logically valid Spark SQL expression.
+
+        Args:
+            message_expr: Message expression
+
+        Raises:
+            InvalidParameterError: If the expression is not a logically valid Spark SQL expression.
+        """
+        try:
+            F.expr(message_expr)
+        except Exception as exc:
+            raise InvalidParameterError(
+                f"Custom message expression '{message_expr}' is not a valid Spark SQL expression."
+            ) from exc
+
 
 @dataclass(frozen=True)
 class DQRowRule(DQRule):
@@ -428,6 +461,7 @@ class DQForEachColRule(DQRuleTypeMixin):
     check_func_args: list[Any] = field(default_factory=list)
     check_func_kwargs: dict[str, Any] = field(default_factory=dict)
     user_metadata: dict[str, str] | None = None
+    message_expr: str | Column | None = None
 
     def get_rules(self) -> list[DQRule]:
         """Build a list of rules for a set of columns.
@@ -453,6 +487,7 @@ def get_rules(self) -> list[DQRule]:
                         criticality=self.criticality,
                         filter=self.filter,
                         user_metadata=self.user_metadata,
+                        message_expr=self.message_expr,
                     )
                 )
             else:  # default to row-level rule
@@ -467,6 +502,7 @@ def get_rules(self) -> list[DQRule]:
                         criticality=self.criticality,
                         filter=self.filter,
                         user_metadata=self.user_metadata,
+                        message_expr=self.message_expr,
                     )
                 )
         return rules

tests/integration/conftest.py

@@ -121,6 +121,30 @@ def build_quality_violation(
     }
 
 
+def build_skipped_violation(
+    name: str,
+    message: str,
+    columns: list[str] | None,
+    *,
+    function: str = "is_not_null",
+    filter_expr: str | None = None,
+    user_metadata: dict | None = None,
+) -> dict[str, Any]:
+    """Helper for constructing expected entries for checks that were skipped during evaluation."""
+
+    return {
+        "name": name,
+        "message": message,
+        "columns": columns,
+        "filter": filter_expr,
+        "function": function,
+        "run_time": RUN_TIME,
+        "run_id": RUN_ID,
+        "user_metadata": user_metadata or {},
+        "skipped": True,
+    }
+
+
 def assert_check_and_split_results(
     checked: DataFrame,
     good_df: DataFrame,