diff --git a/docs/dqx/docs/reference/quality_checks.mdx b/docs/dqx/docs/reference/quality_checks.mdx
index 099f13996..e101bfdd3 100644
--- a/docs/dqx/docs/reference/quality_checks.mdx
+++ b/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.
+## 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.
+
+
+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`).
+
+
+
+
+ ```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 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 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"),
+ ),
+ )
+ ]
+ ```
+
+
+ ```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 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
+ ```
+
+
+
## 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.
diff --git a/src/databricks/labs/dqx/checks_serializer.py b/src/databricks/labs/dqx/checks_serializer.py
index 8167817d4..df489dd26 100644
--- a/src/databricks/labs/dqx/checks_serializer.py
+++ b/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,
)
)
diff --git a/src/databricks/labs/dqx/manager.py b/src/databricks/labs/dqx/manager.py
index aa9a3f6e6..bfb76bc21 100644
--- a/src/databricks/labs/dqx/manager.py
+++ b/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}'"
diff --git a/src/databricks/labs/dqx/rule.py b/src/databricks/labs/dqx/rule.py
index 329ab815e..68e9ce9f3 100644
--- a/src/databricks/labs/dqx/rule.py
+++ b/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
diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py
index 5749278d6..1d6ad4024 100644
--- a/tests/integration/conftest.py
+++ b/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,
diff --git a/tests/integration/test_apply_checks.py b/tests/integration/test_apply_checks.py
index 57a567d48..d7a48ccd0 100755
--- a/tests/integration/test_apply_checks.py
+++ b/tests/integration/test_apply_checks.py
@@ -28,6 +28,7 @@
EXTRA_PARAMS,
RUN_ID,
build_quality_violation,
+ build_skipped_violation,
assert_check_and_split_results,
assert_df_equality_ignore_fingerprints as assert_df_equality,
generate_checks_with_rule_and_set_fingerprint_from_rules,
@@ -10014,29 +10015,19 @@ def test_apply_checks_with_has_valid_schema_extra_columns_in_params(ws, spark):
expected_schema = schema + REPORTING_COLUMNS
- expected_skip_strict = {
- "name": "has_valid_schema_strict",
- "message": "Check evaluation skipped due to invalid check columns: ['missing_col']",
- "columns": ["id", "v1", "missing_col"],
- "filter": None,
- "function": "has_valid_schema",
- "run_time": RUN_TIME,
- "run_id": RUN_ID,
- "user_metadata": {},
- "skipped": True,
- }
-
- expected_skip_permissive = {
- "name": "has_valid_schema_permissive",
- "message": "Check evaluation skipped due to invalid check columns: ['missing_col']",
- "columns": ["id", "v1", "missing_col"],
- "filter": None,
- "function": "has_valid_schema",
- "run_time": RUN_TIME,
- "run_id": RUN_ID,
- "user_metadata": {},
- "skipped": True,
- }
+ expected_skip_strict = build_skipped_violation(
+ name="has_valid_schema_strict",
+ message="Check evaluation skipped due to invalid check columns: ['missing_col']",
+ columns=["id", "v1", "missing_col"],
+ function="has_valid_schema",
+ )
+
+ expected_skip_permissive = build_skipped_violation(
+ name="has_valid_schema_permissive",
+ message="Check evaluation skipped due to invalid check columns: ['missing_col']",
+ columns=["id", "v1", "missing_col"],
+ function="has_valid_schema",
+ )
expected = spark.createDataFrame(
[
@@ -10163,76 +10154,49 @@ def test_apply_checks_skip_checks_with_missing_columns(ws, spark):
{"key1": 1},
{"field1": 1},
[
- {
- "name": "b_is_null_or_empty",
- "message": "Check evaluation skipped due to invalid check filter: 'missing_col > 0'",
- "columns": ["b"],
- "filter": "missing_col > 0",
- "function": "is_not_null_and_not_empty",
- "run_time": RUN_TIME,
- "run_id": RUN_ID,
- "user_metadata": {},
- "skipped": True,
- },
- {
- "name": "missing_col_is_null",
- "message": "Check evaluation skipped due to invalid check columns: ['missing_col']",
- "columns": ["missing_col"],
- "filter": None,
- "function": "is_not_null",
- "run_time": RUN_TIME,
- "run_id": RUN_ID,
- "user_metadata": {},
- "skipped": True,
- },
- {
- "name": "missing_col_sql_expression",
- "message": "Check evaluation skipped due to invalid check columns: ['missing_col']; "
+ build_skipped_violation(
+ name="b_is_null_or_empty",
+ message="Check evaluation skipped due to invalid check filter: 'missing_col > 0'",
+ columns=["b"],
+ function="is_not_null_and_not_empty",
+ filter_expr="missing_col > 0",
+ ),
+ build_skipped_violation(
+ name="missing_col_is_null",
+ message="Check evaluation skipped due to invalid check columns: ['missing_col']",
+ columns=["missing_col"],
+ ),
+ build_skipped_violation(
+ name="missing_col_sql_expression",
+ message="Check evaluation skipped due to invalid check columns: ['missing_col']; "
"Check evaluation skipped due to invalid sql expression: 'missing_col > 0'",
- "columns": ["missing_col"],
- "filter": None,
- "function": "sql_expression",
- "run_time": RUN_TIME,
- "run_id": RUN_ID,
- "user_metadata": {},
- "skipped": True,
- },
- {
- "name": "missing_col_is_unique",
- "message": "Check evaluation skipped due to invalid check columns: ['missing_col']; "
+ columns=["missing_col"],
+ function="sql_expression",
+ ),
+ build_skipped_violation(
+ name="missing_col_is_unique",
+ message="Check evaluation skipped due to invalid check columns: ['missing_col']; "
"Check evaluation skipped due to invalid check filter: 'missing_col > 0'",
- "columns": ["missing_col"],
- "filter": "missing_col > 0",
- "function": "is_unique",
- "run_time": RUN_TIME,
- "run_id": RUN_ID,
- "user_metadata": {},
- "skipped": True,
- },
- {
- "name": "invalid_col_sql_expression",
- "message": "Check evaluation skipped due to invalid sql expression: 'missing_col > 0'",
- "columns": None,
- "filter": None,
- "function": "sql_expression",
- "run_time": RUN_TIME,
- "run_id": RUN_ID,
- "user_metadata": {},
- "skipped": True,
- },
+ columns=["missing_col"],
+ function="is_unique",
+ filter_expr="missing_col > 0",
+ ),
+ build_skipped_violation(
+ name="invalid_col_sql_expression",
+ message="Check evaluation skipped due to invalid sql expression: 'missing_col > 0'",
+ columns=None,
+ function="sql_expression",
+ ),
],
[
- {
- "name": "missing_col_is_null_or_empty",
- "message": "Check evaluation skipped due to invalid check columns: ['missing_col']",
- "columns": ["missing_col"],
- "filter": "a > 0",
- "function": "is_not_null_and_not_empty",
- "run_time": RUN_TIME,
- "run_id": RUN_ID,
- "user_metadata": {"tag1": "value1", "tag2": "value2"},
- "skipped": True,
- },
+ build_skipped_violation(
+ name="missing_col_is_null_or_empty",
+ message="Check evaluation skipped due to invalid check columns: ['missing_col']",
+ columns=["missing_col"],
+ function="is_not_null_and_not_empty",
+ filter_expr="a > 0",
+ user_metadata={"tag1": "value1", "tag2": "value2"},
+ ),
],
]
],
@@ -10347,76 +10311,49 @@ def test_apply_checks_by_metadata_skip_checks_with_missing_columns(ws, spark):
{"key1": 1},
{"field1": 1},
[
- {
- "name": "b_is_null_or_empty",
- "message": "Check evaluation skipped due to invalid check filter: 'missing_col > 0'",
- "columns": ["b"],
- "filter": "missing_col > 0",
- "function": "is_not_null_and_not_empty",
- "run_time": RUN_TIME,
- "run_id": RUN_ID,
- "user_metadata": {},
- "skipped": True,
- },
- {
- "name": "missing_col_is_null",
- "message": "Check evaluation skipped due to invalid check columns: ['missing_col']",
- "columns": ["missing_col"],
- "filter": None,
- "function": "is_not_null",
- "run_time": RUN_TIME,
- "run_id": RUN_ID,
- "user_metadata": {},
- "skipped": True,
- },
- {
- "name": "missing_col_sql_expression",
- "message": "Check evaluation skipped due to invalid check columns: ['missing_col']; "
+ build_skipped_violation(
+ name="b_is_null_or_empty",
+ message="Check evaluation skipped due to invalid check filter: 'missing_col > 0'",
+ columns=["b"],
+ function="is_not_null_and_not_empty",
+ filter_expr="missing_col > 0",
+ ),
+ build_skipped_violation(
+ name="missing_col_is_null",
+ message="Check evaluation skipped due to invalid check columns: ['missing_col']",
+ columns=["missing_col"],
+ ),
+ build_skipped_violation(
+ name="missing_col_sql_expression",
+ message="Check evaluation skipped due to invalid check columns: ['missing_col']; "
"Check evaluation skipped due to invalid sql expression: 'missing_col > 0'",
- "columns": ["missing_col"],
- "filter": None,
- "function": "sql_expression",
- "run_time": RUN_TIME,
- "run_id": RUN_ID,
- "user_metadata": {},
- "skipped": True,
- },
- {
- "name": "missing_col_is_unique",
- "message": "Check evaluation skipped due to invalid check columns: ['missing_col']; "
+ columns=["missing_col"],
+ function="sql_expression",
+ ),
+ build_skipped_violation(
+ name="missing_col_is_unique",
+ message="Check evaluation skipped due to invalid check columns: ['missing_col']; "
"Check evaluation skipped due to invalid check filter: 'missing_col > 0'",
- "columns": ["missing_col"],
- "filter": "missing_col > 0",
- "function": "is_unique",
- "run_time": RUN_TIME,
- "run_id": RUN_ID,
- "user_metadata": {},
- "skipped": True,
- },
- {
- "name": "invalid_col_sql_expression",
- "message": "Check evaluation skipped due to invalid sql expression: 'missing_col > 0'",
- "columns": None,
- "filter": None,
- "function": "sql_expression",
- "run_time": RUN_TIME,
- "run_id": RUN_ID,
- "user_metadata": {},
- "skipped": True,
- },
+ columns=["missing_col"],
+ function="is_unique",
+ filter_expr="missing_col > 0",
+ ),
+ build_skipped_violation(
+ name="invalid_col_sql_expression",
+ message="Check evaluation skipped due to invalid sql expression: 'missing_col > 0'",
+ columns=None,
+ function="sql_expression",
+ ),
],
[
- {
- "name": "missing_col_is_null_or_empty",
- "message": "Check evaluation skipped due to invalid check columns: ['missing_col']",
- "columns": ["missing_col"],
- "filter": "a > 0",
- "function": "is_not_null_and_not_empty",
- "run_time": RUN_TIME,
- "run_id": RUN_ID,
- "user_metadata": {"tag1": "value1", "tag2": "value2"},
- "skipped": True,
- },
+ build_skipped_violation(
+ name="missing_col_is_null_or_empty",
+ message="Check evaluation skipped due to invalid check columns: ['missing_col']",
+ columns=["missing_col"],
+ function="is_not_null_and_not_empty",
+ filter_expr="a > 0",
+ user_metadata={"tag1": "value1", "tag2": "value2"},
+ ),
],
]
],
diff --git a/tests/integration/test_custom_messages.py b/tests/integration/test_custom_messages.py
new file mode 100644
index 000000000..6e6d65ffa
--- /dev/null
+++ b/tests/integration/test_custom_messages.py
@@ -0,0 +1,559 @@
+"""Integration tests for custom message expressions on DQRule."""
+
+import pyspark.sql.functions as F
+
+from databricks.labs.dqx.engine import DQEngine
+from databricks.labs.dqx.rule import DQRowRule, DQForEachColRule
+from databricks.labs.dqx import check_funcs
+from tests.integration.conftest import (
+ EXTRA_PARAMS,
+ assert_df_equality_ignore_fingerprints as assert_df_equality,
+ REPORTING_COLUMNS,
+ build_quality_violation,
+ build_skipped_violation,
+)
+
+SCHEMA = "a: int, b: int, c: int"
+EXPECTED_SCHEMA = SCHEMA + REPORTING_COLUMNS
+
+
+def test_apply_checks_with_static_custom_message(ws, spark):
+ """A plain SQL literal message should appear in the result DataFrame."""
+ dq_engine = DQEngine(workspace_client=ws, extra_params=EXTRA_PARAMS)
+ test_df = spark.createDataFrame([[1, 3, None]], SCHEMA)
+
+ rules = [
+ DQRowRule(
+ name="c_not_null",
+ criticality="error",
+ check_func=check_funcs.is_not_null,
+ column="c",
+ message_expr="'Custom error: c_not_null'",
+ ),
+ ]
+
+ checked_df = dq_engine.apply_checks(test_df, rules)
+ expected_errors = [
+ build_quality_violation(
+ name="c_not_null",
+ message="Custom error: c_not_null",
+ columns=["c"],
+ function="is_not_null",
+ )
+ ]
+ expected_df = spark.createDataFrame(
+ [[1, 3, None, expected_errors, None]],
+ EXPECTED_SCHEMA,
+ )
+ assert_df_equality(checked_df, expected_df)
+
+
+def test_apply_checks_with_dynamic_column_value_message(ws, spark):
+ """SQL expression referencing the actual column should produce dynamic messages."""
+ dq_engine = DQEngine(workspace_client=ws, extra_params=EXTRA_PARAMS)
+ test_df = spark.createDataFrame([[1, None, 3]], SCHEMA)
+
+ rules = [
+ DQRowRule(
+ name="b_not_null",
+ criticality="warn",
+ check_func=check_funcs.is_not_null,
+ column="b",
+ message_expr=(
+ "concat('Rule b_not_null (is_not_null) failed for value: '," " coalesce(cast(b as string), 'null'))"
+ ),
+ ),
+ ]
+
+ checked_df = dq_engine.apply_checks(test_df, rules)
+ expected_warnings = [
+ build_quality_violation(
+ name="b_not_null",
+ message="Rule b_not_null (is_not_null) failed for value: null",
+ columns=["b"],
+ function="is_not_null",
+ )
+ ]
+ expected_df = spark.createDataFrame(
+ [[1, None, 3, None, expected_warnings]],
+ EXPECTED_SCHEMA,
+ )
+ assert_df_equality(checked_df, expected_df)
+
+
+def test_apply_checks_passing_rows_have_no_custom_message(ws, spark):
+ """Rows that pass the check should not have a message even with a custom message."""
+ dq_engine = DQEngine(workspace_client=ws, extra_params=EXTRA_PARAMS)
+ test_df = spark.createDataFrame([[1, 3, 5]], SCHEMA)
+
+ rules = [
+ DQRowRule(
+ name="c_not_null",
+ criticality="error",
+ check_func=check_funcs.is_not_null,
+ column="c",
+ message_expr="'Custom error: c_not_null'",
+ ),
+ ]
+
+ checked_df = dq_engine.apply_checks(test_df, rules)
+ expected_df = spark.createDataFrame(
+ [[1, 3, 5, None, None]],
+ EXPECTED_SCHEMA,
+ )
+ assert_df_equality(checked_df, expected_df)
+
+
+def test_for_each_col_rule_with_custom_message(ws, spark):
+ """DQForEachColRule with message_expr should propagate to generated rules.
+
+ The same expression is used for every generated rule. To produce a per-column
+ message, reference each column inline (e.g. concat with a per-column literal) or
+ construct rules individually.
+ """
+ dq_engine = DQEngine(workspace_client=ws, extra_params=EXTRA_PARAMS)
+ test_df = spark.createDataFrame([[None, None, 3]], SCHEMA)
+
+ rules = DQForEachColRule(
+ columns=["a", "b"],
+ check_func=check_funcs.is_not_null,
+ criticality="error",
+ message_expr="'Custom error: column missing'",
+ ).get_rules()
+
+ checked_df = dq_engine.apply_checks(test_df, rules)
+ expected_errors = [
+ build_quality_violation(
+ name="a_is_null",
+ message="Custom error: column missing",
+ columns=["a"],
+ function="is_not_null",
+ ),
+ build_quality_violation(
+ name="b_is_null",
+ message="Custom error: column missing",
+ columns=["b"],
+ function="is_not_null",
+ ),
+ ]
+ expected_df = spark.createDataFrame(
+ [[None, None, 3, expected_errors, None]],
+ EXPECTED_SCHEMA,
+ )
+ assert_df_equality(checked_df, expected_df)
+
+
+def test_apply_checks_with_column_value_non_null(ws, spark):
+ """When a check fails with a non-null value, the message can include that value."""
+ dq_engine = DQEngine(workspace_client=ws, extra_params=EXTRA_PARAMS)
+ test_df = spark.createDataFrame([[1, -5, 3]], SCHEMA)
+
+ rules = [
+ DQRowRule(
+ name="b_positive",
+ criticality="error",
+ check_func=check_funcs.is_not_less_than,
+ column="b",
+ check_func_kwargs={"limit": 0},
+ message_expr=("concat('b_positive: value ', coalesce(cast(b as string), 'null')," " ' is not positive')"),
+ ),
+ ]
+
+ checked_df = dq_engine.apply_checks(test_df, rules)
+ expected_errors = [
+ build_quality_violation(
+ name="b_positive",
+ message="b_positive: value -5 is not positive",
+ columns=["b"],
+ function="is_not_less_than",
+ )
+ ]
+ expected_df = spark.createDataFrame(
+ [[1, -5, 3, expected_errors, None]],
+ EXPECTED_SCHEMA,
+ )
+ assert_df_equality(checked_df, expected_df)
+
+
+def test_apply_checks_simple_literal_message(ws, spark):
+ """A plain string literal (no expression) should work as a static message."""
+ dq_engine = DQEngine(workspace_client=ws, extra_params=EXTRA_PARAMS)
+ test_df = spark.createDataFrame([[1, 3, None]], SCHEMA)
+
+ rules = [
+ DQRowRule(
+ name="c_not_null",
+ criticality="error",
+ check_func=check_funcs.is_not_null,
+ column="c",
+ message_expr="'Column c must not be null'",
+ ),
+ ]
+
+ checked_df = dq_engine.apply_checks(test_df, rules)
+ expected_errors = [
+ build_quality_violation(
+ name="c_not_null",
+ message="Column c must not be null",
+ columns=["c"],
+ function="is_not_null",
+ )
+ ]
+ expected_df = spark.createDataFrame(
+ [[1, 3, None, expected_errors, None]],
+ EXPECTED_SCHEMA,
+ )
+ assert_df_equality(checked_df, expected_df)
+
+
+def test_apply_checks_with_column_message_expr(ws, spark):
+ """A Spark Column passed as message_expr should be used directly without conversion."""
+ dq_engine = DQEngine(workspace_client=ws, extra_params=EXTRA_PARAMS)
+ test_df = spark.createDataFrame([[1, 3, None]], SCHEMA)
+
+ rules = [
+ DQRowRule(
+ name="c_not_null",
+ criticality="error",
+ check_func=check_funcs.is_not_null,
+ column="c",
+ message_expr=F.concat(
+ F.lit("Custom error: c_not_null (value="),
+ F.coalesce(F.col("c").cast("string"), F.lit("null")),
+ F.lit(")"),
+ ),
+ ),
+ ]
+
+ checked_df = dq_engine.apply_checks(test_df, rules)
+ expected_errors = [
+ build_quality_violation(
+ name="c_not_null",
+ message="Custom error: c_not_null (value=null)",
+ columns=["c"],
+ function="is_not_null",
+ )
+ ]
+ expected_df = spark.createDataFrame(
+ [[1, 3, None, expected_errors, None]],
+ EXPECTED_SCHEMA,
+ )
+ assert_df_equality(checked_df, expected_df)
+
+
+def test_metadata_static_custom_message(ws, spark):
+ """Static message defined in YAML-style metadata should appear in the result."""
+ dq_engine = DQEngine(workspace_client=ws, extra_params=EXTRA_PARAMS)
+ test_df = spark.createDataFrame([[1, 3, None]], SCHEMA)
+
+ checks = [
+ {
+ "name": "c_not_null",
+ "criticality": "error",
+ "message_expr": "'Custom error: c_not_null'",
+ "check": {"function": "is_not_null", "arguments": {"column": "c"}},
+ }
+ ]
+
+ checked_df = dq_engine.apply_checks_by_metadata(test_df, checks)
+ expected_errors = [
+ build_quality_violation(
+ name="c_not_null",
+ message="Custom error: c_not_null",
+ columns=["c"],
+ function="is_not_null",
+ )
+ ]
+ expected_df = spark.createDataFrame(
+ [[1, 3, None, expected_errors, None]],
+ EXPECTED_SCHEMA,
+ )
+ assert_df_equality(checked_df, expected_df)
+
+
+def test_metadata_dynamic_column_value_message(ws, spark):
+ """Dynamic message expression referencing the column from metadata should resolve correctly."""
+ dq_engine = DQEngine(workspace_client=ws, extra_params=EXTRA_PARAMS)
+ test_df = spark.createDataFrame([[1, None, 3]], SCHEMA)
+
+ checks = [
+ {
+ "name": "b_not_null",
+ "criticality": "warn",
+ "message_expr": (
+ "concat('Rule b_not_null (is_not_null) failed for value: '," " coalesce(cast(b as string), 'null'))"
+ ),
+ "check": {"function": "is_not_null", "arguments": {"column": "b"}},
+ }
+ ]
+
+ checked_df = dq_engine.apply_checks_by_metadata(test_df, checks)
+ expected_warnings = [
+ build_quality_violation(
+ name="b_not_null",
+ message="Rule b_not_null (is_not_null) failed for value: null",
+ columns=["b"],
+ function="is_not_null",
+ )
+ ]
+ expected_df = spark.createDataFrame(
+ [[1, None, 3, None, expected_warnings]],
+ EXPECTED_SCHEMA,
+ )
+ assert_df_equality(checked_df, expected_df)
+
+
+def test_metadata_without_message_uses_default(ws, spark):
+ """Metadata checks without a message_expr field should produce the default message."""
+ dq_engine = DQEngine(workspace_client=ws, extra_params=EXTRA_PARAMS)
+ test_df = spark.createDataFrame([[1, 3, None]], SCHEMA)
+
+ checks = [
+ {
+ "name": "c_not_null",
+ "criticality": "error",
+ "check": {"function": "is_not_null", "arguments": {"column": "c"}},
+ }
+ ]
+
+ checked_df = dq_engine.apply_checks_by_metadata(test_df, checks)
+ expected_errors = [
+ build_quality_violation(
+ name="c_not_null",
+ message="Column 'c' value is null",
+ columns=["c"],
+ function="is_not_null",
+ )
+ ]
+ expected_df = spark.createDataFrame(
+ [[1, 3, None, expected_errors, None]],
+ EXPECTED_SCHEMA,
+ )
+ assert_df_equality(checked_df, expected_df)
+
+
+def test_metadata_for_each_column_with_custom_message(ws, spark):
+ """for_each_column in metadata with message_expr should propagate to all generated rules."""
+ dq_engine = DQEngine(workspace_client=ws, extra_params=EXTRA_PARAMS)
+ test_df = spark.createDataFrame([[None, None, 3]], SCHEMA)
+
+ checks = [
+ {
+ "criticality": "error",
+ "message_expr": "'Custom error: column missing'",
+ "check": {
+ "function": "is_not_null",
+ "for_each_column": ["a", "b"],
+ },
+ }
+ ]
+
+ checked_df = dq_engine.apply_checks_by_metadata(test_df, checks)
+ expected_errors = [
+ build_quality_violation(
+ name="a_is_null",
+ message="Custom error: column missing",
+ columns=["a"],
+ function="is_not_null",
+ ),
+ build_quality_violation(
+ name="b_is_null",
+ message="Custom error: column missing",
+ columns=["b"],
+ function="is_not_null",
+ ),
+ ]
+ expected_df = spark.createDataFrame(
+ [[None, None, 3, expected_errors, None]],
+ EXPECTED_SCHEMA,
+ )
+ assert_df_equality(checked_df, expected_df)
+
+
+def test_apply_checks_skip_message_for_invalid_custom_message(ws, spark):
+ """A check with a valid evaluation but an unresolvable custom message expression should be skipped."""
+ dq_engine = DQEngine(workspace_client=ws, extra_params=EXTRA_PARAMS)
+ test_df = spark.createDataFrame([[1, 3, 5]], SCHEMA)
+
+ rules = [
+ DQRowRule(
+ name="c_not_null",
+ criticality="error",
+ check_func=check_funcs.is_not_null,
+ column="c",
+ message_expr="concat('Custom error for value: ', missing_col)",
+ user_metadata={"should_be": "IGNORED"},
+ ),
+ ]
+
+ checked_df = dq_engine.apply_checks(test_df, rules)
+ expected_errors = [
+ build_skipped_violation(
+ name="c_not_null",
+ message="Check evaluation skipped due to invalid custom message expression: "
+ "'concat('Custom error for value: ', missing_col)'",
+ columns=["c"],
+ user_metadata={"should_be": "IGNORED"},
+ )
+ ]
+ expected_df = spark.createDataFrame(
+ [[1, 3, 5, expected_errors, None]],
+ EXPECTED_SCHEMA,
+ )
+ assert_df_equality(checked_df, expected_df)
+
+
+def test_metadata_skip_message_for_invalid_custom_message(ws, spark):
+ """A metadata-defined check with an unresolvable custom message expression should be skipped."""
+ dq_engine = DQEngine(workspace_client=ws, extra_params=EXTRA_PARAMS)
+ test_df = spark.createDataFrame([[1, 3, 5]], SCHEMA)
+
+ checks = [
+ {
+ "name": "c_not_null",
+ "criticality": "error",
+ "message_expr": "concat('Custom error for value: ', missing_col)",
+ "check": {"function": "is_not_null", "arguments": {"column": "c"}},
+ }
+ ]
+
+ checked_df = dq_engine.apply_checks_by_metadata(test_df, checks)
+ expected_errors = [
+ build_skipped_violation(
+ name="c_not_null",
+ message="Check evaluation skipped due to invalid custom message expression: "
+ "'concat('Custom error for value: ', missing_col)'",
+ columns=["c"],
+ )
+ ]
+ expected_df = spark.createDataFrame(
+ [[1, 3, 5, expected_errors, None]],
+ EXPECTED_SCHEMA,
+ )
+ assert_df_equality(checked_df, expected_df)
+
+
+def test_apply_checks_skip_message_for_invalid_column_message_expr(ws, spark):
+ """A Column message_expr that cannot be resolved should be skipped without injecting the Column repr."""
+ dq_engine = DQEngine(workspace_client=ws, extra_params=EXTRA_PARAMS)
+ test_df = spark.createDataFrame([[1, 3, 5]], SCHEMA)
+
+ rules = [
+ DQRowRule(
+ name="c_not_null",
+ criticality="error",
+ check_func=check_funcs.is_not_null,
+ column="c",
+ message_expr=F.col("missing_col"),
+ ),
+ ]
+
+ checked_df = dq_engine.apply_checks(test_df, rules)
+ expected_errors = [
+ build_skipped_violation(
+ name="c_not_null",
+ message="Check evaluation skipped due to invalid custom message expression",
+ columns=["c"],
+ )
+ ]
+ expected_df = spark.createDataFrame(
+ [[1, 3, 5, expected_errors, None]],
+ EXPECTED_SCHEMA,
+ )
+ assert_df_equality(checked_df, expected_df)
+
+
+def test_apply_checks_skip_message_combines_invalid_columns_and_custom_message(ws, spark):
+ """When both the check column and the custom message are invalid, both reasons should be reported."""
+ dq_engine = DQEngine(workspace_client=ws, extra_params=EXTRA_PARAMS)
+ test_df = spark.createDataFrame([[1, 3, 5]], SCHEMA)
+
+ rules = [
+ DQRowRule(
+ name="missing_not_null",
+ criticality="error",
+ check_func=check_funcs.is_not_null,
+ column="missing_col",
+ message_expr="concat('Custom error for value: ', another_missing_col)",
+ ),
+ ]
+
+ checked_df = dq_engine.apply_checks(test_df, rules)
+ expected_errors = [
+ build_skipped_violation(
+ name="missing_not_null",
+ message="Check evaluation skipped due to invalid check columns: ['missing_col']; "
+ "Check evaluation skipped due to invalid custom message expression: "
+ "'concat('Custom error for value: ', another_missing_col)'",
+ columns=["missing_col"],
+ )
+ ]
+ expected_df = spark.createDataFrame(
+ [[1, 3, 5, expected_errors, None]],
+ EXPECTED_SCHEMA,
+ )
+ assert_df_equality(checked_df, expected_df)
+
+
+def test_apply_checks_valid_custom_message_not_skipped(ws, spark):
+ """A check without invalid fields and with a valid custom message must not be marked skipped."""
+ dq_engine = DQEngine(workspace_client=ws, extra_params=EXTRA_PARAMS)
+ test_df = spark.createDataFrame([[1, 3, None]], SCHEMA)
+
+ rules = [
+ DQRowRule(
+ name="c_not_null",
+ criticality="error",
+ check_func=check_funcs.is_not_null,
+ column="c",
+ message_expr="concat('Custom error for c: ', coalesce(cast(c as string), 'null'))",
+ ),
+ ]
+
+ checked_df = dq_engine.apply_checks(test_df, rules)
+ expected_errors = [
+ build_quality_violation(
+ name="c_not_null",
+ message="Custom error for c: null",
+ columns=["c"],
+ function="is_not_null",
+ )
+ ]
+ expected_df = spark.createDataFrame(
+ [[1, 3, None, expected_errors, None]],
+ EXPECTED_SCHEMA,
+ )
+ assert_df_equality(checked_df, expected_df)
+
+
+def test_apply_checks_skip_message_overrides_custom_message_for_invalid_sql_expression(ws, spark):
+ """test that message_expr is overridden when a sql_expression check is skipped due to an invalid expression"""
+ dq_engine = DQEngine(workspace_client=ws, extra_params=EXTRA_PARAMS)
+ test_df = spark.createDataFrame([[1, 3, 5]], SCHEMA)
+
+ rules = [
+ DQRowRule(
+ name="invalid_sql_expression",
+ criticality="error",
+ check_func=check_funcs.sql_expression,
+ check_func_kwargs={"expression": "missing_col > 0"},
+ message_expr="'Custom error: this message must be overridden by the skip message'",
+ user_metadata={"should_be": "IGNORED"},
+ ),
+ ]
+
+ checked_df = dq_engine.apply_checks(test_df, rules)
+ expected_errors = [
+ build_skipped_violation(
+ name="invalid_sql_expression",
+ message="Check evaluation skipped due to invalid sql expression: 'missing_col > 0'",
+ columns=None,
+ function="sql_expression",
+ user_metadata={"should_be": "IGNORED"},
+ )
+ ]
+ expected_df = spark.createDataFrame(
+ [[1, 3, 5, expected_errors, None]],
+ EXPECTED_SCHEMA,
+ )
+ assert_df_equality(checked_df, expected_df)
diff --git a/tests/unit/test_custom_messages.py b/tests/unit/test_custom_messages.py
new file mode 100644
index 000000000..5e25d3e76
--- /dev/null
+++ b/tests/unit/test_custom_messages.py
@@ -0,0 +1,143 @@
+"""Tests for custom message expressions on DQRule."""
+
+import logging
+
+import pyspark.sql.functions as F
+
+from databricks.labs.dqx.check_funcs import is_not_null, is_unique
+from databricks.labs.dqx.rule import DQRowRule, DQForEachColRule, DQDatasetRule
+
+
+def test_dq_row_rule_accepts_message_expr_string():
+ """DQRowRule with message_expr should accept a SQL expression string."""
+ rule = DQRowRule(
+ check_func=is_not_null,
+ column="id",
+ message_expr="concat('Failed: ', 'id_not_null')",
+ )
+ assert rule.message_expr == "concat('Failed: ', 'id_not_null')"
+
+
+def test_dq_row_rule_accepts_message_expr_column():
+ """DQRowRule with message_expr should accept a Spark Column object."""
+ column_expr = F.concat(F.lit("Failed: "), F.lit("id_not_null"))
+ rule = DQRowRule(
+ check_func=is_not_null,
+ column="id",
+ message_expr=column_expr,
+ )
+ assert rule.message_expr is not None
+
+
+def test_dq_row_rule_message_expr_defaults_to_none():
+ """DQRowRule without message_expr should default to None."""
+ rule = DQRowRule(check_func=is_not_null, column="id")
+ assert rule.message_expr is None
+
+
+def test_dq_dataset_rule_message_expr_defaults_to_none():
+ """DQDatasetRule without message_expr should default to None."""
+ rule = DQDatasetRule(check_func=is_unique, columns=["id"])
+ assert rule.message_expr is None
+
+
+def test_dq_for_each_col_rule_propagates_message_expr():
+ """DQForEachColRule should propagate message_expr to generated DQRowRules."""
+ msg = "concat('Check failed for ', 'rule_name')"
+ for_each_rule = DQForEachColRule(
+ columns=["a", "b"],
+ check_func=is_not_null,
+ message_expr=msg,
+ )
+ rules = for_each_rule.get_rules()
+ assert len(rules) == 2
+ for rule in rules:
+ assert rule.message_expr == msg
+
+
+def test_dq_for_each_col_rule_message_expr_defaults_to_none():
+ """DQForEachColRule without message_expr should produce rules with message_expr=None."""
+ for_each_rule = DQForEachColRule(
+ columns=["a", "b"],
+ check_func=is_not_null,
+ )
+ rules = for_each_rule.get_rules()
+ for rule in rules:
+ assert rule.message_expr is None
+
+
+def test_dq_rule_to_dict_includes_message_expr_when_string():
+ """to_dict() should include the message_expr key when set to a SQL string."""
+ msg = "'Custom error for id_not_null'"
+ rule = DQRowRule(
+ check_func=is_not_null,
+ column="id",
+ name="id_not_null",
+ message_expr=msg,
+ )
+ rule_dict = rule.to_dict()
+ assert rule_dict["message_expr"] == msg
+
+
+def test_dq_rule_to_dict_omits_message_expr_when_none():
+ """to_dict() should not include message_expr key when it is None."""
+ rule = DQRowRule(
+ check_func=is_not_null,
+ column="id",
+ name="id_not_null",
+ )
+ rule_dict = rule.to_dict()
+ assert "message_expr" not in rule_dict
+
+
+def test_dq_rule_to_dict_omits_message_expr_when_column():
+ """Column message_expr is in-process only and is excluded from serialised metadata."""
+ rule = DQRowRule(
+ check_func=is_not_null,
+ column="id",
+ name="id_not_null",
+ message_expr=F.lit("Custom error"),
+ )
+ rule_dict = rule.to_dict()
+ assert "message_expr" not in rule_dict
+
+
+def test_dq_rule_to_dict_warns_when_message_expr_is_column(caplog):
+ """Serialising a Column message_expr should warn that it cannot be round-tripped."""
+ rule = DQRowRule(
+ check_func=is_not_null,
+ column="id",
+ name="id_not_null",
+ message_expr=F.lit("Custom error"),
+ )
+ with caplog.at_level(logging.WARNING, logger="databricks.labs.dqx.rule"):
+ rule.to_dict()
+ assert "cannot be serialized" in caplog.text
+
+
+def test_dq_rule_to_dict_does_not_warn_when_message_expr_is_string(caplog):
+ """A string message_expr is serializable, so no serialization warning should be emitted."""
+ rule = DQRowRule(
+ check_func=is_not_null,
+ column="id",
+ name="id_not_null",
+ message_expr="'Custom error for id_not_null'",
+ )
+ with caplog.at_level(logging.WARNING, logger="databricks.labs.dqx.rule"):
+ rule.to_dict()
+ assert "cannot be serialized" not in caplog.text
+
+
+def test_dq_for_each_col_rule_to_dict_warns_for_each_column_when_message_expr_is_column(caplog):
+ """Column message_expr propagated by DQForEachColRule should warn once per generated rule."""
+ for_each_rule = DQForEachColRule(
+ columns=["a", "b"],
+ check_func=is_not_null,
+ message_expr=F.lit("Custom error"),
+ )
+ rules = for_each_rule.get_rules()
+ with caplog.at_level(logging.WARNING, logger="databricks.labs.dqx.rule"):
+ for rule in rules:
+ rule_dict = rule.to_dict()
+ assert "message_expr" not in rule_dict
+ assert caplog.text.count("cannot be serialized") == len(rules)