diff --git a/docs/dqx/docs/reference/quality_checks.mdx b/docs/dqx/docs/reference/quality_checks.mdx index 41441037d..395b6c80b 100644 --- a/docs/dqx/docs/reference/quality_checks.mdx +++ b/docs/dqx/docs/reference/quality_checks.mdx @@ -30,8 +30,8 @@ You can also define your own custom checks in Python (see [Creating custom check | `is_not_null` | Checks whether the values in the input column are not null. | `column`: column to check (can be a string column name or a column expression) | | `is_not_empty` | Checks whether the values in the input column are not empty (but may be null). | `column`: column to check (can be a string column name or a column expression) | | `is_not_null_and_not_empty` | Checks whether the values in the input column are not null and not empty. | `column`: column to check (can be a string column name or a column expression); `trim_strings`: optional boolean flag to trim spaces from strings | -| `is_in_list` | Checks whether the values in the input column are present in the list of allowed values (null values are allowed). This check is not suited for large lists of allowed values. In such cases, it’s recommended to use the `foreign_key` dataset-level check instead. | `column`: column to check (can be a string column name or a column expression); `allowed`: list of allowed values | -| `is_not_null_and_is_in_list` | Checks whether the values in the input column are not null and present in the list of allowed values. This check is not suited for large lists of allowed values. In such cases, it’s recommended to use the `foreign_key` dataset-level check instead. | `column`: column to check (can be a string column name or a column expression); `allowed`: list of allowed values | +| `is_in_list` | Checks whether the values in the input column are present in the list of allowed values (null values are allowed). This check is not suited for large lists of allowed values. In such cases, it’s recommended to use the `foreign_key` dataset-level check instead. This check is not suited for `MapType` or `StructType` columns. | `column`: column to check (can be a string column name or a column expression); `allowed`: list of allowed values; `case_sensitive`: optional boolean flag for case-sensitive comparison (default: True) | +| `is_not_null_and_is_in_list` | Checks whether the values in the input column are not null and present in the list of allowed values. This check is not suited for large lists of allowed values. In such cases, it’s recommended to use the `foreign_key` dataset-level check instead. This check is not suited for `MapType` or `StructType` columns. | `column`: column to check (can be a string column name or a column expression); `allowed`: list of allowed values; `case_sensitive`: optional boolean flag for case-sensitive comparison (default: True) | | `is_not_null_and_not_empty_array` | Checks whether the values in the array input column are not null and not empty. | `column`: column to check (can be a string column name or a column expression) | | `is_in_range` | Checks whether the values in the input column are in the provided range (inclusive of both boundaries). | `column`: column to check (can be a string column name or a column expression); `min_limit`: min limit as number, date, timestamp, column name or sql expression; `max_limit`: max limit as number, date, timestamp, column name or sql expression | | `is_not_in_range` | Checks whether the values in the input column are outside the provided range (inclusive of both boundaries). | `column`: column to check (can be a string column name or a column expression); `min_limit`: min limit as number, date, timestamp, column name or sql expression; `max_limit`: max limit as number, date, timestamp, column name or sql expression | diff --git a/src/databricks/labs/dqx/check_funcs.py b/src/databricks/labs/dqx/check_funcs.py index 96bd960c6..5ab24df1f 100644 --- a/src/databricks/labs/dqx/check_funcs.py +++ b/src/databricks/labs/dqx/check_funcs.py @@ -18,6 +18,7 @@ is_sql_query_safe, normalize_col_str, get_columns_as_strings, + to_lowercase, ) from databricks.labs.dqx.errors import MissingParameterError, InvalidParameterError, UnsafeSqlQueryError @@ -129,12 +130,15 @@ def is_not_null(column: str | Column) -> Column: @register_rule("row") -def is_not_null_and_is_in_list(column: str | Column, allowed: list) -> Column: +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. + Can optionally perform a case-insensitive comparison. + This check is not suited for `MapType` or `StructType` columns. Args: column: column to check; can be a string column name or a column expression allowed: list of allowed values (actual values or Column objects) + case_sensitive: whether to perform a case-sensitive comparison (default: True) Returns: Column object for condition @@ -145,16 +149,26 @@ def is_not_null_and_is_in_list(column: str | Column, allowed: list) -> Column: """ if allowed is None: raise MissingParameterError("allowed list is not provided.") - if not isinstance(allowed, list): raise InvalidParameterError(f"allowed parameter must be a list, got {str(type(allowed))} instead.") - if not allowed: raise InvalidParameterError("allowed list must not be empty.") allowed_cols = [item if isinstance(item, Column) else F.lit(item) for item in allowed] col_str_norm, col_expr_str, col_expr = _get_normalized_column_and_expr(column) - condition = col_expr.isNull() | ~col_expr.isin(*allowed_cols) + + # Apply case-insensitive transformation if needed + if not case_sensitive: + has_arrays = any(isinstance(item, (list, tuple)) for item in allowed if not isinstance(item, Column)) + col_expr_compare = to_lowercase(col_expr, is_array=has_arrays) + allowed_cols_compare = [ + to_lowercase(c, is_array=isinstance(allowed[i], (list, tuple))) for i, c in enumerate(allowed_cols) + ] + else: + col_expr_compare, allowed_cols_compare = col_expr, allowed_cols + + condition = col_expr.isNull() | ~col_expr_compare.isin(*allowed_cols_compare) + return make_condition( condition, F.concat_ws( @@ -162,7 +176,7 @@ def is_not_null_and_is_in_list(column: str | Column, allowed: list) -> Column: F.lit("Value '"), F.when(col_expr.isNull(), F.lit("null")).otherwise(col_expr.cast("string")), F.lit(f"' in Column '{col_expr_str}' is null or not in the allowed list: ["), - F.concat_ws(", ", *allowed_cols), + F.concat_ws(", ", *[c.cast("string") for c in allowed_cols]), F.lit("]"), ), f"{col_str_norm}_is_null_or_is_not_in_the_list", @@ -170,13 +184,15 @@ def is_not_null_and_is_in_list(column: str | Column, allowed: list) -> Column: @register_rule("row") -def is_in_list(column: str | Column, allowed: list) -> Column: +def is_in_list(column: str | Column, allowed: list, case_sensitive: bool = True) -> Column: """Checks whether the values in the input column are present in the list of allowed values - (null values are allowed). + (null values are allowed). Can optionally perform a case-insensitive comparison. + This check is not suited for `MapType` or `StructType` columns. Args: column: column to check; can be a string column name or a column expression allowed: list of allowed values (actual values or Column objects) + case_sensitive: whether to perform a case-sensitive comparison (default: True) Returns: Column object for condition @@ -187,16 +203,26 @@ def is_in_list(column: str | Column, allowed: list) -> Column: """ if allowed is None: raise MissingParameterError("allowed list is not provided.") - if not isinstance(allowed, list): raise InvalidParameterError(f"allowed parameter must be a list, got {str(type(allowed))} instead.") - if not allowed: raise InvalidParameterError("allowed list must not be empty.") allowed_cols = [item if isinstance(item, Column) else F.lit(item) for item in allowed] col_str_norm, col_expr_str, col_expr = _get_normalized_column_and_expr(column) - condition = ~col_expr.isin(*allowed_cols) + + # Apply case-insensitive transformation if needed + if not case_sensitive: + has_arrays = any(isinstance(item, (list, tuple)) for item in allowed if not isinstance(item, Column)) + col_expr_compare = to_lowercase(col_expr, is_array=has_arrays) + allowed_cols_compare = [ + to_lowercase(c, is_array=isinstance(allowed[i], (list, tuple))) for i, c in enumerate(allowed_cols) + ] + else: + col_expr_compare, allowed_cols_compare = col_expr, allowed_cols + + condition = ~col_expr_compare.isin(*allowed_cols_compare) + return make_condition( condition, F.concat_ws( @@ -204,7 +230,7 @@ def is_in_list(column: str | Column, allowed: list) -> Column: F.lit("Value '"), F.when(col_expr.isNull(), F.lit("null")).otherwise(col_expr.cast("string")), F.lit(f"' in Column '{col_expr_str}' is not in the allowed list: ["), - F.concat_ws(", ", *allowed_cols), + F.concat_ws(", ", *[c.cast("string") for c in allowed_cols]), F.lit("]"), ), f"{col_str_norm}_is_not_in_the_list", diff --git a/src/databricks/labs/dqx/utils.py b/src/databricks/labs/dqx/utils.py index 48fd9de20..ce6df48a7 100644 --- a/src/databricks/labs/dqx/utils.py +++ b/src/databricks/labs/dqx/utils.py @@ -14,12 +14,12 @@ except ImportError: ConnectColumn = None # type: ignore +import pyspark.sql.functions as F from databricks.sdk import WorkspaceClient from databricks.labs.blueprint.limiter import rate_limited from databricks.labs.dqx.errors import InvalidParameterError from databricks.sdk.errors import NotFound - logger = logging.getLogger(__name__) @@ -450,3 +450,18 @@ def _match_table_patterns(table: str, patterns: list[str]) -> bool: bool: True if the table name matches any of the patterns, False otherwise. """ return any(fnmatch(table, pattern) for pattern in patterns) + + +def to_lowercase(col_expr: Column, is_array: bool = False) -> Column: + """Converts a column expression to lowercase, handling both scalar and array types. + + Args: + col_expr: Column expression to convert + is_array: Whether the column contains array values + + Returns: + Column expression with lowercase transformation applied + """ + if is_array: + return F.transform(col_expr, F.lower) + return F.lower(col_expr) diff --git a/tests/integration/test_row_checks.py b/tests/integration/test_row_checks.py index 2fc713a03..da5f0bd81 100644 --- a/tests/integration/test_row_checks.py +++ b/tests/integration/test_row_checks.py @@ -3,6 +3,8 @@ import pytest import pyspark.sql.functions as F from chispa.dataframe_comparer import assert_df_equality # type: ignore +from pyspark.errors import AnalysisException + from databricks.labs.dqx.check_funcs import ( is_equal_to, is_not_equal_to, @@ -161,6 +163,8 @@ def test_col_is_not_null_and_is_in_list(spark): ["str1", 1, {"val": "a"}, ["a", "b"]], ["str2", None, {"val": "str2"}, [None, "a"]], [" ", 3, {"val": " "}, [None, " "]], + ["STR1", 4, {"val": "A"}, ["B", "c"]], + ["str5", 5, {"val": "e"}, ["b", "C"]], ], input_schema, ) @@ -170,28 +174,81 @@ def test_col_is_not_null_and_is_in_list(spark): is_not_null_and_is_in_list("b", [F.lit(3)]), is_not_null_and_is_in_list(F.col("c").getItem("val"), [F.lit("a")]), is_not_null_and_is_in_list(F.try_element_at("d", F.lit(2)), ["b"]), + is_not_null_and_is_in_list("a", ["str1"], case_sensitive=False), + is_not_null_and_is_in_list(F.col("c").getItem("val"), [F.lit("a")], case_sensitive=False), + is_not_null_and_is_in_list(F.try_element_at("d", F.lit(2)), ["b"], case_sensitive=False), + is_not_null_and_is_in_list("d", [["a", "b"], ["B", "c"]]), + is_not_null_and_is_in_list("d", [["a", "b"], ["B", "c"]], case_sensitive=False), ) checked_schema = ( "a_is_null_or_is_not_in_the_list: string, " + "b_is_null_or_is_not_in_the_list: string, " + "unresolvedextractvalue_c_val_is_null_or_is_not_in_the_list: string, " - + "try_element_at_d_2_is_null_or_is_not_in_the_list: string" + + "try_element_at_d_2_is_null_or_is_not_in_the_list: string, " + + "a_is_null_or_is_not_in_the_list: string, " + + "unresolvedextractvalue_c_val_is_null_or_is_not_in_the_list: string, " + + "try_element_at_d_2_is_null_or_is_not_in_the_list: string, " + + "d_is_null_or_is_not_in_the_list: string, " + + "d_is_null_or_is_not_in_the_list: string" ) + col_d_null = None expected = spark.createDataFrame( [ - [None, "Value '1' in Column 'b' is null or not in the allowed list: [3]", None, None], + [ + None, + "Value '1' in Column 'b' is null or not in the allowed list: [3]", + None, + None, + None, + None, + None, + None, + col_d_null, + ], [ "Value 'str2' in Column 'a' is null or not in the allowed list: [str1]", "Value 'null' in Column 'b' is null or not in the allowed list: [3]", "Value 'str2' in Column 'UnresolvedExtractValue(c, val)' is null or not in the allowed list: [a]", "Value 'a' in Column 'try_element_at(d, 2)' is null or not in the allowed list: [b]", + "Value 'str2' in Column 'a' is null or not in the allowed list: [str1]", + "Value 'str2' in Column 'UnresolvedExtractValue(c, val)' is null or not in the allowed list: [a]", + "Value 'a' in Column 'try_element_at(d, 2)' is null or not in the allowed list: [b]", + "Value '[null, a]' in Column 'd' is null or not in the allowed list: [[a, b], [B, c]]", + "Value '[null, a]' in Column 'd' is null or not in the allowed list: [[a, b], [B, c]]", ], [ "Value ' ' in Column 'a' is null or not in the allowed list: [str1]", None, "Value ' ' in Column 'UnresolvedExtractValue(c, val)' is null or not in the allowed list: [a]", "Value ' ' in Column 'try_element_at(d, 2)' is null or not in the allowed list: [b]", + "Value ' ' in Column 'a' is null or not in the allowed list: [str1]", + "Value ' ' in Column 'UnresolvedExtractValue(c, val)' is null or not in the allowed list: [a]", + "Value ' ' in Column 'try_element_at(d, 2)' is null or not in the allowed list: [b]", + "Value '[null, ]' in Column 'd' is null or not in the allowed list: [[a, b], [B, c]]", + "Value '[null, ]' in Column 'd' is null or not in the allowed list: [[a, b], [B, c]]", + ], + [ + "Value 'STR1' in Column 'a' is null or not in the allowed list: [str1]", + "Value '4' in Column 'b' is null or not in the allowed list: [3]", + "Value 'A' in Column 'UnresolvedExtractValue(c, val)' is null or not in the allowed list: [a]", + "Value 'c' in Column 'try_element_at(d, 2)' is null or not in the allowed list: [b]", + None, + None, + "Value 'c' in Column 'try_element_at(d, 2)' is null or not in the allowed list: [b]", + None, + None, + ], + [ + "Value 'str5' in Column 'a' is null or not in the allowed list: [str1]", + "Value '5' in Column 'b' is null or not in the allowed list: [3]", + "Value 'e' in Column 'UnresolvedExtractValue(c, val)' is null or not in the allowed list: [a]", + "Value 'C' in Column 'try_element_at(d, 2)' is null or not in the allowed list: [b]", + "Value 'str5' in Column 'a' is null or not in the allowed list: [str1]", + "Value 'e' in Column 'UnresolvedExtractValue(c, val)' is null or not in the allowed list: [a]", + "Value 'C' in Column 'try_element_at(d, 2)' is null or not in the allowed list: [b]", + "Value '[b, C]' in Column 'd' is null or not in the allowed list: [[a, b], [B, c]]", + None, ], ], checked_schema, @@ -207,37 +264,79 @@ def test_col_is_not_in_list(spark): ["str1", 1, {"val": "a"}, ["a", "b"]], ["str2", None, {"val": "str2"}, [None, "a"]], [" ", 3, {"val": None}, [None, "a"]], + ["STR1", 4, {"val": "A"}, ["B", "c"]], + ["str5", 5, {"val": "e"}, ["b", "C"]], ], input_schema, ) - actual = test_df.select( is_in_list("a", ["str1"]), is_in_list("b", [F.lit(3)]), is_in_list(F.col("c").getItem("val"), [F.lit("a")]), is_in_list(F.try_element_at("d", F.lit(2)), ["b"]), + is_in_list("a", ["str1"], case_sensitive=False), + is_in_list(F.col("c").getItem("val"), [F.lit("a")], case_sensitive=False), + is_in_list(F.try_element_at("d", F.lit(2)), ["b"], case_sensitive=False), + is_in_list("d", [["a", "b"], ["B", "c"]]), + is_in_list("d", [["a", "b"], ["B", "c"]], case_sensitive=False), ) - checked_schema = ( "a_is_not_in_the_list: string, " + "b_is_not_in_the_list: string, " + "unresolvedextractvalue_c_val_is_not_in_the_list: string, " - + "try_element_at_d_2_is_not_in_the_list: string" + + "try_element_at_d_2_is_not_in_the_list: string, " + + "a_is_not_in_the_list: string, " + + "unresolvedextractvalue_c_val_is_not_in_the_list: string, " + + "try_element_at_d_2_is_not_in_the_list: string, " + + "d_is_not_in_the_list: string, " + + "d_is_not_in_the_list: string" ) expected = spark.createDataFrame( [ - [None, "Value '1' in Column 'b' is not in the allowed list: [3]", None, None], + [None, "Value '1' in Column 'b' is not in the allowed list: [3]", None, None, None, None, None, None, None], [ "Value 'str2' in Column 'a' is not in the allowed list: [str1]", None, "Value 'str2' in Column 'UnresolvedExtractValue(c, val)' is not in the allowed list: [a]", "Value 'a' in Column 'try_element_at(d, 2)' is not in the allowed list: [b]", + "Value 'str2' in Column 'a' is not in the allowed list: [str1]", + "Value 'str2' in Column 'UnresolvedExtractValue(c, val)' is not in the allowed list: [a]", + "Value 'a' in Column 'try_element_at(d, 2)' is not in the allowed list: [b]", + "Value '[null, a]' in Column 'd' is not in the allowed list: [[a, b], [B, c]]", + "Value '[null, a]' in Column 'd' is not in the allowed list: [[a, b], [B, c]]", ], [ "Value ' ' in Column 'a' is not in the allowed list: [str1]", None, None, "Value 'a' in Column 'try_element_at(d, 2)' is not in the allowed list: [b]", + "Value ' ' in Column 'a' is not in the allowed list: [str1]", + None, + "Value 'a' in Column 'try_element_at(d, 2)' is not in the allowed list: [b]", + "Value '[null, a]' in Column 'd' is not in the allowed list: [[a, b], [B, c]]", + "Value '[null, a]' in Column 'd' is not in the allowed list: [[a, b], [B, c]]", + ], + [ + "Value 'STR1' in Column 'a' is not in the allowed list: [str1]", + "Value '4' in Column 'b' is not in the allowed list: [3]", + "Value 'A' in Column 'UnresolvedExtractValue(c, val)' is not in the allowed list: [a]", + "Value 'c' in Column 'try_element_at(d, 2)' is not in the allowed list: [b]", + None, + None, + "Value 'c' in Column 'try_element_at(d, 2)' is not in the allowed list: [b]", + None, + None, + ], + [ + "Value 'str5' in Column 'a' is not in the allowed list: [str1]", + "Value '5' in Column 'b' is not in the allowed list: [3]", + "Value 'e' in Column 'UnresolvedExtractValue(c, val)' is not in the allowed list: [a]", + "Value 'C' in Column 'try_element_at(d, 2)' is not in the allowed list: [b]", + "Value 'str5' in Column 'a' is not in the allowed list: [str1]", + "Value 'e' in Column 'UnresolvedExtractValue(c, val)' is not in the allowed list: [a]", + "Value 'C' in Column 'try_element_at(d, 2)' is not in the allowed list: [b]", + "Value '[b, C]' in Column 'd' is not in the allowed list: [[a, b], [B, c]]", + None, ], ], checked_schema, @@ -246,6 +345,36 @@ def test_col_is_not_in_list(spark): assert_df_equality(actual, expected, ignore_nullable=True) +def test_col_is_not_in_list_mismatch_datatype(spark): + input_schema = "a: string, d: array" + test_df = spark.createDataFrame( + [ + ["str1", ["a", "b"]], + ], + input_schema, + ) + actual = test_df.select( + is_in_list("d", ["a", "b"]), # wrong data type + ) + with pytest.raises(AnalysisException, match="[DATATYPE_MISMATCH.DATA_DIFF_TYPES]"): + actual.count() + + +def test_col_is_not_null_and_is_in_list_mismatch_datatype(spark): + input_schema = "a: string, d: array" + test_df = spark.createDataFrame( + [ + ["str1", ["a", "b"]], + ], + input_schema, + ) + actual = test_df.select( + is_not_null_and_is_in_list("d", ["a", "b"]), # wrong data type + ) + with pytest.raises(AnalysisException, match="[DATATYPE_MISMATCH.DATA_DIFF_TYPES]"): + actual.count() + + def test_col_sql_expression(spark): test_df = spark.createDataFrame([["str1", 1, 1], ["str2", None, None], ["", 2, 3]], SCHEMA + ", c: string") diff --git a/tests/unit/test_build_rules.py b/tests/unit/test_build_rules.py index ca9607554..00c4f3fe0 100644 --- a/tests/unit/test_build_rules.py +++ b/tests/unit/test_build_rules.py @@ -890,13 +890,13 @@ def test_build_checks_by_metadata_logging_debug_calls(caplog): def test_validate_check_func_arguments_too_many_positional(): - with pytest.raises(TypeError, match="takes 2 positional arguments but 3 were given"): + with pytest.raises(TypeError, match="takes from 2 to 3 positional arguments but 4 were given"): DQRowRule( name="col1_is_not_in_the_list", criticality="error", check_func=is_in_list, column="col1", - check_func_args=[[1, 2], "extra_arg"], + check_func_args=[[1, 2], True, "extra_arg"], )