diff --git a/docs/dqx/docs/guide/data_contract_quality_rules_generation.mdx b/docs/dqx/docs/guide/data_contract_quality_rules_generation.mdx index da2a15f27..c31cb2f21 100644 --- a/docs/dqx/docs/guide/data_contract_quality_rules_generation.mdx +++ b/docs/dqx/docs/guide/data_contract_quality_rules_generation.mdx @@ -535,6 +535,249 @@ If you don't have LLM dependencies installed or want to skip text processing: +## Metric Rule Generation + + + + + +ODCS quality entries with `type: library` (or, per the spec, no `type` at all as long as `metric` is set) reference a named, standard data quality metric instead of a natural-language description or a hand-written DQX check. DQX recognizes five of these metrics — `rowCount`, `nullValues`, `missingValues`, `invalidValues`, and `duplicateValues` — and maps each one onto the DQX check(s) that best express it. + + +Library metrics exist for **portability**: the same five-metric vocabulary is understood by any ODCS-compliant tool, including [datacontract-cli](https://github.com/datacontract/datacontract-cli). They are not a capability DQX otherwise lacks — every one of the five is already expressible directly with `is_aggr_*` checks, `is_unique`, and `column: "*"` — and they cover a narrow slice of what DQX can check (no outlier detection, freshness, foreign keys, schema validation, or dataset comparison). For anything beyond simple row-count/null/duplicate/allowlist checks on a contract that must stay portable across tools, prefer [explicit rules](#explicit-rule-generation) with `type: custom` and `engine: dqx`, which give you the full DQX check surface. Treat `type: library` as the option for the simple, common cases, not the default way to express quality in a contract. + +This five-metric surface is intentionally closed: DQX does not plan to grow this mapping into a general ODCS interpreter for arbitrary quality vocabularies. + + +Every `type: library` entry is processed whenever `generate_rules_from_contract` runs, unless disabled with `process_library_rules=False` (default `True`), matching the `generate_predefined_rules`/`process_text_rules` opt-out pattern used elsewhere. An entry that DQX cannot map is skipped with a logged warning instead of failing the whole contract (see [Malformed and Unrecognized Entries](#malformed-and-unrecognized-entries)). + + +Several of the mapping decisions below are not settled by the ODCS spec text. Where that's the case, DQX matches [datacontract-cli](https://github.com/datacontract/datacontract-cli)'s reference mapping (onto [Soda Core](https://github.com/sodadata/soda-core) checks) rather than inventing its own reading, so a contract produces the same verdict regardless of which tool evaluates it. Each judgment call is called out explicitly where it applies: `duplicateValues`'s counting unit, `mustBeBetween`/`mustNotBeBetween`'s bound inclusivity, and `invalidValues`'s combined-condition handling of `validValues` + `pattern`. + + +### Defining Metric Rules in ODCS v3.x Contracts + +A library quality entry sets `type: library` and a `metric` name in the `quality` section of a property (for `nullValues`, `missingValues`, `invalidValues`, and the single-column form of `duplicateValues`) or a schema (for `rowCount` and the composite-key form of `duplicateValues`): + +```yaml +quality: + - type: library + metric: nullValues + mustBe: 0 + dimension: completeness # optional; defaults per metric, see below + severity: critical # optional; recorded verbatim, DQX defines no vocabulary for it +``` + +Every metric shares the same **threshold fields** (one of which must be set) and an optional **`unit`** — see [Threshold Fields and the SQL Fallback](#threshold-fields-and-the-sql-fallback) and [Rows vs. Percent](#rows-vs-percent) below. `dimension` and `severity` are recorded in `user_metadata` for every metric rule; `dimension` falls back to a per-metric ODCS default when the contract omits it, and `severity` is only present when the contract sets it. + +### Supported Metrics + +| Metric | Level | Default `dimension` | What it validates | +|--------|-------|----------------------|--------------------| +| `rowCount` | Schema | `completeness` | The total row count of the dataset. | +| `nullValues` | Property | `completeness` | The count or percentage of `NULL` values in a column. | +| `missingValues` | Property | `completeness` | The count or percentage of `NULL`s plus contract-supplied sentinel values (e.g. `""`, `"N/A"`) in a column. | +| `invalidValues` | Property | `conformity` | The count or percentage of values in a column that fail an allowlist and/or a regex pattern. | +| `duplicateValues` | Property (single column) or Schema (composite key) | `uniqueness` | The count or percentage of rows that duplicate another row on one or more key columns. | + +`nullValues`, `missingValues`, and `invalidValues` are property-level only; a schema-level entry for one of these is skipped with a warning. `rowCount` is schema-level only. `duplicateValues` supports both: a property-level entry with no arguments checks that single column, while a schema-level entry reads a composite key from `arguments.properties`. + +### Threshold Fields and the SQL Fallback + +Every library metric is evaluated against one of eight ODCS threshold fields, tried in this order — the first one present on the entry is used: `mustBe`, `mustNotBe`, `mustBeGreaterOrEqualTo`, `mustBeLessOrEqualTo`, `mustBeGreaterThan`, `mustBeLessThan`, `mustBeBetween`, `mustNotBeBetween`. + +- **`mustBe: 0`** is special-cased for `nullValues`, `missingValues`, `invalidValues`, and `duplicateValues`: it maps onto a cheap **row-level** check (`is_not_null`, `is_not_in_list`, `is_in_list`/`regex_match`, or `is_unique`) that pinpoints the offending rows, rather than a dataset-level count. +- **`mustBe`, `mustNotBe`, `mustBeGreaterOrEqualTo`, `mustBeLessOrEqualTo`** (including `mustBe` with a non-zero value) map onto exact-fit **dataset-level aggregate checks** — `is_aggr_equal`, `is_aggr_not_equal`, `is_aggr_not_less_than`, `is_aggr_not_greater_than` — over the metric's count or percentage, for `rowCount`, `nullValues`, `missingValues`, and `invalidValues`. +- **`mustBeGreaterThan`, `mustBeLessThan`, `mustBeBetween`, `mustNotBeBetween`** have no strict-inequality/range equivalent among DQX's aggregate checks, so they fall back to a dataset-level [`sql_query`](/docs/reference/quality_checks#using-sql-query) check with `condition_column: "condition"` (`true` means a violation). For `mustBeBetween`/`mustNotBeBetween`, **both bounds are inclusive** — a value exactly equal to either bound counts as being "between" them. The ODCS spec text doesn't settle this either way; DQX matches datacontract-cli's reference mapping (SodaCL's plain `between`, which is inclusive on both ends unless a bound is written with a round bracket) rather than picking its own reading. +- **`duplicateValues` is the one exception**: every non-`mustBe: 0` threshold (not just the four strict-inequality/range ones above) falls back to `sql_query`. The duplicate count/percentage is computed via a `GROUP BY ... HAVING` subquery rather than a plain aggregate expression, since Spark rejects a window function (the `PARTITION BY` used to detect duplicates) nested inside an aggregate function (`SUM`/`AVG`) — see [duplicateValues](#duplicatevalues) below for what it counts. + + +An entry with none of the eight fields set is skipped with a warning naming all eight, so a contract author can spot a typo (e.g. `mustbe` instead of `mustBe`) without reading DQX source. + + +### Rows vs. Percent + +The optional `unit` field controls whether a non-`mustBe: 0` threshold is evaluated as a raw row count (`unit: rows`, the default) or a percentage of all rows (`unit: percent`). This applies to `nullValues`, `missingValues`, `invalidValues`, and `duplicateValues`; `rowCount` is inherently a count and does not branch on `unit` (whatever value is set is still recorded in `user_metadata`, but does not change the generated check). For `nullValues`, `missingValues`, `invalidValues`, and `duplicateValues`, an unrecognized `unit` value (anything other than `rows` or `percent`) causes the entry to be skipped with a warning. + +### rowCount + + + + ```yaml + schema: + - name: orders + physicalType: table + quality: + - type: library + metric: rowCount + mustBeGreaterOrEqualTo: 1 # → is_aggr_not_less_than(column="*", aggr_type="count") + ``` + + + +A `mustBeGreaterThan`, `mustBeLessThan`, `mustBeBetween`, or `mustNotBeBetween` threshold on `rowCount` falls back to a `sql_query` check counting `COUNT(*)` over the whole dataset. + +### nullValues + + + + ```yaml + properties: + - name: customer_email + logicalType: string + physicalType: string + quality: + - type: library + metric: nullValues + mustBe: 0 # → row-level is_not_null(column="customer_email") + + - name: middle_name + logicalType: string + physicalType: string + quality: + - type: library + metric: nullValues + mustBeLessOrEqualTo: 5 + unit: percent # → is_aggr_not_greater_than over the null-percentage indicator + ``` + + + +### missingValues + +`missingValues` combines real `NULL`s with a contract-supplied sentinel list, so it needs `arguments.missingValues` — a list that may include a literal `null` alongside sentinel strings: + + + + ```yaml + properties: + - name: country_code + logicalType: string + physicalType: string + quality: + - type: library + metric: missingValues + mustBe: 0 + arguments: + missingValues: [null, "", "N/A", "UNKNOWN"] + # → row-level is_not_null (for the null entry) + # + row-level is_not_in_list(forbidden=["", "N/A", "UNKNOWN"]) + ``` + + + +`NULL` is always counted as missing, whether or not a literal `null` appears in `arguments.missingValues` — listing it is redundant, not required. With `mustBe: 0`, a row-level `is_not_null` rule is always generated, plus a row-level `is_not_in_list` rule when non-null sentinels are present; both rules share the same `user_metadata` and are OR'd together via DQX's own per-row error/warning union, since `is_not_in_list`'s `forbidden` list can never itself match a real SQL `NULL`. Any other threshold routes through a dataset-level count/percentage of rows matching `column IS NULL OR column IN (...)` — the same unconditional-`NULL` condition, so the metric means the same thing at every threshold. + +### invalidValues + +`invalidValues` accepts an allowlist (`arguments.validValues`), a regex (`arguments.pattern`), or both — a row fails if it matches neither. When both are present, DQX combines them into a single "invalid" condition (`NOT (in the allowlist) OR NOT (matches the pattern)`) rather than treating them as two independent thresholds, so `mustBeLessOrEqualTo: 5` means at most 5 rows failing *either* criterion, not up to 5 failing each independently. This combination is a DQX judgment call — datacontract-cli's reference mapping ignores `arguments.pattern` for `invalidValues` entirely and only honors `validValues`, so `pattern` support here is a DQX-only extension beyond what the reference implementation covers, even though the spec documents the field: + + + + ```yaml + properties: + - name: order_status + logicalType: string + physicalType: string + quality: + - type: library + metric: invalidValues + mustBe: 0 + arguments: + validValues: ["pending", "confirmed", "shipped", "delivered", "cancelled"] + # → row-level is_in_list(allowed=[...], case_sensitive=true) + + - name: postal_code + logicalType: string + physicalType: string + quality: + - type: library + metric: invalidValues + mustBeLessOrEqualTo: 0.5 + unit: percent + arguments: + pattern: '^[0-9]{5}$' + # → is_aggr_not_greater_than over the invalid-percentage indicator + ``` + + + + +`arguments.pattern` is checked at generation time for patterns that could cause catastrophic regex backtracking (ReDoS). A pattern that fails this check is treated as absent — its own warning is logged, and the entry still proceeds using `validValues` alone if present. Keep patterns simple and avoid nested or alternation-heavy quantifiers. + + +### duplicateValues + +The single-column, argument-less form is a property-level entry; the composite-key form is a schema-level entry with `arguments.properties`. + + +For a non-`mustBe: 0` threshold, `duplicateValues` counts the number of *distinct values (or key combinations) that recur* — for a column holding `[A, A, A, B, B, C]` that's 2 (`A` and `B` each recur), not 5 (the rows sitting in those two groups). `unit: percent` divides that count by the total row count. This matches datacontract-cli's reference mapping onto Soda's `duplicate_count`; the ODCS spec text ("Counts duplicate values in a column") doesn't settle it either way, so DQX follows the reference rather than counting affected rows. `mustBe: 0` is unaffected by this choice, since both readings agree at zero. + + + + + ```yaml + properties: + - name: order_id + logicalType: string + physicalType: string + quality: + - type: library + metric: duplicateValues + mustBe: 0 # → row-level is_unique(columns=["order_id"]) + ``` + + + ```yaml + schema: + - name: order_lines + physicalType: table + quality: + - type: library + metric: duplicateValues + mustBeLessOrEqualTo: 1 + unit: percent + arguments: + properties: [order_id, line_number] + # → sql_query over the percentage of distinct recurring (order_id, line_number) + # combinations, grouped by (order_id, line_number) + ``` + + + +### Malformed and Unrecognized Entries + +DQX **warns and skips** an individual `type: library` entry — logging the reason and continuing to process the rest of the contract — rather than failing generation, whenever it encounters: + +- A missing or unrecognized `metric` name (the warning names all five supported metrics). +- No recognized threshold field set (none of the eight `mustBe...`/`mustNotBe...` fields). +- An unrecognized `unit` (anything other than `rows` or `percent`) — see [Rows vs. Percent](#rows-vs-percent). +- A schema-level entry for a property-only metric (`nullValues`, `missingValues`, `invalidValues`). +- Malformed `arguments` — not a dict, a missing or wrong-typed key (e.g. `arguments.missingValues` not a list), or an empty list where a non-empty one is required. +- For `invalidValues`, an entry with neither a usable `arguments.validValues` list nor a usable `arguments.pattern` string (after the pattern safety check, described above). + +This per-entry warn-and-skip behavior is independent of `process_library_rules`: even with `process_library_rules=True` (the default), an individual malformed entry is skipped rather than failing the whole contract. Set `process_library_rules=False` to skip *all* `type: library` entries instead — e.g. if a contract's library entries need to be ignored altogether. + + +DQX does not deduplicate library metric rules against predefined rules generated from schema constraints. A property marked `required: true` (which generates a predefined `is_not_null` rule) that also carries a `nullValues`/`mustBe: 0` entry produces two separate `is_not_null` rules under different names. Both are harmless (they agree on every row), but if you want only one, disable one of the two sources — e.g. `generate_predefined_rules=False` or omit the redundant `nullValues` entry. + + +### Disabling Library Metric Processing + +If a contract's `type: library` entries misbehave, or you'd rather express quality entirely through [explicit rules](#explicit-rule-generation): + + + + ```python + # Skip type: library processing, generate only predefined/explicit/text rules + rules = generator.generate_rules_from_contract( + contract_file="contract.yaml", + process_library_rules=False # Skip library metric mapping + ) + ``` + + + ## Complete Usage Example Here's a complete example showing contract-based rule generation with all three rule types: @@ -692,9 +935,15 @@ All generated rules include rich metadata that traces them back to the source co | `contract_version` | Contract version | `2.1.0` | | `odcs_version` | ODCS API version (standard version) | `v3.0.2` | | `schema` | Schema/table name (ODCS v3.x) | `sensor_readings` | -| `field` | Property/column name (if property-level rule) | `sensor_id` | -| `rule_type` | How rule was generated | `predefined`, `explicit`, or `text_llm` | +| `field` | Property/column name (single-field property-level rule) | `sensor_id` | +| `fields` | Composite key columns (schema-level `duplicateValues` metric rule only) | `["order_id", "line_number"]` | +| `rule_type` | How rule was generated | `predefined`, `explicit`, `text_llm`, or `metric` (see [Metric Rule Generation](#metric-rule-generation)) | | `text_expectation` | Original text (for text-based rules only) | Natural language description | +| `metric` | ODCS library metric name (`rule_type: "metric"` only) | `rowCount`, `nullValues`, `missingValues`, `invalidValues`, or `duplicateValues` | +| `threshold_field` | ODCS threshold field that produced the check (`rule_type: "metric"` only) | `mustBe`, `mustBeGreaterOrEqualTo`, `mustBeBetween`, etc. | +| `unit` | Threshold unit for the metric (`rule_type: "metric"` only) | `rows` (default) or `percent` | +| `dimension` | Data quality dimension — the contract's own `dimension` if set, else the metric's ODCS default | `completeness`, `uniqueness`, or `conformity` | +| `severity` | Contract's own `severity` value, recorded verbatim for audit only (never used to derive `criticality`); present only when the contract sets it | Any contract-defined string | This metadata enables: - **Traceability**: Link quality check results back to contract specifications @@ -957,4 +1206,6 @@ pip install 'databricks-labs-dqx[datacontract]' - Dataset-level quality checks must be defined as explicit DQX rules (custom quality checks). - Generated rules quality depends on the completeness and correctness of the contract definition. - Complex business logic (text based rules) may still require manual rule definition or refinement. +- `type: library` covers five metrics only (see [Metric Rule Generation](#metric-rule-generation)) and is not intended to grow into a general ODCS interpreter; use `type: custom` with `engine: dqx` for anything beyond simple row-count/null/duplicate/allowlist checks. +- `type: library` entries are matched on the current `metric` field only; the deprecated ODCS 3.0.x `rule` key (superseded by `metric` in 3.1+) is not recognized, so a contract still authored against 3.0.x semantics needs its `rule` entries migrated to `metric` before DQX will map them. diff --git a/src/databricks/labs/dqx/datacontract/contract_rules_generator.py b/src/databricks/labs/dqx/datacontract/contract_rules_generator.py index 6514e672b..60083d2f7 100644 --- a/src/databricks/labs/dqx/datacontract/contract_rules_generator.py +++ b/src/databricks/labs/dqx/datacontract/contract_rules_generator.py @@ -12,7 +12,7 @@ import re from collections.abc import Callable from pathlib import Path -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING, Any, TypeVar import yaml @@ -33,6 +33,7 @@ from databricks.labs.dqx.errors import InvalidPhysicalTypeError, ODCSContractError, ParameterError from databricks.labs.dqx.telemetry import telemetry_logger from databricks.labs.dqx.package_utils import missing_required_packages +from databricks.labs.dqx.utils import sanitize_for_logging # DQLLMEngine is referenced only as a type annotation. Eagerly importing it # requires installation of [llm] extras which may not be installed or wanted @@ -43,6 +44,8 @@ logger = logging.getLogger(__name__) +_T = TypeVar("_T") + class DataContractRulesGenerator(DQEngineBase): """ @@ -90,6 +93,7 @@ def generate_rules_from_contract( contract_format: str = "odcs", generate_predefined_rules: bool = True, process_text_rules: bool = True, + process_library_rules: bool = True, generate_schema_validation: bool = True, strict_schema_validation: bool = True, default_criticality: str = "error", @@ -119,6 +123,7 @@ def generate_rules_from_contract( contract_format: Contract format specification (default is "odcs"). Only "odcs" is supported. generate_predefined_rules: Whether to generate rules from schema properties (default True). Set to False to only generate explicit rules. process_text_rules: Whether to process text-based expectations using LLM (default True). Requires llm_engine to be provided in __init__. + process_library_rules: Whether to generate rules from ODCS ``type: library`` quality metric entries (default True). Set to False to skip this mapping entirely, e.g. if a contract's library entries misbehave. generate_schema_validation: Whether to generate dataset-level has_valid_schema rules from the contract schema (default True). strict_schema_validation: Passed as the strict argument to has_valid_schema (default True = exact columns, order, types; False = permissive). default_criticality: Default criticality level for generated rules (default is "error"). @@ -140,6 +145,7 @@ def generate_rules_from_contract( odcs, generate_predefined_rules, process_text_rules, + process_library_rules, generate_schema_validation, strict_schema_validation, default_criticality, @@ -240,6 +246,7 @@ def _generate_all_rules( odcs: OpenDataContractStandard, generate_predefined_rules: bool, process_text_rules: bool, + process_library_rules: bool, generate_schema_validation: bool, strict_schema_validation: bool, default_criticality: str, @@ -270,6 +277,12 @@ def _generate_all_rules( explicit_rules = self._process_explicit_rules_for_schema(schema_obj, schema_name, odcs, default_criticality) dq_rules.extend(explicit_rules) + if process_library_rules: + library_rules = self._process_library_rules_for_schema( + schema_obj, schema_name, odcs, default_criticality + ) + dq_rules.extend(library_rules) + return dq_rules def _validate_generated_rules(self, dq_rules: list[dict]) -> list[dict]: @@ -1459,3 +1472,1464 @@ def _build_rule_dict( if owner: rule["owner"] = owner return rule + + # ODCS type: library quality metric support. + # + # Sibling dispatch to the explicit-rule extractors above: explicit rules fail fast on + # malformed structure and dispatch on a single generic `implementation` dict, while library + # rules warn-and-skip per-entry and dispatch per-metric to distinct builders. Conflating the + # two would mix incompatible error philosophies into one method. + + _SUPPORTED_LIBRARY_METRICS: tuple[str, ...] = ( + "rowCount", + "nullValues", + "missingValues", + "invalidValues", + "duplicateValues", + ) + + # Per-metric default `user_metadata["dimension"]` when the contract doesn't set its own, + # drawn from ODCS's own dimension vocabulary. + _LIBRARY_METRIC_DEFAULT_DIMENSIONS: dict[str, str] = { + "rowCount": "completeness", + "nullValues": "completeness", + "missingValues": "completeness", + "invalidValues": "conformity", + "duplicateValues": "uniqueness", + } + + _SAFE_SQL_IDENTIFIER_PATTERN = re.compile(r"^[a-zA-Z_][a-zA-Z0-9_]*$") + + # The eight ODCS threshold fields shared by every library metric (rowCount, duplicateValues, ...). + _THRESHOLD_FIELDS: tuple[str, ...] = ( + "mustBe", + "mustNotBe", + "mustBeGreaterThan", + "mustBeGreaterOrEqualTo", + "mustBeLessThan", + "mustBeLessOrEqualTo", + "mustBeBetween", + "mustNotBeBetween", + ) + + _MAX_LIBRARY_PATTERN_LENGTH = 200 + _NESTED_QUANTIFIER_PATTERN = re.compile(r"\([^()]*[+*]\)[+*{]") + _ALTERNATION_QUANTIFIER_PATTERN = re.compile(r"\([^()]*\|[^()]*\)[+*{]") + + def _is_dqx_library_rule(self, quality_rule: DataQuality) -> bool: + """Check if a quality rule is an ODCS type: library quality metric entry. + + Per the ODCS spec, ``type`` "can be omitted, if a metric property is defined" -- every + library example in the spec omits it. An explicit ``type: library`` is also accepted for + contracts that set it anyway. + """ + if quality_rule.type == 'library': + return True + return quality_rule.type is None and quality_rule.metric is not None + + def _process_library_rules_for_schema( + self, schema_obj: SchemaObject, schema_name: str, odcs: OpenDataContractStandard, default_criticality: str + ) -> list[dict]: + """Process ODCS type: library quality metric entries from an ODCS schema.""" + rules: list[dict] = [] + + # Process property-level library rules + for prop in schema_obj.properties or []: + if prop.quality: + rules.extend(self._extract_property_library_rules(prop, schema_name, odcs, default_criticality)) + + # Process schema-level library rules (e.g. rowCount) + if schema_obj.quality: + rules.extend(self._extract_schema_library_rules(schema_obj.quality, schema_name, odcs, default_criticality)) + + return rules + + def _extract_property_library_rules( + self, prop: SchemaProperty, schema_name: str, odcs: OpenDataContractStandard, default_criticality: str + ) -> list[dict]: + """Extract DQX rules from property-level type: library quality metric entries.""" + rules: list[dict] = [] + + if prop.quality is None: + return rules + + for quality_rule in prop.quality: + if not self._is_dqx_library_rule(quality_rule): + continue + try: + metric = self._resolve_library_metric(quality_rule, schema_name, prop.name) + if metric is None: + continue + rules.extend( + self._build_library_rules_for_metric( + quality_rule, metric, prop.name, schema_name, odcs, default_criticality + ) + ) + except (AttributeError, KeyError, TypeError) as e: + logger.warning( + f"Skipping malformed type: library quality entry on property " + f"'{sanitize_for_logging(prop.name or 'unknown')}' in schema " + f"'{sanitize_for_logging(schema_name)}': {sanitize_for_logging(str(e))}" + ) + return rules + + def _extract_schema_library_rules( + self, + quality_list: list[DataQuality], + schema_name: str, + odcs: OpenDataContractStandard, + default_criticality: str, + ) -> list[dict]: + """Extract DQX rules from schema-level type: library quality metric entries.""" + rules: list[dict] = [] + for quality_rule in quality_list: + if not self._is_dqx_library_rule(quality_rule): + continue + try: + metric = self._resolve_library_metric(quality_rule, schema_name) + if metric is None: + continue + rules.extend( + self._build_library_rules_for_metric( + quality_rule, metric, None, schema_name, odcs, default_criticality + ) + ) + except (AttributeError, KeyError, TypeError) as e: + logger.warning( + f"Skipping malformed type: library quality entry in schema " + f"'{sanitize_for_logging(schema_name)}': {sanitize_for_logging(str(e))}" + ) + return rules + + def _resolve_library_metric( + self, quality_rule: DataQuality, schema_name: str, property_name: str | None = None + ) -> str | None: + """Validate quality_rule.metric against the five supported ODCS library metrics. + + Warns and returns None for a missing or unrecognized metric name so the caller can skip + the entry without raising. The five supported metrics are named in the warning so a + contract author can self-correct a typo without reading DQX source. + """ + metric = quality_rule.metric + supported = ", ".join(self._SUPPORTED_LIBRARY_METRICS) + location = ( + f"property '{sanitize_for_logging(property_name)}' in schema '{sanitize_for_logging(schema_name)}'" + if property_name + else f"schema '{sanitize_for_logging(schema_name)}'" + ) + + if not metric: + logger.warning( + f"Missing 'metric' on type: library quality entry on {location}; skipping this quality check. " + f"Supported metrics: {supported}." + ) + return None + + if metric not in self._SUPPORTED_LIBRARY_METRICS: + logger.warning( + f"Unrecognized library metric '{sanitize_for_logging(metric)}' on {location}; skipping this " + f"quality check. Supported metrics: {supported}." + ) + return None + + return metric + + def _build_library_rules_for_metric( + self, + quality_rule: DataQuality, + metric: str, + property_name: str | None, + schema_name: str, + odcs: OpenDataContractStandard, + default_criticality: str, + ) -> list[dict]: + """Dispatch a recognized type: library metric to its rule-building implementation. + + Per-metric builders (rowCount, nullValues, missingValues, invalidValues, duplicateValues) + are added incrementally; a recognized metric with no builder yet produces no rules rather + than raising or warning. + """ + if metric == "rowCount": + return self._build_row_count_rules(quality_rule, schema_name, odcs, default_criticality) + if metric == "nullValues": + return self._build_nullvalues_rules(quality_rule, property_name, schema_name, odcs, default_criticality) + if metric == "missingValues": + return self._build_missing_values_rules(quality_rule, property_name, schema_name, odcs, default_criticality) + if metric == "invalidValues": + return self._build_invalid_values_rules(quality_rule, property_name, schema_name, odcs, default_criticality) + if metric == "duplicateValues": + return self._build_duplicate_values_rules( + quality_rule, property_name, schema_name, odcs, default_criticality + ) + return [] + + # rowCount ODCS threshold field -> (threshold_field, DQX check dict) builder, tried in order. + # mustBe/mustNotBe/mustBeGreaterOrEqualTo/mustBeLessOrEqualTo map onto exact-fit dataset-level + # aggregate checks; strict inequalities and both range forms have no aggregate equivalent and + # fall back to the dataset-level sql_query escape hatch. mustBeBetween/mustNotBeBetween treat + # both bounds as inclusive: the ODCS spec text doesn't settle it, and we match + # datacontract-cli's reference mapping (SodaCL's plain `between`, which is inclusive on both + # ends unless a bound is written with a round bracket) rather than pick our own reading. + + def _build_row_count_rules( + self, + quality_rule: DataQuality, + schema_name: str, + odcs: OpenDataContractStandard, + default_criticality: str, + ) -> list[dict]: + """Build the single DQX dataset-level row-count rule for an ODCS rowCount library entry.""" + resolved = self._row_count_check(quality_rule, schema_name) + if resolved is None: + return [] + threshold_field, check_dict = resolved + + user_metadata = { + "contract_id": odcs.id or "unknown", + "contract_version": odcs.version or "unknown", + "odcs_version": odcs.apiVersion or "unknown", + "schema": schema_name, + "rule_type": "metric", + "metric": "rowCount", + "threshold_field": threshold_field, + "unit": quality_rule.unit or "rows", + "dimension": self._library_dimension(quality_rule, "rowCount"), + **self._library_severity_metadata(quality_rule), + } + + return [ + { + "check": check_dict, + # threshold_field is included since a schema can carry multiple rowCount entries + # (e.g. a lower and an upper bound), which would otherwise collide on rule name. + "name": f"{schema_name}_rowCount_{threshold_field}", + "criticality": default_criticality, + "user_metadata": user_metadata, + } + ] + + def _row_count_check(self, quality_rule: DataQuality, schema_name: str) -> tuple[str, dict] | None: + """Resolve the first set ODCS threshold field on a rowCount entry to (field name, check dict). + + Returns None (after logging) when none of the eight ODCS threshold fields are set. + """ + if quality_rule.mustBe is not None: + return "mustBe", self._row_count_aggregate_check("is_aggr_equal", quality_rule.mustBe) + if quality_rule.mustNotBe is not None: + return "mustNotBe", self._row_count_aggregate_check("is_aggr_not_equal", quality_rule.mustNotBe) + if quality_rule.mustBeGreaterOrEqualTo is not None: + return "mustBeGreaterOrEqualTo", self._row_count_aggregate_check( + "is_aggr_not_less_than", quality_rule.mustBeGreaterOrEqualTo + ) + if quality_rule.mustBeLessOrEqualTo is not None: + return "mustBeLessOrEqualTo", self._row_count_aggregate_check( + "is_aggr_not_greater_than", quality_rule.mustBeLessOrEqualTo + ) + if quality_rule.mustBeGreaterThan is not None: + return "mustBeGreaterThan", self._library_sql_query_check( + f"SELECT COUNT(*) <= {quality_rule.mustBeGreaterThan} AS condition FROM {{{{ input_view }}}}" + ) + if quality_rule.mustBeLessThan is not None: + return "mustBeLessThan", self._library_sql_query_check( + f"SELECT COUNT(*) >= {quality_rule.mustBeLessThan} AS condition FROM {{{{ input_view }}}}" + ) + if quality_rule.mustBeBetween is not None: + min_val, max_val = quality_rule.mustBeBetween + return "mustBeBetween", self._library_sql_query_check( + f"SELECT NOT (COUNT(*) >= {min_val} AND COUNT(*) <= {max_val}) AS condition FROM {{{{ input_view }}}}" + ) + if quality_rule.mustNotBeBetween is not None: + min_val, max_val = quality_rule.mustNotBeBetween + return "mustNotBeBetween", self._library_sql_query_check( + f"SELECT (COUNT(*) >= {min_val} AND COUNT(*) <= {max_val}) AS condition FROM {{{{ input_view }}}}" + ) + + logger.warning( + f"type: library rowCount entry on schema '{sanitize_for_logging(schema_name)}' has no recognized " + "threshold field set (mustBe, mustNotBe, mustBeGreaterOrEqualTo, mustBeLessOrEqualTo, " + "mustBeGreaterThan, mustBeLessThan, mustBeBetween, mustNotBeBetween); skipping this quality check." + ) + return None + + @staticmethod + def _row_count_aggregate_check(function: str, limit: Any) -> dict: + """Build a dataset-level count-aggregate check dict (is_aggr_equal / is_aggr_not_equal / etc.).""" + return { + "function": function, + "arguments": {"column": "*", "limit": limit, "aggr_type": "count"}, + } + + @staticmethod + def _library_sql_query_check(query: str, *, row_filter: str | None = None) -> dict: + """Build a dataset-level sql_query check dict. condition_column semantics: true = violation. + + Shared strict/between fallback across library metrics (rowCount, nullValues, + duplicateValues, ...). row_filter, when provided, narrows the view to the rows the query's + aggregate should be scoped to (e.g. only the null rows for a nullValues check). + """ + arguments: dict[str, Any] = {"query": query, "condition_column": "condition"} + if row_filter is not None: + arguments["row_filter"] = row_filter + return { + "function": "sql_query", + "arguments": arguments, + } + + # duplicateValues: single-property (argument-less) and composite (arguments.properties) forms + # share one mapping, since is_unique's `columns` argument already accepts a list. mustBe: 0 maps + # directly to is_unique; every other threshold routes through the sql_query escape hatch, since + # the duplicate count must be computed via a GROUP BY subquery (see + # _duplicate_values_count_expr) rather than a window function nested in an is_aggr_* `column` + # expression: SUM(CASE WHEN ... COUNT(*) OVER (PARTITION BY ...) ...) is rejected by Spark at + # apply time (a window function can't be nested inside an aggregate function). The count itself + # matches datacontract-cli's reference mapping (Soda's duplicate_count: distinct recurring + # values, not rows-in-group) rather than our own reading of the ODCS spec text, which does not + # settle it -- see _duplicate_values_count_expr's own docstring for the full rationale. + + def _build_duplicate_values_rules( + self, + quality_rule: DataQuality, + property_name: str | None, + schema_name: str, + odcs: OpenDataContractStandard, + default_criticality: str, + ) -> list[dict]: + """Build the DQX dataset-level uniqueness rule for an ODCS duplicateValues library entry. + + property_name is set for the property-level, single-column, argument-less form; None for + the schema-level entry, whose composite key is read from arguments.properties. + """ + key_columns = self._duplicate_values_key_columns(quality_rule, property_name, schema_name) + if key_columns is None: + return [] + + resolved = self._duplicate_values_check(quality_rule, key_columns, schema_name) + if resolved is None: + return [] + threshold_field, check_dict = resolved + + user_metadata: dict[str, Any] = { + "contract_id": odcs.id or "unknown", + "contract_version": odcs.version or "unknown", + "odcs_version": odcs.apiVersion or "unknown", + "schema": schema_name, + "rule_type": "metric", + "metric": "duplicateValues", + "threshold_field": threshold_field, + "unit": quality_rule.unit or "rows", + "dimension": self._library_dimension(quality_rule, "duplicateValues"), + **self._library_severity_metadata(quality_rule), + } + if property_name is not None: + user_metadata["field"] = property_name + rule_name = f"{property_name}_duplicateValues" + else: + user_metadata["fields"] = key_columns + rule_name = f"{schema_name}_duplicateValues" + + return [ + { + "check": check_dict, + "name": rule_name, + "criticality": default_criticality, + "user_metadata": user_metadata, + } + ] + + def _duplicate_values_key_columns( + self, quality_rule: DataQuality, property_name: str | None, schema_name: str + ) -> list[str] | None: + """Resolve the duplicateValues key columns. + + Returns [property_name] for the property-level form. For the schema-level form + (property_name is None), reads the composite key from arguments.properties, warning and + returning None if it is missing, not a list, empty, or contains a non-string entry. + """ + if property_name is not None: + return [property_name] + + properties = self._read_library_argument(quality_rule.arguments, "properties", list) + if properties is None: + return None + if not all(isinstance(entry, str) for entry in properties): + logger.warning( + f"'arguments.properties' for type: library duplicateValues entry in schema " + f"'{sanitize_for_logging(schema_name)}' must be a list of property name strings; " + "skipping this quality check." + ) + return None + return properties + + def _duplicate_values_check( + self, quality_rule: DataQuality, key_columns: list[str], schema_name: str + ) -> tuple[str, dict] | None: + """Resolve the first set ODCS threshold field on a duplicateValues entry to (field name, check dict). + + mustBe: 0 maps directly to is_unique (unit-independent: 0 rows = 0% either way). Every other + threshold requires a resolved unit (rows/percent) and routes through the sql_query escape + hatch, keyed on the GROUP BY-based duplicate count/percentage from + _duplicate_values_count_expr (see the class comment above this metric's section for why a + plain is_aggr_* aggregate can't be used here). Returns None (after logging) when no + threshold field is set, or when unit is missing/unrecognized. + """ + if self._is_zero_threshold(quality_rule.mustBe): + return "mustBe", {"function": "is_unique", "arguments": {"columns": key_columns}} + + if not self._has_any_threshold_field(quality_rule): + logger.warning( + f"type: library duplicateValues entry in schema '{sanitize_for_logging(schema_name)}' has no " + "recognized threshold field set (mustBe, mustNotBe, mustBeGreaterOrEqualTo, mustBeLessOrEqualTo, " + "mustBeGreaterThan, mustBeLessThan, mustBeBetween, mustNotBeBetween); skipping this quality check." + ) + return None + + unit = self._resolve_duplicate_values_unit(quality_rule, schema_name) + if unit is None: + return None + + count_expr = self._duplicate_values_count_expr(key_columns, unit) + + # count_expr is a self-contained scalar subquery (it references {{ input_view }} itself + # for its own GROUP BY), so the comparison is selected with no outer FROM at all -- adding + # one (e.g. `FROM {{ input_view }}`) would re-introduce a row per input row and trip + # sql_query's "dataset-level query must return exactly one row" check. + if quality_rule.mustBe is not None: + return "mustBe", self._library_sql_query_check(f"SELECT {count_expr} <> {quality_rule.mustBe} AS condition") + if quality_rule.mustNotBe is not None: + return "mustNotBe", self._library_sql_query_check( + f"SELECT {count_expr} = {quality_rule.mustNotBe} AS condition" + ) + if quality_rule.mustBeGreaterOrEqualTo is not None: + return "mustBeGreaterOrEqualTo", self._library_sql_query_check( + f"SELECT {count_expr} < {quality_rule.mustBeGreaterOrEqualTo} AS condition" + ) + if quality_rule.mustBeLessOrEqualTo is not None: + return "mustBeLessOrEqualTo", self._library_sql_query_check( + f"SELECT {count_expr} > {quality_rule.mustBeLessOrEqualTo} AS condition" + ) + if quality_rule.mustBeGreaterThan is not None: + return "mustBeGreaterThan", self._library_sql_query_check( + f"SELECT {count_expr} <= {quality_rule.mustBeGreaterThan} AS condition" + ) + if quality_rule.mustBeLessThan is not None: + return "mustBeLessThan", self._library_sql_query_check( + f"SELECT {count_expr} >= {quality_rule.mustBeLessThan} AS condition" + ) + if quality_rule.mustBeBetween is not None: + min_val, max_val = quality_rule.mustBeBetween + return "mustBeBetween", self._library_sql_query_check( + f"SELECT NOT ({count_expr} >= {min_val} AND {count_expr} <= {max_val}) AS condition" + ) + if quality_rule.mustNotBeBetween is not None: + min_val, max_val = quality_rule.mustNotBeBetween + return "mustNotBeBetween", self._library_sql_query_check( + f"SELECT ({count_expr} >= {min_val} AND {count_expr} <= {max_val}) AS condition" + ) + + # Unreachable: _has_any_threshold_field guarantees one of the eight fields above is set. + raise AssertionError("Unreachable: no duplicateValues threshold field matched despite a prior check.") + + @classmethod + def _has_any_threshold_field(cls, quality_rule: DataQuality) -> bool: + """Return True if any of the eight ODCS threshold fields is set on quality_rule.""" + return any(getattr(quality_rule, field) is not None for field in cls._THRESHOLD_FIELDS) + + @staticmethod + def _is_zero_threshold(value: Any) -> bool: # value: mustBe is Any-typed on the ODCS DataQuality model + """Return True only for a genuine numeric zero threshold. + + A plain `value == 0` would also match the boolean `False` (`False == 0` is `True` in + Python) and would miss a numeric string like `"0"`. Bool is checked before the numeric + branch since `bool` is a subclass of `int`. + """ + if isinstance(value, bool): + return False + if isinstance(value, (int, float)): + return value == 0 + if isinstance(value, str): + try: + return float(value) == 0 + except ValueError: + return False + return False + + def _resolve_duplicate_values_unit(self, quality_rule: DataQuality, schema_name: str) -> str | None: + """Resolve the unit for a non-zero duplicateValues threshold: 'rows' (default) or 'percent'. + + Warns and returns None for any other value (unrecognized unit, general library-metric policy). + """ + unit = quality_rule.unit or "rows" + if unit not in ("rows", "percent"): + logger.warning( + f"Unrecognized unit '{sanitize_for_logging(unit)}' on type: library duplicateValues entry in " + f"schema '{sanitize_for_logging(schema_name)}'; expected 'rows' or 'percent'. Skipping this " + "quality check." + ) + return None + return unit + + @classmethod + def _duplicate_values_count_expr(cls, key_columns: list[str], unit: str) -> str: + """Build a scalar SQL expression for the duplicate value count (unit: rows) or percentage + (unit: percent) of key_columns. + + Matches datacontract-cli's reference mapping of duplicateValues to Soda's + duplicate_count: the number of *distinct values (key combinations) that recur*, not the + number of rows sitting in a duplicated group -- for `[A, A, A, B, B, C]` this is 2 (A and + B recur), not 5 (rows in a duplicated group). Percent divides that count by the total row + count, again matching the reference (`duplicate_count * 100 / row_count`), not by the + number of duplicated rows. + + Computed via a GROUP BY ... HAVING subquery -- a genuine aggregate, not a window function + -- so it can be safely nested inside the enclosing sql_query comparison. An earlier version + built this indicator with `COUNT(*) OVER (PARTITION BY ...)` passed as an is_aggr_* `column` + argument; wrapping that in SUM/AVG produces a window function nested inside an aggregate + function, which Spark rejects at apply time for every threshold except mustBe: 0. A row in + an all-non-null key group is only "in the group" with rows sharing its identical non-null + key, so gating the GROUP BY's source rows on every key column being non-null is sufficient. + Every key column is quoted via _safe_sql_identifier before interpolation. + """ + quoted = [cls._safe_sql_identifier(col) for col in key_columns] + not_null_clause = " AND ".join(f"{col} IS NOT NULL" for col in quoted) + partition_by = ", ".join(quoted) + duplicate_values = ( + f"(SELECT COUNT(*) FROM (SELECT 1 FROM {{{{ input_view }}}} WHERE {not_null_clause} " + f"GROUP BY {partition_by} HAVING COUNT(*) > 1) AS dqx_dup_groups)" + ) + if unit == "rows": + return duplicate_values + return f"(100.0 * {duplicate_values} / NULLIF((SELECT COUNT(*) FROM {{{{ input_view }}}}), 0))" + + # nullValues ODCS threshold field -> (threshold_field, DQX check dict) builder, tried in order. + # mustBe: 0 is a special case mapping onto the row-level is_not_null check (cheaper, pinpoints + # the offending row, and is unit-independent: 0 nulls and 0% nulls are the same fact). Every + # other threshold (including mustBe with N > 0) maps onto a dataset-level null-count/percentage + # aggregate for the entry's unit: rows (the default, count(*) over rows where the column IS + # NULL) or percent (AVG of a CASE WHEN ... SQL string indicator, since is_aggr_*'s row_filter+"*" + # mechanism can only count rows, not express a percentage). Strict inequalities and both range + # forms have no aggregate equivalent and fall back to the dataset-level sql_query escape hatch + # for both units. mustBeBetween/mustNotBeBetween treat both bounds as inclusive, matching + # datacontract-cli's reference mapping rather than our own reading of the spec text (see the + # rowCount section above for the full reasoning). + + def _build_nullvalues_rules( + self, + quality_rule: DataQuality, + property_name: str | None, + schema_name: str, + odcs: OpenDataContractStandard, + default_criticality: str, + ) -> list[dict]: + """Build the single DQX rule for an ODCS nullValues library entry. + + nullValues is a property-level ODCS metric; a schema-level entry (no property) is skipped. + """ + if not property_name: + logger.warning( + f"type: library nullValues entry in schema '{sanitize_for_logging(schema_name)}' has no " + "property; nullValues is a property-level metric. Skipping this quality check." + ) + return [] + + resolved = self._nullvalues_check(quality_rule, property_name, schema_name) + if resolved is None: + return [] + threshold_field, check_dict = resolved + + user_metadata = { + "contract_id": odcs.id or "unknown", + "contract_version": odcs.version or "unknown", + "odcs_version": odcs.apiVersion or "unknown", + "schema": schema_name, + "rule_type": "metric", + "metric": "nullValues", + "threshold_field": threshold_field, + "unit": quality_rule.unit or "rows", + "dimension": self._library_dimension(quality_rule, "nullValues"), + "field": property_name, + **self._library_severity_metadata(quality_rule), + } + + return [ + { + "check": check_dict, + "name": f"{property_name}_nullValues", + "criticality": default_criticality, + "user_metadata": user_metadata, + } + ] + + def _nullvalues_check( + self, quality_rule: DataQuality, property_name: str, schema_name: str + ) -> tuple[str, dict] | None: + """Resolve the first set ODCS threshold field on a nullValues entry to (field name, check dict). + + mustBe: 0 is checked before consulting unit at all. Every other threshold routes through + the null-count (unit: rows) or null-percentage (unit: percent) mechanism. Returns None + (after logging) when no threshold field is set, or when unit is missing/unrecognized. + """ + if self._is_zero_threshold(quality_rule.mustBe): + return "mustBe", {"function": "is_not_null", "arguments": {"column": property_name}} + + if not self._has_any_threshold_field(quality_rule): + logger.warning( + f"type: library nullValues entry on property '{sanitize_for_logging(property_name)}' in schema " + f"'{sanitize_for_logging(schema_name)}' has no recognized threshold field set (mustBe, mustNotBe, " + "mustBeGreaterOrEqualTo, mustBeLessOrEqualTo, mustBeGreaterThan, mustBeLessThan, mustBeBetween, " + "mustNotBeBetween); skipping this quality check." + ) + return None + + unit = quality_rule.unit or "rows" + if unit not in ("rows", "percent"): + logger.warning( + f"Unrecognized unit '{sanitize_for_logging(unit)}' on type: library nullValues entry on " + f"property '{sanitize_for_logging(property_name)}' in schema " + f"'{sanitize_for_logging(schema_name)}'; expected 'rows' or 'percent'. Skipping this quality " + "check." + ) + return None + + quoted_column = self._safe_sql_identifier(property_name) + if unit == "percent": + return self._nullvalues_percent_check(quality_rule, quoted_column) + return self._nullvalues_rows_check(quality_rule, quoted_column) + + def _nullvalues_rows_check(self, quality_rule: DataQuality, quoted_column: str) -> tuple[str, dict]: + """Build the unit: rows (default) null-count check: count(*) over rows where column IS NULL.""" + row_filter = f"{quoted_column} IS NULL" + + if quality_rule.mustBe is not None: + return "mustBe", self._nullvalues_aggregate_check("is_aggr_equal", row_filter, quality_rule.mustBe) + if quality_rule.mustNotBe is not None: + return "mustNotBe", self._nullvalues_aggregate_check( + "is_aggr_not_equal", row_filter, quality_rule.mustNotBe + ) + if quality_rule.mustBeGreaterOrEqualTo is not None: + return "mustBeGreaterOrEqualTo", self._nullvalues_aggregate_check( + "is_aggr_not_less_than", row_filter, quality_rule.mustBeGreaterOrEqualTo + ) + if quality_rule.mustBeLessOrEqualTo is not None: + return "mustBeLessOrEqualTo", self._nullvalues_aggregate_check( + "is_aggr_not_greater_than", row_filter, quality_rule.mustBeLessOrEqualTo + ) + if quality_rule.mustBeGreaterThan is not None: + return "mustBeGreaterThan", self._library_sql_query_check( + f"SELECT COUNT(*) <= {quality_rule.mustBeGreaterThan} AS condition FROM {{{{ input_view }}}}", + row_filter=row_filter, + ) + if quality_rule.mustBeLessThan is not None: + return "mustBeLessThan", self._library_sql_query_check( + f"SELECT COUNT(*) >= {quality_rule.mustBeLessThan} AS condition FROM {{{{ input_view }}}}", + row_filter=row_filter, + ) + if quality_rule.mustBeBetween is not None: + min_val, max_val = quality_rule.mustBeBetween + return "mustBeBetween", self._library_sql_query_check( + f"SELECT NOT (COUNT(*) >= {min_val} AND COUNT(*) <= {max_val}) AS condition FROM {{{{ input_view }}}}", + row_filter=row_filter, + ) + assert quality_rule.mustNotBeBetween is not None # only remaining field per _has_any_threshold_field + min_val, max_val = quality_rule.mustNotBeBetween + return "mustNotBeBetween", self._library_sql_query_check( + f"SELECT (COUNT(*) >= {min_val} AND COUNT(*) <= {max_val}) AS condition FROM {{{{ input_view }}}}", + row_filter=row_filter, + ) + + def _nullvalues_percent_check(self, quality_rule: DataQuality, quoted_column: str) -> tuple[str, dict]: + """Build the unit: percent null-percentage check. + + Uses a CASE WHEN indicator SQL expression (100.0 when null, else 0.0), passed as the + *column* argument (resolved via F.expr at check-execution time), with aggr_type="avg" for + the exact-fit thresholds -- is_aggr_*'s row_filter+"*" mechanism can only count rows, not + express a percentage. A raw Column object is deliberately avoided here: it can't round-trip + through YAML/JSON (breaking save_checks()), and ChecksSemanticValidator's conflict-key + hashing isn't Column-aware either. A string routes through F.expr() at apply time instead, + producing the identical Spark expression without either problem. No row_filter is used for + the percentage fallback either, so the denominator stays the full row count rather than + shrinking to just the null rows. + """ + indicator_sql = f"CASE WHEN {quoted_column} IS NULL THEN 100.0 ELSE 0.0 END" + + if quality_rule.mustBe is not None: + return "mustBe", self._nullvalues_percent_aggregate_check( + "is_aggr_equal", indicator_sql, quality_rule.mustBe + ) + if quality_rule.mustNotBe is not None: + return "mustNotBe", self._nullvalues_percent_aggregate_check( + "is_aggr_not_equal", indicator_sql, quality_rule.mustNotBe + ) + if quality_rule.mustBeGreaterOrEqualTo is not None: + return "mustBeGreaterOrEqualTo", self._nullvalues_percent_aggregate_check( + "is_aggr_not_less_than", indicator_sql, quality_rule.mustBeGreaterOrEqualTo + ) + if quality_rule.mustBeLessOrEqualTo is not None: + return "mustBeLessOrEqualTo", self._nullvalues_percent_aggregate_check( + "is_aggr_not_greater_than", indicator_sql, quality_rule.mustBeLessOrEqualTo + ) + + percent_expr = f"AVG(CASE WHEN {quoted_column} IS NULL THEN 100.0 ELSE 0.0 END)" + if quality_rule.mustBeGreaterThan is not None: + return "mustBeGreaterThan", self._library_sql_query_check( + f"SELECT {percent_expr} <= {quality_rule.mustBeGreaterThan} AS condition FROM {{{{ input_view }}}}" + ) + if quality_rule.mustBeLessThan is not None: + return "mustBeLessThan", self._library_sql_query_check( + f"SELECT {percent_expr} >= {quality_rule.mustBeLessThan} AS condition FROM {{{{ input_view }}}}" + ) + if quality_rule.mustBeBetween is not None: + min_val, max_val = quality_rule.mustBeBetween + return "mustBeBetween", self._library_sql_query_check( + f"SELECT NOT ({percent_expr} >= {min_val} AND {percent_expr} <= {max_val}) " + f"AS condition FROM {{{{ input_view }}}}" + ) + assert quality_rule.mustNotBeBetween is not None # only remaining field per _has_any_threshold_field + min_val, max_val = quality_rule.mustNotBeBetween + return "mustNotBeBetween", self._library_sql_query_check( + f"SELECT ({percent_expr} >= {min_val} AND {percent_expr} <= {max_val}) AS condition FROM {{{{ input_view }}}}" + ) + + @staticmethod + def _nullvalues_aggregate_check(function: str, row_filter: str, limit: Any) -> dict: + """Build a dataset-level null-count aggregate check dict (is_aggr_equal / is_aggr_not_equal / etc.).""" + return { + "function": function, + "arguments": {"column": "*", "limit": limit, "aggr_type": "count", "row_filter": row_filter}, + } + + @staticmethod + def _nullvalues_percent_aggregate_check(function: str, indicator_sql: str, limit: Any) -> dict: + """Build a dataset-level null-percentage aggregate check dict (is_aggr_equal / is_aggr_not_equal / etc.).""" + return { + "function": function, + "arguments": {"column": indicator_sql, "limit": limit, "aggr_type": "avg"}, + } + + # invalidValues ODCS threshold field -> (threshold_field, DQX check dict) builder, tried in + # order. mustBe: 0 maps onto row-level is_in_list / regex_match checks -- one per present + # mechanism (validValues allowlist and/or pattern), OR'd via DQX's own per-row + # _errors/_warnings union when both are present. Every other threshold routes through a + # dataset-level aggregate over the OR'd "invalid" SQL condition (NOT IN the allowlist and/or + # NOT RLIKE the pattern), for the entry's unit: rows (row_filter+"*"+count, matching + # nullValues) or percent (a CASE WHEN indicator SQL string with aggr_type="avg", since + # row_filter+"*" can't express a percentage -- a raw Column indicator is deliberately avoided, + # since DQEngine.validate_checks' semantic conflict detection isn't Column-aware and raises on + # one, the same issue missingValues works around below). Strict inequalities and both range + # forms fall back to the dataset-level sql_query escape hatch for both units. + # mustBeBetween/mustNotBeBetween treat both bounds as inclusive, matching datacontract-cli's + # reference mapping rather than our own reading of the spec text (see the rowCount section + # above for the full reasoning). validValues and pattern, when both present, are combined into + # a single OR'd "invalid" condition rather than two independent thresholds, so "at most N + # invalid" means at most N rows failing either criterion, not N failing each independently. + + def _build_invalid_values_rules( + self, + quality_rule: DataQuality, + property_name: str | None, + schema_name: str, + odcs: OpenDataContractStandard, + default_criticality: str, + ) -> list[dict]: + """Build the DQX rule(s) for an ODCS invalidValues library entry. + + invalidValues is a property-level ODCS metric; a schema-level entry (no property) is skipped. + """ + if not property_name: + logger.warning( + f"type: library invalidValues entry in schema '{sanitize_for_logging(schema_name)}' has no " + "property; invalidValues is a property-level metric. Skipping this quality check." + ) + return [] + + valid_values, pattern = self._invalid_values_arguments(quality_rule, property_name, schema_name) + if valid_values is None and pattern is None: + return [] + + contract_metadata = { + "contract_id": odcs.id or "unknown", + "contract_version": odcs.version or "unknown", + "odcs_version": odcs.apiVersion or "unknown", + "schema": schema_name, + "rule_type": "metric", + "metric": "invalidValues", + "unit": quality_rule.unit or "rows", + "dimension": self._library_dimension(quality_rule, "invalidValues"), + "field": property_name, + **self._library_severity_metadata(quality_rule), + } + + if self._is_zero_threshold(quality_rule.mustBe): + return self._invalid_values_row_level_rules( + valid_values, pattern, property_name, contract_metadata, default_criticality + ) + + resolved = self._invalid_values_check(quality_rule, valid_values, pattern, property_name, schema_name) + if resolved is None: + return [] + threshold_field, check_dict = resolved + + return [ + { + "check": check_dict, + "name": f"{property_name}_invalidValues", + "criticality": default_criticality, + "user_metadata": {**contract_metadata, "threshold_field": threshold_field}, + } + ] + + def _invalid_values_arguments( + self, quality_rule: DataQuality, property_name: str, schema_name: str + ) -> tuple[list | None, str | None]: + """Read arguments.validValues and/or arguments.pattern for an invalidValues library entry. + + Either, neither, or both may be present. A syntactically valid pattern that fails the + generation-time ReDoS safety guard is treated as absent (its own warning is logged) + rather than as a malformed-arguments error. Warns once more, generically, only when + neither mechanism yields a usable value overall. + """ + arguments = quality_rule.arguments + + valid_values: list | None = None + if isinstance(arguments, dict) and "validValues" in arguments: + valid_values = self._read_library_argument(arguments, "validValues", list) + + pattern: str | None = None + if isinstance(arguments, dict) and "pattern" in arguments: + raw_pattern = self._read_library_argument(arguments, "pattern", str) + if raw_pattern is not None: + if self._is_library_pattern_safe(raw_pattern): + pattern = raw_pattern + else: + logger.warning( + f"'arguments.pattern' on property '{sanitize_for_logging(property_name)}' in schema " + f"'{sanitize_for_logging(schema_name)}' failed the ReDoS safety guard (200-character " + "cap, nested-quantifier check, alternation-quantifier check, or re.compile); skipping " + "the pattern-based check for this entry." + ) + + if valid_values is None and pattern is None: + logger.warning( + f"type: library invalidValues entry on property '{sanitize_for_logging(property_name)}' in " + f"schema '{sanitize_for_logging(schema_name)}' has neither a usable 'arguments.validValues' " + "list nor a usable 'arguments.pattern' string; skipping this quality check." + ) + + return valid_values, pattern + + def _invalid_values_row_level_rules( + self, + valid_values: list | None, + pattern: str | None, + property_name: str, + contract_metadata: dict, + default_criticality: str, + ) -> list[dict]: + """Build the row-level rule(s) for mustBe: 0 -- one per present mechanism (validValues + allowlist and/or pattern). Emitting both as separate rules (sharing identical + user_metadata, only *name* differs) reproduces "invalid if either criterion fails" via + DQX's own per-row _errors/_warnings union, the same OR-of-rules treatment missingValues + uses for its null/sentinel split. + """ + user_metadata = {**contract_metadata, "threshold_field": "mustBe"} + rules = [] + if valid_values is not None: + rules.append( + { + "check": { + "function": "is_in_list", + "arguments": { + "column": property_name, + "allowed": [self._is_in_list_literal(value) for value in valid_values], + "case_sensitive": True, + }, + }, + "name": f"{property_name}_invalidValues_allowed", + "criticality": default_criticality, + "user_metadata": dict(user_metadata), + } + ) + if pattern is not None: + rules.append( + { + "check": {"function": "regex_match", "arguments": {"column": property_name, "regex": pattern}}, + "name": f"{property_name}_invalidValues_pattern", + "criticality": default_criticality, + "user_metadata": dict(user_metadata), + } + ) + return rules + + @staticmethod + def _is_in_list_literal(value: Any) -> Any: # value/return: any contract-supplied scalar (str, number, bool) + """Render a validValues entry as an is_in_list *allowed* list literal. + + is_in_list resolves each *allowed* entry like a comparison-check limit: a bare string is + parsed as a **column expression** via F.expr(), not a string literal (see + check_funcs.is_in_list's own docstring) -- so a contract-supplied string value must be + single-quoted to compare correctly, with embedded backslashes doubled first (then embedded + single quotes), matching _sql_scalar_literal: with Spark's default + spark.sql.parser.escapedStringLiterals=false, an unescaped backslash in a SQL string + literal is not preserved as-is, which would otherwise disagree with the aggregate paths' + NOT IN condition for the same sentinel value. Non-string values are passed through + unchanged; get_limit_expr already resolves them via F.lit(). + """ + if isinstance(value, str): + escaped = value.replace("\\", "\\\\").replace("'", "''") + return f"'{escaped}'" + return value + + def _invalid_values_check( + self, + quality_rule: DataQuality, + valid_values: list | None, + pattern: str | None, + property_name: str, + schema_name: str, + ) -> tuple[str, dict] | None: + """Resolve a non-zero-threshold invalidValues entry to (threshold_field, check dict). + + Routes through the invalid-row-count (unit: rows) or invalid-row-percentage (unit: + percent) mechanism. Returns None (after logging) when no threshold field is set, or when + unit is missing/unrecognized. + """ + if not self._has_any_threshold_field(quality_rule): + logger.warning( + f"type: library invalidValues entry on property '{sanitize_for_logging(property_name)}' in " + f"schema '{sanitize_for_logging(schema_name)}' has no recognized threshold field set (mustBe, " + "mustNotBe, mustBeGreaterOrEqualTo, mustBeLessOrEqualTo, mustBeGreaterThan, mustBeLessThan, " + "mustBeBetween, mustNotBeBetween); skipping this quality check." + ) + return None + + unit = quality_rule.unit or "rows" + if unit not in ("rows", "percent"): + logger.warning( + f"Unrecognized unit '{sanitize_for_logging(unit)}' on type: library invalidValues entry on " + f"property '{sanitize_for_logging(property_name)}' in schema " + f"'{sanitize_for_logging(schema_name)}'; expected 'rows' or 'percent'. Skipping this quality " + "check." + ) + return None + + invalid_condition = self._invalid_values_condition_sql(property_name, valid_values, pattern) + if unit == "percent": + return self._invalid_values_percent_check(quality_rule, invalid_condition) + return self._invalid_values_rows_check(quality_rule, invalid_condition) + + def _invalid_values_rows_check(self, quality_rule: DataQuality, invalid_condition: str) -> tuple[str, dict] | None: + """Build the unit: rows (default) invalid-row-count check: count(*) over invalid rows. + + Always returns a check dict in practice -- callers only reach this after + _has_any_threshold_field confirms at least one of the eight fields is set -- but the + return type stays Optional so mypy can verify the final mustNotBeBetween branch without an + unreachable-code assertion. + """ + if quality_rule.mustBe is not None: + return "mustBe", self._invalid_values_aggregate_check( + "is_aggr_equal", invalid_condition, quality_rule.mustBe + ) + if quality_rule.mustNotBe is not None: + return "mustNotBe", self._invalid_values_aggregate_check( + "is_aggr_not_equal", invalid_condition, quality_rule.mustNotBe + ) + if quality_rule.mustBeGreaterOrEqualTo is not None: + return "mustBeGreaterOrEqualTo", self._invalid_values_aggregate_check( + "is_aggr_not_less_than", invalid_condition, quality_rule.mustBeGreaterOrEqualTo + ) + if quality_rule.mustBeLessOrEqualTo is not None: + return "mustBeLessOrEqualTo", self._invalid_values_aggregate_check( + "is_aggr_not_greater_than", invalid_condition, quality_rule.mustBeLessOrEqualTo + ) + if quality_rule.mustBeGreaterThan is not None: + return "mustBeGreaterThan", self._library_sql_query_check( + f"SELECT COUNT(*) <= {quality_rule.mustBeGreaterThan} AS condition FROM {{{{ input_view }}}}", + row_filter=invalid_condition, + ) + if quality_rule.mustBeLessThan is not None: + return "mustBeLessThan", self._library_sql_query_check( + f"SELECT COUNT(*) >= {quality_rule.mustBeLessThan} AS condition FROM {{{{ input_view }}}}", + row_filter=invalid_condition, + ) + if quality_rule.mustBeBetween is not None: + min_val, max_val = quality_rule.mustBeBetween + return "mustBeBetween", self._library_sql_query_check( + f"SELECT NOT (COUNT(*) >= {min_val} AND COUNT(*) <= {max_val}) AS condition FROM {{{{ input_view }}}}", + row_filter=invalid_condition, + ) + if quality_rule.mustNotBeBetween is not None: + min_val, max_val = quality_rule.mustNotBeBetween + return "mustNotBeBetween", self._library_sql_query_check( + f"SELECT (COUNT(*) >= {min_val} AND COUNT(*) <= {max_val}) AS condition FROM {{{{ input_view }}}}", + row_filter=invalid_condition, + ) + return None + + def _invalid_values_percent_check( + self, quality_rule: DataQuality, invalid_condition: str + ) -> tuple[str, dict] | None: + """Build the unit: percent invalid-row-percentage check. + + Uses a CASE WHEN indicator SQL expression (100.0 when invalid, else 0.0), passed as the + *column* argument (resolved via F.expr at check-execution time), with aggr_type="avg" for + the exact-fit thresholds -- is_aggr_*'s row_filter+"*" mechanism can only count rows, not + express a percentage. A raw Column object is deliberately avoided here: DQEngine.validate_checks' + semantic conflict detection isn't Column-aware and raises when a check argument is one. No + row_filter is used for the percentage fallback either, so the denominator stays the full + row count rather than shrinking to just the invalid rows. Always returns a check dict in + practice, per _invalid_values_rows_check's own docstring note about the Optional return type. + """ + indicator_sql = f"CASE WHEN {invalid_condition} THEN 100.0 ELSE 0.0 END" + + if quality_rule.mustBe is not None: + return "mustBe", self._invalid_values_percent_aggregate_check( + "is_aggr_equal", indicator_sql, quality_rule.mustBe + ) + if quality_rule.mustNotBe is not None: + return "mustNotBe", self._invalid_values_percent_aggregate_check( + "is_aggr_not_equal", indicator_sql, quality_rule.mustNotBe + ) + if quality_rule.mustBeGreaterOrEqualTo is not None: + return "mustBeGreaterOrEqualTo", self._invalid_values_percent_aggregate_check( + "is_aggr_not_less_than", indicator_sql, quality_rule.mustBeGreaterOrEqualTo + ) + if quality_rule.mustBeLessOrEqualTo is not None: + return "mustBeLessOrEqualTo", self._invalid_values_percent_aggregate_check( + "is_aggr_not_greater_than", indicator_sql, quality_rule.mustBeLessOrEqualTo + ) + + percent_expr = f"AVG({indicator_sql})" + if quality_rule.mustBeGreaterThan is not None: + return "mustBeGreaterThan", self._library_sql_query_check( + f"SELECT {percent_expr} <= {quality_rule.mustBeGreaterThan} AS condition FROM {{{{ input_view }}}}" + ) + if quality_rule.mustBeLessThan is not None: + return "mustBeLessThan", self._library_sql_query_check( + f"SELECT {percent_expr} >= {quality_rule.mustBeLessThan} AS condition FROM {{{{ input_view }}}}" + ) + if quality_rule.mustBeBetween is not None: + min_val, max_val = quality_rule.mustBeBetween + return "mustBeBetween", self._library_sql_query_check( + f"SELECT NOT ({percent_expr} >= {min_val} AND {percent_expr} <= {max_val}) " + f"AS condition FROM {{{{ input_view }}}}" + ) + if quality_rule.mustNotBeBetween is not None: + min_val, max_val = quality_rule.mustNotBeBetween + return "mustNotBeBetween", self._library_sql_query_check( + f"SELECT ({percent_expr} >= {min_val} AND {percent_expr} <= {max_val}) " + f"AS condition FROM {{{{ input_view }}}}" + ) + return None + + @staticmethod + def _invalid_values_aggregate_check( + function: str, + row_filter: str, + limit: Any, # limit: the contract's own mustBe/mustNotBe/etc. value (int | float) + ) -> dict: + """Build a dataset-level invalid-row-count aggregate check dict (is_aggr_equal / etc.).""" + return { + "function": function, + "arguments": {"column": "*", "limit": limit, "aggr_type": "count", "row_filter": row_filter}, + } + + @staticmethod + def _invalid_values_percent_aggregate_check( + function: str, indicator_sql: str, limit: Any # limit: the contract's own threshold value (int | float) + ) -> dict: + """Build a dataset-level invalid-row-percentage aggregate check dict (is_aggr_equal / etc.).""" + return { + "function": function, + "arguments": {"column": indicator_sql, "limit": limit, "aggr_type": "avg"}, + } + + def _invalid_values_condition_sql(self, property_name: str, valid_values: list | None, pattern: str | None) -> str: + """Build the SQL boolean condition matching an "invalid" row for property_name. + + NOT IN the allowlist and/or NOT RLIKE the pattern, OR'd together when both mechanisms are + present; each is independently null-tolerant (NULL NOT IN (...) and NOT (NULL RLIKE ...) + are both NULL, not TRUE), matching the row-level is_in_list/regex_match checks' own + null-tolerant semantics. + """ + quoted_column = self._safe_sql_identifier(property_name) + clauses = [] + if valid_values is not None: + escaped_values = ", ".join(self._sql_scalar_literal(value) for value in valid_values) + clauses.append(f"{quoted_column} NOT IN ({escaped_values})") + if pattern is not None: + clauses.append(f"NOT ({quoted_column} RLIKE {self._sql_scalar_literal(pattern)})") + return " OR ".join(clauses) + + # missingValues ODCS threshold field -> (threshold_field, DQX check dict) builder, tried in + # order. missingValues is distinct from nullValues: it combines real NULLs with a + # contract-supplied sentinel list (arguments.missingValues, e.g. [null, "", "N/A"]) read via + # _read_library_argument. mustBe: 0 splits into up to two independent row-level rules (one + # is_not_null, one is_not_in_list) -- present only for the sentinel kinds actually listed -- + # OR'd via DQX's own per-row _errors/_warnings union, the same treatment invalidValues uses + # for its allowed-list/pattern split. Every other threshold routes through a dataset-level + # aggregate over the "col IS NULL [OR col IN (...)]" SQL condition, for the entry's unit: rows + # (row_filter+"*"+count, matching nullValues/invalidValues) or percent. The percent path uses + # an `AVG(CASE WHEN ... THEN 100.0 ELSE 0.0 END)` SQL *string* expression passed as `column`, + # deliberately not an F.when(...) Column: ChecksSemanticValidator._conflict_key evaluates + # `arguments.get("column") or ...`, and Column.__bool__ raises PySparkValueError, which crashes + # rule generation outright. A string routes through F.expr() at apply time instead, producing + # the identical Spark expression without tripping that truthiness check. Strict inequalities + # and both range forms fall back to the dataset-level sql_query escape hatch for both units. + # mustBeBetween/mustNotBeBetween treat both bounds as inclusive, matching datacontract-cli's + # reference mapping rather than our own reading of the spec text (see the rowCount section + # above for the full reasoning). NULL is counted unconditionally on every threshold, including + # mustBe: 0 -- an explicit `null` entry in arguments.missingValues is redundant, not required. + + def _build_missing_values_rules( + self, + quality_rule: DataQuality, + property_name: str | None, + schema_name: str, + odcs: OpenDataContractStandard, + default_criticality: str, + ) -> list[dict]: + """Build the DQX rule(s) for an ODCS missingValues library entry. + + missingValues is a property-level ODCS metric; a schema-level entry (no property) is skipped. + """ + if not property_name: + logger.warning( + f"type: library missingValues entry in schema '{sanitize_for_logging(schema_name)}' has no " + "property; missingValues is a property-level metric. Skipping this quality check." + ) + return [] + + sentinel_list = self._read_library_argument(quality_rule.arguments, "missingValues", list) + if sentinel_list is None: + return [] + + # NULL is always counted as missing regardless of whether `null` appears in the list (it + # matches every other threshold's `col IS NULL OR col IN (...)` condition); an explicit + # `null` entry is therefore redundant, not required. + non_null_sentinels = [value for value in sentinel_list if value is not None] + + contract_metadata = { + "contract_id": odcs.id or "unknown", + "contract_version": odcs.version or "unknown", + "odcs_version": odcs.apiVersion or "unknown", + "schema": schema_name, + "rule_type": "metric", + "metric": "missingValues", + "unit": quality_rule.unit or "rows", + "dimension": self._library_dimension(quality_rule, "missingValues"), + "field": property_name, + **self._library_severity_metadata(quality_rule), + } + + if self._is_zero_threshold(quality_rule.mustBe): + return self._missing_values_row_level_rules( + non_null_sentinels, property_name, contract_metadata, default_criticality + ) + + resolved = self._missing_values_check(quality_rule, property_name, schema_name, non_null_sentinels) + if resolved is None: + return [] + threshold_field, check_dict = resolved + + return [ + { + "check": check_dict, + "name": f"{property_name}_missingValues", + "criticality": default_criticality, + "user_metadata": {**contract_metadata, "threshold_field": threshold_field}, + } + ] + + def _missing_values_row_level_rules( + self, + non_null_sentinels: list, + property_name: str, + contract_metadata: dict, + default_criticality: str, + ) -> list[dict]: + """Build the row-level rule(s) for mustBe: 0 -- an unconditional NULL check, plus a + sentinel check when non-null sentinels are present. Emitting both as separate rules + (sharing identical user_metadata, only `name` differs) reproduces "missing if either + criterion fails" via DQX's own per-row _errors/_warnings union: is_not_in_list's + `forbidden` list can never itself catch a real SQL NULL (`x IN (...)` is NULL, not TRUE, + whenever x IS NULL), so the two conditions cannot be folded into a single check. NULL is + always checked here, matching every other threshold's unconditional `col IS NULL` -- + whether `null` was explicitly listed in arguments.missingValues does not matter. + """ + user_metadata = {**contract_metadata, "threshold_field": "mustBe"} + rules = [ + { + "check": {"function": "is_not_null", "arguments": {"column": property_name}}, + "name": f"{property_name}_missingValues_null", + "criticality": default_criticality, + "user_metadata": dict(user_metadata), + } + ] + if non_null_sentinels: + rules.append( + { + "check": { + "function": "is_not_in_list", + "arguments": { + "column": property_name, + "forbidden": [self._is_in_list_literal(value) for value in non_null_sentinels], + "case_sensitive": True, + }, + }, + "name": f"{property_name}_missingValues_sentinel", + "criticality": default_criticality, + "user_metadata": dict(user_metadata), + } + ) + return rules + + def _missing_values_check( + self, quality_rule: DataQuality, property_name: str, schema_name: str, non_null_sentinels: list + ) -> tuple[str, dict] | None: + """Resolve a non-zero-threshold missingValues entry to (threshold_field, check dict). + + Routes through the missing-row-count (unit: rows) or missing-row-percentage (unit: + percent) mechanism. Returns None (after logging) when no threshold field is set, or when + unit is missing/unrecognized. + """ + if not self._has_any_threshold_field(quality_rule): + logger.warning( + f"type: library missingValues entry on property '{sanitize_for_logging(property_name)}' in " + f"schema '{sanitize_for_logging(schema_name)}' has no recognized threshold field set (mustBe, " + "mustNotBe, mustBeGreaterOrEqualTo, mustBeLessOrEqualTo, mustBeGreaterThan, mustBeLessThan, " + "mustBeBetween, mustNotBeBetween); skipping this quality check." + ) + return None + + unit = quality_rule.unit or "rows" + if unit not in ("rows", "percent"): + logger.warning( + f"Unrecognized unit '{sanitize_for_logging(unit)}' on type: library missingValues entry on " + f"property '{sanitize_for_logging(property_name)}' in schema " + f"'{sanitize_for_logging(schema_name)}'; expected 'rows' or 'percent'. Skipping this quality " + "check." + ) + return None + + missing_condition = self._missing_values_condition_sql(property_name, non_null_sentinels) + if unit == "percent": + return self._missing_values_percent_check(quality_rule, missing_condition) + return self._missing_values_rows_check(quality_rule, missing_condition) + + def _missing_values_rows_check(self, quality_rule: DataQuality, missing_condition: str) -> tuple[str, dict]: + """Build the unit: rows (default) missing-row-count check: count(*) over missing rows.""" + if quality_rule.mustBe is not None: + return "mustBe", self._missing_values_aggregate_check( + "is_aggr_equal", missing_condition, quality_rule.mustBe + ) + if quality_rule.mustNotBe is not None: + return "mustNotBe", self._missing_values_aggregate_check( + "is_aggr_not_equal", missing_condition, quality_rule.mustNotBe + ) + if quality_rule.mustBeGreaterOrEqualTo is not None: + return "mustBeGreaterOrEqualTo", self._missing_values_aggregate_check( + "is_aggr_not_less_than", missing_condition, quality_rule.mustBeGreaterOrEqualTo + ) + if quality_rule.mustBeLessOrEqualTo is not None: + return "mustBeLessOrEqualTo", self._missing_values_aggregate_check( + "is_aggr_not_greater_than", missing_condition, quality_rule.mustBeLessOrEqualTo + ) + if quality_rule.mustBeGreaterThan is not None: + return "mustBeGreaterThan", self._library_sql_query_check( + f"SELECT COUNT(*) <= {quality_rule.mustBeGreaterThan} AS condition FROM {{{{ input_view }}}}", + row_filter=missing_condition, + ) + if quality_rule.mustBeLessThan is not None: + return "mustBeLessThan", self._library_sql_query_check( + f"SELECT COUNT(*) >= {quality_rule.mustBeLessThan} AS condition FROM {{{{ input_view }}}}", + row_filter=missing_condition, + ) + if quality_rule.mustBeBetween is not None: + min_val, max_val = quality_rule.mustBeBetween + return "mustBeBetween", self._library_sql_query_check( + f"SELECT NOT (COUNT(*) >= {min_val} AND COUNT(*) <= {max_val}) AS condition FROM {{{{ input_view }}}}", + row_filter=missing_condition, + ) + assert quality_rule.mustNotBeBetween is not None # only remaining field per _has_any_threshold_field + min_val, max_val = quality_rule.mustNotBeBetween + return "mustNotBeBetween", self._library_sql_query_check( + f"SELECT (COUNT(*) >= {min_val} AND COUNT(*) <= {max_val}) AS condition FROM {{{{ input_view }}}}", + row_filter=missing_condition, + ) + + def _missing_values_percent_check(self, quality_rule: DataQuality, missing_condition: str) -> tuple[str, dict]: + """Build the unit: percent missing-row-percentage check. + + Uses an `AVG(CASE WHEN ... THEN 100.0 ELSE 0.0 END)` SQL string expression passed as the + `column` argument (not an F.when(...) Column -- see the class comment above this metric's + section for why) for the exact-fit thresholds, since is_aggr_*'s row_filter+"*" mechanism + can only count rows, not express a percentage. No row_filter is used for the percentage + fallback either, so the denominator stays the full row count rather than shrinking to just + the missing rows. + """ + indicator_expr = f"CASE WHEN {missing_condition} THEN 100.0 ELSE 0.0 END" + + if quality_rule.mustBe is not None: + return "mustBe", self._missing_values_percent_aggregate_check( + "is_aggr_equal", indicator_expr, quality_rule.mustBe + ) + if quality_rule.mustNotBe is not None: + return "mustNotBe", self._missing_values_percent_aggregate_check( + "is_aggr_not_equal", indicator_expr, quality_rule.mustNotBe + ) + if quality_rule.mustBeGreaterOrEqualTo is not None: + return "mustBeGreaterOrEqualTo", self._missing_values_percent_aggregate_check( + "is_aggr_not_less_than", indicator_expr, quality_rule.mustBeGreaterOrEqualTo + ) + if quality_rule.mustBeLessOrEqualTo is not None: + return "mustBeLessOrEqualTo", self._missing_values_percent_aggregate_check( + "is_aggr_not_greater_than", indicator_expr, quality_rule.mustBeLessOrEqualTo + ) + + percent_expr = f"AVG({indicator_expr})" + if quality_rule.mustBeGreaterThan is not None: + return "mustBeGreaterThan", self._library_sql_query_check( + f"SELECT {percent_expr} <= {quality_rule.mustBeGreaterThan} AS condition FROM {{{{ input_view }}}}" + ) + if quality_rule.mustBeLessThan is not None: + return "mustBeLessThan", self._library_sql_query_check( + f"SELECT {percent_expr} >= {quality_rule.mustBeLessThan} AS condition FROM {{{{ input_view }}}}" + ) + if quality_rule.mustBeBetween is not None: + min_val, max_val = quality_rule.mustBeBetween + return "mustBeBetween", self._library_sql_query_check( + f"SELECT NOT ({percent_expr} >= {min_val} AND {percent_expr} <= {max_val}) " + f"AS condition FROM {{{{ input_view }}}}" + ) + assert quality_rule.mustNotBeBetween is not None # only remaining field per _has_any_threshold_field + min_val, max_val = quality_rule.mustNotBeBetween + return "mustNotBeBetween", self._library_sql_query_check( + f"SELECT ({percent_expr} >= {min_val} AND {percent_expr} <= {max_val}) AS condition FROM {{{{ input_view }}}}" + ) + + @staticmethod + def _missing_values_aggregate_check( + function: str, row_filter: str, limit: Any # limit: the contract's own threshold value (int | float) + ) -> dict: + """Build a dataset-level missing-row-count aggregate check dict (is_aggr_equal / etc.).""" + return { + "function": function, + "arguments": {"column": "*", "limit": limit, "aggr_type": "count", "row_filter": row_filter}, + } + + @staticmethod + def _missing_values_percent_aggregate_check( + function: str, indicator_expr: str, limit: Any # limit: the contract's own threshold value (int | float) + ) -> dict: + """Build a dataset-level missing-row-percentage aggregate check dict (is_aggr_equal / etc.).""" + return { + "function": function, + "arguments": {"column": indicator_expr, "limit": limit, "aggr_type": "avg"}, + } + + def _missing_values_condition_sql(self, property_name: str, non_null_sentinels: list) -> str: + """Build the SQL boolean condition matching a "missing" row for property_name: real NULL, + plus an IN (...) match against any contract-supplied non-null sentinel values. + """ + quoted_column = self._safe_sql_identifier(property_name) + condition = f"{quoted_column} IS NULL" + if non_null_sentinels: + escaped_values = ", ".join(self._sql_scalar_literal(value) for value in non_null_sentinels) + condition += f" OR {quoted_column} IN ({escaped_values})" + return condition + + def _read_library_argument( + self, arguments: dict[str, Any] | None, key: str, expected_type: type[_T], *, allow_empty: bool = False + ) -> _T | None: + """Validate and read arguments[key] from a type: library quality entry's arguments dict. + + Returns None (after logging a targeted warning) if *arguments* isn't a dict, *key* is + absent, the value isn't an instance of *expected_type*, or (unless allow_empty) the value + is empty. Callers treat None as "skip this quality entry." + """ + if not isinstance(arguments, dict): + logger.warning(f"Missing or malformed 'arguments' for library metric argument '{key}'; expected a dict.") + return None + + value = arguments.get(key) + if not isinstance(value, expected_type): + logger.warning( + f"Missing or malformed 'arguments.{key}': expected {expected_type.__name__}, " + f"got {type(value).__name__}." + ) + return None + + if not allow_empty and hasattr(value, "__len__") and len(value) == 0: + logger.warning(f"'arguments.{key}' must not be empty.") + return None + + return value + + @classmethod + def _safe_sql_identifier(cls, column_path: str) -> str: + """Quote column_path for safe interpolation into a row_filter/sql_query/PARTITION BY fragment. + + Splits on '.', backtick-quotes any segment that isn't a bare identifier + (`^[a-zA-Z_][a-zA-Z0-9_]*$`), escaping any embedded backtick first, and rejoins. Every + input is handled by quoting rather than rejecting, so this never raises. + """ + segments = column_path.split(".") + quoted_segments = [ + segment if cls._SAFE_SQL_IDENTIFIER_PATTERN.match(segment) else f"`{segment.replace('`', '``')}`" + for segment in segments + ] + return ".".join(quoted_segments) + + @staticmethod + def _sql_scalar_literal(value: Any) -> str: # value: any contract-supplied scalar (str, number, bool) + """Render a contract-supplied scalar as a SQL literal for safe interpolation into an + invalidValues/missingValues NOT IN/IN/RLIKE condition. + + A number passes through unquoted (e.g. `123`, `1.5`) so a numeric column compares + numerically, matching the row-level is_in_list/regex_match path's own handling of + `arguments.validValues` (see _is_in_list_literal). Bool is checked before the numeric + branch since `bool` is a subclass of `int` in Python. Every other value (in practice, + always a string) is rendered as a single-quoted SQL string literal: embedded backslashes + are doubled first, then embedded single quotes, so a literal backslash in the value (e.g. + a RLIKE pattern such as `\\d+`) survives Spark's string-literal escape processing intact -- + with Spark's default spark.sql.parser.escapedStringLiterals=false, an unescaped backslash + in a SQL string literal is not preserved as-is. + """ + if isinstance(value, bool): + return "TRUE" if value else "FALSE" + if isinstance(value, (int, float)): + return str(value) + escaped = str(value).replace("\\", "\\\\").replace("'", "''") + return f"'{escaped}'" + + @classmethod + def _is_library_pattern_safe(cls, pattern: str) -> bool: + """Best-effort, generation-time-only ReDoS guard for a contract-supplied arguments.pattern. + + Rejects patterns over 200 characters, nested-quantifier (`(a+)+`-shaped) or + alternation-plus-quantifier (`(a|a)*`-shaped) structures, and anything that fails to + compile. This is a heuristic, not a formal linear-time guarantee. + """ + if len(pattern) > cls._MAX_LIBRARY_PATTERN_LENGTH: + return False + if cls._NESTED_QUANTIFIER_PATTERN.search(pattern) or cls._ALTERNATION_QUANTIFIER_PATTERN.search(pattern): + return False + try: + re.compile(pattern) + except re.error: + return False + return True + + def _library_dimension(self, quality_rule: DataQuality, metric: str) -> str: + """Return the contract's own dimension when present, else the per-metric ODCS default.""" + return quality_rule.dimension or self._LIBRARY_METRIC_DEFAULT_DIMENSIONS[metric] + + @staticmethod + def _library_severity_metadata(quality_rule: DataQuality) -> dict[str, str]: + """Return {"severity": ...} verbatim when the contract sets DataQuality.severity, else {}. + + ODCS defines no vocabulary for severity, so it is recorded as-is for audit purposes and + never used to derive Criticality. + """ + if quality_rule.severity: + return {"severity": quality_rule.severity} + return {} diff --git a/tests/integration/test_datacontract_integration.py b/tests/integration/test_datacontract_integration.py index f87b87b7c..f2c4c1d8b 100644 --- a/tests/integration/test_datacontract_integration.py +++ b/tests/integration/test_datacontract_integration.py @@ -17,7 +17,7 @@ import pytest import yaml from datacontract.data_contract import DataContract -from pyspark.sql import SparkSession +from pyspark.sql import Row, SparkSession from pyspark.sql import types as spark_types from databricks.sdk import WorkspaceClient @@ -48,6 +48,21 @@ def _generate_rules_from_temp_contract( pass +def _flagged_names(row: Row) -> set[str]: + """Return the set of check names present in a row's _errors array (empty set if none).""" + errors = row["_errors"] + return {e["name"] for e in errors} if errors else set() + + +def _assert_flagged_only(flagged_by_row: dict[int, set[str]], check_name: str, expected_row_ids: set[int]) -> None: + """Assert check_name is flagged for exactly expected_row_ids among flagged_by_row's rows.""" + for row_id, names in flagged_by_row.items(): + if row_id in expected_row_ids: + assert check_name in names, f"row {row_id}: expected '{check_name}' violation" + else: + assert check_name not in names, f"row {row_id}: unexpected '{check_name}' violation" + + @contextlib.contextmanager def _temp_contract_file(content: str): """Write raw content to a temp YAML file and yield its path. Cleans up on exit.""" @@ -714,3 +729,139 @@ def test_error_unknown_physical_type_raises(self, ws, spark): ) msg = str(exc_info.value).lower() assert "not a valid" in msg or "unity catalog" in msg or "NOT_A_UC_TYPE" in str(exc_info.value) + + +class TestLibraryMetricsIntegration: + """End-to-end integration coverage for ODCS `type: library` quality metric rule generation.""" + + @pytest.fixture + def library_metrics_contract_path(self): + """Path to the type: library quality metrics fixture contract.""" + tests_dir = os.path.dirname(os.path.dirname(__file__)) + return os.path.join(tests_dir, "resources", "sample_datacontract_library_metrics.yaml") + + def test_generate_and_apply_library_metric_rules_end_to_end(self, ws, spark, library_metrics_contract_path): + """Generate rules from a contract covering all five library metrics and apply them to real data.""" + generator = DQGenerator(workspace_client=ws, spark=spark) + rules = generator.generate_rules_from_contract( + contract_file=library_metrics_contract_path, + generate_predefined_rules=False, + process_text_rules=False, + generate_schema_validation=False, + ) + + status = DQEngine.validate_checks(rules) + assert not status.has_errors, f"Generated rules have validation errors: {status.errors}" + + generated_metrics = {r["user_metadata"]["metric"] for r in rules} + assert generated_metrics == {"rowCount", "nullValues", "missingValues", "invalidValues", "duplicateValues"} + + missing_values_rule = next(r for r in rules if r["user_metadata"]["metric"] == "missingValues") + assert ( + missing_values_rule["check"]["function"] == "sql_query" + ), "missingValues mustBeGreaterThan has no aggregate equivalent and must fall back to sql_query" + + schema = "row_id: int, customer_id: string, email: string, status: string, phone: string" + test_df = spark.createDataFrame( + [ + [1, "C1", "a@example.com", "ACTIVE", "111-111-1111"], + [2, "C2", None, "ACTIVE", "222-222-2222"], + [3, "C2", "c@example.com", "ACTIVE", "333-333-3333"], + [4, "C3", "d@example.com", "PENDING", None], + [5, "C4", "e@example.com", "INACTIVE", ""], + ], + schema, + ) + + dq_engine = DQEngine(workspace_client=ws) + checked = dq_engine.apply_checks_by_metadata(test_df, rules) + + flagged_by_row = {row["row_id"]: _flagged_names(row) for row in checked.collect()} + + # rowCount (mustBeGreaterOrEqualTo: 10, no row_filter) is a dataset-wide aggregate: the + # contract's 5-row dataset fails count >= 10, so every row is flagged with it. + _assert_flagged_only(flagged_by_row, "customers_rowCount_mustBeGreaterOrEqualTo", {1, 2, 3, 4, 5}) + + # nullValues (mustBe: 0) is row-level: only the row with a null email is flagged. + _assert_flagged_only(flagged_by_row, "email_nullValues", {2}) + + # invalidValues (mustBe: 0) is row-level: only the row with an out-of-list status is flagged. + _assert_flagged_only(flagged_by_row, "status_invalidValues_allowed", {4}) + + # duplicateValues (mustBe: 0) flags only the rows sharing the duplicated customer_id. + _assert_flagged_only(flagged_by_row, "customer_id_duplicateValues", {2, 3}) + + # missingValues (mustBeGreaterThan: 3, sql_query fallback): the contract's 2 missing phone + # values don't exceed 3, so the dataset-level condition is a violation -- but the sql_query + # row_filter restricts the broadcast result to just the rows matching the missing condition. + _assert_flagged_only(flagged_by_row, "phone_missingValues", {4, 5}) + + # Row 1 satisfies every property-level metric; only the dataset-wide rowCount check applies. + assert flagged_by_row[1] == {"customers_rowCount_mustBeGreaterOrEqualTo"} + + def test_apply_non_zero_duplicate_values_threshold_does_not_raise(self, ws, spark): + """A non-zero duplicateValues threshold must actually apply against a real DataFrame + without raising an AnalysisException, and must count distinct recurring values (matching + datacontract-cli's reference mapping), not rows sitting in a duplicated group. + + The duplicate count is computed via a GROUP BY ... HAVING subquery selected with no outer + FROM (rather than a window function passed as an is_aggr_* `column` argument, or a scalar + subquery re-selected FROM the input view, both of which raise at apply time -- the window + form because Spark rejects a window function nested inside an aggregate, the outer-FROM + form because sql_query's dataset-level check requires exactly one result row). Every + non-mustBe:0 duplicateValues threshold hits this path, and it was previously only ever + asserted as a generated SQL string in unit tests, never executed. + """ + contract = { + "kind": "DataContract", + "apiVersion": "v3.0.2", + "id": "test:duplicate_values_non_zero", + "name": "Duplicate Values Non-Zero Threshold", + "version": "1.0.0", + "status": "active", + "schema": [ + { + "name": "sales_orders", + "physicalType": "table", + "properties": [ + { + "name": "order_ref", + "physicalType": "STRING", + "quality": [ + {"type": "library", "metric": "duplicateValues", "mustBeLessOrEqualTo": 4}, + ], + }, + ], + }, + ], + } + rules = _generate_rules_from_temp_contract( + ws, + spark, + contract, + generate_predefined_rules=False, + process_text_rules=False, + generate_schema_validation=False, + ) + assert len(rules) == 1 + assert rules[0]["check"]["function"] == "sql_query" + + status = DQEngine.validate_checks(rules) + assert not status.has_errors, f"Generated rule has validation errors: {status.errors}" + + # order_ref has two recurring values ("A" recurs 3x, "B" recurs 2x; "C" is unique). The + # duplicate value count is 2 (A and B), not the 5 rows sitting in those two groups -- 2 is + # within the mustBeLessOrEqualTo: 4 threshold, so nothing is flagged. A regression to + # summing group sizes instead of counting distinct recurring values would report 5 here, + # which exceeds 4 and would incorrectly flag every row. + test_df = spark.createDataFrame( + [["A"], ["A"], ["A"], ["B"], ["B"], ["C"]], + "order_ref: string", + ) + dq_engine = DQEngine(workspace_client=ws) + checked = dq_engine.apply_checks_by_metadata(test_df, rules) + rows = checked.collect() + + assert len(rows) == 6 + for row in rows: + assert _flagged_names(row) == set() diff --git a/tests/resources/sample_datacontract_library_metrics.yaml b/tests/resources/sample_datacontract_library_metrics.yaml new file mode 100644 index 000000000..51680842e --- /dev/null +++ b/tests/resources/sample_datacontract_library_metrics.yaml @@ -0,0 +1,71 @@ +# Dedicated ODCS v3.x fixture exercising the five `type: library` quality metrics +# (rowCount, nullValues, missingValues, invalidValues, duplicateValues) end-to-end +# against a real DataFrame. Kept separate from sample_datacontract.yaml so it doesn't +# perturb the fixed rule counts / expected-rule lists asserted against that fixture +# elsewhere (see tests/integration/test_datacontract_integration.py and +# tests/unit/test_datacontract_generator.py). + +kind: DataContract +apiVersion: v3.0.2 +id: urn:datacontract:tests:library_metrics +name: Library Metrics Test Contract +version: 1.0.0 +status: active + +schema: + - name: customers + physicalType: table + description: Customer records used to exercise ODCS type:library quality metrics end-to-end. + + properties: + # duplicateValues (mustBe: 0) -> row-level-flagging is_unique dataset check + - name: customer_id + physicalType: STRING + logicalType: string + required: true + quality: + - type: library + metric: duplicateValues + mustBe: 0 + + # nullValues (mustBe: 0) -> row-level is_not_null check + - name: email + physicalType: STRING + logicalType: string + quality: + - type: library + metric: nullValues + mustBe: 0 + + # invalidValues (mustBe: 0) -> row-level is_in_list check + - name: status + physicalType: STRING + logicalType: string + quality: + - type: library + metric: invalidValues + mustBe: 0 + arguments: + validValues: + - ACTIVE + - INACTIVE + + # missingValues (mustBeGreaterThan, a strict threshold) -> sql_query fallback + - name: phone + physicalType: STRING + logicalType: string + quality: + - type: library + metric: missingValues + mustBeGreaterThan: 3 + arguments: + missingValues: + - null + - "" + - N/A + + # rowCount (mustBeGreaterOrEqualTo) -> dataset-level aggregate check, schema-level entry + quality: + - type: library + metric: rowCount + mustBeGreaterOrEqualTo: 10 diff --git a/tests/unit/test_datacontract_generator.py b/tests/unit/test_datacontract_generator.py index 95dde10d3..3eba9a331 100644 --- a/tests/unit/test_datacontract_generator.py +++ b/tests/unit/test_datacontract_generator.py @@ -13,6 +13,7 @@ from databricks.sdk.errors import NotFound import databricks.labs.dqx.profiler.generator as generator_module from databricks.labs.dqx.check_funcs import make_condition, register_rule +from databricks.labs.dqx.checks_semantic_validator import ChecksSemanticValidator import databricks.labs.dqx.datacontract.contract_rules_generator as contract_rules_generator_module from databricks.labs.dqx.datacontract.contract_rules_generator import DataContractRulesGenerator from databricks.labs.dqx.engine import DQEngine @@ -3565,3 +3566,1304 @@ def test_schema_with_no_properties(self, generator): assert len(rules) == 0 finally: os.unlink(temp_path) + + +class TestDataContractGeneratorLibraryRules(DataContractGeneratorTestBase): + """Tests for type: library quality metric dispatch and warn-and-skip handling.""" + + _SUPPORTED_METRICS_TEXT = "rowCount, nullValues, missingValues, invalidValues, duplicateValues" + + def test_unrecognized_metric_on_property_is_skipped_with_warning(self, generator, caplog): + """A type: library entry with an unrecognized metric is skipped; no rule is generated.""" + contract_dict = self.create_contract_with_quality( + property_name="email", + logical_type="string", + quality_checks=[{"type": "library", "metric": "notARealMetric"}], + ) + temp_path = self.create_test_contract_file(custom_contract=contract_dict) + + try: + with caplog.at_level(logging.WARNING): + rules = generator.generate_rules_from_contract( + contract_file=temp_path, + generate_predefined_rules=False, + process_text_rules=False, + generate_schema_validation=False, + ) + + assert rules == [] + assert "Unrecognized library metric 'notARealMetric'" in caplog.text + assert "property 'email'" in caplog.text + assert "schema 'test_table'" in caplog.text + assert self._SUPPORTED_METRICS_TEXT in caplog.text + finally: + os.unlink(temp_path) + + def test_missing_metric_on_property_is_skipped_with_warning(self, generator, caplog): + """A type: library entry with no 'metric' field is skipped, not raised.""" + contract_dict = self.create_contract_with_quality( + property_name="email", + logical_type="string", + quality_checks=[{"type": "library"}], + ) + temp_path = self.create_test_contract_file(custom_contract=contract_dict) + + try: + with caplog.at_level(logging.WARNING): + rules = generator.generate_rules_from_contract( + contract_file=temp_path, + generate_predefined_rules=False, + process_text_rules=False, + generate_schema_validation=False, + ) + + assert rules == [] + assert "Missing 'metric' on type: library quality entry" in caplog.text + assert "property 'email'" in caplog.text + assert self._SUPPORTED_METRICS_TEXT in caplog.text + finally: + os.unlink(temp_path) + + def test_unrecognized_metric_on_schema_omits_property_clause(self, generator, caplog): + """A schema-level type: library entry's warning omits the property clause entirely.""" + contract_dict = self.create_basic_contract( + schema_name="orders", + properties=[{"name": "order_id", "physicalType": "STRING"}], + ) + contract_dict["schema"][0]["quality"] = [{"type": "library", "metric": "bogusMetric"}] + temp_path = self.create_test_contract_file(custom_contract=contract_dict) + + try: + with caplog.at_level(logging.WARNING): + rules = generator.generate_rules_from_contract( + contract_file=temp_path, + generate_predefined_rules=False, + process_text_rules=False, + generate_schema_validation=False, + ) + + assert rules == [] + assert "Unrecognized library metric 'bogusMetric'" in caplog.text + assert "schema 'orders'" in caplog.text + assert "property" not in caplog.text + finally: + os.unlink(temp_path) + + def test_missing_metric_on_schema_is_skipped_with_warning(self, generator, caplog): + """A schema-level type: library entry missing 'metric' is skipped, not raised.""" + contract_dict = self.create_basic_contract( + schema_name="orders", + properties=[{"name": "order_id", "physicalType": "STRING"}], + ) + contract_dict["schema"][0]["quality"] = [{"type": "library"}] + temp_path = self.create_test_contract_file(custom_contract=contract_dict) + + try: + with caplog.at_level(logging.WARNING): + rules = generator.generate_rules_from_contract( + contract_file=temp_path, + generate_predefined_rules=False, + process_text_rules=False, + generate_schema_validation=False, + ) + + assert rules == [] + assert "Missing 'metric' on type: library quality entry" in caplog.text + assert "schema 'orders'" in caplog.text + finally: + os.unlink(temp_path) + + def test_recognized_metric_does_not_raise_and_yields_no_rule_yet(self, generator, caplog): + """A recognized metric name (e.g. nullValues) is accepted without warning. + + Per-metric rule construction lands with each metric's own ticket; nullValues now has one + (see TestDataContractGeneratorLibraryRulesNullValues), so it generates a rule here too. + """ + contract_dict = self.create_contract_with_quality( + property_name="email", + logical_type="string", + quality_checks=[{"type": "library", "metric": "nullValues", "mustBe": 0}], + ) + temp_path = self.create_test_contract_file(custom_contract=contract_dict) + + try: + with caplog.at_level(logging.WARNING): + rules = generator.generate_rules_from_contract( + contract_file=temp_path, + generate_predefined_rules=False, + process_text_rules=False, + generate_schema_validation=False, + ) + + assert len(rules) == 1 + assert "nullValues" not in caplog.text + finally: + os.unlink(temp_path) + + def test_omitted_type_with_metric_set_is_still_recognized(self, generator): + """Per ODCS, `type` "can be omitted, if a metric property is defined" -- every library + example in the spec omits it. An entry with no `type` at all but a recognized `metric` + must still be processed as a library rule, not silently dropped.""" + contract_dict = self.create_contract_with_quality( + property_name="email", + logical_type="string", + quality_checks=[{"metric": "nullValues", "mustBe": 0}], + ) + temp_path = self.create_test_contract_file(custom_contract=contract_dict) + + try: + rules = generator.generate_rules_from_contract( + contract_file=temp_path, + generate_predefined_rules=False, + process_text_rules=False, + generate_schema_validation=False, + ) + + assert len(rules) == 1 + assert rules[0]["check"] == {"function": "is_not_null", "arguments": {"column": "email"}} + finally: + os.unlink(temp_path) + + def test_explicit_non_library_type_is_not_treated_as_library_rule(self, generator): + """An entry with a `type` set to something other than 'library' (e.g. an explicit DQX + implementation entry) is left alone by the library-metric path even if it happens to + carry a `metric`-shaped field.""" + contract_dict = self.create_contract_with_quality( + property_name="email", + logical_type="string", + quality_checks=[{"type": "text", "description": "some text expectation"}], + ) + temp_path = self.create_test_contract_file(custom_contract=contract_dict) + + try: + rules = generator.generate_rules_from_contract( + contract_file=temp_path, + generate_predefined_rules=False, + process_text_rules=False, + generate_schema_validation=False, + ) + + assert rules == [] + finally: + os.unlink(temp_path) + + def test_mixed_valid_and_malformed_library_entries_preserve_other_rules(self, generator, caplog): + """One malformed type: library entry never blocks the rest of the contract's rules.""" + contract_dict = { + "kind": "DataContract", + "apiVersion": "v3.0.2", + "id": "urn:datacontract:test:mixed", + "name": "Mixed Rule Sources Contract", + "version": "1.0.0", + "status": "active", + "schema": [ + { + "name": "customers", + "physicalType": "table", + "properties": [ + { + "name": "customer_id", + "physicalType": "STRING", + "logicalType": "string", + "required": True, + }, + { + "name": "email", + "physicalType": "STRING", + "logicalType": "string", + "quality": [ + {"type": "library"}, + {"type": "library", "metric": "notARealMetric"}, + { + "type": "custom", + "engine": "dqx", + "implementation": { + "name": "email_not_empty", + "criticality": "error", + "check": { + "function": "is_not_null_and_not_empty", + "arguments": {"column": "email"}, + }, + }, + }, + ], + }, + ], + } + ], + } + temp_path = self.create_test_contract_file(custom_contract=contract_dict) + + try: + with caplog.at_level(logging.WARNING): + rules = generator.generate_rules_from_contract( + contract_file=temp_path, + generate_predefined_rules=True, + process_text_rules=False, + generate_schema_validation=True, + ) + self._assert_mixed_library_rules_preserved(rules, caplog) + finally: + os.unlink(temp_path) + + @staticmethod + def _assert_mixed_library_rules_preserved(rules, caplog): + """Assert other rule sources still generated despite malformed library entries.""" + rule_types = {rule["user_metadata"].get("rule_type") for rule in rules} + assert rule_types == {"schema_validation", "predefined", "explicit"} + + not_null_rules = [r for r in rules if r["name"] == "customer_id_is_null"] + assert len(not_null_rules) == 1 + + explicit_rules = [r for r in rules if r["name"] == "email_not_empty"] + assert len(explicit_rules) == 1 + + assert "Missing 'metric' on type: library quality entry" in caplog.text + assert "Unrecognized library metric 'notARealMetric'" in caplog.text + + +class TestDataContractGeneratorLibraryRulesRowCount(DataContractGeneratorTestBase): + """Tests for the type: library rowCount metric mapping onto DQX dataset-level checks.""" + + def _contract_with_row_count(self, quality_entry: dict, schema_name: str = "orders") -> dict: + """Build an ODCS contract with a single schema-level rowCount library quality entry.""" + contract_dict = self.create_basic_contract( + schema_name=schema_name, + properties=[{"name": "order_id", "physicalType": "STRING"}], + ) + contract_dict["schema"][0]["quality"] = [{"type": "library", "metric": "rowCount", **quality_entry}] + return contract_dict + + def _generate(self, generator, contract_dict) -> list[dict]: + temp_path = self.create_test_contract_file(custom_contract=contract_dict) + try: + return generator.generate_rules_from_contract( + contract_file=temp_path, + generate_predefined_rules=False, + process_text_rules=False, + generate_schema_validation=False, + ) + finally: + os.unlink(temp_path) + + def test_must_be_generates_is_aggr_equal(self, generator): + """mustBe maps onto an is_aggr_equal count-aggregate check with full lineage metadata.""" + rules = self._generate(generator, self._contract_with_row_count({"mustBe": 100})) + + assert len(rules) == 1 + rule = rules[0] + assert rule["name"] == "orders_rowCount_mustBe" + assert rule["check"]["function"] == "is_aggr_equal" + assert rule["check"]["arguments"] == {"column": "*", "limit": 100, "aggr_type": "count"} + user_metadata = rule["user_metadata"] + assert user_metadata["rule_type"] == "metric" + assert user_metadata["metric"] == "rowCount" + assert user_metadata["threshold_field"] == "mustBe" + assert user_metadata["unit"] == "rows" + assert user_metadata["dimension"] == "completeness" + assert "field" not in user_metadata + assert "fields" not in user_metadata + assert "severity" not in user_metadata + + def test_must_not_be_generates_is_aggr_not_equal(self, generator): + """mustNotBe maps onto an is_aggr_not_equal count-aggregate check.""" + rules = self._generate(generator, self._contract_with_row_count({"mustNotBe": 0})) + + assert len(rules) == 1 + rule = rules[0] + assert rule["check"]["function"] == "is_aggr_not_equal" + assert rule["check"]["arguments"] == {"column": "*", "limit": 0, "aggr_type": "count"} + assert rule["user_metadata"]["threshold_field"] == "mustNotBe" + + def test_must_be_greater_or_equal_to_generates_inclusive_bound_check(self, generator): + """mustBeGreaterOrEqualTo maps onto the matching inclusive-bound aggregate check.""" + rules = self._generate(generator, self._contract_with_row_count({"mustBeGreaterOrEqualTo": 10})) + + assert len(rules) == 1 + rule = rules[0] + assert rule["check"]["function"] == "is_aggr_not_less_than" + assert rule["check"]["arguments"] == {"column": "*", "limit": 10, "aggr_type": "count"} + assert rule["user_metadata"]["threshold_field"] == "mustBeGreaterOrEqualTo" + + def test_must_be_greater_than_falls_back_to_sql_query(self, generator): + """mustBeGreaterThan (strict) has no aggregate equivalent, so it falls back to sql_query.""" + rules = self._generate(generator, self._contract_with_row_count({"mustBeGreaterThan": 100})) + + assert len(rules) == 1 + rule = rules[0] + assert rule["check"]["function"] == "sql_query" + assert rule["check"]["arguments"]["query"] == "SELECT COUNT(*) <= 100 AS condition FROM {{ input_view }}" + assert rule["check"]["arguments"]["condition_column"] == "condition" + assert rule["user_metadata"]["threshold_field"] == "mustBeGreaterThan" + + def test_must_be_between_falls_back_to_sql_query_with_inclusive_bounds(self, generator): + """mustBeBetween has no aggregate equivalent, so it falls back to sql_query. Both bounds + are inclusive, matching datacontract-cli's reference mapping (the ODCS spec text doesn't + settle it).""" + rules = self._generate(generator, self._contract_with_row_count({"mustBeBetween": [10, 20]})) + + assert len(rules) == 1 + rule = rules[0] + assert rule["check"]["function"] == "sql_query" + assert rule["check"]["arguments"]["query"] == ( + "SELECT NOT (COUNT(*) >= 10 AND COUNT(*) <= 20) AS condition FROM {{ input_view }}" + ) + assert rule["user_metadata"]["threshold_field"] == "mustBeBetween" + + def test_severity_passed_through_verbatim_when_present(self, generator): + """DataQuality.severity is copied verbatim into user_metadata when the contract sets it.""" + rules = self._generate(generator, self._contract_with_row_count({"mustBe": 5, "severity": "critical"})) + + assert rules[0]["user_metadata"]["severity"] == "critical" + + def test_contract_dimension_overrides_default(self, generator): + """DataQuality.dimension overrides the rowCount default of 'completeness'.""" + rules = self._generate(generator, self._contract_with_row_count({"mustBe": 5, "dimension": "accuracy"})) + + assert rules[0]["user_metadata"]["dimension"] == "accuracy" + + def test_no_threshold_field_set_is_skipped_with_warning(self, generator, caplog): + """A rowCount entry with none of the eight threshold fields set is skipped, not raised.""" + with caplog.at_level(logging.WARNING): + rules = self._generate(generator, self._contract_with_row_count({})) + + assert rules == [] + assert "rowCount entry on schema 'orders' has no recognized threshold field set" in caplog.text + + def test_multiple_row_count_entries_get_distinct_rule_names(self, generator): + """Two rowCount entries on one schema (e.g. a lower and an upper bound) must not collide + on rule name -- each carries its own threshold_field suffix.""" + contract_dict = self.create_basic_contract( + schema_name="orders", + properties=[{"name": "order_id", "physicalType": "STRING"}], + ) + contract_dict["schema"][0]["quality"] = [ + {"type": "library", "metric": "rowCount", "mustBeGreaterOrEqualTo": 10}, + {"type": "library", "metric": "rowCount", "mustBeLessOrEqualTo": 1000}, + ] + rules = self._generate(generator, contract_dict) + + assert len(rules) == 2 + names = {rule["name"] for rule in rules} + assert names == {"orders_rowCount_mustBeGreaterOrEqualTo", "orders_rowCount_mustBeLessOrEqualTo"} + + +class TestDataContractGeneratorLibraryRulesNullValues(DataContractGeneratorTestBase): + """Tests for the type: library nullValues metric mapping onto DQX checks.""" + + def _contract_with_nullvalues( + self, quality_entry: dict, property_name: str = "email", schema_name: str = "test_table" + ) -> dict: + """Build an ODCS contract with a single property-level nullValues library quality entry.""" + return self.create_contract_with_quality( + property_name=property_name, + logical_type="string", + quality_checks=[{"type": "library", "metric": "nullValues", **quality_entry}], + schema_name=schema_name, + ) + + def _generate(self, generator, contract_dict) -> list[dict]: + temp_path = self.create_test_contract_file(custom_contract=contract_dict) + try: + return generator.generate_rules_from_contract( + contract_file=temp_path, + generate_predefined_rules=False, + process_text_rules=False, + generate_schema_validation=False, + ) + finally: + os.unlink(temp_path) + + def test_must_be_zero_generates_row_level_is_not_null(self, generator): + """mustBe: 0 maps onto the cheaper, row-pinpointing is_not_null check rather than a + dataset-level aggregate, unit-independent.""" + rules = self._generate(generator, self._contract_with_nullvalues({"mustBe": 0})) + + assert len(rules) == 1 + rule = rules[0] + assert rule["name"] == "email_nullValues" + assert rule["check"] == {"function": "is_not_null", "arguments": {"column": "email"}} + user_metadata = rule["user_metadata"] + assert user_metadata["rule_type"] == "metric" + assert user_metadata["metric"] == "nullValues" + assert user_metadata["threshold_field"] == "mustBe" + assert user_metadata["unit"] == "rows" + assert user_metadata["dimension"] == "completeness" + assert user_metadata["field"] == "email" + assert "severity" not in user_metadata + + def test_nonzero_rows_threshold_generates_null_count_aggregate(self, generator): + """A non-zero unit: rows threshold generates a dataset-level null-count aggregate, scoped + to null rows via a row_filter built from a safely quoted column.""" + rules = self._generate(generator, self._contract_with_nullvalues({"mustBeLessOrEqualTo": 5})) + + assert len(rules) == 1 + rule = rules[0] + assert rule["check"]["function"] == "is_aggr_not_greater_than" + assert rule["check"]["arguments"] == { + "column": "*", + "limit": 5, + "aggr_type": "count", + "row_filter": "email IS NULL", + } + assert rule["user_metadata"]["threshold_field"] == "mustBeLessOrEqualTo" + assert rule["user_metadata"]["unit"] == "rows" + + def test_percent_column_is_a_serializable_sql_string_not_a_pyspark_column(self, generator): + """The percent path's `column` argument must be a plain SQL string, not a live PySpark + Column: a Column can't round-trip through YAML/JSON (breaking save_checks()) and is + unhashable (breaking ChecksSemanticValidator's conflict-key grouping, which silently skips + conflict detection on a TypeError). Two conflicting percent-unit nullValues entries on the + same property must therefore now be flagged as a conflict. + """ + contract_dict = self.create_contract_with_quality( + property_name="email", + logical_type="string", + quality_checks=[ + {"type": "library", "metric": "nullValues", "mustBe": 5, "unit": "percent"}, + {"type": "library", "metric": "nullValues", "mustBe": 10, "unit": "percent"}, + ], + schema_name="test_table", + ) + rules = self._generate(generator, contract_dict) + + assert len(rules) == 2 + for rule in rules: + assert isinstance(rule["check"]["arguments"]["column"], str) + + issues = ChecksSemanticValidator.detect_conflicts(rules) + assert any("Conflicting rules detected" in issue for issue in issues) + + def test_unit_percent_generates_avg_indicator_check(self, generator): + """unit: percent compares the percentage of null rows via an AVG-of-indicator aggregate, + not the row_filter+'*' mechanism (which cannot express a percentage).""" + rules = self._generate(generator, self._contract_with_nullvalues({"mustBeLessOrEqualTo": 1, "unit": "percent"})) + + assert len(rules) == 1 + rule = rules[0] + assert rule["check"]["function"] == "is_aggr_not_greater_than" + args = rule["check"]["arguments"] + assert args["aggr_type"] == "avg" + assert args["limit"] == 1 + assert "row_filter" not in args + assert "email IS NULL" in str(args["column"]) + assert rule["user_metadata"]["unit"] == "percent" + + def test_must_be_greater_than_falls_back_to_sql_query(self, generator): + """mustBeGreaterThan (strict) has no aggregate equivalent, so it falls back to sql_query.""" + rules = self._generate(generator, self._contract_with_nullvalues({"mustBeGreaterThan": 3})) + + assert len(rules) == 1 + rule = rules[0] + assert rule["check"]["function"] == "sql_query" + assert rule["check"]["arguments"]["query"] == "SELECT COUNT(*) <= 3 AS condition FROM {{ input_view }}" + assert rule["check"]["arguments"]["row_filter"] == "email IS NULL" + assert rule["user_metadata"]["threshold_field"] == "mustBeGreaterThan" + + def test_quoted_identifier_for_property_needing_quoting(self, generator): + """A property name requiring SQL-identifier quoting is safely backtick-quoted in the + generated row_filter, not left to produce a broken query.""" + rules = self._generate( + generator, + self._contract_with_nullvalues({"mustBeLessOrEqualTo": 5}, property_name="order-id"), + ) + + assert len(rules) == 1 + rule = rules[0] + assert rule["check"]["arguments"]["row_filter"] == "`order-id` IS NULL" + + def test_no_threshold_field_set_is_skipped_with_warning(self, generator, caplog): + """A nullValues entry with none of the eight threshold fields set is skipped, not raised.""" + with caplog.at_level(logging.WARNING): + rules = self._generate(generator, self._contract_with_nullvalues({})) + + assert rules == [] + assert "nullValues entry on property 'email'" in caplog.text + assert "has no recognized threshold field set" in caplog.text + + def test_unrecognized_unit_on_non_zero_threshold_is_skipped_with_warning(self, generator, caplog): + """A non-zero threshold with an unrecognized unit is skipped, not silently treated as + unit: rows, matching duplicateValues/invalidValues/missingValues's own unit handling.""" + with caplog.at_level(logging.WARNING): + rules = self._generate(generator, self._contract_with_nullvalues({"mustBe": 5, "unit": "percentage"})) + + assert rules == [] + assert "Unrecognized unit 'percentage'" in caplog.text + + def test_schema_level_entry_is_skipped_with_warning(self, generator, caplog): + """nullValues is a property-level metric; a schema-level entry (no property) is skipped.""" + contract_dict = self.create_basic_contract( + schema_name="orders", + properties=[{"name": "order_id", "physicalType": "STRING"}], + ) + contract_dict["schema"][0]["quality"] = [{"type": "library", "metric": "nullValues", "mustBe": 0}] + temp_path = self.create_test_contract_file(custom_contract=contract_dict) + + try: + with caplog.at_level(logging.WARNING): + rules = generator.generate_rules_from_contract( + contract_file=temp_path, + generate_predefined_rules=False, + process_text_rules=False, + generate_schema_validation=False, + ) + + assert rules == [] + assert "nullValues entry in schema 'orders' has no property" in caplog.text + finally: + os.unlink(temp_path) + + +class TestDataContractGeneratorLibraryRulesMissingValues(DataContractGeneratorTestBase): + """Tests for the type: library missingValues metric mapping onto DQX checks.""" + + def _contract_with_missing_values( + self, quality_entry: dict, property_name: str = "email", schema_name: str = "test_table" + ) -> dict: + """Build an ODCS contract with a single property-level missingValues library quality entry.""" + return self.create_contract_with_quality( + property_name=property_name, + logical_type="string", + quality_checks=[{"type": "library", "metric": "missingValues", **quality_entry}], + schema_name=schema_name, + ) + + def _generate(self, generator, contract_dict) -> list[dict]: + temp_path = self.create_test_contract_file(custom_contract=contract_dict) + try: + return generator.generate_rules_from_contract( + contract_file=temp_path, + generate_predefined_rules=False, + process_text_rules=False, + generate_schema_validation=False, + ) + finally: + os.unlink(temp_path) + + def test_must_be_zero_with_null_and_sentinel_generates_two_row_level_rules(self, generator): + """mustBe: 0 with both a null and a non-null sentinel listed splits into two independent + row-level rules (is_not_null, is_not_in_list) that share identical user_metadata, only + `name` differing -- OR'd via DQX's own per-row _errors/_warnings union.""" + rules = self._generate( + generator, + self._contract_with_missing_values({"mustBe": 0, "arguments": {"missingValues": [None, "", "N/A"]}}), + ) + + assert len(rules) == 2 + null_rule, sentinel_rule = rules + + assert null_rule["name"] == "email_missingValues_null" + assert null_rule["check"] == {"function": "is_not_null", "arguments": {"column": "email"}} + + assert sentinel_rule["name"] == "email_missingValues_sentinel" + assert sentinel_rule["check"]["function"] == "is_not_in_list" + sentinel_args = sentinel_rule["check"]["arguments"] + assert sentinel_args["column"] == "email" + assert sentinel_args["case_sensitive"] is True + # Plain quoted string literals (not F.lit(...) Columns), so the rule stays + # YAML/JSON-serializable through save_checks() and hashable for conflict detection. + assert sentinel_args["forbidden"] == ["''", "'N/A'"] + + for rule in rules: + user_metadata = rule["user_metadata"] + assert user_metadata["rule_type"] == "metric" + assert user_metadata["metric"] == "missingValues" + assert user_metadata["threshold_field"] == "mustBe" + assert user_metadata["unit"] == "rows" + assert user_metadata["dimension"] == "completeness" + assert user_metadata["field"] == "email" + assert "severity" not in user_metadata + + # user_metadata is shared verbatim between the two rules; only the top-level `name` differs. + assert null_rule["user_metadata"] == sentinel_rule["user_metadata"] + + def test_must_be_zero_null_only_generates_single_row_level_rule(self, generator): + """A sentinel list containing only `null` emits just the is_not_null row rule.""" + rules = self._generate( + generator, + self._contract_with_missing_values({"mustBe": 0, "arguments": {"missingValues": [None]}}), + ) + + assert len(rules) == 1 + assert rules[0]["name"] == "email_missingValues_null" + assert rules[0]["check"] == {"function": "is_not_null", "arguments": {"column": "email"}} + + def test_must_be_zero_without_explicit_null_still_checks_null(self, generator): + """NULL is always counted as missing, even when `null` is not itself listed in + arguments.missingValues -- matching the non-zero-threshold condition + (`col IS NULL OR col IN (...)`), which never gated NULL on the list either.""" + rules = self._generate( + generator, + self._contract_with_missing_values({"mustBe": 0, "arguments": {"missingValues": ["N/A"]}}), + ) + + assert len(rules) == 2 + names = {rule["name"] for rule in rules} + assert names == {"email_missingValues_null", "email_missingValues_sentinel"} + null_rule = next(rule for rule in rules if rule["name"] == "email_missingValues_null") + assert null_rule["check"] == {"function": "is_not_null", "arguments": {"column": "email"}} + + def test_nonzero_rows_threshold_generates_missing_count_aggregate(self, generator): + """A non-zero unit: rows threshold generates a dataset-level count aggregate scoped to + rows matching either the null condition or the sentinel condition, OR'd in the row_filter, + with the column name safely quoted via _safe_sql_identifier.""" + rules = self._generate( + generator, + self._contract_with_missing_values( + {"mustBeLessOrEqualTo": 5, "arguments": {"missingValues": [None, "", "N/A"]}} + ), + ) + + assert len(rules) == 1 + rule = rules[0] + assert rule["name"] == "email_missingValues" + assert rule["check"]["function"] == "is_aggr_not_greater_than" + assert rule["check"]["arguments"] == { + "column": "*", + "limit": 5, + "aggr_type": "count", + "row_filter": "email IS NULL OR email IN ('', 'N/A')", + } + assert rule["user_metadata"]["threshold_field"] == "mustBeLessOrEqualTo" + assert rule["user_metadata"]["unit"] == "rows" + + def test_unit_percent_generates_avg_indicator_check(self, generator): + """unit: percent compares the percentage of missing rows via an AVG-of-indicator + aggregate string expression combining both conditions with OR, not the row_filter+'*' + mechanism (which cannot express a percentage).""" + rules = self._generate( + generator, + self._contract_with_missing_values( + { + "mustBeLessOrEqualTo": 1, + "unit": "percent", + "arguments": {"missingValues": [None, "N/A"]}, + } + ), + ) + + assert len(rules) == 1 + rule = rules[0] + assert rule["check"]["function"] == "is_aggr_not_greater_than" + args = rule["check"]["arguments"] + assert args["aggr_type"] == "avg" + assert args["limit"] == 1 + assert "row_filter" not in args + assert args["column"] == "CASE WHEN email IS NULL OR email IN ('N/A') THEN 100.0 ELSE 0.0 END" + assert rule["user_metadata"]["unit"] == "percent" + + def test_must_be_greater_than_falls_back_to_sql_query(self, generator): + """mustBeGreaterThan (strict) has no aggregate equivalent, so it falls back to sql_query.""" + rules = self._generate( + generator, + self._contract_with_missing_values({"mustBeGreaterThan": 3, "arguments": {"missingValues": [None]}}), + ) + + assert len(rules) == 1 + rule = rules[0] + assert rule["check"]["function"] == "sql_query" + assert rule["check"]["arguments"]["query"] == "SELECT COUNT(*) <= 3 AS condition FROM {{ input_view }}" + assert rule["check"]["arguments"]["row_filter"] == "email IS NULL" + assert rule["user_metadata"]["threshold_field"] == "mustBeGreaterThan" + + def test_must_be_between_falls_back_to_sql_query_with_inclusive_bounds_and_percent_unit(self, generator): + """mustBeBetween has no aggregate equivalent, so it falls back to sql_query under unit: + percent too, using the AVG(CASE WHEN ...) expression. Both bounds are inclusive, matching + datacontract-cli's reference mapping.""" + rules = self._generate( + generator, + self._contract_with_missing_values( + { + "mustBeBetween": [1, 5], + "unit": "percent", + "arguments": {"missingValues": [None]}, + } + ), + ) + + assert len(rules) == 1 + rule = rules[0] + assert rule["check"]["function"] == "sql_query" + assert rule["check"]["arguments"]["query"] == ( + "SELECT NOT (AVG(CASE WHEN email IS NULL THEN 100.0 ELSE 0.0 END) >= 1 AND " + "AVG(CASE WHEN email IS NULL THEN 100.0 ELSE 0.0 END) <= 5) AS condition FROM {{ input_view }}" + ) + assert rule["user_metadata"]["threshold_field"] == "mustBeBetween" + + def test_missing_arguments_is_skipped_with_warning(self, generator, caplog): + """No 'arguments' block at all is skipped, not raised, naming the malformed key.""" + with caplog.at_level(logging.WARNING): + rules = self._generate(generator, self._contract_with_missing_values({"mustBe": 0})) + + assert rules == [] + assert "arguments" in caplog.text + assert "missingValues" in caplog.text + + def test_wrong_type_arguments_missing_values_is_skipped_with_warning(self, generator, caplog): + """arguments.missingValues of the wrong type (not a list) is skipped, not raised, naming + the malformed key and its expected shape.""" + with caplog.at_level(logging.WARNING): + rules = self._generate( + generator, + self._contract_with_missing_values({"mustBe": 0, "arguments": {"missingValues": "not-a-list"}}), + ) + + assert rules == [] + assert "arguments.missingValues" in caplog.text + assert "expected list" in caplog.text + + def test_empty_arguments_missing_values_is_skipped_with_warning(self, generator, caplog): + """An empty arguments.missingValues list is skipped, not raised.""" + with caplog.at_level(logging.WARNING): + rules = self._generate( + generator, + self._contract_with_missing_values({"mustBe": 0, "arguments": {"missingValues": []}}), + ) + + assert rules == [] + assert "arguments.missingValues" in caplog.text + + def test_schema_level_entry_is_skipped_with_warning(self, generator, caplog): + """missingValues is a property-level metric; a schema-level entry (no property) is skipped.""" + contract_dict = self.create_basic_contract( + schema_name="orders", + properties=[{"name": "order_id", "physicalType": "STRING"}], + ) + contract_dict["schema"][0]["quality"] = [ + {"type": "library", "metric": "missingValues", "mustBe": 0, "arguments": {"missingValues": [None]}} + ] + temp_path = self.create_test_contract_file(custom_contract=contract_dict) + + try: + with caplog.at_level(logging.WARNING): + rules = generator.generate_rules_from_contract( + contract_file=temp_path, + generate_predefined_rules=False, + process_text_rules=False, + generate_schema_validation=False, + ) + + assert rules == [] + assert "missingValues entry in schema 'orders' has no property" in caplog.text + finally: + os.unlink(temp_path) + + def test_no_threshold_field_set_is_skipped_with_warning(self, generator, caplog): + """A missingValues entry with none of the eight threshold fields set is skipped, not raised.""" + with caplog.at_level(logging.WARNING): + rules = self._generate( + generator, self._contract_with_missing_values({"arguments": {"missingValues": [None]}}) + ) + + assert rules == [] + assert "missingValues entry on property 'email'" in caplog.text + assert "has no recognized threshold field set" in caplog.text + + def test_quoted_identifier_for_property_needing_quoting(self, generator): + """A property name requiring SQL-identifier quoting is safely backtick-quoted in the + generated row_filter, not left to produce a broken query.""" + rules = self._generate( + generator, + self._contract_with_missing_values( + {"mustBeLessOrEqualTo": 5, "arguments": {"missingValues": [None]}}, + property_name="order-id", + ), + ) + + assert len(rules) == 1 + rule = rules[0] + assert rule["check"]["arguments"]["row_filter"] == "`order-id` IS NULL" + + +class TestDataContractGeneratorLibraryRulesInvalidValues(DataContractGeneratorTestBase): + """Tests for the type: library invalidValues metric mapping onto DQX checks.""" + + def _contract_with_invalid_values(self, quality_entry: dict, property_name: str = "status") -> dict: + """Build an ODCS contract with a single property-level invalidValues library quality entry.""" + return self.create_contract_with_quality( + property_name=property_name, + logical_type="string", + quality_checks=[{"type": "library", "metric": "invalidValues", **quality_entry}], + ) + + def _generate(self, generator, contract_dict) -> list[dict]: + temp_path = self.create_test_contract_file(custom_contract=contract_dict) + try: + return generator.generate_rules_from_contract( + contract_file=temp_path, + generate_predefined_rules=False, + process_text_rules=False, + generate_schema_validation=False, + ) + finally: + os.unlink(temp_path) + + def test_must_be_zero_with_valid_values_only_generates_is_in_list_row_rule(self, generator): + """mustBe: 0 with only arguments.validValues generates a single row-level is_in_list check, + quoting each string entry so is_in_list resolves it as a literal, not a column reference.""" + rules = self._generate( + generator, + self._contract_with_invalid_values({"mustBe": 0, "arguments": {"validValues": ["ACTIVE", "INACTIVE"]}}), + ) + + assert len(rules) == 1 + rule = rules[0] + assert rule["name"] == "status_invalidValues_allowed" + assert rule["check"]["function"] == "is_in_list" + assert rule["check"]["arguments"] == { + "column": "status", + "allowed": ["'ACTIVE'", "'INACTIVE'"], + "case_sensitive": True, + } + user_metadata = rule["user_metadata"] + assert user_metadata["rule_type"] == "metric" + assert user_metadata["metric"] == "invalidValues" + assert user_metadata["threshold_field"] == "mustBe" + assert user_metadata["unit"] == "rows" + assert user_metadata["dimension"] == "conformity" + assert user_metadata["field"] == "status" + assert "severity" not in user_metadata + + def test_must_be_zero_with_pattern_only_generates_regex_match_row_rule(self, generator): + """mustBe: 0 with only arguments.pattern generates a single row-level regex_match check.""" + rules = self._generate( + generator, + self._contract_with_invalid_values({"mustBe": 0, "arguments": {"pattern": "^[A-Z]{3}$"}}), + ) + + assert len(rules) == 1 + rule = rules[0] + assert rule["name"] == "status_invalidValues_pattern" + assert rule["check"]["function"] == "regex_match" + assert rule["check"]["arguments"] == {"column": "status", "regex": "^[A-Z]{3}$"} + assert rule["user_metadata"]["threshold_field"] == "mustBe" + + def test_must_be_zero_with_both_mechanisms_generates_both_row_rules(self, generator): + """mustBe: 0 with both validValues and pattern generates both row-level rules, sharing + identical user_metadata (only *name* differs) -- a value is invalid if it fails either.""" + rules = self._generate( + generator, + self._contract_with_invalid_values( + {"mustBe": 0, "arguments": {"validValues": ["ACTIVE"], "pattern": "^[A-Z]+$"}} + ), + ) + + assert len(rules) == 2 + names = {rule["name"] for rule in rules} + assert names == {"status_invalidValues_allowed", "status_invalidValues_pattern"} + + allowed_rule = next(rule for rule in rules if rule["name"] == "status_invalidValues_allowed") + pattern_rule = next(rule for rule in rules if rule["name"] == "status_invalidValues_pattern") + assert allowed_rule["check"]["function"] == "is_in_list" + assert pattern_rule["check"]["function"] == "regex_match" + + allowed_metadata = dict(allowed_rule["user_metadata"]) + pattern_metadata = dict(pattern_rule["user_metadata"]) + assert allowed_metadata == pattern_metadata + + def test_non_zero_threshold_unit_rows_generates_row_filter_count_aggregate(self, generator): + """A non-zero mustBe threshold with unit: rows (default) generates a count(*) aggregate + over an OR'd NOT IN / NOT RLIKE row_filter.""" + rules = self._generate( + generator, + self._contract_with_invalid_values( + {"mustBe": 5, "arguments": {"validValues": ["ACTIVE"], "pattern": "^[A-Z]+$"}} + ), + ) + + assert len(rules) == 1 + rule = rules[0] + assert rule["name"] == "status_invalidValues" + assert rule["check"]["function"] == "is_aggr_equal" + assert rule["check"]["arguments"] == { + "column": "*", + "limit": 5, + "aggr_type": "count", + "row_filter": "status NOT IN ('ACTIVE') OR NOT (status RLIKE '^[A-Z]+$')", + } + user_metadata = rule["user_metadata"] + assert user_metadata["threshold_field"] == "mustBe" + assert user_metadata["unit"] == "rows" + assert user_metadata["field"] == "status" + + def test_unit_percent_generates_indicator_column_avg_aggregate(self, generator): + """unit: percent generates a CASE WHEN indicator SQL expression with aggr_type='avg', + since row_filter+'*' cannot express a percentage.""" + rules = self._generate( + generator, + self._contract_with_invalid_values( + {"mustNotBe": 10, "unit": "percent", "arguments": {"validValues": ["ACTIVE"]}} + ), + ) + + assert len(rules) == 1 + rule = rules[0] + assert rule["check"]["function"] == "is_aggr_not_equal" + assert rule["check"]["arguments"] == { + "column": "CASE WHEN status NOT IN ('ACTIVE') THEN 100.0 ELSE 0.0 END", + "limit": 10, + "aggr_type": "avg", + } + assert rule["user_metadata"]["unit"] == "percent" + + def test_strict_threshold_falls_back_to_sql_query(self, generator): + """mustBeGreaterThan (strict) has no aggregate equivalent, so it falls back to sql_query + with the invalid condition passed as row_filter.""" + rules = self._generate( + generator, + self._contract_with_invalid_values( + {"mustBeGreaterThan": 3, "arguments": {"validValues": ["ACTIVE", "INACTIVE"]}} + ), + ) + + assert len(rules) == 1 + rule = rules[0] + assert rule["check"]["function"] == "sql_query" + assert rule["check"]["arguments"]["query"] == ("SELECT COUNT(*) <= 3 AS condition FROM {{ input_view }}") + assert rule["check"]["arguments"]["row_filter"] == "status NOT IN ('ACTIVE', 'INACTIVE')" + assert rule["user_metadata"]["threshold_field"] == "mustBeGreaterThan" + + def test_malformed_arguments_is_skipped_with_warning(self, generator, caplog): + """A malformed arguments.validValues (wrong type) with no usable pattern is skipped, not raised.""" + with caplog.at_level(logging.WARNING): + rules = self._generate( + generator, + self._contract_with_invalid_values({"mustBe": 0, "arguments": {"validValues": "not-a-list"}}), + ) + + assert rules == [] + assert "Missing or malformed 'arguments.validValues': expected list, got str" in caplog.text + assert "neither a usable 'arguments.validValues' list nor a usable 'arguments.pattern'" in caplog.text + + def test_neither_valid_values_nor_pattern_present_is_skipped_with_warning(self, generator, caplog): + """An invalidValues entry with no arguments at all is skipped, not raised.""" + with caplog.at_level(logging.WARNING): + rules = self._generate(generator, self._contract_with_invalid_values({"mustBe": 0})) + + assert rules == [] + assert "neither a usable 'arguments.validValues' list nor a usable 'arguments.pattern'" in caplog.text + + def test_unrecognized_unit_on_non_zero_threshold_is_skipped_with_warning(self, generator, caplog): + """A non-zero threshold with an unrecognized unit is skipped, not raised, per the general + unmapped-metric unit policy (only 'rows' and 'percent' are recognized).""" + with caplog.at_level(logging.WARNING): + rules = self._generate( + generator, + self._contract_with_invalid_values( + {"mustBe": 5, "unit": "bogus", "arguments": {"validValues": ["ACTIVE"]}} + ), + ) + + assert rules == [] + assert "Unrecognized unit 'bogus'" in caplog.text + assert "expected 'rows' or 'percent'" in caplog.text + + def test_pattern_exceeding_length_cap_is_skipped_with_warning(self, generator, caplog): + """A pattern longer than 200 characters fails the ReDoS length-cap guard and is skipped, + never compiled into a check.""" + long_pattern = "a" * 201 + with caplog.at_level(logging.WARNING): + rules = self._generate( + generator, self._contract_with_invalid_values({"mustBe": 0, "arguments": {"pattern": long_pattern}}) + ) + + assert rules == [] + assert "failed the ReDoS safety guard" in caplog.text + + def test_pattern_with_nested_quantifier_is_skipped_with_warning(self, generator, caplog): + """A nested-quantifier pattern like (a+)+ fails the ReDoS guard and is skipped.""" + with caplog.at_level(logging.WARNING): + rules = self._generate( + generator, self._contract_with_invalid_values({"mustBe": 0, "arguments": {"pattern": "(a+)+"}}) + ) + + assert rules == [] + assert "failed the ReDoS safety guard" in caplog.text + + def test_pattern_with_alternation_quantifier_is_skipped_with_warning(self, generator, caplog): + """An alternation-plus-quantifier pattern like (a|a)* fails the ReDoS guard and is skipped.""" + with caplog.at_level(logging.WARNING): + rules = self._generate( + generator, self._contract_with_invalid_values({"mustBe": 0, "arguments": {"pattern": "(a|a)*"}}) + ) + + assert rules == [] + assert "failed the ReDoS safety guard" in caplog.text + + def test_pattern_failing_to_compile_is_skipped_with_warning(self, generator, caplog): + """A syntactically invalid regex fails the re.compile() proxy check and is skipped.""" + with caplog.at_level(logging.WARNING): + rules = self._generate( + generator, self._contract_with_invalid_values({"mustBe": 0, "arguments": {"pattern": "a("}}) + ) + + assert rules == [] + assert "failed the ReDoS safety guard" in caplog.text + + def test_numeric_valid_values_are_unquoted_in_not_in_clause(self, generator): + """Numeric validValues entries are rendered unquoted, so the NOT IN clause compares + numerically -- matching the row-level is_in_list path's own numeric handling -- rather + than stringified (which would silently mismatch on a numeric column).""" + rules = self._generate( + generator, + self._contract_with_invalid_values({"mustBeGreaterThan": 3, "arguments": {"validValues": [1, 2, 3]}}), + ) + + assert len(rules) == 1 + assert rules[0]["check"]["arguments"]["row_filter"] == "status NOT IN (1, 2, 3)" + + def test_pattern_backslash_is_escaped_for_rlike(self, generator): + """A pattern containing a backslash (e.g. \\d+) has it doubled before interpolation, so it + survives Spark's string-literal escape processing and reaches RLIKE as the literal + backslash the row-level regex_match path (mustBe: 0) would also see.""" + rules = self._generate( + generator, + self._contract_with_invalid_values({"mustBeGreaterThan": 3, "arguments": {"pattern": r"\d+"}}), + ) + + assert len(rules) == 1 + assert rules[0]["check"]["arguments"]["row_filter"] == r"NOT (status RLIKE '\\d+')" + + def test_must_be_zero_valid_values_backslash_is_escaped_like_non_zero_path(self, generator): + """A validValues string containing a backslash is escaped identically on the mustBe: 0 + row-level is_in_list path and the non-zero aggregate/sql_query path (both eventually parse + the literal as Spark SQL), so a `\\N` sentinel compares the same way regardless of + threshold.""" + rules = self._generate( + generator, + self._contract_with_invalid_values({"mustBe": 0, "arguments": {"validValues": [r"\N"]}}), + ) + + assert len(rules) == 1 + assert rules[0]["check"]["arguments"]["allowed"] == [r"'\\N'"] + + +class TestDataContractGeneratorLibraryRulesDuplicateValues(DataContractGeneratorTestBase): + """Tests for the type: library duplicateValues metric mapping onto DQX checks.""" + + def _contract_with_duplicate_values_property( + self, quality_entry: dict, property_name: str = "order_id", schema_name: str = "orders" + ) -> dict: + """Build an ODCS contract with a single-property (argument-less) duplicateValues entry.""" + contract_dict = self.create_basic_contract( + schema_name=schema_name, + properties=[{"name": property_name, "physicalType": "STRING"}], + ) + contract_dict["schema"][0]["properties"][0]["quality"] = [ + {"type": "library", "metric": "duplicateValues", **quality_entry} + ] + return contract_dict + + def _contract_with_duplicate_values_schema(self, quality_entry: dict, schema_name: str = "orders") -> dict: + """Build an ODCS contract with a schema-level (composite-key) duplicateValues entry.""" + contract_dict = self.create_basic_contract( + schema_name=schema_name, + properties=[ + {"name": "tenant_id", "physicalType": "STRING"}, + {"name": "order_id", "physicalType": "STRING"}, + ], + ) + contract_dict["schema"][0]["quality"] = [{"type": "library", "metric": "duplicateValues", **quality_entry}] + return contract_dict + + def _generate(self, generator, contract_dict) -> list[dict]: + temp_path = self.create_test_contract_file(custom_contract=contract_dict) + try: + return generator.generate_rules_from_contract( + contract_file=temp_path, + generate_predefined_rules=False, + process_text_rules=False, + generate_schema_validation=False, + ) + finally: + os.unlink(temp_path) + + def test_must_be_zero_single_property_generates_is_unique(self, generator): + """A single-property duplicateValues entry with mustBe: 0 generates is_unique on that column.""" + rules = self._generate(generator, self._contract_with_duplicate_values_property({"mustBe": 0})) + + assert len(rules) == 1 + rule = rules[0] + assert rule["name"] == "order_id_duplicateValues" + assert rule["check"] == {"function": "is_unique", "arguments": {"columns": ["order_id"]}} + user_metadata = rule["user_metadata"] + assert user_metadata["rule_type"] == "metric" + assert user_metadata["metric"] == "duplicateValues" + assert user_metadata["threshold_field"] == "mustBe" + assert user_metadata["unit"] == "rows" + assert user_metadata["dimension"] == "uniqueness" + assert user_metadata["field"] == "order_id" + assert "fields" not in user_metadata + assert "severity" not in user_metadata + + def test_must_be_false_does_not_use_is_unique_fast_path(self, generator): + """mustBe: false (a boolean, not a genuine numeric zero) must not be special-cased into + is_unique -- `False == 0` is True in Python, so a naive `mustBe == 0` check would wrongly + treat it as mustBe: 0.""" + rules = self._generate(generator, self._contract_with_duplicate_values_property({"mustBe": False})) + + assert len(rules) == 1 + assert rules[0]["check"]["function"] != "is_unique" + assert rules[0]["user_metadata"]["threshold_field"] == "mustBe" + + def test_must_be_numeric_string_zero_uses_is_unique_fast_path(self, generator): + """mustBe: "0" (a numeric string, not a genuine numeric zero) must still be special-cased + into is_unique -- a naive `mustBe == 0` check would miss it since `"0" == 0` is False.""" + rules = self._generate(generator, self._contract_with_duplicate_values_property({"mustBe": "0"})) + + assert len(rules) == 1 + assert rules[0]["check"] == {"function": "is_unique", "arguments": {"columns": ["order_id"]}} + + def test_composite_properties_must_be_zero_generates_is_unique_across_columns(self, generator): + """A schema-level arguments.properties composite key with mustBe: 0 generates is_unique + across every listed column.""" + rules = self._generate( + generator, + self._contract_with_duplicate_values_schema( + {"mustBe": 0, "arguments": {"properties": ["tenant_id", "order_id"]}} + ), + ) + + assert len(rules) == 1 + rule = rules[0] + assert rule["name"] == "orders_duplicateValues" + assert rule["check"] == {"function": "is_unique", "arguments": {"columns": ["tenant_id", "order_id"]}} + user_metadata = rule["user_metadata"] + assert user_metadata["fields"] == ["tenant_id", "order_id"] + assert "field" not in user_metadata + + @staticmethod + def _duplicate_count_expr(group_by_clause: str, where_clause: str) -> str: + """Build the expected GROUP BY ... HAVING-based duplicate value-count scalar subquery + expression: the number of distinct recurring values (matching datacontract-cli's + reference mapping), not the number of rows sitting in a duplicated group.""" + return ( + f"(SELECT COUNT(*) FROM (SELECT 1 FROM {{{{ input_view }}}} WHERE {where_clause} " + f"GROUP BY {group_by_clause} HAVING COUNT(*) > 1) AS dqx_dup_groups)" + ) + + def test_non_zero_unit_rows_falls_back_to_sql_query(self, generator): + """A non-zero threshold with unit: rows (or absent) falls back to sql_query, keyed on a + GROUP BY ... HAVING-based duplicate value count (not a window function, which Spark + rejects when nested inside an aggregate, and not re-selected FROM the input view, which + would return one row per input row instead of the required single result row).""" + rules = self._generate(generator, self._contract_with_duplicate_values_property({"mustBeLessOrEqualTo": 5})) + + assert len(rules) == 1 + rule = rules[0] + assert rule["check"]["function"] == "sql_query" + count_expr = self._duplicate_count_expr("order_id", "order_id IS NOT NULL") + assert rule["check"]["arguments"]["query"] == f"SELECT {count_expr} > 5 AS condition" + assert rule["check"]["arguments"]["condition_column"] == "condition" + assert rule["user_metadata"]["unit"] == "rows" + assert rule["user_metadata"]["threshold_field"] == "mustBeLessOrEqualTo" + + def test_unit_percent_falls_back_to_sql_query(self, generator): + """unit: percent divides the same duplicate value count by the total row count, times 100.""" + rules = self._generate( + generator, self._contract_with_duplicate_values_property({"mustBe": 10, "unit": "percent"}) + ) + + assert len(rules) == 1 + rule = rules[0] + assert rule["check"]["function"] == "sql_query" + count_expr = self._duplicate_count_expr("order_id", "order_id IS NOT NULL") + percent_expr = f"(100.0 * {count_expr} / NULLIF((SELECT COUNT(*) FROM {{{{ input_view }}}}), 0))" + assert rule["check"]["arguments"]["query"] == f"SELECT {percent_expr} <> 10 AS condition" + assert rule["user_metadata"]["unit"] == "percent" + + def test_must_be_greater_than_falls_back_to_sql_query(self, generator): + """A strict threshold (mustBeGreaterThan) has no aggregate equivalent, so it falls back to + sql_query, keyed on the same duplicate-count expression.""" + rules = self._generate(generator, self._contract_with_duplicate_values_property({"mustBeGreaterThan": 3})) + + assert len(rules) == 1 + rule = rules[0] + assert rule["check"]["function"] == "sql_query" + count_expr = self._duplicate_count_expr("order_id", "order_id IS NOT NULL") + assert rule["check"]["arguments"]["query"] == f"SELECT {count_expr} <= 3 AS condition" + assert rule["check"]["arguments"]["condition_column"] == "condition" + assert rule["user_metadata"]["threshold_field"] == "mustBeGreaterThan" + + def test_composite_between_falls_back_to_sql_query_with_inclusive_bounds(self, generator): + """A composite-key mustBeBetween falls back to sql_query with inclusive bounds (matching + datacontract-cli's reference mapping), grouped by every listed column.""" + rules = self._generate( + generator, + self._contract_with_duplicate_values_schema( + {"mustBeBetween": [1, 4], "arguments": {"properties": ["tenant_id", "order_id"]}} + ), + ) + + assert len(rules) == 1 + rule = rules[0] + assert rule["check"]["function"] == "sql_query" + count_expr = self._duplicate_count_expr("tenant_id, order_id", "tenant_id IS NOT NULL AND order_id IS NOT NULL") + assert ( + rule["check"]["arguments"]["query"] == f"SELECT NOT ({count_expr} >= 1 AND {count_expr} <= 4) AS condition" + ) + assert rule["user_metadata"]["threshold_field"] == "mustBeBetween" + + def test_malformed_arguments_properties_wrong_type_is_skipped_with_warning(self, generator, caplog): + """A schema-level entry whose arguments.properties isn't a list is skipped, not raised.""" + contract_dict = self._contract_with_duplicate_values_schema({"mustBe": 0, "arguments": {"properties": "x"}}) + with caplog.at_level(logging.WARNING): + rules = self._generate(generator, contract_dict) + + assert rules == [] + assert "arguments.properties" in caplog.text + + def test_malformed_arguments_properties_empty_is_skipped_with_warning(self, generator, caplog): + """A schema-level entry with an empty arguments.properties list is skipped, not raised.""" + contract_dict = self._contract_with_duplicate_values_schema({"mustBe": 0, "arguments": {"properties": []}}) + with caplog.at_level(logging.WARNING): + rules = self._generate(generator, contract_dict) + + assert rules == [] + assert "arguments.properties" in caplog.text + + def test_malformed_arguments_properties_non_string_entry_is_skipped_with_warning(self, generator, caplog): + """A schema-level entry whose arguments.properties list contains a non-string entry is + skipped, not raised.""" + contract_dict = self._contract_with_duplicate_values_schema( + {"mustBe": 0, "arguments": {"properties": ["tenant_id", 123]}} + ) + with caplog.at_level(logging.WARNING): + rules = self._generate(generator, contract_dict) + + assert rules == [] + assert "arguments.properties" in caplog.text + + def test_unrecognized_unit_on_non_zero_threshold_is_skipped_with_warning(self, generator, caplog): + """A non-zero threshold with an unrecognized unit value is skipped, not defaulted.""" + with caplog.at_level(logging.WARNING): + rules = self._generate( + generator, self._contract_with_duplicate_values_property({"mustBe": 5, "unit": "bogus"}) + ) + + assert rules == [] + assert "Unrecognized unit 'bogus'" in caplog.text + + def test_severity_passed_through_verbatim_when_present(self, generator): + """DataQuality.severity is copied verbatim into user_metadata when the contract sets it.""" + rules = self._generate( + generator, self._contract_with_duplicate_values_property({"mustBe": 0, "severity": "critical"}) + ) + + assert rules[0]["user_metadata"]["severity"] == "critical" + + def test_contract_dimension_overrides_default(self, generator): + """DataQuality.dimension overrides the duplicateValues default of 'uniqueness'.""" + rules = self._generate( + generator, self._contract_with_duplicate_values_property({"mustBe": 0, "dimension": "accuracy"}) + ) + + assert rules[0]["user_metadata"]["dimension"] == "accuracy" + + def test_no_threshold_field_set_is_skipped_with_warning(self, generator, caplog): + """A duplicateValues entry with none of the eight threshold fields set is skipped, not raised.""" + with caplog.at_level(logging.WARNING): + rules = self._generate(generator, self._contract_with_duplicate_values_property({})) + + assert rules == [] + assert "duplicateValues entry in schema 'orders' has no recognized threshold field set" in caplog.text