Add valid string case check (#1335)

docs/dqx/docs/reference/quality_checks.mdx

@@ -60,6 +60,7 @@ You can also define your own custom checks in Python (see [Creating custom check
 | `is_older_than_col2_for_n_days`    | Checks whether the values in one input column are at least N days older than the values in another column.                                                                                                                                                                                                                                                                                                                        | `column1`: first column to check (can be a string column name or a column expression); `column2`: second column to check (can be a string column name or a column expression); `days`: number of days; `negate`: if the condition should be negated                                                                                                                                                                                                                                                                                                  |
 | `regex_match`                      | Checks whether the values in the input column match a given regex.                                                                                                                                                                                                                                                                                                                                                                | `column`: column to check (can be a string column name or a column expression); regex: regex to check; `negate`: if the condition should be negated (true) or not                                                                                                                                                                                                                                                                                                                                                                                    |
 | `is_valid_email`                   | Checks whether the values in the input column have valid email address format.                                                                                                                                                                                                                                                                                                                                                    | `column`: column to check (can be a string column name or a column expression)                                                                                                                                                                                                                                                                                                                                                                                                                                                                       |
+| `has_valid_string_case`            | Checks whether the values match the requested string case. `upper` and `lower` use the corresponding Spark case functions; `title` uses Spark `initcap`; `sentence` splits on literal `.` characters, cases each segment independently, and preserves delimiters and surrounding whitespace. | `column`: column to check (can be a string column name or a column expression); `case`: one of `upper`, `lower`, `title`, or `sentence` |
 | `is_valid_ipv4_address`            | Checks whether the values in the input column have valid IPv4 address format.                                                                                                                                                                                                                                                                                                                                                     | `column` to check (can be a string column name or a column expression)                                                                                                                                                                                                                                                                                                                                                                                                                                                                               |
 | `is_ipv4_address_in_cidr`          | Checks whether the values in the input column have valid IPv4 address format and fall within the given CIDR block.                                                                                                                                                                                                                                                                                                                | `column`: column to check (can be a string column name or a column expression); `cidr_block`: CIDR block string                                                                                                                                                                                                                                                                                                                                                                                                                                      |
 | `is_valid_ipv6_address`            | Checks whether the values in the input column have valid IPv6 address format.                                                                                                                                                                                                                                                                                                                                                     | `column` to check (can be a string column name or a column expression)                                                                                                                                                                                                                                                                                                                                                                                                                                                                               |
@@ -532,6 +533,14 @@ For brevity, the `name` field in the examples is omitted and it will be auto-gen
     arguments:
       column: col1
 
+# has_valid_string_case check
+- criticality: error
+  check:
+    function: has_valid_string_case
+    arguments:
+      column: col1
+      case: lower
+
 # is_valid_ipv4_address check
 - criticality: error
   check:
@@ -1279,6 +1288,14 @@ checks = [
       column="col1"
     ),
 
+    # has_valid_string_case check
+    DQRowRule(
+      criticality="error",
+      check_func=check_funcs.has_valid_string_case,
+      column="col1",
+      check_func_kwargs={"case": "lower"}
+    ),
+
     # is_valid_ipv4_address check
     DQRowRule(
       criticality="error",

skills/dqx-define-checks/SKILL.md

@@ -86,7 +86,7 @@ Full reference: <https://databrickslabs.github.io/dqx/docs/reference/quality_che
 
 ## Fallback: custom SQL
 
-**Search `check_funcs` first** — the built-ins cover null/empty, range, set membership, regex, referential, aggregate, uniqueness, schema, freshness, comparison, and outlier cases with typed error messages and tested edge handling. Drop down to SQL only when no built-in fits.
+**Search `check_funcs` first** - the built-ins cover null/empty, string case, range, set membership, regex, referential, aggregate, uniqueness, schema, freshness, comparison, and outlier cases with typed error messages and tested edge handling. Use `has_valid_string_case` for `upper`, `lower`, `title`, or `sentence` string-case validation. Drop down to SQL only when no built-in fits.
 
 - `sql_expression` — **row-level** SQL boolean expression. Use when one row's validity depends on its own columns.
 - `sql_query` — **dataset-level** SQL query against `{{ input_view }}`. Use for cross-row aggregates, joins to reference DataFrames, or anything needing `GROUP BY`. Queries are validated by `is_sql_query_safe()` — read-only `SELECT`, no DDL/DML.

src/databricks/labs/dqx/check_funcs.py

@@ -28,6 +28,7 @@
 _IPV4_CIDR_SUFFIX = r"(3[0-2]|[12]?\d)"
 IPV4_MAX_OCTET_COUNT = 4
 IPV4_BIT_LENGTH = 32
+_VALID_STRING_CASES = {"upper", "lower", "title", "sentence"}
 
 # Email helpers (RFC 5322 §3.2.3, §3.2.4 + RFC 5321 §4.1.3, §4.5.3.1).
 _EMAIL_ATEXT = r"[A-Za-z0-9!#$%&'*+/=?^_`{|}~-]"
@@ -249,6 +250,61 @@ def is_null_or_empty(column: str | Column, trim_strings: bool | None = False) ->
     )
 
 
+@register_rule("row")
+def has_valid_string_case(column: str | Column, case: str) -> Column:
+    """Checks whether string values use the requested letter case.
+
+    Sentence case capitalizes the first non-whitespace character of every
+    period-delimited segment and lowercases the remainder of that segment.
+
+    Args:
+        column: column to check; can be a string column name or a column expression
+        case: expected case; one of *upper*, *lower*, *title*, or *sentence*
+
+    Returns:
+        Column object for condition
+
+    Raises:
+        InvalidParameterError: If *case* is not a supported string case.
+    """
+    if not isinstance(case, str):
+        raise InvalidParameterError(f"'case' must be a string, got {type(case)} instead.")
+    if case not in _VALID_STRING_CASES:
+        raise InvalidParameterError(f"'case' must be one of {sorted(_VALID_STRING_CASES)}, got '{case}'")
+
+    col_str_norm, col_expr_str, col_expr = get_normalized_column_and_expr(column)
+    col_expr_cast = col_expr.cast("string")
+
+    if case == "upper":
+        normalized_case = F.upper(col_expr_cast)
+    elif case == "lower":
+        normalized_case = F.lower(col_expr_cast)
+    elif case == "title":
+        normalized_case = F.initcap(col_expr_cast)
+    else:
+        segments = F.split(col_expr_cast, r"\.")
+        normalized_case = F.array_join(
+            F.transform(
+                segments,
+                lambda segment: F.concat(
+                    F.regexp_extract(segment, r"^(\s*)", 1),
+                    F.upper(F.substring(F.regexp_replace(segment, r"^\s*", ""), 1, 1)),
+                    F.lower(F.substring(F.regexp_replace(segment, r"^\s*", ""), 2, 2147483647)),
+                ),
+            ),
+            ".",
+        )
+
+    condition = F.when(col_expr.isNotNull(), col_expr_cast != normalized_case).otherwise(F.lit(None))
+    message = F.concat_ws(
+        "",
+        F.lit("Value '"),
+        col_expr.cast("string"),
+        F.lit(f"' in Column '{col_expr_str}' does not have valid '{case}' string case"),
+    )
+    return make_condition(condition, message, f"{col_str_norm}_has_invalid_{case}_string_case")
+
+
 @register_rule("row")
 def is_not_null_and_is_in_list(column: str | Column, allowed: list, case_sensitive: bool = True) -> Column:
     """Checks whether the values in the input column are not null and present in the list of allowed values.

tests/integration/test_row_checks.py

@@ -39,13 +39,71 @@
     is_null,
     is_empty,
     is_null_or_empty,
+    has_valid_string_case,
 )
 from databricks.labs.dqx.pii import pii_detection_funcs
 from databricks.labs.dqx.errors import InvalidParameterError
 
 SCHEMA = "a: string, b: int"
 
 
+def test_has_valid_string_case(spark):
+    test_df = spark.createDataFrame(
+        [
+            ["UPPER", "lower", "Spark Sql", "First segment. Second segment.", {"value": "nested"}, 123],
+            ["Upper", "Lower", "Spark SQL", "first SEGMENT. second SEGMENT", {"value": "Nested"}, None],
+            [None, "", "   ", "123!?. #$", {"value": "nested"}, 123],
+            [" UPPER ", " lower ", " spark sql ", " first SEGMENT. second SEGMENT ", {"value": "nested"}, 123],
+            ["123!?", "123!?", "hello-world", "Hello 123!. Second segment", {"value": "NESTED"}, 123],
+        ],
+        "upper: string, lower: string, title: string, sentence: string, nested: struct<value:string>, numeric: int",
+    )
+
+    actual = test_df.select(
+        has_valid_string_case("upper", "upper"),
+        has_valid_string_case("lower", "lower"),
+        has_valid_string_case("title", "title"),
+        has_valid_string_case("sentence", "sentence"),
+        has_valid_string_case(F.col("nested").getItem("value"), "lower"),
+        has_valid_string_case("numeric", "lower"),
+    )
+
+    def violation(value: str, column: str, case: str) -> str:
+        return f"Value '{value}' in Column '{column}' does not have valid '{case}' string case"
+
+    expected = spark.createDataFrame(
+        [
+            [None, None, None, None, None, None],
+            [
+                violation("Upper", "upper", "upper"),
+                violation("Lower", "lower", "lower"),
+                violation("Spark SQL", "title", "title"),
+                violation("first SEGMENT. second SEGMENT", "sentence", "sentence"),
+                violation("Nested", "nested['value']", "lower"),
+                None,
+            ],
+            [None, None, None, None, None, None],
+            [None, None, None, violation(" first SEGMENT. second SEGMENT ", "sentence", "sentence"), None, None],
+            [
+                None,
+                None,
+                violation("hello-world", "title", "title"),
+                None,
+                violation("NESTED", "nested['value']", "lower"),
+                None,
+            ],
+        ],
+        "upper_has_invalid_upper_string_case: string, "
+        "lower_has_invalid_lower_string_case: string, "
+        "title_has_invalid_title_string_case: string, "
+        "sentence_has_invalid_sentence_string_case: string, "
+        "nested_value_has_invalid_lower_string_case: string, "
+        "numeric_has_invalid_lower_string_case: string",
+    )
+
+    assertDataFrameEqual(actual, expected)
+
+
 def test_col_is_not_null_and_not_empty(spark):
     input_schema = "a: string, b: int, c: map<string, string>, d: array<string>, e: string"
     test_df = spark.createDataFrame(

tests/resources/all_row_checks.yaml

@@ -483,3 +483,11 @@
     function: is_valid_email
     arguments:
       column: col_email
+
+# has_valid_string_case check
+- criticality: error
+  check:
+    function: has_valid_string_case
+    arguments:
+      column: col1
+      case: lower

tests/unit/test_build_rules.py

@@ -24,6 +24,7 @@
     foreign_key,
     is_valid_ipv4_address,
     is_valid_email,
+    has_valid_string_case,
     is_ipv4_address_in_cidr,
     is_not_less_than,
     is_not_greater_than,
@@ -55,6 +56,35 @@
 SCHEMA = "a: int, b: int, c: int"
 
 
+def test_has_valid_string_case_rule_round_trip():
+    rule = DQRowRule(
+        criticality="warn",
+        check_func=has_valid_string_case,
+        column="a",
+        check_func_kwargs={"case": "lower"},
+    )
+    expected_metadata = [
+        {
+            "name": "a_has_invalid_lower_string_case",
+            "criticality": "warn",
+            "check": {
+                "function": "has_valid_string_case",
+                "arguments": {"column": "a", "case": "lower"},
+            },
+        }
+    ]
+
+    assert serialize_checks([rule]) == expected_metadata
+
+    deserialized_rules = deserialize_checks(expected_metadata)
+    assert len(deserialized_rules) == 1
+    deserialized_rule = deserialized_rules[0]
+    assert deserialized_rule.name == "a_has_invalid_lower_string_case"
+    assert deserialized_rule.check_func is has_valid_string_case
+    assert deserialized_rule.column == "a"
+    assert deserialized_rule.check_func_kwargs == {"case": "lower"}
+
+
 def test_get_for_each_rules():
     actual_rules = (
         # set of columns for the same check

tests/unit/test_row_checks.py

@@ -1,3 +1,4 @@
+from typing import cast
 import pytest
 from databricks.labs.dqx.utils import get_column_name_or_alias
 from databricks.labs.dqx.check_funcs import (
@@ -10,6 +11,7 @@
     is_in_list,
     is_not_null_and_is_in_list,
     is_aggr_not_greater_than,
+    has_valid_string_case,
     is_ipv4_address_in_cidr,
     is_ipv6_address_in_cidr,
     sql_expression,
@@ -52,6 +54,23 @@ def test_col_is_in_list_missing_allowed_list():
         is_in_list("a", allowed=[])
 
 
+@pytest.mark.parametrize(
+    "case, expected_message",
+    [
+        ("camel", "'case' must be one of ['lower', 'sentence', 'title', 'upper'], got 'camel'"),
+        ("", "'case' must be one of ['lower', 'sentence', 'title', 'upper'], got ''"),
+        ("Upper", "'case' must be one of ['lower', 'sentence', 'title', 'upper'], got 'Upper'"),
+        (None, "'case' must be a string, got <class 'NoneType'> instead."),
+        (1, "'case' must be a string, got <class 'int'> instead."),
+    ],
+)
+def test_has_valid_string_case_rejects_invalid_case(case: object, expected_message: str):
+    with pytest.raises(InvalidParameterError) as error:
+        has_valid_string_case("a", cast(str, case))
+
+    assert str(error.value) == expected_message
+
+
 def test_incorrect_aggr_type():
     # With new implementation, invalid aggr_type triggers a warning (not immediate error)
     # The error occurs at runtime when the apply function is called