diff --git a/docs/dqx/docs/reference/profiler.mdx b/docs/dqx/docs/reference/profiler.mdx index fd94ce9d3..ddc30694c 100644 --- a/docs/dqx/docs/reference/profiler.mdx +++ b/docs/dqx/docs/reference/profiler.mdx @@ -16,19 +16,19 @@ When running the code on a Databricks workspace, the workspace client is automat You only need the following code to create the workspace client if you run DQX on Databricks workspace: - - ```python - from databricks.sdk import WorkspaceClient - from databricks.labs.dqx.profiler.profiler import DQProfiler - from databricks.labs.dqx.profiler.generator import DQGenerator - from databricks.labs.dqx.profiler.dlt_generator import DQDltGenerator - - ws = WorkspaceClient() - profiler = DQProfiler(ws) - generator = DQGenerator(ws) - dlt_generator = DQDltGenerator(ws) - ``` - + + ```python + from databricks.sdk import WorkspaceClient + from databricks.labs.dqx.profiler.profiler import DQProfiler + from databricks.labs.dqx.profiler.generator import DQGenerator + from databricks.labs.dqx.profiler.dlt_generator import DQDltGenerator + + ws = WorkspaceClient() + profiler = DQProfiler(ws) + generator = DQGenerator(ws) + dlt_generator = DQDltGenerator(ws) + ``` + For external environments, such as CI servers or local machines, you can authenticate to Databricks using any method supported by the Databricks SDK. For detailed instructions, refer to the [default authentication flow](https://databricks-sdk-py.readthedocs.io/en/latest/authentication.html#default-authentication-flow). @@ -68,6 +68,14 @@ The profiler supports extensive configuration options to customize behavior: | `limit` | `1000` | Maximum number of records to analyze | | `filter` | `None` | Filter for the input data as a string SQL expression | | `llm_primary_key_detection` | `True` | Detect primary keys for generation of `is_unique` rule | +| `outliers_ratio` | `0.01` | Generate `has_no_outliers` rule if the fraction of outliers (detected via MAD) is at or below this threshold | +| `has_no_outliers` | `False` | Enable or disable `has_no_outliers` rule generation globally for all columns. Disabled by default due to performance considerations. | +| `has_no_outliers_allow_columns` | `[]` | Generate `has_no_outliers` rule only for the listed columns; all other columns are skipped. Mutually exclusive with `has_no_outliers_deny_columns` | +| `has_no_outliers_deny_columns` | `[]` | Skip `has_no_outliers` rule generation for the listed columns; all other columns are processed. Mutually exclusive with `has_no_outliers_allow_columns` | + + +Ratio-based decisions such as `outliers_ratio` (and the other `*_ratio` options) are evaluated on the **sampled** data (see `sample_fraction` / `limit`), not the full dataset. A `has_no_outliers` rule is therefore emitted based on the outlier fraction observed in the sample; the generated check itself still runs against the full dataset at apply time. For more representative rule generation on skewed data, increase `sample_fraction` or `limit`. + ## DQProfile Structure @@ -88,14 +96,15 @@ class DQProfile: Profiling data generates several types of `DQProfile` objects. Each profile can be used to generate a corresponding `DQRule` consisting of an input column, built-in DQX check function, and arguments inferred during profiling. -| Profile Type | DQX Check Function | Supported Data Types | Profiled Values | Notes | -| ----------------------------- | ----------------------------- | --------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| is_not_null | `is_not_null` | All | | Generated when `null_ratio <= max_null_ratio` for non-text columns, or when `null_ratio <= max_null_ratio` but `empty_ratio > max_empty_ratio` for text columns | -| is_not_null_or_empty | `is_not_null_and_not_empty` | `StringType`, `CharType`, or `VarcharType` | `trim_strings`: Whether to trim whitespace before checking for empty strings | Generated for text columns when both `null_ratio <= max_null_ratio` and `empty_ratio <= max_empty_ratio` | -| is_not_empty | `is_not_empty` | `StringType`, `CharType`, or `VarcharType` | `trim_strings`: Whether to trim whitespace before checking for empty strings | Generated for text columns when `null_ratio > max_null_ratio` (nulls allowed) but `empty_ratio <= max_empty_ratio` | -| is_in | `is_in_list` | `StringType`, `CharType`, `VarcharType`, `IntegerType`, or `LongType` | `in`: List of distinct values from the profiled data | List of values checks are inferred based on the `max_in_count` and `distinct_ratio` in the profiler options | -| min_max | `is_in_range` | `NumericType`, `DateType`, `TimestampType`, or `TimestampNTZType` | `min`: Minimum value from the profiled data; `max`: Maximum value from the profiled data | Values can be rounded by setting `"round": True` in the profiler options; Outlier values may be ignored based on the profiler options (e.g. `remove_outliers`, `num_sigmas`, and `outlier_columns`) | -| is_unique | `is_unique` | All | `reasoning`: Reason for inferring that a column is a primary key; `confidence`: Confidence score of the LLM-generated result | Requires installation of [LLM Extensions](/docs/installation#optional-dependencies); Set `llm_primary_key_detection: True` in the profiler options to enable primary key detection | +| Profile Type | DQX Check Function | Supported Data Types | Profiled Values | Notes | +|----------------------|-----------------------------|-----------------------------------------------------------------------|------------------------------------------------------------------------------------------------------------------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| +| is_not_null | `is_not_null` | All | | Generated when `null_ratio <= max_null_ratio` for non-text columns, or when `null_ratio <= max_null_ratio` but `empty_ratio > max_empty_ratio` for text columns | +| is_not_null_or_empty | `is_not_null_and_not_empty` | `StringType`, `CharType`, or `VarcharType` | `trim_strings`: Whether to trim whitespace before checking for empty strings | Generated for text columns when both `null_ratio <= max_null_ratio` and `empty_ratio <= max_empty_ratio` | +| is_not_empty | `is_not_empty` | `StringType`, `CharType`, or `VarcharType` | `trim_strings`: Whether to trim whitespace before checking for empty strings | Generated for text columns when `null_ratio > max_null_ratio` (nulls allowed) but `empty_ratio <= max_empty_ratio` | +| is_in | `is_in_list` | `StringType`, `CharType`, `VarcharType`, `IntegerType`, or `LongType` | `in`: List of distinct values from the profiled data | List of values checks are inferred based on the `max_in_count` and `distinct_ratio` in the profiler options | +| min_max | `is_in_range` | `NumericType`, `DateType`, `TimestampType`, or `TimestampNTZType` | `min`: Minimum value from the profiled data; `max`: Maximum value from the profiled data | Values can be rounded by setting `"round": True` in the profiler options; Outlier values may be ignored based on the profiler options (e.g. `remove_outliers`, `num_sigmas`, and `outlier_columns`) | +| has_no_outliers | `has_no_outliers` | `NumericType` (all numeric types) | | Generated when the fraction of outliers detected via the Median Absolute Deviation (MAD) method is at or below `outliers_ratio`; outliers are values outside `median ± 3.5 × MAD` | +| is_unique | `is_unique` | All | `reasoning`: Reason for inferring that a column is a primary key; `confidence`: Confidence score of the LLM-generated result | Requires installation of [LLM Extensions](/docs/installation#optional-dependencies); Set `llm_primary_key_detection: True` in the profiler options to enable primary key detection | ## DQGenerator methods @@ -116,5 +125,6 @@ The `DQDltGenerator` class creates Delta Live Tables expectation statements from | `generate_dlt_rules` | Generates Delta Live Table rules in the specified language. | `rules`: List of DQProfile objects; `action`: Optional violation action ("drop", "fail", or None); `language`: Target language ("SQL", "Python", or "Python_Dict"). | Yes | -For comprehensive examples, advanced options, and best practices, see the [Data Profiling Guide](/docs/guide/data_profiling). + For comprehensive examples, advanced options, and best practices, see the [Data Profiling + Guide](/docs/guide/data_profiling). diff --git a/src/databricks/labs/dqx/check_funcs.py b/src/databricks/labs/dqx/check_funcs.py index 87a09e788..523d7d706 100644 --- a/src/databricks/labs/dqx/check_funcs.py +++ b/src/databricks/labs/dqx/check_funcs.py @@ -14,6 +14,8 @@ from pyspark.sql import types from pyspark.sql import Column, DataFrame, SparkSession from pyspark.sql.window import Window + +from databricks.labs.dqx.profiling_utils import calculate_median_absolute_deviation_bounds from databricks.labs.dqx.rule import register_rule, register_for_original_columns_preselection from databricks.labs.dqx.utils import ( get_column_name_or_alias, @@ -1311,13 +1313,9 @@ def apply(df: DataFrame) -> DataFrame: # Validate and compile the filter (rejecting unsafe SQL); keep None when no filter is given so # the MAD calculation stays a true no-op instead of filtering on F.lit(True). filter_condition = safe_filter_expr(row_filter) if row_filter else None - median, mad = _calculate_median_absolute_deviation(df, col_expr_str, filter_condition) - if median is not None and mad is not None: - median = float(median) - mad = float(mad) - # Create outlier condition - lower_bound = median - (3.5 * mad) - upper_bound = median + (3.5 * mad) + bounds = calculate_median_absolute_deviation_bounds(df, col_expr_str, filter_condition) + if bounds is not None: + lower_bound, upper_bound = bounds lower_bound_expr = get_limit_expr(lower_bound) upper_bound_expr = get_limit_expr(upper_bound) @@ -4206,38 +4204,3 @@ def _validate_sql_query_params(query: str, merge_columns: list[str] | None) -> N invalid_columns = [col for col in merge_columns if not isinstance(col, str) or not col] if invalid_columns: raise InvalidParameterError("'merge_columns' entries must be non-empty strings.") - - -def _calculate_median_absolute_deviation( - df: DataFrame, column: str, filter_condition: Column | None -) -> tuple[Any, Any]: - """ - Calculate the Median Absolute Deviation (MAD) for a numeric column. - - The MAD is a robust measure of variability based on the median, calculated as: - MAD = median(|X_i - median(X)|) - - This is useful for outlier detection as it is more robust to outliers than - standard deviation. - - Args: - df: PySpark DataFrame - column: Name of the numeric column to calculate MAD for - filter_condition: Filter to apply before calculation (optional) - - Returns: - The Median and Absolute Deviation values - """ - if filter_condition is not None: - df = df.filter(filter_condition) - - # Step 1: Calculate the median of the column - median_value = df.agg(F.percentile_approx(column, 0.5)).collect()[0][0] - - # Step 2: Calculate absolute deviations from the median - df_with_deviations = df.select(F.abs(F.col(column) - F.lit(median_value)).alias("absolute_deviation")) - - # Step 3: Calculate the median of absolute deviations - mad = df_with_deviations.agg(F.percentile_approx("absolute_deviation", 0.5)).collect()[0][0] - - return median_value, mad diff --git a/src/databricks/labs/dqx/config.py b/src/databricks/labs/dqx/config.py index 82b6ba619..335be4f89 100644 --- a/src/databricks/labs/dqx/config.py +++ b/src/databricks/labs/dqx/config.py @@ -92,6 +92,7 @@ class ProfilerConfig: # Override profiler default thresholds max_null_ratio: float | None = None max_empty_ratio: float | None = None + outliers_ratio: float | None = None @dataclass diff --git a/src/databricks/labs/dqx/profiler/generator.py b/src/databricks/labs/dqx/profiler/generator.py index 5b08f3b45..4b9e1cf28 100644 --- a/src/databricks/labs/dqx/profiler/generator.py +++ b/src/databricks/labs/dqx/profiler/generator.py @@ -443,6 +443,31 @@ def dq_generate_is_unique(column: str, criticality: str = "error", **params: dic "user_metadata": user_metadata, } + @staticmethod + def dq_generate_has_no_outliers(column: str, criticality: str = "error", **params: dict): + """Generates a data quality rule to check if a column's values contain no outliers. + + Uses the MAD (Median Absolute Deviation) method via the *has_no_outliers* check function. + Values outside median ± 3.5 * MAD are flagged as outliers. + + Args: + column: The name of the column to check. + criticality: The criticality of the rule as "warn" or "error" (default is "error"). + params: Additional parameters (unused; left for compatibility with other functions). + + Returns: + A dictionary representing the data quality rule. + """ + params = params or {} # consume the shared **params dispatch arg (see _checks_mapping call site) + return { + "check": { + "function": "has_no_outliers", + "arguments": {"column": column}, + }, + "name": f"{column}_has_no_outliers", + "criticality": criticality, + } + _checks_mapping = { "is_not_null": dq_generate_is_not_null, "is_in": dq_generate_is_in, @@ -450,4 +475,5 @@ def dq_generate_is_unique(column: str, criticality: str = "error", **params: dic "is_not_null_or_empty": dq_generate_is_not_null_or_empty, "is_not_empty": dq_generate_is_not_empty, "is_unique": dq_generate_is_unique, + "has_no_outliers": dq_generate_has_no_outliers, } diff --git a/src/databricks/labs/dqx/profiler/profile_builder.py b/src/databricks/labs/dqx/profiler/profile_builder.py index 8ddd8d2fd..b75fbf503 100644 --- a/src/databricks/labs/dqx/profiler/profile_builder.py +++ b/src/databricks/labs/dqx/profiler/profile_builder.py @@ -7,13 +7,34 @@ from pyspark.sql import DataFrame from pyspark.sql import types as T, functions as F + +from databricks.labs.dqx.check_funcs import get_limit_expr from databricks.labs.dqx.errors import InvalidParameterError from databricks.labs.dqx.profiler.profile import DQProfile, DQProfileBuilder +from databricks.labs.dqx.profiling_utils import calculate_median_absolute_deviation_bounds +from databricks.labs.dqx.profiler.profile_options import ( + PROFILE_OPTION_DISTINCT_RATIO, + PROFILE_OPTION_FILTER, + PROFILE_OPTION_MAX_EMPTY_RATIO, + PROFILE_OPTION_MAX_IN_COUNT, + PROFILE_OPTION_MAX_NULL_RATIO, + PROFILE_OPTION_NUM_SIGMAS, + PROFILE_OPTION_OUTLIER_COLUMNS, + PROFILE_OPTION_OUTLIERS_RATIO, + PROFILE_OPTION_REMOVE_OUTLIERS, + PROFILE_OPTION_ROUND, + PROFILE_OPTION_TRIM_STRINGS, + PROFILE_OPTION_HAS_NO_OUTLIERS, + PROFILE_OPTION_HAS_NO_OUTLIERS_ALLOW_COLUMNS, + PROFILE_OPTION_HAS_NO_OUTLIERS_DENY_COLUMNS, + DEFAULT_PROFILE_OPTIONS, +) # Type alias for annotations; use TEXT_TYPES for isinstance() checks. TextType = T.CharType | T.StringType | T.VarcharType TEXT_TYPES: tuple[type[TextType], ...] = (T.CharType, T.StringType, T.VarcharType) + PROFILE_BUILDER_REGISTRY: dict[str, DQProfileBuilder] = {} logger = logging.getLogger(__name__) @@ -81,8 +102,8 @@ def make_is_in_profile( if total_count == 0: return None - max_in_count = profiler_options.get("max_in_count", 0) - max_distinct_ratio = profiler_options.get("distinct_ratio", 0.0) + max_in_count = profiler_options.get(PROFILE_OPTION_MAX_IN_COUNT, 0) + max_distinct_ratio = profiler_options.get(PROFILE_OPTION_DISTINCT_RATIO, 0.0) col = df.columns[0] distinct_values = [row[0] for row in df.select(col).distinct().collect()] @@ -99,7 +120,7 @@ def make_is_in_profile( name="is_in", column=column_name, parameters={"in": distinct_values}, - filter=profiler_options.get("filter", None), + filter=profiler_options.get(PROFILE_OPTION_FILTER, None), ) return None @@ -177,9 +198,9 @@ def _make_null_or_empty_profile( null_ratio = null_count / total_count empty_count = profiler_metrics.get("empty_count", 0) empty_ratio = empty_count / total_count - max_null_ratio = profiler_options.get("max_null_ratio", 0.0) - max_empty_ratio = profiler_options.get("max_empty_ratio", 0.0) - trim_strings = profiler_options.get("trim_strings", True) + max_null_ratio = profiler_options.get(PROFILE_OPTION_MAX_NULL_RATIO, 0.0) + max_empty_ratio = profiler_options.get(PROFILE_OPTION_MAX_EMPTY_RATIO, 0.0) + trim_strings = profiler_options.get(PROFILE_OPTION_TRIM_STRINGS, True) not_null = null_ratio <= max_null_ratio not_empty = empty_ratio <= max_empty_ratio @@ -196,7 +217,7 @@ def _make_null_or_empty_profile( column=column_name, description=description, parameters={"trim_strings": trim_strings}, - filter=profiler_options.get("filter", None), + filter=profiler_options.get(PROFILE_OPTION_FILTER, None), ) if not_null: @@ -209,7 +230,7 @@ def _make_null_or_empty_profile( name="is_not_null", column=column_name, description=description, - filter=profiler_options.get("filter", None), + filter=profiler_options.get(PROFILE_OPTION_FILTER, None), ) if not_empty: @@ -222,7 +243,7 @@ def _make_null_or_empty_profile( else None ), parameters={"trim_strings": trim_strings}, - filter=profiler_options.get("filter", None), + filter=profiler_options.get(PROFILE_OPTION_FILTER, None), ) return None @@ -247,7 +268,7 @@ def _make_null_profile( if total_count == 0: return None null_ratio = null_count / total_count - max_null_ratio = profiler_options.get("max_null_ratio", 0.0) + max_null_ratio = profiler_options.get(PROFILE_OPTION_MAX_NULL_RATIO, 0.0) if null_ratio <= max_null_ratio: return DQProfile( @@ -258,7 +279,7 @@ def _make_null_profile( if null_count > 0 else None ), - filter=profiler_options.get("filter", None), + filter=profiler_options.get(PROFILE_OPTION_FILTER, None), ) return None @@ -303,16 +324,111 @@ def _remove_outliers(column_name: str, profiler_options: dict[str, Any]) -> bool Returns: True if outlier removal should be applied to this column, otherwise False. """ - remove_outliers = profiler_options.get("remove_outliers", True) + remove_outliers = profiler_options.get(PROFILE_OPTION_REMOVE_OUTLIERS, True) if not remove_outliers: return False - outlier_columns = profiler_options.get("outlier_columns", []) + outlier_columns = profiler_options.get(PROFILE_OPTION_OUTLIER_COLUMNS, []) if not outlier_columns: return True # empty list means apply to all columns return column_name in outlier_columns +def _is_has_no_outliers_enabled(column_name: str, profiler_options: dict[str, Any]) -> bool: + """ + Checks if *has_no_outliers* profiling is enabled for given column. + + Args: + column_name: Input column name + profiler_options: Configuration options for the DQProfiler + + Returns: + True if *has_no_outliers* profile is enabled to this column, otherwise False. + """ + return _is_profile_enabled( + column_name, + PROFILE_OPTION_HAS_NO_OUTLIERS, + PROFILE_OPTION_HAS_NO_OUTLIERS_ALLOW_COLUMNS, + PROFILE_OPTION_HAS_NO_OUTLIERS_DENY_COLUMNS, + profiler_options, + ) + + +def _is_profile_enabled( + column_name: str, + profile_enabled_option_name: str, + profile_allow_columns_option_name: str, + profile_deny_columns_option_name: str, + profiler_options: dict[str, Any], +) -> bool: + """ + Checks if a profiler builder is enabled for the given column. + + The builder can be switched on or off for all columns via *profile_enabled_option_name*. + When enabled globally, *profile_allow_columns_option_name* restricts it to a specific set of + columns, while *profile_deny_columns_option_name* excludes specific columns and applies the + builder to all others. The two lists are mutually exclusive. + + Args: + column_name: Input column name. + profile_enabled_option_name: Option key whose value is a bool controlling global enablement. + profile_allow_columns_option_name: Option key whose value is a list of columns to include. + profile_deny_columns_option_name: Option key whose value is a list of columns to exclude. + profiler_options: Configuration options for the DQProfiler. + + Returns: + True if the profiler builder should run for this column, otherwise False. + + Raises: + InvalidParameterError: if both *profile_allow_columns_option_name* and + *profile_deny_columns_option_name* are provided at the same time. + """ + profiler_enabled = profiler_options.get(profile_enabled_option_name, True) + if not profiler_enabled: + return False + + allowed_columns = profiler_options.get(profile_allow_columns_option_name, []) + denied_columns = profiler_options.get(profile_deny_columns_option_name, []) + + if not denied_columns and not allowed_columns: + return True + + if allowed_columns and denied_columns: + raise InvalidParameterError( + f"Values for both '{profile_allow_columns_option_name}' and '{profile_deny_columns_option_name}' are provided in the configuration. Please provide only one of them." + ) + + if allowed_columns and column_name not in allowed_columns: + return False + + return column_name not in denied_columns + + +def validate_profile_options(profiler_options: dict[str, Any]) -> None: + """ + Validates profiler options once, up front, before any profiling work is done. + + Currently checks the allow/deny column-list pairs that are mutually exclusive, so a + misconfiguration fails fast at option-build time rather than partway through a profiling run + (which would happen if the check only ran per column inside *_is_profile_enabled*). + + Args: + profiler_options: Configuration options for the DQProfiler (merged with defaults). + + Raises: + InvalidParameterError: if both the allow-columns and deny-columns option of a builder are set. + """ + mutually_exclusive_pairs = [ + (PROFILE_OPTION_HAS_NO_OUTLIERS_ALLOW_COLUMNS, PROFILE_OPTION_HAS_NO_OUTLIERS_DENY_COLUMNS), + ] + for allow_option, deny_option in mutually_exclusive_pairs: + if profiler_options.get(allow_option) and profiler_options.get(deny_option): + raise InvalidParameterError( + f"Values for both '{allow_option}' and '{deny_option}' are provided in the configuration. " + "Please provide only one of them." + ) + + def _make_min_max_profile_with_outlier_removal( df: DataFrame, column_name: str, @@ -363,7 +479,7 @@ def _make_min_max_profile_with_outlier_removal( column=column_name, description=description, parameters={"min": min_limit, "max": max_limit}, - filter=profiler_options.get("filter", None), + filter=profiler_options.get(PROFILE_OPTION_FILTER, None), ) @@ -444,7 +560,7 @@ def _make_min_max_profile_without_outlier_removal( column=column_name, parameters={"min": min_value, "max": max_value}, description="Real min/max values were used", - filter=profiler_options.get("filter", None), + filter=profiler_options.get(PROFILE_OPTION_FILTER, None), ) @@ -467,7 +583,7 @@ def _get_min_max_limits( max_value = aggregates.get("max_value") mean_value = aggregates.get("mean_value") stddev_value = aggregates.get("stddev_value") - num_sigmas = profiler_options.get("num_sigmas", 3) + num_sigmas = profiler_options.get(PROFILE_OPTION_NUM_SIGMAS, 3) if mean_value is None or stddev_value is None: adjusted_min_value, adjusted_max_value = _adjust_min_max_limits( @@ -560,7 +676,7 @@ def _round_value(value: Any, rounding_direction: str, profiler_options: dict[str Returns: The rounded value, or the original value if rounding is not enabled. """ - if value is None or not profiler_options.get("round", False): + if value is None or not profiler_options.get(PROFILE_OPTION_ROUND, False): return value if isinstance(value, datetime.datetime): @@ -646,3 +762,86 @@ def _round_decimal(value: decimal.Decimal, rounding_direction: str) -> decimal.D if rounding_direction == "up": return value.to_integral_value(rounding=decimal.ROUND_CEILING) return value + + +@register_profile_builder("has_no_outliers") +def make_has_no_outliers_profile( + df: DataFrame, + column_name: str, + column_type: T.DataType, + profiler_metrics: dict[str, Any], + profiler_options: dict[str, Any], +) -> DQProfile | None: + """ + Creates a *has_no_outliers* profile using the same MAD method as the *has_no_outliers* check rule. + + A profile is returned when all the following conditions are met: + - The column type is child of `pyspark.sql.types.NumericType`. + - The DataFrame is non-empty. + - The fraction of outliers (values outside *median* ± 3.5 × MAD) is at or below *outliers_ratio*. + - Profile generation is enabled at configuration level for all columns or given column. + + Args: + df: The DataFrame to create the profile for. + column_name: Input column name + column_type: Input column type + profiler_metrics: Column-level statistics computed by the DQProfiler + profiler_options: Configuration options for the DQProfiler + + Returns: + A DQProfile if all conditions are met, otherwise None. + """ + if not isinstance(column_type, T.NumericType): + return None + + if not _is_has_no_outliers_enabled(column_name, profiler_options): + return None + + total_non_null_count = profiler_metrics.get("count_non_null", 0) + if total_non_null_count == 0: + logger.info(f"Column '{column_name}' has no non-null values. Skipping `has_no_outliers` profile generation") + return None + + bounds = calculate_median_absolute_deviation_bounds(df, column_name) + if bounds is None: + logger.info( + f"MAD bounds were not calculated for column '{column_name}'. Skipping `has_no_outliers` profile generation" + ) + return None + + lower_bound, upper_bound = bounds + # Skip degenerate distributions. Exactly-equal bounds mean MAD == 0 (a constant column). A + # tiny-but-nonzero MAD collapses the band to a near-zero width relative to the column's scale, + # which would emit a rule that flags almost every row as an outlier at apply time. Guard both + # with a scale-relative check so only meaningfully-wide bands produce a rule. + band_width = upper_bound - lower_bound + scale = max(abs(lower_bound), abs(upper_bound)) + if band_width <= 0 or (scale > 0 and band_width <= 1e-12 * scale): + logger.info( + f"MAD band is degenerate (near-zero width) for column '{column_name}'. " + "The distribution is (near-)constant. Skipping profile generation." + ) + return None + + below_lower_bound_expr = F.col(column_name) < get_limit_expr(lower_bound) + above_upper_bound_expr = F.col(column_name) > get_limit_expr(upper_bound) + outside_bounds_expr = below_lower_bound_expr | above_upper_bound_expr + outliers_count = df.filter(outside_bounds_expr).count() + + outliers_ratio = float(outliers_count) / total_non_null_count + outliers_ratio_threshold = profiler_options.get( + PROFILE_OPTION_OUTLIERS_RATIO, DEFAULT_PROFILE_OPTIONS[PROFILE_OPTION_OUTLIERS_RATIO] + ) + + safe_column_name = column_name.replace("\n", " ").replace("\r", " ") + # Inclusive (<=) to match the sibling ratio gates (max_null_ratio / max_empty_ratio), so a + # column whose outlier fraction exactly equals the configured threshold still emits a rule. + if outliers_ratio <= outliers_ratio_threshold: + return DQProfile( + name="has_no_outliers", + description=f"Column {safe_column_name} has {outliers_ratio * 100:.1f}% of outliers (allowed: {outliers_ratio_threshold * 100:.1f}%). Lower boundary - {lower_bound}, upper boundary - {upper_bound}.", + column=column_name, + filter=profiler_options.get(PROFILE_OPTION_FILTER, None), + ) + + return None diff --git a/src/databricks/labs/dqx/profiler/profile_options.py b/src/databricks/labs/dqx/profiler/profile_options.py new file mode 100644 index 000000000..136272b99 --- /dev/null +++ b/src/databricks/labs/dqx/profiler/profile_options.py @@ -0,0 +1,85 @@ +"""Constants for DQProfiler option keys and default option values.""" + +# Round min/max boundary values away from the mean when generating *min_max* rules. +PROFILE_OPTION_ROUND = "round" + +# Maximum number of distinct values allowed for an *is_in* rule to be emitted. +PROFILE_OPTION_MAX_IN_COUNT = "max_in_count" + +# Maximum ratio of distinct-to-total rows (0–1) for an *is_in* rule to be emitted. +PROFILE_OPTION_DISTINCT_RATIO = "distinct_ratio" + +# Maximum acceptable fraction of null values for a *is_not_null* rule to be emitted. +PROFILE_OPTION_MAX_NULL_RATIO = "max_null_ratio" + +# Maximum acceptable fraction of empty-string values for an *is_not_empty* rule to be emitted. +PROFILE_OPTION_MAX_EMPTY_RATIO = "max_empty_ratio" + +# Whether to cap *min_max* boundaries using ``mean ± num_sigmas * stddev`` outlier removal. +PROFILE_OPTION_REMOVE_OUTLIERS = "remove_outliers" + +# Explicit list of column names on which to apply outlier removal; empty list means all columns. +PROFILE_OPTION_OUTLIER_COLUMNS = "outlier_columns" + +# Number of standard deviations used to cap outlier boundaries. +PROFILE_OPTION_NUM_SIGMAS = "num_sigmas" + +# Whether to trim leading/trailing whitespace before analysing string columns. +PROFILE_OPTION_TRIM_STRINGS = "trim_strings" + +# Fraction of rows to sample (float) or per-stratum fractions (dict) before profiling. +PROFILE_OPTION_SAMPLE_FRACTION = "sample_fraction" + +# Random seed for reproducible sampling; *None* means non-deterministic. +PROFILE_OPTION_SAMPLE_SEED = "sample_seed" + +# Column name used to stratify the sample; requires *PROFILE_OPTION_SAMPLE_FRACTION*. +PROFILE_OPTION_SAMPLE_BY_COLUMN = "sample_by_column" + +# Maximum number of distinct stratum values collected for uniform stratified sampling. +PROFILE_OPTION_SAMPLE_BY_VALUES_LIMIT = "sample_by_values_limit" + +# Hard row limit applied after sampling via ``DataFrame.limit``; *None* disables it. +PROFILE_OPTION_LIMIT = "limit" + +# SQL expression used to pre-filter the DataFrame before profiling. +PROFILE_OPTION_FILTER = "filter" + +# Whether to invoke the LLM engine to detect primary-key columns and emit an *is_unique* rule. +PROFILE_OPTION_LLM_PRIMARY_KEY_DETECTION = "llm_primary_key_detection" + +# Outlier ratio threshold; emit a *has_no_outliers* profile only when the outlier fraction is below this value. +PROFILE_OPTION_OUTLIERS_RATIO = "outliers_ratio" + +# Whether generate *has_no_outliers* profile +PROFILE_OPTION_HAS_NO_OUTLIERS = "has_no_outliers" + +# List of allowed columns to generate *has_no_outliers* profile for +PROFILE_OPTION_HAS_NO_OUTLIERS_ALLOW_COLUMNS = "has_no_outliers_allow_columns" + +# List of denied columns to generate *has_no_outliers* profile for +PROFILE_OPTION_HAS_NO_OUTLIERS_DENY_COLUMNS = "has_no_outliers_deny_columns" + +# Default values for all DQProfiler options. +DEFAULT_PROFILE_OPTIONS: dict[str, None | bool | int | float | str | list[int] | list[float] | list[str]] = { + PROFILE_OPTION_ROUND: True, + PROFILE_OPTION_MAX_IN_COUNT: 10, + PROFILE_OPTION_DISTINCT_RATIO: 0.05, + PROFILE_OPTION_MAX_NULL_RATIO: 0.01, + PROFILE_OPTION_REMOVE_OUTLIERS: True, + PROFILE_OPTION_OUTLIER_COLUMNS: [], + PROFILE_OPTION_NUM_SIGMAS: 3, + PROFILE_OPTION_TRIM_STRINGS: True, + PROFILE_OPTION_MAX_EMPTY_RATIO: 0.01, + PROFILE_OPTION_SAMPLE_FRACTION: 0.3, + PROFILE_OPTION_SAMPLE_SEED: None, + PROFILE_OPTION_SAMPLE_BY_COLUMN: None, + PROFILE_OPTION_SAMPLE_BY_VALUES_LIMIT: 1000, + PROFILE_OPTION_LIMIT: 1000, + PROFILE_OPTION_FILTER: None, + PROFILE_OPTION_LLM_PRIMARY_KEY_DETECTION: True, + PROFILE_OPTION_OUTLIERS_RATIO: 0.01, + PROFILE_OPTION_HAS_NO_OUTLIERS: False, + PROFILE_OPTION_HAS_NO_OUTLIERS_ALLOW_COLUMNS: [], + PROFILE_OPTION_HAS_NO_OUTLIERS_DENY_COLUMNS: [], +} diff --git a/src/databricks/labs/dqx/profiler/profiler.py b/src/databricks/labs/dqx/profiler/profiler.py index b99c033a1..c88e93d97 100644 --- a/src/databricks/labs/dqx/profiler/profiler.py +++ b/src/databricks/labs/dqx/profiler/profiler.py @@ -18,7 +18,18 @@ from databricks.labs.dqx.errors import MissingParameterError, InvalidConfigError from databricks.labs.dqx.io import read_input_data, STORAGE_PATH_PATTERN from databricks.labs.dqx.profiler.profile import DQProfile -from databricks.labs.dqx.profiler.profile_builder import PROFILE_BUILDER_REGISTRY, TEXT_TYPES +from databricks.labs.dqx.profiler.profile_builder import PROFILE_BUILDER_REGISTRY, TEXT_TYPES, validate_profile_options +from databricks.labs.dqx.profiler.profile_options import ( + DEFAULT_PROFILE_OPTIONS, + PROFILE_OPTION_FILTER, + PROFILE_OPTION_LIMIT, + PROFILE_OPTION_LLM_PRIMARY_KEY_DETECTION, + PROFILE_OPTION_SAMPLE_BY_COLUMN, + PROFILE_OPTION_SAMPLE_BY_VALUES_LIMIT, + PROFILE_OPTION_SAMPLE_FRACTION, + PROFILE_OPTION_SAMPLE_SEED, + PROFILE_OPTION_TRIM_STRINGS, +) from databricks.labs.dqx.utils import list_tables from databricks.labs.dqx.telemetry import telemetry_logger @@ -47,25 +58,6 @@ def __init__( llm_model_config = llm_model_config or LLMModelConfig() self.llm_engine = DQLLMEngine(model_config=llm_model_config, spark=self.spark) if LLM_ENABLED else None - default_profile_options = { - "round": True, # round the min/max values - "max_in_count": 10, # generate is_in if we have less than 1 percent of distinct values - "distinct_ratio": 0.05, # generate is_in if we have less than 1 percent of distinct values - "max_null_ratio": 0.01, # generate is_not_null if we have less than 1 percent of nulls - "remove_outliers": True, # remove outliers - "outlier_columns": [], # remove outliers in the columns - "num_sigmas": 3, # number of sigmas to use when remove_outliers is True - "trim_strings": True, # trim whitespace from strings - "max_empty_ratio": 0.01, # generate is_not_null_or_empty rule if we have less than 1 percent of empty strings - "sample_fraction": 0.3, # fraction of data to sample. float for uniform sample, or dict for per-stratum sample (requires sample_by_column) - "sample_seed": None, # seed for sampling - "sample_by_column": None, # column used to stratify the sample via DataFrame.sampleBy - "sample_by_values_limit": 1000, # max distinct sample_by_column values collected when sampling uniformly - "limit": 1000, # limit the number of samples - "filter": None, # filter to apply to the dataset - "llm_primary_key_detection": True, # detect primary keys - } - @staticmethod def get_columns_or_fields(columns: list[T.StructField]) -> list[T.StructField]: """ @@ -111,7 +103,8 @@ def profile( if options is None: options = {} - options = {**self.default_profile_options, **options} # merge default options with user-provided options + options = {**DEFAULT_PROFILE_OPTIONS, **options} # merge default options with user-provided options + validate_profile_options(options) # fail fast on misconfiguration before any profiling work df = self._sample(df, options) dq_rules: list[DQProfile] = [] @@ -297,7 +290,7 @@ def _build_options_from_list(table: str, options: list[dict[str, Any]]) -> dict[ """ matched_options = DQProfiler._match_options_list(table, options) sorted_options = DQProfiler._sort_options_list(table, matched_options) - built_options = DQProfiler.default_profile_options.copy() + built_options = DEFAULT_PROFILE_OPTIONS.copy() for opt in sorted_options: if opt and isinstance(opt, dict): built_options |= opt.get("options") or {} @@ -340,12 +333,12 @@ def _sort_options_list(table: str, options: list[dict[str, Any]]) -> list[dict[s @staticmethod def _sample(df: DataFrame, opts: dict[str, Any]) -> DataFrame: - sample_fraction = opts.get("sample_fraction", None) - sample_seed = opts.get("sample_seed", None) - sample_by_column = opts.get("sample_by_column", None) - sample_by_values_limit = opts.get("sample_by_values_limit", None) - limit = opts.get("limit", None) - filter_dataset = opts.get("filter", None) + sample_fraction = opts.get(PROFILE_OPTION_SAMPLE_FRACTION, None) + sample_seed = opts.get(PROFILE_OPTION_SAMPLE_SEED, None) + sample_by_column = opts.get(PROFILE_OPTION_SAMPLE_BY_COLUMN, None) + sample_by_values_limit = opts.get(PROFILE_OPTION_SAMPLE_BY_VALUES_LIMIT, None) + limit = opts.get(PROFILE_OPTION_LIMIT, None) + filter_dataset = opts.get(PROFILE_OPTION_FILTER, None) if filter_dataset: df = df.filter(filter_dataset) @@ -444,7 +437,7 @@ def _profile( summary_stats: Summary statistics dictionary to update with profiler results. total_count: Total number of rows in the input DataFrame. """ - trim_strings = opts.get("trim_strings", True) + trim_strings = opts.get(PROFILE_OPTION_TRIM_STRINGS, True) for field in self.get_columns_or_fields(df_cols): field_name = field.name @@ -489,8 +482,10 @@ def _build_profiles_for_column( """Run registered profile builders for a column and append profiles. Builders are invoked in PROFILE_BUILDER_REGISTRY insertion order (null_or_empty → - is_in → min_max). Preserving that order matters: the min_max builder reads summary-stats - metrics written by earlier passes, so it must run last among the built-in builders. + is_in → min_max → has_no_outliers). Preserving that order matters: the min_max builder + reads summary-stats metrics written by earlier passes, so it must run after them. Any + builder that depends on min_max's resolved min/max (written back into *metrics* below) + must be registered after min_max. After a min_max profile is produced, its resolved min/max values are written back into *metrics* so that downstream consumers (e.g. LLM primary-key detection) can read them @@ -521,7 +516,7 @@ def _add_llm_primary_key_for_dataframe( summary_stats: Summary statistics dictionary to update with PK detection results opts: A dictionary of options for profiling. """ - if not LLM_ENABLED or not opts.get("llm_primary_key_detection", False): + if not LLM_ENABLED or not opts.get(PROFILE_OPTION_LLM_PRIMARY_KEY_DETECTION, False): return logger.info("🤖 Starting LLM-based primary key detection for DataFrame") @@ -551,7 +546,7 @@ def _add_llm_primary_key_for_dataframe( name="is_unique", column=",".join(valid_columns), parameters={"nulls_distinct": False, "reasoning": reasoning, "confidence": confidence}, - filter=opts.get("filter", None), + filter=opts.get(PROFILE_OPTION_FILTER, None), description=f"LLM-detected primary key columns: {', '.join(valid_columns)}", ) ) diff --git a/src/databricks/labs/dqx/profiler/profiler_runner.py b/src/databricks/labs/dqx/profiler/profiler_runner.py index b0211a58f..4281348e0 100644 --- a/src/databricks/labs/dqx/profiler/profiler_runner.py +++ b/src/databricks/labs/dqx/profiler/profiler_runner.py @@ -16,6 +16,18 @@ from databricks.labs.dqx.engine import DQEngine from databricks.labs.dqx.io import read_input_data from databricks.labs.dqx.profiler.generator import DQGenerator +from databricks.labs.dqx.profiler.profile_options import ( + PROFILE_OPTION_FILTER, + PROFILE_OPTION_LIMIT, + PROFILE_OPTION_LLM_PRIMARY_KEY_DETECTION, + PROFILE_OPTION_MAX_EMPTY_RATIO, + PROFILE_OPTION_MAX_NULL_RATIO, + PROFILE_OPTION_SAMPLE_BY_COLUMN, + PROFILE_OPTION_SAMPLE_BY_VALUES_LIMIT, + PROFILE_OPTION_SAMPLE_FRACTION, + PROFILE_OPTION_SAMPLE_SEED, + PROFILE_OPTION_OUTLIERS_RATIO, +) from databricks.labs.dqx.profiler.profiler import DQProfiler from databricks.labs.blueprint.installation import Installation @@ -65,18 +77,23 @@ def run( df = read_input_data(self.spark, run_config.input_config) options = { - "sample_fraction": run_config.profiler_config.sample_fraction, - "sample_seed": run_config.profiler_config.sample_seed, - "sample_by_column": run_config.profiler_config.sample_by_column, - "sample_by_values_limit": run_config.profiler_config.sample_by_values_limit, - "limit": run_config.profiler_config.limit, - "filter": run_config.profiler_config.filter, - "llm_primary_key_detection": run_config.profiler_config.llm_primary_key_detection, + PROFILE_OPTION_SAMPLE_FRACTION: run_config.profiler_config.sample_fraction, + PROFILE_OPTION_SAMPLE_SEED: run_config.profiler_config.sample_seed, + PROFILE_OPTION_SAMPLE_BY_COLUMN: run_config.profiler_config.sample_by_column, + PROFILE_OPTION_SAMPLE_BY_VALUES_LIMIT: run_config.profiler_config.sample_by_values_limit, + PROFILE_OPTION_LIMIT: run_config.profiler_config.limit, + PROFILE_OPTION_FILTER: run_config.profiler_config.filter, + PROFILE_OPTION_LLM_PRIMARY_KEY_DETECTION: run_config.profiler_config.llm_primary_key_detection, } if run_config.profiler_config.max_null_ratio is not None: - options["max_null_ratio"] = run_config.profiler_config.max_null_ratio + options[PROFILE_OPTION_MAX_NULL_RATIO] = run_config.profiler_config.max_null_ratio + if run_config.profiler_config.max_empty_ratio is not None: - options["max_empty_ratio"] = run_config.profiler_config.max_empty_ratio + options[PROFILE_OPTION_MAX_EMPTY_RATIO] = run_config.profiler_config.max_empty_ratio + + if run_config.profiler_config.outliers_ratio is not None: + options[PROFILE_OPTION_OUTLIERS_RATIO] = run_config.profiler_config.outliers_ratio + summary_stats, profiles = self.profiler.profile(df, options=options) checks = generator.generate_dq_rules(profiles, criticality=run_config.profiler_config.criticality) logger.info(f"Using options: \n{run_config.profiler_config}") @@ -119,14 +136,16 @@ def run_for_patterns( max_parallelism: Maximum number of parallel threads to use for profiling. """ pattern_options = { - "sample_fraction": run_config.profiler_config.sample_fraction, - "sample_seed": run_config.profiler_config.sample_seed, - "limit": run_config.profiler_config.limit, + PROFILE_OPTION_SAMPLE_FRACTION: run_config.profiler_config.sample_fraction, + PROFILE_OPTION_SAMPLE_SEED: run_config.profiler_config.sample_seed, + PROFILE_OPTION_LIMIT: run_config.profiler_config.limit, } if run_config.profiler_config.max_null_ratio is not None: - pattern_options["max_null_ratio"] = run_config.profiler_config.max_null_ratio + pattern_options[PROFILE_OPTION_MAX_NULL_RATIO] = run_config.profiler_config.max_null_ratio if run_config.profiler_config.max_empty_ratio is not None: - pattern_options["max_empty_ratio"] = run_config.profiler_config.max_empty_ratio + pattern_options[PROFILE_OPTION_MAX_EMPTY_RATIO] = run_config.profiler_config.max_empty_ratio + if run_config.profiler_config.outliers_ratio is not None: + pattern_options[PROFILE_OPTION_OUTLIERS_RATIO] = run_config.profiler_config.outliers_ratio options = [{"table": "*", "options": pattern_options}] logger.info(f"Using options: \n{options}") diff --git a/src/databricks/labs/dqx/profiling_utils.py b/src/databricks/labs/dqx/profiling_utils.py index 2d853b168..b573ffdf6 100644 --- a/src/databricks/labs/dqx/profiling_utils.py +++ b/src/databricks/labs/dqx/profiling_utils.py @@ -5,7 +5,7 @@ import collections.abc import pyspark.sql.functions as F -from pyspark.sql import DataFrame +from pyspark.sql import Column, DataFrame def compute_null_and_distinct_counts( @@ -49,3 +49,61 @@ def compute_exact_distinct_counts( stats_row = df.agg(*distinct_exprs).first() assert stats_row is not None, "Failed to compute distinct counts" return {col_name: stats_row[col_name] for col_name in columns} + + +def calculate_median_absolute_deviation_bounds( + df: DataFrame, column: str, filter_condition: str | Column | None = None +) -> tuple[float, float] | None: + """ + Calculates the lower and upper bounds using the median absolute deviation of a numeric column. + + Bounds are defined as *median* ± 3.5 × MAD. Returns None if the filtered DataFrame + is empty and the median cannot be computed. + + Args: + df: PySpark DataFrame + column: Name of the numeric column to calculate MAD for + filter_condition: Filter to apply before calculation (optional), as a SQL expression string or a + pre-compiled Column (e.g. a validated filter from *safe_filter_expr*). + + Returns: + A (lower_bound, upper_bound) tuple, or None if bounds cannot be calculated. + """ + median, mad = calculate_median_absolute_deviation(df, column, filter_condition) + if median is not None and mad is not None: + median = float(median) + mad = float(mad) + lower_bound = median - (3.5 * mad) + upper_bound = median + (3.5 * mad) + return lower_bound, upper_bound + + return None + + +def calculate_median_absolute_deviation( + df: DataFrame, column: str, filter_condition: str | Column | None +) -> tuple[float | None, float | None]: + """ + Calculates the Median Absolute Deviation (MAD) for a numeric column. + + MAD is a robust measure of variability: MAD = median(|X_i - median(X)|). + Computation applies *filter_condition* first, then computes the column median, + then the median of the absolute deviations from that median. + + Args: + df: PySpark DataFrame + column: Name of the numeric column to calculate MAD for + filter_condition: Filter to apply before calculation (optional), as a SQL expression string or a + pre-compiled Column (e.g. a validated filter from *safe_filter_expr*). + + Returns: + A (median, mad) tuple. Both values are None when the filtered DataFrame is empty. + """ + if filter_condition is not None: + df = df.filter(filter_condition) + + median_value = df.agg(F.percentile_approx(column, 0.5)).collect()[0][0] + df_with_deviations = df.select(F.abs(F.col(column) - F.lit(median_value)).alias("absolute_deviation")) + mad = df_with_deviations.agg(F.percentile_approx("absolute_deviation", 0.5)).collect()[0][0] + + return median_value, mad diff --git a/tests/integration/test_generator.py b/tests/integration/test_generator.py index 040706b82..b197dfe26 100644 --- a/tests/integration/test_generator.py +++ b/tests/integration/test_generator.py @@ -42,6 +42,11 @@ description="Real min/max values were used", parameters={"min": Decimal("0.01"), "max": Decimal("999.99")}, ), + DQProfile( + name="has_no_outliers", + column="price", + description="Price has no outliers", + ), ] @@ -110,6 +115,14 @@ def test_generate_dq_rules(ws, spark): "name": "price_isnt_in_range", "criticality": "error", }, + { + "check": { + "function": "has_no_outliers", + "arguments": {"column": "price"}, + }, + "name": "price_has_no_outliers", + "criticality": "error", + }, ] assert expectations == expected, f"Actual expectations: {expectations}" @@ -179,6 +192,14 @@ def test_generate_dq_rules_warn(ws, spark): "name": "price_isnt_in_range", "criticality": "warn", }, + { + "check": { + "function": "has_no_outliers", + "arguments": {"column": "price"}, + }, + "name": "price_has_no_outliers", + "criticality": "warn", + }, ] assert expectations == expected, f"Actual expectations: {expectations}" @@ -250,6 +271,7 @@ def test_generate_dq_rules_dataframe_filter(ws, spark): description=None, ), DQProfile(name="is_not_null_or_empty", column="vendor_id", parameters={"trim_strings": True}), + DQProfile(name="has_no_outliers", column="price"), ] expectations = generator.generate_dq_rules(test_rules_filter) @@ -289,6 +311,14 @@ def test_generate_dq_rules_dataframe_filter(ws, spark): "name": "vendor_id_is_null_or_empty", "criticality": "error", }, + { + "check": { + "function": "has_no_outliers", + "arguments": {"column": "price"}, + }, + "name": "price_has_no_outliers", + "criticality": "error", + }, ] assert expectations == expected @@ -315,6 +345,7 @@ def test_generate_dq_rules_dataframe_filter_none(ws, spark): filter=None, ), DQProfile(name="is_not_null_or_empty", column="vendor_id", parameters={"trim_strings": True}, filter=None), + DQProfile(name="has_no_outliers", column="price"), ] expectations = generator.generate_dq_rules(test_rules_no_filter) @@ -342,6 +373,14 @@ def test_generate_dq_rules_dataframe_filter_none(ws, spark): "name": "vendor_id_is_null_or_empty", "criticality": "error", }, + { + "check": { + "function": "has_no_outliers", + "arguments": {"column": "price"}, + }, + "name": "price_has_no_outliers", + "criticality": "error", + }, ] assert expectations == expected diff --git a/tests/integration/test_profile_builder.py b/tests/integration/test_profile_builder.py new file mode 100644 index 000000000..3ba005a6e --- /dev/null +++ b/tests/integration/test_profile_builder.py @@ -0,0 +1,160 @@ +import decimal + +import pyspark.sql.types as T +import pytest + +from databricks.labs.dqx.profiler.profile_builder import make_has_no_outliers_profile + + +@pytest.mark.parametrize( + "col_type", + [ + (T.ByteType()), + (T.IntegerType()), + (T.LongType()), + (T.ShortType()), + (T.FloatType()), + (T.DoubleType()), + (T.DecimalType(10, 2)), + ], +) +def test_make_has_no_outliers_profile_empty_data_frame(spark, col_type): + """No profile when count_non_null is zero (early-exit path).""" + df = spark.createDataFrame([], T.StructType([T.StructField("col", col_type)])) + profiler_metrics = {"count_non_null": 0} + profiler_options = {"outliers_ratio": 0.01} + profile = make_has_no_outliers_profile(df, "col", col_type, profiler_metrics, profiler_options) + assert profile is None + + +def test_make_has_no_outliers_profile_bounds_none(spark): + """No profile when MAD bounds cannot be computed because the effective DataFrame is empty. + + count_non_null is non-zero so the early-exit check is bypassed; the empty df makes + calculate_median_absolute_deviation_bounds return None, exercising the bounds=None path. + """ + df = spark.createDataFrame([], T.StructType([T.StructField("col", T.IntegerType())])) + profiler_metrics = {"count_non_null": 5} + profiler_options = {"outliers_ratio": 0.01} + profile = make_has_no_outliers_profile(df, "col", T.IntegerType(), profiler_metrics, profiler_options) + assert profile is None + + +@pytest.mark.parametrize( + "col_type,data", + [ + (T.ByteType(), [(1,), (2,), (3,), (4,), (5,), (6,), (7,), (8,), (9,), (10,), (127,)]), + (T.IntegerType(), [(1,), (2,), (3,), (4,), (5,), (6,), (7,), (8,), (9,), (10,), (1000,)]), + (T.LongType(), [(1,), (2,), (3,), (4,), (5,), (6,), (7,), (8,), (9,), (10,), (1000,)]), + (T.ShortType(), [(1,), (2,), (3,), (4,), (5,), (6,), (7,), (8,), (9,), (10,), (1000,)]), + (T.FloatType(), [(1,), (2,), (3,), (4,), (5,), (6,), (7,), (8,), (9,), (10,), (1000,)]), + (T.DoubleType(), [(1,), (2,), (3,), (4,), (5,), (6,), (7,), (8,), (9,), (10,), (1000,)]), + ( + T.DecimalType(10, 2), + [ + (decimal.Decimal("1"),), + (decimal.Decimal("2"),), + (decimal.Decimal("3"),), + (decimal.Decimal("4"),), + (decimal.Decimal("5"),), + (decimal.Decimal("6"),), + (decimal.Decimal("7"),), + (decimal.Decimal("8"),), + (decimal.Decimal("9"),), + (decimal.Decimal("10"),), + (decimal.Decimal("1000"),), + ], + ), + ], +) +def test_make_has_no_outliers_profile_outliers_below_threshold(spark, col_type, data): + """ + Test expects profile to be created for the following case: 11 values with one extreme value. + 1 outlier out of 11 ≈ 9 %. The threshold configuration is 10 %, hence profile is expected to return. + MAD bounds: median=6; MAD=3; lower=-4.5; upper=16.5; Hence only 1000 is an outlier. + """ + df = spark.createDataFrame(data, T.StructType([T.StructField("col", col_type)])) + profiler_metrics = {"count_non_null": len(data)} + profiler_options = {"outliers_ratio": 0.1} + profile = make_has_no_outliers_profile(df, "col", col_type, profiler_metrics, profiler_options) + assert profile is not None + assert profile.name == "has_no_outliers" + assert profile.column == "col" + assert profile.filter is None + + +@pytest.mark.parametrize( + "col_type,data", + [ + (T.ByteType(), [(1,), (2,), (3,), (4,), (100,), (120,), (127,)]), + (T.IntegerType(), [(1,), (2,), (3,), (4,), (100,), (200,), (300,)]), + (T.LongType(), [(1,), (2,), (3,), (4,), (100,), (200,), (300,)]), + (T.ShortType(), [(1,), (2,), (3,), (4,), (100,), (200,), (300,)]), + (T.FloatType(), [(1,), (2,), (3,), (4,), (100,), (200,), (300,)]), + (T.DoubleType(), [(1,), (2,), (3,), (4,), (100,), (200,), (300,)]), + ( + T.DecimalType(10, 2), + [ + (decimal.Decimal("1"),), + (decimal.Decimal("2"),), + (decimal.Decimal("3"),), + (decimal.Decimal("4"),), + (decimal.Decimal("100"),), + (decimal.Decimal("200"),), + (decimal.Decimal("300"),), + ], + ), + ], +) +def test_make_has_no_outliers_profile_outliers_above_threshold(spark, col_type, data): + """ + Test expects no profile for the following case: 7 values with three extreme values. + 3 outliers out of 7 ≈ 43 %. The threshold configuration is 10 %, hence no profile is expected to return. + MAD bounds: median=4; MAD=3; lower=-6.5; upper=14.5; Hence 100, 200, 300 are outliers. + """ + # [1..4] + three extreme values → 3 outliers out of 7 ≈ 43 %, threshold 10 % → None + # MAD bounds: median=4, MAD=3 → lower=-6.5, upper=14.5 → 100, 200, 300 are outliers + df = spark.createDataFrame(data, T.StructType([T.StructField("col", col_type)])) + profiler_metrics = {"count_non_null": len(data)} + profiler_options = {"outliers_ratio": 0.1} + profile = make_has_no_outliers_profile(df, "col", col_type, profiler_metrics, profiler_options) + assert profile is None + + +def test_make_has_no_outliers_profile_bounds_equal(spark): + """No profile when all values are identical. + + MAD = 0 when every value equals the median, so lower_bound == upper_bound == median. + The function detects this and returns None to avoid a degenerate rule that would flag + every value as an outlier. + """ + data = [(5,), (5,), (5,), (5,), (5,)] + df = spark.createDataFrame(data, T.StructType([T.StructField("col", T.IntegerType())])) + profiler_metrics = {"count_non_null": len(data)} + profiler_options = {"outliers_ratio": 0.1} + profile = make_has_no_outliers_profile(df, "col", T.IntegerType(), profiler_metrics, profiler_options) + assert profile is None + + +@pytest.mark.parametrize( + "col_type", + [ + (T.ByteType()), + (T.IntegerType()), + (T.LongType()), + (T.ShortType()), + (T.FloatType()), + (T.DoubleType()), + (T.DecimalType(10, 2)), + ], +) +def test_make_has_no_outliers_profile_outliers_null_values(spark, col_type): + """ + Test expects no profile for the following case: data frame is not empty but consists of null values. + """ + data = [(None,), (None,)] + df = spark.createDataFrame(data, T.StructType([T.StructField("col", col_type)])) + profiler_metrics = {"count_non_null": 0} + profiler_options = {"outliers_ratio": 0.1} + profile = make_has_no_outliers_profile(df, "col", col_type, profiler_metrics, profiler_options) + assert profile is None diff --git a/tests/integration/test_profiler.py b/tests/integration/test_profiler.py index b585d392a..1ae3c3d3e 100644 --- a/tests/integration/test_profiler.py +++ b/tests/integration/test_profiler.py @@ -71,13 +71,22 @@ def test_profiler(spark, ws): ) profiler = DQProfiler(ws) - stats, profiles = profiler.profile(inp_df, options={"sample_fraction": None, "llm_primary_key_detection": False}) + stats, profiles = profiler.profile( + inp_df, options={"sample_fraction": None, "llm_primary_key_detection": False, "has_no_outliers": True} + ) expected_profiles = [ DQProfile(name="is_not_null", column="t1", description=None, parameters=None), DQProfile( name="min_max", column="t1", description="Real min/max values were used", parameters={"min": 1, "max": 3} ), + DQProfile( + name='has_no_outliers', + column='t1', + description='Column t1 has 0.0% of outliers (allowed: 1.0%). Lower boundary - -1.5, upper boundary - 5.5.', + parameters=None, + filter=None, + ), DQProfile(name="is_not_null", column="d1", description=None, parameters=None), DQProfile( name="min_max", @@ -210,13 +219,22 @@ def test_profiler_rounding_midnight_behavior(spark, ws): ) profiler = DQProfiler(ws) - stats, profiles = profiler.profile(inp_df, options={"sample_fraction": None, "llm_primary_key_detection": False}) + stats, profiles = profiler.profile( + inp_df, options={"sample_fraction": None, "llm_primary_key_detection": False, "has_no_outliers": True} + ) expected_profiles = [ DQProfile(name="is_not_null", column="t1", description=None, parameters=None), DQProfile( name="min_max", column="t1", description="Real min/max values were used", parameters={"min": 1, "max": 3} ), + DQProfile( + name='has_no_outliers', + column='t1', + description='Column t1 has 0.0% of outliers (allowed: 1.0%). Lower boundary - -1.5, upper boundary - 5.5.', + parameters=None, + filter=None, + ), DQProfile(name="is_not_null", column="d1", description=None, parameters=None), DQProfile( name="min_max", @@ -326,6 +344,7 @@ def test_profiler_non_default_profile_options(spark, ws): "limit": 1000, # limit the number of samples "filter": "t1 > 0", # filter out the first row "llm_primary_key_detection": False, # disable pk detection + "has_no_outliers": True, # generate has_no_outliers profile } stats, profiles = profiler.profile(input_df, columns=input_df.columns, options=profile_options) @@ -339,6 +358,13 @@ def test_profiler_non_default_profile_options(spark, ws): parameters={"min": 1, "max": 3}, filter="t1 > 0", ), + DQProfile( + name='has_no_outliers', + column='t1', + description='Column t1 has 0.0% of outliers (allowed: 1.0%). Lower boundary - -1.5, upper boundary - 5.5.', + parameters=None, + filter="t1 > 0", + ), DQProfile( name="is_not_empty", # Column t2 contains null values column="t2", @@ -441,6 +467,7 @@ def test_profiler_non_default_profile_options_remove_outliers_no_outlier_columns "sample_seed": None, # seed for sampling "limit": 1000, # limit the number of samples "llm_primary_key_detection": False, # disable pk detection + "has_no_outliers": True, # generate has_no_outliers profile } stats, profiles = profiler.profile(inp_df, columns=inp_df.columns, options=profile_options) @@ -450,6 +477,13 @@ def test_profiler_non_default_profile_options_remove_outliers_no_outlier_columns DQProfile( name="min_max", column="t1", description="Real min/max values were used", parameters={"min": 1, "max": 3} ), + DQProfile( + name='has_no_outliers', + column='t1', + description='Column t1 has 0.0% of outliers (allowed: 1.0%). Lower boundary - -1.5, upper boundary - 5.5.', + parameters=None, + filter=None, + ), DQProfile( name="is_not_empty", column="t2", description=None, parameters={"trim_strings": False} ), # t2 contains null values @@ -544,6 +578,7 @@ def test_profiler_non_default_profile_options_with_rounding_enabled(spark, ws): "sample_seed": None, # seed for sampling "limit": 1000, # limit the number of samples "llm_primary_key_detection": False, # disable pk detection + "has_no_outliers": True, # generate has_no_outliers profile } stats, profiles = profiler.profile(inp_df, columns=inp_df.columns, options=profile_options) @@ -553,6 +588,13 @@ def test_profiler_non_default_profile_options_with_rounding_enabled(spark, ws): DQProfile( name="min_max", column="t1", description="Real min/max values were used", parameters={"min": 1, "max": 3} ), + DQProfile( + name='has_no_outliers', + column='t1', + description='Column t1 has 0.0% of outliers (allowed: 1.0%). Lower boundary - -1.5, upper boundary - 5.5.', + parameters=None, + filter=None, + ), DQProfile(name="is_not_empty", column="t2", description=None, parameters={"trim_strings": False}), DQProfile(name="is_not_null", column="s1.ns1", description=None, parameters=None), DQProfile( @@ -1664,6 +1706,7 @@ def test_profile_with_dataset_filter(spark, ws): "limit": None, "filter": "machine_id IN ('MCH-002', 'MCH-003') AND maintenance_type = 'preventive'", "llm_primary_key_detection": False, + "has_no_outliers": True, } profiler = DQProfiler(ws) @@ -1714,6 +1757,13 @@ def test_profile_with_dataset_filter(spark, ws): filter="machine_id IN ('MCH-002', 'MCH-003') AND maintenance_type = 'preventive'", description="Real min/max values were used", ), + DQProfile( + name='has_no_outliers', + column='cost', + description='Column cost has 0.0% of outliers (allowed: 1.0%). Lower boundary - -25.0, upper boundary - 325.0.', + parameters=None, + filter="machine_id IN ('MCH-002', 'MCH-003') AND maintenance_type = 'preventive'", + ), DQProfile( name="is_not_null", column="next_scheduled_date", @@ -1914,7 +1964,7 @@ def test_profiler_with_pk_detection(spark, ws): llm_model_config = LLMModelConfig() profiler = DQProfiler(ws, llm_model_config=llm_model_config) - stats, profiles = profiler.profile(input_df, options={"sample_fraction": None}) + stats, profiles = profiler.profile(input_df, options={"sample_fraction": None, "has_no_outliers": True}) expected_profiles = [ DQProfile(name="is_not_null", column="order_id", description=None, parameters=None), @@ -1924,6 +1974,13 @@ def test_profiler_with_pk_detection(spark, ws): description="Real min/max values were used", parameters={"max": 5, "min": 1}, ), + DQProfile( + name='has_no_outliers', + column='order_id', + description='Column order_id has 0.0% of outliers (allowed: 1.0%). Lower boundary - -0.5, upper boundary - 6.5.', + parameters=None, + filter=None, + ), DQProfile(name="is_not_null", column="customer_id", description=None, parameters=None), DQProfile( name="min_max", @@ -1931,6 +1988,13 @@ def test_profiler_with_pk_detection(spark, ws): description="Real min/max values were used", parameters={"max": 102, "min": 100}, ), + DQProfile( + name='has_no_outliers', + column='customer_id', + description='Column customer_id has 0.0% of outliers (allowed: 1.0%). Lower boundary - 97.5, upper boundary - 104.5.', + parameters=None, + filter=None, + ), DQProfile(name="is_not_null", column="amount", description=None, parameters=None), DQProfile( name="min_max", @@ -1938,6 +2002,13 @@ def test_profiler_with_pk_detection(spark, ws): description="Real min/max values were used", parameters={"max": 90, "min": 45}, ), + DQProfile( + name='has_no_outliers', + column='amount', + description='Column amount has 0.0% of outliers (allowed: 1.0%). Lower boundary - 7.5, upper boundary - 112.5.', + parameters=None, + filter=None, + ), DQProfile(name="is_not_null_or_empty", column="status", description=None, parameters={"trim_strings": True}), DQProfile( name="is_unique", @@ -2254,6 +2325,59 @@ def test_profiler_count_distinct_computed(spark, ws): assert stats["value"]["count_distinct"] == 3 +def test_profiler_generates_has_no_outliers_for_clean_numeric_data(spark, ws): + """End-to-end: has_no_outliers profile is emitted when the outlier fraction is below the threshold. + + 20 clean integer values → 0 outliers → 0 % < 1 % threshold → profile generated. + MAD bounds for [1..20]: median=10.5, MAD=5.0, bounds=(-7.0, 28.0) → no values outside. + """ + schema = T.StructType([T.StructField("value", T.IntegerType())]) + data = [(i,) for i in range(1, 21)] + input_df = spark.createDataFrame(data, schema) + + profiler = DQProfiler(ws) + _, profiles = profiler.profile( + input_df, + options={ + "sample_fraction": None, + "llm_primary_key_detection": False, + "remove_outliers": False, + "outliers_ratio": 0.01, + "has_no_outliers": True, + }, + ) + + has_no_outliers_profiles = [p for p in profiles if p.name == "has_no_outliers"] + assert len(has_no_outliers_profiles) == 1 + assert has_no_outliers_profiles[0].column == "value" + assert has_no_outliers_profiles[0].filter is None + + +def test_profiler_no_has_no_outliers_when_outliers_exceed_threshold(spark, ws): + """End-to-end: has_no_outliers profile is suppressed when the outlier fraction exceeds the threshold. + + 7 values — 4 normal + 3 extreme → 3/7 ≈ 43 % > 10 % threshold → no profile. + MAD bounds: median=4, MAD=3, bounds=(-6.5, 14.5) → 100, 200, 300 are outliers. + """ + schema = T.StructType([T.StructField("value", T.IntegerType())]) + data = [(1,), (2,), (3,), (4,), (100,), (200,), (300,)] + input_df = spark.createDataFrame(data, schema) + + profiler = DQProfiler(ws) + _, profiles = profiler.profile( + input_df, + options={ + "sample_fraction": None, + "llm_primary_key_detection": False, + "outliers_ratio": 0.1, + "has_no_outliers": True, + }, + ) + + has_no_outliers_profiles = [p for p in profiles if p.name == "has_no_outliers"] + assert len(has_no_outliers_profiles) == 0 + + def _round_stats( stats: dict[str, dict[str, int | float | None]], precision: int = 10 ) -> dict[str, dict[str, int | float | None]]: diff --git a/tests/unit/test_generator.py b/tests/unit/test_generator.py index e5e96071d..988b5b453 100644 --- a/tests/unit/test_generator.py +++ b/tests/unit/test_generator.py @@ -6,6 +6,7 @@ from databricks.labs.dqx.errors import MissingParameterError import databricks.labs.dqx.profiler.generator as generator_module +from databricks.labs.dqx.profiler.profile import DQProfile def test_profiler_llm_disabled(generator, monkeypatch): @@ -64,3 +65,13 @@ def test_generate_dq_rules_ai_assisted_keeps_valid_drops_invalid(generator, capl warnings = [r for r in caplog.records if r.levelno == logging.WARNING] assert len(warnings) == 1, f"Expected one WARNING for the invalid rule only, got {warnings}" assert "nonexistent_function_xyz" in warnings[0].getMessage() + + +def test_generate_has_no_outliers(generator, caplog): + """Test to check has_no_outliers rule generated out from profile""" + profiles = [DQProfile(name="has_no_outliers", column="measurement")] + rules = generator.generate_dq_rules(profiles) + assert len(rules) == 1 + assert "has_no_outliers" in rules[0]["check"]["function"] + assert "measurement" in rules[0]["check"]["arguments"]["column"] + assert "error" in rules[0]["criticality"] diff --git a/tests/unit/test_profile_builder.py b/tests/unit/test_profile_builder.py index 51fe853fa..22f2aacd0 100644 --- a/tests/unit/test_profile_builder.py +++ b/tests/unit/test_profile_builder.py @@ -5,13 +5,16 @@ import pyspark.sql.types as T from pyspark.sql import DataFrame +from databricks.labs.dqx.errors import InvalidParameterError from databricks.labs.dqx.profiler.profile import DQProfile from databricks.labs.dqx.profiler.profile_builder import ( PROFILE_BUILDER_REGISTRY, + make_has_no_outliers_profile, make_is_in_profile, make_min_max_profile, make_null_or_empty_profile, register_profile_builder, + validate_profile_options, ) @@ -30,6 +33,7 @@ def test_profile_builder_registry_contains_expected_builders(): assert "null_or_empty" in PROFILE_BUILDER_REGISTRY assert "is_in" in PROFILE_BUILDER_REGISTRY assert "min_max" in PROFILE_BUILDER_REGISTRY + assert "has_no_outliers" in PROFILE_BUILDER_REGISTRY def test_register_profile_builder_registers_and_builder_is_callable(): @@ -504,3 +508,213 @@ def test_min_max_rounding_enabled_for_decimal_type(mock_df): assert profile is not None assert profile.parameters["min"] == decimal.Decimal("1") assert profile.parameters["max"] == decimal.Decimal("10") + + +# --------------------------------------------------------------------------- +# make_has_no_outliers_profile +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize("column_type", [T.StringType(), T.BooleanType(), T.DateType(), T.TimestampType()]) +def test_has_no_outliers_non_numeric_type_returns_none(mock_df, column_type): + profile = make_has_no_outliers_profile( + mock_df, "col", column_type, {"count_non_null": 10}, {"outliers_ratio": 0.01} + ) + assert profile is None + + +def test_has_no_outliers_count_non_null_zero_returns_none(mock_df): + profile = make_has_no_outliers_profile( + mock_df, "col", T.IntegerType(), {"count_non_null": 0}, {"outliers_ratio": 0.01} + ) + assert profile is None + + +def test_has_no_outliers_numeric_no_outliers_returns_profile(mock_df): + # median=3.0, MAD=1.0 → bounds=(-0.5, 6.5); filter returns 0 outliers + mock_df.agg.return_value.collect.return_value = [[3.0]] + mock_df.select.return_value.agg.return_value.collect.return_value = [[1.0]] + mock_df.filter.return_value.count.return_value = 0 + + profile = make_has_no_outliers_profile( + mock_df, "measurement", T.IntegerType(), {"count_non_null": 10}, {"outliers_ratio": 0.05} + ) + + assert profile is not None + assert profile.name == "has_no_outliers" + assert profile.column == "measurement" + assert profile.filter is None + assert "0.0% of outliers" in profile.description + assert "-0.5" in profile.description + assert "6.5" in profile.description + + +def test_has_no_outliers_outliers_exceed_threshold_returns_none(mock_df): + # median=3.0, MAD=1.0 → bounds=(-0.5, 6.5); 2 outliers / 4 non-null = 50% > 10% threshold + mock_df.agg.return_value.collect.return_value = [[3.0]] + mock_df.select.return_value.agg.return_value.collect.return_value = [[1.0]] + mock_df.filter.return_value.count.return_value = 2 + + profile = make_has_no_outliers_profile( + mock_df, "col", T.IntegerType(), {"count_non_null": 4}, {"outliers_ratio": 0.1} + ) + + assert profile is None + + +def test_has_no_outliers_bounds_none_returns_none(mock_df): + # median=None → calculate_median_absolute_deviation_bounds returns None + mock_df.agg.return_value.collect.return_value = [[None]] + mock_df.select.return_value.agg.return_value.collect.return_value = [[None]] + + profile = make_has_no_outliers_profile( + mock_df, "col", T.IntegerType(), {"count_non_null": 5}, {"outliers_ratio": 0.01} + ) + + assert profile is None + + +# --------------------------------------------------------------------------- +# make_has_no_outliers_profile — _is_has_no_outliers_enabled paths +# --------------------------------------------------------------------------- + + +def test_has_no_outliers_disabled_via_option_returns_none(mock_df): + profile = make_has_no_outliers_profile( + mock_df, "col", T.IntegerType(), {"count_non_null": 10}, {"has_no_outliers": False} + ) + assert profile is None + + +def test_has_no_outliers_column_in_allow_columns_returns_profile(mock_df): + mock_df.agg.return_value.collect.return_value = [[3.0]] + mock_df.select.return_value.agg.return_value.collect.return_value = [[1.0]] + mock_df.filter.return_value.count.return_value = 0 + + profile = make_has_no_outliers_profile( + mock_df, + "measurement", + T.IntegerType(), + {"count_non_null": 10}, + {"has_no_outliers_allow_columns": ["measurement"], "outliers_ratio": 0.05}, + ) + + assert profile is not None + assert profile.name == "has_no_outliers" + assert profile.column == "measurement" + + +def test_has_no_outliers_column_not_in_allow_columns_returns_none(mock_df): + profile = make_has_no_outliers_profile( + mock_df, + "other_col", + T.IntegerType(), + {"count_non_null": 10}, + {"has_no_outliers_allow_columns": ["measurement"], "outliers_ratio": 0.05}, + ) + assert profile is None + + +def test_has_no_outliers_column_in_deny_columns_returns_none(mock_df): + profile = make_has_no_outliers_profile( + mock_df, + "measurement", + T.IntegerType(), + {"count_non_null": 10}, + {"has_no_outliers_deny_columns": ["measurement"], "outliers_ratio": 0.05}, + ) + assert profile is None + + +def test_has_no_outliers_column_not_in_deny_columns_returns_profile(mock_df): + mock_df.agg.return_value.collect.return_value = [[3.0]] + mock_df.select.return_value.agg.return_value.collect.return_value = [[1.0]] + mock_df.filter.return_value.count.return_value = 0 + + profile = make_has_no_outliers_profile( + mock_df, + "measurement", + T.IntegerType(), + {"count_non_null": 10}, + {"has_no_outliers_deny_columns": ["other_col"], "outliers_ratio": 0.05}, + ) + + assert profile is not None + assert profile.name == "has_no_outliers" + assert profile.column == "measurement" + + +def test_has_no_outliers_both_allow_and_deny_columns_raises(mock_df): + with pytest.raises(InvalidParameterError): + make_has_no_outliers_profile( + mock_df, + "measurement", + T.IntegerType(), + {"count_non_null": 10}, + { + "has_no_outliers_allow_columns": ["measurement"], + "has_no_outliers_deny_columns": ["other_col"], + }, + ) + + +def test_has_no_outliers_filter_propagated(mock_df): + mock_df.agg.return_value.collect.return_value = [[3.0]] + mock_df.select.return_value.agg.return_value.collect.return_value = [[1.0]] + mock_df.filter.return_value.count.return_value = 0 + + profile = make_has_no_outliers_profile( + mock_df, + "col", + T.IntegerType(), + {"count_non_null": 10}, + {"outliers_ratio": 0.05, "filter": "x > 0"}, + ) + + assert profile is not None + assert profile.filter == "x > 0" + + +def test_has_no_outliers_ratio_equal_to_threshold_emits_profile(mock_df): + # median=3.0, MAD=1.0 → bounds=(-0.5, 6.5); 1 outlier / 10 non-null = 10% == 10% threshold. + # Inclusive gate (<=) must still emit, matching the null/empty ratio gates. + mock_df.agg.return_value.collect.return_value = [[3.0]] + mock_df.select.return_value.agg.return_value.collect.return_value = [[1.0]] + mock_df.filter.return_value.count.return_value = 1 + + profile = make_has_no_outliers_profile( + mock_df, "measurement", T.IntegerType(), {"count_non_null": 10}, {"outliers_ratio": 0.1} + ) + + assert profile is not None + assert profile.name == "has_no_outliers" + + +def test_has_no_outliers_near_degenerate_mad_returns_none(mock_df): + # A tiny-but-nonzero MAD relative to the column scale collapses the band to near-zero width; + # emitting a rule here would flag almost every row at apply time, so it must be skipped. + mock_df.agg.return_value.collect.return_value = [[1_000_000.0]] + mock_df.select.return_value.agg.return_value.collect.return_value = [[1e-9]] + + profile = make_has_no_outliers_profile( + mock_df, "col", T.IntegerType(), {"count_non_null": 10}, {"outliers_ratio": 0.05} + ) + + assert profile is None + + +def test_validate_profile_options_raises_when_allow_and_deny_both_set(): + with pytest.raises(InvalidParameterError, match="Please provide only one of them"): + validate_profile_options( + { + "has_no_outliers_allow_columns": ["a"], + "has_no_outliers_deny_columns": ["b"], + } + ) + + +def test_validate_profile_options_passes_when_only_one_list_set(): + # Only one of the two lists set (or neither) must not raise. + validate_profile_options({"has_no_outliers_allow_columns": ["a"]}) + validate_profile_options({"has_no_outliers_deny_columns": ["b"]}) + validate_profile_options({})