diff --git a/CHANGELOG.md b/CHANGELOG.md index bf05b96da..329c6bac4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added +- `datacontract test` on Databricks and Spark checks nested array and struct properties (#1278 @rob-h-w) + ## [1.1.3] - 2026-09-03 ### Added diff --git a/README.md b/README.md index 62469bf9b..f5abd4337 100644 --- a/README.md +++ b/README.md @@ -349,6 +349,21 @@ Try to sync all groups: uv sync --all-groups --all-extras ``` +#### Linux system package for postgres/psycopg-based tests + +**Ubuntu/Debian:** +```bash +sudo apt-get update +sudo apt-get install -y libpq-dev +``` +**Fedora/RHEL:** +```bash +# Fedora/RHEL: +sudo dnf install -y postgresql-devel +# Arch: +sudo pacman -S postgresql-libs +``` + ### Docker Build ```bash diff --git a/datacontract/engines/checks/create_checks.py b/datacontract/engines/checks/create_checks.py index 40275d27f..721d19951 100644 --- a/datacontract/engines/checks/create_checks.py +++ b/datacontract/engines/checks/create_checks.py @@ -32,6 +32,7 @@ logger = logging.getLogger(__name__) _FILE_SERVER_TYPES = {"local", "s3", "gcs", "azure"} +_NESTED_CHECK_SERVER_TYPES = {"dataframe", "databricks"} # --------------------------------------------------------------------------- @@ -111,6 +112,31 @@ def quality_definition_yaml(quality: DataQuality) -> str: return yaml.safe_dump(quality.model_dump(exclude_none=True), sort_keys=False) +def _property_type(prop: SchemaProperty) -> str: + return normalize_type_name(prop.physicalType or prop.logicalType) + + +def _iter_property_paths( + properties: list[SchemaProperty] | None, + server_type: str | None, + prefix: str | None = None, +): + for prop in properties or []: + field = prop.physicalName or prop.name + field_path = f"{prefix}.{field}" if prefix else field + yield field_path, prop + + prop_type = _property_type(prop) + if server_type in _NESTED_CHECK_SERVER_TYPES and prop_type == "object" and prop.properties: + yield from _iter_property_paths(prop.properties, server_type, field_path) + elif ( + server_type in _NESTED_CHECK_SERVER_TYPES and prop_type == "array" and prop.items and prop.items.properties + ): + # `[]` marks the array hop; the executor turns it into a predicate + # over the elements instead of a column lookup. + yield from _iter_property_paths(prop.items.properties, server_type, f"{field_path}[]") + + _PERCENT_UNITS = {"percent", "percentage", "%"} @@ -214,12 +240,12 @@ def _is_azure_blob_schema(schema_object: SchemaObject, server: Optional[Server]) def _to_schema_checks(schema_object: SchemaObject, server: Optional[Server]) -> List[CheckSpec]: checks: List[CheckSpec] = [] - server_type = server.type if server and server.type else None + server_type = get_server_type(server) if server is not None else None model = to_schema_name(schema_object, server_type) properties = schema_object.properties or [] check_types = is_check_types(server) uses_raw_view = ( - server is not None and server.type in _FILE_SERVER_TYPES and server.format in ("csv", "parquet", "json") + server is not None and server_type in _FILE_SERVER_TYPES and server.format in ("csv", "parquet", "json") ) # A primary key is both not-null and unique. A composite key is unique as a @@ -231,9 +257,8 @@ def _to_schema_checks(schema_object: SchemaObject, server: Optional[Server]) -> ) primary_key_is_composite = len(primary_key_props) > 1 - for prop in properties: + for field, prop in _iter_property_paths(properties, server_type): # ODCS physicalName is the real column; mirror to_schema_name at field level. - field = prop.physicalName or prop.name checks.append( CheckSpec( @@ -641,7 +666,11 @@ def _quality_checks( def _quality_rule_checks( - model: str, field: Optional[str], quality: DataQuality, count: int, server: Optional[Server] + model: str, + field: Optional[str], + quality: DataQuality, + count: int, + server: Optional[Server], ) -> List[CheckSpec]: """The checks of a single ODCS quality rule (``count`` is its index in the list).""" if quality.type == "custom" and quality.engine == "soda" and quality.implementation: @@ -669,6 +698,28 @@ def _quality_rule_checks( else: check_key = f"{model}__{field}__quality_sql_{count}" check_type = "field_quality_sql" + if field is not None and "[]" in field: + # An array item is not a column, so substituting it into the query + # would produce SQL no backend can parse. + return [ + CheckSpec( + key=check_key, + category="quality", + type=check_type, + name=quality.description or "Quality Check", + model=model, + field=field, + metric=MetricType.UNSUPPORTED, + dimension=quality.dimension, + severity=quality.severity, + preset_result="warning", + preset_reason=( + f"'{field}' is an array item, not a column, so it cannot be substituted into a query. " + f"Declare the rule on '{field.split('[]')[0]}' instead and match the elements with array " + f"functions, for example size(filter(...)) > 0." + ), + ) + ] threshold = to_threshold(quality) query = prepare_query(quality, model, field, server) if query is None: diff --git a/datacontract/engines/ibis/ibis_check_execute.py b/datacontract/engines/ibis/ibis_check_execute.py index eca4652b2..7d3e6da16 100644 --- a/datacontract/engines/ibis/ibis_check_execute.py +++ b/datacontract/engines/ibis/ibis_check_execute.py @@ -279,11 +279,11 @@ def model_row_count() -> int: if spec.metric == MetricType.ROW_COUNT: named = t.count().name(spec.key) elif spec.metric == MetricType.MISSING_COUNT: - col = _resolve_col(columns, spec.field) - named = _count_true(_missing_expr(t, col, spec.missing_values)).name(spec.key) + predicate = _row_predicate(t, columns, spec.field, lambda c: _missing_expr(c, spec.missing_values)) + named = _count_true(predicate).name(spec.key) elif spec.metric == MetricType.INVALID_COUNT: - col = _resolve_col(columns, spec.field) - if _has_array_constraints(spec) and not schema[col].is_array(): + dtype = _resolve_dtype(schema, spec.field) + if _has_array_constraints(spec) and not dtype.is_array(): # Silently dropping the constraint would report the check as # passed, which is worse than saying it could not be run. _set_impl(run, spec.key, _describe(spec), None) @@ -291,11 +291,22 @@ def model_row_count() -> int: run, spec.key, ResultEnum.error, - f"Column {spec.field} is {schema[col]}, not an array, so the constraint cannot be measured.", + f"Column {spec.field} is {dtype}, not an array, so the constraint cannot be measured.", ) continue - expr = _invalid_expr(t, col, schema[col], spec) - if expr is None: + unconstrained = [] + + def _invalid(c, _t=t, _dtype=dtype, _spec=spec, _flag=unconstrained): + import ibis + + built = _invalid_expr(_t, c, _dtype, _spec) + if built is None: + _flag.append(True) + return ibis.literal(False) + return built + + expr = _row_predicate(t, columns, spec.field, _invalid) + if unconstrained: # No validity constraints => nothing can be invalid. _set_impl(run, spec.key, "invalid_count = 0 (no validity constraints configured)", None) _evaluate(run, spec, 0, row_count=model_row_count()) @@ -304,13 +315,13 @@ def model_row_count() -> int: elif spec.metric == MetricType.DUPLICATE_COUNT: _run_duplicate(run, t, columns, spec, model_row_count()) elif spec.metric == MetricType.FIELD_PRESENT: - _run_present(run, con, model, columns, spec) + _run_present(run, con, model, columns, schema, spec) elif spec.metric == MetricType.FIELD_TYPE: _run_type(run, schema, columns, spec, structured_types) elif spec.metric == MetricType.FIELD_PHYSICAL_TYPE: - _run_physical_type(run, con, server, schema, columns, native_types, spec, structured_types) + _run_physical_type(run, con, server, schema, native_types, spec, structured_types) elif spec.metric == MetricType.FIELD_NESTED_TYPE: - _run_nested_type(run, schema, columns, spec, structured_types, sqlglot_dialect(con)) + _run_nested_type(run, schema, spec, structured_types, sqlglot_dialect(con)) elif spec.metric in (MetricType.FRESHNESS, MetricType.RETENTION): _run_freshness(run, t, columns, spec) elif spec.metric == MetricType.CUSTOM_SQL: @@ -444,17 +455,31 @@ def _select_columns(columns, sensitive, wanted): def _samples_for(t, columns, schema, spec: CheckSpec, identifiers, sensitive): if spec.metric == MetricType.DUPLICATE_COUNT: - return _duplicate_samples(t, columns, sensitive, spec) - - col = _resolve_col(columns, spec.field) - if spec.metric == MetricType.MISSING_COUNT: - predicate = _missing_expr(t, col, spec.missing_values) + if not _is_item_duplicate(spec): + return _duplicate_samples(t, columns, sensitive, spec) + predicate = _item_duplicate_predicate(t, columns, spec.field) + elif spec.metric == MetricType.MISSING_COUNT: + predicate = _row_predicate(t, columns, spec.field, lambda c: _missing_expr(c, spec.missing_values)) else: # INVALID_COUNT - predicate = _invalid_expr(t, col, schema[col], spec) - if predicate is None: + dtype = _resolve_dtype(schema, spec.field) + unconstrained = [] + + def _invalid(c, _flag=unconstrained): + import ibis + + built = _invalid_expr(t, c, dtype, spec) + if built is None: + _flag.append(True) + return ibis.literal(False) + return built + + predicate = _row_predicate(t, columns, spec.field, _invalid) + if unconstrained: return None - select_cols = _select_columns(columns, sensitive, [*identifiers, spec.field]) + # A nested path is not a column; show the one it starts from. + root = spec.field.split("[]")[0].split(".")[0] if spec.field else None + select_cols = _select_columns(columns, sensitive, [*identifiers, root]) rows = t.filter(predicate) rows = rows.select(select_cols) if select_cols else rows return _df_to_records(rows.limit(_FAILED_SAMPLE_LIMIT).execute()) @@ -506,12 +531,12 @@ def _count_true(bool_expr): return bool_expr.ifelse(1, 0).sum() -def _missing_expr(t, col, missing_values): - cond = t[col].isnull() +def _missing_expr(col, missing_values): + cond = col.isnull() if missing_values: non_null = [v for v in missing_values if v is not None] if non_null: - cond = cond | t[col].isin(non_null) + cond = cond | col.isin(non_null) return cond @@ -612,27 +637,27 @@ def _valid_expr(t, col, dtype, spec: CheckSpec): """Boolean: a non-missing value satisfies all configured validity constraints.""" conds = [] if spec.valid_values is not None: - conds.append(t[col].isin(spec.valid_values)) + conds.append(col.isin(spec.valid_values)) if spec.valid_regex is not None: - conds.append(_regex_search_expr(t, _as_string(t[col], dtype), spec.valid_regex)) + conds.append(_regex_search_expr(t, _as_string(col, dtype), spec.valid_regex)) if spec.valid_min is not None: - conds.append(t[col] >= spec.valid_min) + conds.append(col >= spec.valid_min) if spec.valid_max is not None: - conds.append(t[col] <= spec.valid_max) + conds.append(col <= spec.valid_max) if spec.valid_min_length is not None: - conds.append(_as_string(t[col], dtype).length() >= spec.valid_min_length) + conds.append(_as_string(col, dtype).length() >= spec.valid_min_length) if spec.valid_max_length is not None: - conds.append(_as_string(t[col], dtype).length() <= spec.valid_max_length) + conds.append(_as_string(col, dtype).length() <= spec.valid_max_length) # Array constraints count the elements of the row's array. A column the # contract calls an array but the server does not cannot be measured that # way, so the constraint is left off rather than compiled into invalid SQL. if dtype is not None and dtype.is_array(): if spec.valid_min_items is not None: - conds.append(t[col].length() >= spec.valid_min_items) + conds.append(col.length() >= spec.valid_min_items) if spec.valid_max_items is not None: - conds.append(t[col].length() <= spec.valid_max_items) + conds.append(col.length() <= spec.valid_max_items) if spec.valid_unique_items: - conds.append(t[col].unique().length() == t[col].length()) + conds.append(col.unique().length() == col.length()) if not conds: return None expr = conds[0] @@ -643,13 +668,13 @@ def _valid_expr(t, col, dtype, spec: CheckSpec): def _invalid_expr(t, col, dtype, spec: CheckSpec): """Reproduce soda's invalid_count: NOT missing AND (NOT valid OR in invalid_values).""" - missing = _missing_expr(t, col, spec.missing_values) + missing = _missing_expr(col, spec.missing_values) valid = _valid_expr(t, col, dtype, spec) invalid_terms = [] if valid is not None: invalid_terms.append(~valid) if spec.invalid_values: - invalid_terms.append(t[col].isin(spec.invalid_values)) + invalid_terms.append(col.isin(spec.invalid_values)) if not invalid_terms: return None invalid_any = invalid_terms[0] @@ -698,7 +723,11 @@ def _run_duplicate(run: Run, t, columns, spec: CheckSpec, row_count: int): keys span are what is reported as failed.""" import pandas as pd - cols = [_resolve_col(columns, c) for c in (spec.columns or [spec.field])] + if _is_item_duplicate(spec): + _run_item_duplicate(run, t, columns, spec, row_count) + return + + cols = [_resolve_expr(t, columns, c) for c in (spec.columns or [spec.field])] grouped = t.group_by(cols).aggregate(_dup_n=t.count()) dup_groups = grouped.filter(grouped["_dup_n"] > 1) _record_sql(run, spec, dup_groups) @@ -718,17 +747,48 @@ def _int(value) -> int: _update_diagnostics(run, spec.key, extra) -def _run_present(run: Run, con, model: str, columns, spec: CheckSpec): +def _is_item_duplicate(spec: CheckSpec) -> bool: + return any("[]" in c for c in (spec.columns or [spec.field]) if c) + + +def _item_duplicate_predicate(t, columns, field: str): + """Parent rows whose array repeats a value in the item property ``field``.""" + array_path, _, leaf = field.rpartition("[].") + + def _repeats(array): + values = array.map(lambda element: _struct_path(element, leaf)) + return values.unique().length() < values.length() + + return _row_predicate(t, columns, array_path, _repeats) + + +def _run_item_duplicate(run: Run, t, columns, spec: CheckSpec, row_count: int): + """``unique`` on an array item property: no parent may repeat a value inside its own array. + + Counted in parent rows. + """ + bad = t.filter(_item_duplicate_predicate(t, columns, spec.field)) + _record_sql(run, spec, bad) + repeats = int(bad.count().execute()) + _evaluate(run, spec, repeats, row_count=row_count) + _update_diagnostics(run, spec.key, {"failed_rows": repeats}) + + +def _run_present(run: Run, con, model: str, columns, schema, spec: CheckSpec): target = f"{model}__raw__" if spec.uses_raw_view else model _set_impl(run, spec.key, f"column '{spec.field}' exists in {target}", "introspection") - present = set(columns.keys()) if spec.uses_raw_view: try: - raw = con.table(f"{model}__raw__") - present = {c.lower() for c in raw.columns} + raw = _resolve_table(con, f"{model}__raw__") + table = raw except Exception: - pass - ok = spec.field.lower() in present + table = _resolve_table(con, model) + target_schema = table.schema() + else: + # Reuse the already-resolved model schema to avoid an extra lookup that + # can fail on case-sensitive backends (for example Oracle). + target_schema = schema + ok = _field_present(target_schema, spec.field) _set_diagnostics(run, spec.key, _diag(metric="field_present", field=spec.field, present=ok)) set_result( run, @@ -745,12 +805,11 @@ def _run_type(run: Run, schema, columns, spec: CheckSpec, structured_types: dict f"type of '{spec.field}' is compatible with '{spec.expected_type_label}'", "introspection", ) - actual_col = columns.get(spec.field.lower()) - if actual_col is None: + dtype = _resolve_dtype(schema, spec.field) + if dtype is None: _set_diagnostics(run, spec.key, _diag(metric="field_type", field=spec.field, expected=spec.expected_type_label)) set_result(run, spec.key, ResultEnum.failed, f"Column '{spec.field}' is missing") return - dtype = schema[actual_col] # Snowflake structured types come back collapsed from ibis; prefer the nested # tree recovered from SHOW COLUMNS when available. structured_prop = structured_types.get(spec.field.lower()) if structured_types else None @@ -782,7 +841,6 @@ def _run_physical_type( con, server, schema, - columns, native_types, spec: CheckSpec, structured_types: dict[str, SchemaProperty] | None = None, @@ -800,8 +858,8 @@ def _run_physical_type( f"physical type of '{spec.field}' is '{spec.expected_physical_type}'", "introspection", ) - actual_col = columns.get(spec.field.lower()) - if actual_col is None: + dtype = _resolve_dtype(schema, spec.field) + if dtype is None: _set_diagnostics( run, spec.key, _diag(metric="field_physical_type", field=spec.field, expected=spec.expected_physical_type) ) @@ -839,16 +897,28 @@ def _run_physical_type( set_result(run, spec.key, ResultEnum.failed, reason) return + # Native types are read per top-level column, so a nested path never has one. + # The logicalType fallback would report `passed` for a physical type nothing + # compared, so skip instead. + if actual_native is None and ("." in spec.field or "[]" in spec.field): + set_result( + run, + spec.key, + ResultEnum.warning, + f"The native type of '{spec.field}' is not read for nested paths; skipping the physical type check", + ) + return + # result is None: the physical type could not be evaluated. Fall back to the # logicalType category check when the property declares one. fallback = spec.expected_schema_property if fallback is not None and fallback.logicalType is not None: - actual_prop = structured_prop or ibis_dtype_to_schema_property(schema[actual_col]) + actual_prop = structured_prop or ibis_dtype_to_schema_property(dtype) if schema_property_matches(fallback, actual_prop): set_result(run, spec.key, ResultEnum.passed, None) else: mismatch = schema_property_mismatch_reason(fallback, actual_prop) - actual_label = actual_native or schema[actual_col] + actual_label = actual_native or dtype set_result( run, spec.key, @@ -863,7 +933,6 @@ def _run_physical_type( def _run_nested_type( run: Run, schema, - columns, spec: CheckSpec, structured_types: dict[str, SchemaProperty] | None = None, dialect=None, @@ -873,13 +942,11 @@ def _run_nested_type( """ metric = spec.type _set_impl(run, spec.key, f"nested types of '{spec.field}' match the contract", "introspection") - actual_col = columns.get(spec.field.lower()) - if actual_col is None: + dtype = _resolve_dtype(schema, spec.field) + if dtype is None: _set_diagnostics(run, spec.key, _diag(metric=metric, field=spec.field, expected=spec.expected_type_label)) set_result(run, spec.key, ResultEnum.failed, f"Column '{spec.field}' is missing") return - - dtype = schema[actual_col] structured_prop = structured_types.get(spec.field.lower()) if structured_types else None actual_prop = structured_prop or ibis_dtype_to_schema_property(dtype) actual_label = (structured_prop.physicalType if structured_prop else None) or str(dtype) @@ -931,8 +998,8 @@ def _run_nested_type( def _run_freshness(run: Run, t, columns, spec: CheckSpec): import pandas as pd - col = _resolve_col(columns, spec.field) - reduction = t[col].min() if spec.metric == MetricType.RETENTION else t[col].max() + col = _resolve_expr(t, columns, spec.field) + reduction = col.min() if spec.metric == MetricType.RETENTION else col.max() _record_sql(run, spec, t.aggregate(value=reduction)) raw = reduction.execute() if raw is None or pd.isna(raw): @@ -1149,6 +1216,92 @@ def _resolve_col(columns: dict, field: str) -> str: return actual +def _resolve_expr(t, columns: dict, field: str): + if field is None: + raise _ColumnNotFound("Column 'None' not found") + if "." not in field: + return t[_resolve_col(columns, field)] + return _resolve_nested_expr(t, field, columns) + + +def _resolve_nested_expr(t, field: str, columns: dict): + head, _, tail = field.partition(".") + return _struct_path(t[_resolve_col(columns, head)], tail) + + +def _row_predicate(t, columns: dict, field: str, build): + """A boolean over *parent* rows for ``build`` applied at ``field``. + + An array hop (``items[].sku``) becomes "some element satisfies build", so the + row count never changes and an empty array is never a violation. + """ + head, marker, tail = field.partition("[].") + if not marker: + return build(_resolve_expr(t, columns, field)) + return _resolve_expr(t, columns, head).filter(lambda i: _element_predicate(i, tail, build)).length() > 0 + + +def _element_predicate(element, path: str, build): + head, marker, tail = path.partition("[].") + if not marker: + return build(_struct_path(element, path)) + return _struct_path(element, head).filter(lambda i: _element_predicate(i, tail, build)).length() > 0 + + +def _struct_path(value, path: str): + for part in path.split("."): + value = value[_struct_field_name(value, part)] + return value + + +def _struct_field_name(value, name: str) -> str: + """The struct's own spelling of ``name``, which a backend may report in another case.""" + try: + names = value.type().names + except Exception: + return name + if name in names: + return name + return next((n for n in names if n.lower() == name.lower()), name) + + +def _resolve_dtype(schema, field: str): + if field is None: + return None + current = schema + parts = field.split(".") + dtype = None + for idx, part in enumerate(parts): + name, marker, _ = part.partition("[]") + try: + dtype = current[name] + except Exception: + # Backends that report uppercase names (Snowflake, Oracle, Databricks) + # must still match a contract that spells the field in lower case. + actual = next((k for k in current.keys() if k.lower() == name.lower()), None) + if actual is None: + return None + dtype = current[actual] + if marker: + # `items[]` names the element type, not the array's own type. + try: + dtype = dtype.value_type + except Exception: + return None + if idx < len(parts) - 1: + try: + current = dtype.fields + except Exception: + return None + return dtype + + +def _field_present(schema, field: str) -> bool: + if field is None: + return False + return _resolve_dtype(schema, field) is not None + + def _table_database(con, server: Optional[Server]) -> Optional[str]: """The schema to qualify the table with during introspection, or ``None``. @@ -1203,7 +1356,7 @@ def _apply_row_filter(t, model: str, predicate: str): def _resolve_table(con, model: str, database: Optional[str] = None): - """Resolve a table by name, tolerating case differences across dialects.""" + """Resolve a table by name, tolerating case differences.""" if getattr(con, "name", None) == "pyspark": return _pyspark_table_unconvertible_as_unknown(con, model) kwargs = {"database": database} if database else {} diff --git a/tests/test_create_checks_nested.py b/tests/test_create_checks_nested.py new file mode 100644 index 000000000..eeba26412 --- /dev/null +++ b/tests/test_create_checks_nested.py @@ -0,0 +1,187 @@ +from open_data_contract_standard.model import Server + +from datacontract.data_contract import DataContract +from datacontract.engines.checks.check_spec import MetricType +from datacontract.engines.checks.create_checks import create_checks + +CONTRACT = """ +apiVersion: v3.0.2 +kind: DataContract +id: nested-checks +version: 1.0.0 +status: active +schema: + - name: orders + properties: + - name: id + logicalType: string + required: true + - name: user + logicalType: object + properties: + - name: email + logicalType: string + required: true + logicalTypeOptions: + pattern: ^.+@.+$ + - name: status + logicalType: string + quality: + - type: sql + query: SELECT COUNT(*) FROM {model} WHERE {field} NOT IN ('active', 'inactive') + mustBe: 0 + - name: emails + logicalType: array + items: + logicalType: object + properties: + - name: address + logicalType: string + required: true + - name: line_items + logicalType: array + items: + logicalType: object + properties: + - name: sku + logicalType: string + required: true + - name: product + logicalType: object + properties: + - name: tags + logicalType: array + items: + logicalType: object + properties: + - name: tag_id + logicalType: string + required: true +""" + + +def _checks(server_type: str): + odcs = DataContract(data_contract_str=CONTRACT).get_data_contract() + return create_checks(odcs, Server(type=server_type)) + + +def test_create_checks_recurses_for_dataframe_nested_structs_and_arrays(): + checks = _checks("dataframe") + + assert any(c.field == "user.email" and c.type == "field_required" and c.model == "orders" for c in checks) + assert any(c.field == "user.email" and c.type == "field_regex" and c.model == "orders" for c in checks) + assert any(c.field == "line_items[].sku" and c.type == "field_required" and c.model == "orders" for c in checks) + + nested_sql = next(c for c in checks if c.type == "field_quality_sql") + assert nested_sql.field == "user.status" + assert nested_sql.model == "orders" + assert nested_sql.metric == MetricType.CUSTOM_SQL + assert "user.status" in (nested_sql.query or "") + + +def test_create_checks_skips_nested_checks_for_unverified_backends(): + checks = _checks("postgres") + + assert not any(c.field == "user.email" for c in checks) + assert not any(c.field == "line_items[].sku" for c in checks) + + +def test_create_checks_marks_array_hops_in_the_field_path(): + checks = _checks("dataframe") + + assert any( + c.model == "orders" and c.field == "user.emails[].address" and c.type == "field_required" for c in checks + ) + assert any( + c.model == "orders" and c.field == "line_items[].product.tags[].tag_id" and c.type == "field_required" + for c in checks + ) + + +PHYSICAL_TYPE_CONTRACT = """ +apiVersion: v3.0.2 +kind: DataContract +id: nested-physical-types +version: 1.0.0 +status: active +schema: + - name: orders + properties: + - name: customer + logicalType: object + physicalType: STRUCT + properties: + - name: name + logicalType: string + physicalType: STRING + - name: line_items + logicalType: array + physicalType: ARRAY + items: + logicalType: object + physicalType: STRUCT + properties: + - name: sku + logicalType: string + physicalType: STRING +""" + + +def test_create_checks_keeps_array_item_checks_on_the_real_model(): + odcs = DataContract(data_contract_str=PHYSICAL_TYPE_CONTRACT).get_data_contract() + checks = create_checks(odcs, Server(type="databricks")) + + sku = next(c for c in checks if c.type == "field_physical_type" and c.field == "line_items[].sku") + assert sku.model == "orders" + + # A struct field keeps the parent model and is addressed by its dotted path. + name = next(c for c in checks if c.type == "field_physical_type" and c.field == "customer.name") + assert name.model == "orders" + + +ARRAY_QUALITY_CONTRACT = """ +apiVersion: v3.0.2 +kind: DataContract +id: array-quality +version: 1.0.0 +status: active +schema: + - name: orders + properties: + - name: customer + logicalType: object + properties: + - name: email + logicalType: string + quality: + - type: sql + query: SELECT COUNT(*) FROM {model} WHERE {field} IS NULL + mustBe: 0 + - name: items + logicalType: array + items: + logicalType: object + properties: + - name: sku + logicalType: string + quality: + - type: sql + query: SELECT COUNT(*) FROM {model} WHERE {field} IS NULL + mustBe: 0 +""" + + +def test_a_quality_query_on_an_array_item_warns_instead_of_emitting_unparseable_sql(): + odcs = DataContract(data_contract_str=ARRAY_QUALITY_CONTRACT).get_data_contract() + checks = {c.field: c for c in create_checks(odcs, Server(type="databricks")) if c.type == "field_quality_sql"} + + # a struct path is a legal column reference, so it still becomes a real query + assert checks["customer.email"].metric != MetricType.UNSUPPORTED + assert "customer.email" in checks["customer.email"].query + + # an array item is not, so the rule is reported as unsupported rather than run + item = checks["items[].sku"] + assert item.metric == MetricType.UNSUPPORTED + assert item.preset_result == "warning" + assert "items[].sku" in item.preset_reason + assert "Declare the rule on 'items'" in item.preset_reason diff --git a/tests/test_create_checks_nested_type.py b/tests/test_create_checks_nested_type.py index 0e7db942f..afd6bd687 100644 --- a/tests/test_create_checks_nested_type.py +++ b/tests/test_create_checks_nested_type.py @@ -12,7 +12,7 @@ Server, ) -from datacontract.engines.checks.check_spec import MetricType +from datacontract.engines.checks.check_spec import CheckSpec, MetricType from datacontract.engines.checks.create_checks import create_checks from datacontract.engines.ibis.ibis_check_execute import _run_nested_type, build_check_stubs from datacontract.engines.ibis.snowflake_structured_types import _to_property @@ -188,7 +188,7 @@ def _run(prop: SchemaProperty, dtype: str, server_type: str = "local", fmt: str run.checks = build_check_stubs(specs) spec = _nested(specs) field = prop.physicalName or prop.name - _run_nested_type(run, ibis.schema({field: dtype}), {field.lower(): field}, spec) + _run_nested_type(run, ibis.schema({field: dtype}), spec) return next(c for c in run.checks if c.key == spec.key) @@ -295,7 +295,7 @@ def _run_snowflake(prop: SchemaProperty, data_type: dict = _SHOW_COLUMNS_SIC_COD spec = _nested(specs) field = prop.physicalName or prop.name structured_types = {field.lower(): _to_property(data_type)} - _run_nested_type(run, ibis.schema({field: dtype}), {field.lower(): field}, spec, structured_types, "snowflake") + _run_nested_type(run, ibis.schema({field: dtype}), spec, structured_types, "snowflake") return next(c for c in run.checks if c.key == spec.key) @@ -436,3 +436,28 @@ def test_the_mismatch_reason_does_not_repeat_the_columns_own_structure(): assert check.result == ResultEnum.failed assert check.reason == "Cannot verify the nested types of 'primary_sic_code': the column is not an object" assert check.diagnostics["actual"] == "array>" + + +def test_nested_type_resolves_a_dotted_struct_path(): + spec = CheckSpec( + key="k", + category="schema", + type="field_nested_type", + name="nested type", + model="orders", + field="customer.address", + metric=MetricType.FIELD_TYPE, + expected_type_label="object", + expected_schema_property=SchemaProperty( + name="address", + logicalType="object", + properties=[SchemaProperty(name="city", logicalType="string")], + ), + ) + run = Run.create_run() + run.checks = build_check_stubs([spec]) + schema = ibis.schema({"customer": "struct>"}) + + _run_nested_type(run, schema, spec) + + assert run.checks[0].result == ResultEnum.passed diff --git a/tests/test_ibis_check_execute.py b/tests/test_ibis_check_execute.py new file mode 100644 index 000000000..e50e6c238 --- /dev/null +++ b/tests/test_ibis_check_execute.py @@ -0,0 +1,152 @@ +from datacontract.engines.checks.check_spec import CheckSpec, MetricType +from datacontract.engines.ibis.ibis_check_execute import _run_present +from datacontract.model.run import Check, ResultEnum, Run + + +class _FakeTable: + def __init__(self, schema): + self._schema = schema + + def schema(self): + return self._schema + + +class _NoLookupConnection: + def table(self, _name): + raise AssertionError("table() should not be called for non-raw field presence checks") + + +class _CaseSensitiveConnection: + def __init__(self, tables): + self._tables = tables + + def table(self, name): + if name in self._tables: + return self._tables[name] + raise KeyError(name) + + def list_tables(self): + return list(self._tables.keys()) + + +def _run_with_stubbed_check(key: str = "k") -> Run: + run = Run.create_run() + run.checks = [Check(type="field_is_present", key=key)] + return run + + +def test_run_present_uses_resolved_schema_without_extra_lookup(): + run = _run_with_stubbed_check() + spec = CheckSpec( + key="k", + category="schema", + type="field_is_present", + name="field is present", + model="checks_testcase", + field="CTC_ID", + metric=MetricType.FIELD_PRESENT, + ) + + _run_present(run, _NoLookupConnection(), "checks_testcase", {"ctc_id": "CTC_ID"}, {"CTC_ID": "int64"}, spec) + + assert run.checks[0].result == ResultEnum.passed + + +def test_run_present_raw_view_falls_back_to_model_with_case_insensitive_resolution(): + run = _run_with_stubbed_check() + spec = CheckSpec( + key="k", + category="schema", + type="field_is_present", + name="field is present", + model="checks_testcase", + field="CTC_ID", + metric=MetricType.FIELD_PRESENT, + uses_raw_view=True, + ) + con = _CaseSensitiveConnection({"CHECKS_TESTCASE": _FakeTable({"CTC_ID": "int64"})}) + + _run_present(run, con, "checks_testcase", {"ctc_id": "CTC_ID"}, {"IGNORED": "int64"}, spec) + + assert run.checks[0].result == ResultEnum.passed + + +def test_run_present_matches_uppercase_column_for_lowercase_contract_field(): + run = _run_with_stubbed_check() + spec = CheckSpec( + key="k", + category="schema", + type="field_is_present", + name="field is present", + model="checks_testcase", + field="ctc_id", + metric=MetricType.FIELD_PRESENT, + ) + + _run_present(run, _NoLookupConnection(), "checks_testcase", {"ctc_id": "CTC_ID"}, {"CTC_ID": "int64"}, spec) + + assert run.checks[0].result == ResultEnum.passed + + +def test_run_present_matches_uppercase_nested_field_for_lowercase_contract_path(): + import ibis + + run = _run_with_stubbed_check() + spec = CheckSpec( + key="k", + category="schema", + type="field_is_present", + name="field is present", + model="checks_testcase", + field="customer.name", + metric=MetricType.FIELD_PRESENT, + ) + schema = ibis.memtable({"CUSTOMER": [{"NAME": "a"}]}).schema() + + _run_present(run, _NoLookupConnection(), "checks_testcase", {"customer": "CUSTOMER"}, schema, spec) + + assert run.checks[0].result == ResultEnum.passed + + +def _physical_type_spec(field: str) -> CheckSpec: + from open_data_contract_standard.model import SchemaProperty + + return CheckSpec( + key="k", + category="schema", + type="field_physical_type", + name="physical type", + model="orders", + field=field, + metric=MetricType.FIELD_PHYSICAL_TYPE, + expected_category="DECIMAL(10,2)", + expected_physical_type="DECIMAL(10,2)", + expected_type_label="DECIMAL(10,2)", + expected_schema_property=SchemaProperty(name=field, logicalType="number"), + ) + + +def test_physical_type_of_a_nested_path_is_skipped_rather_than_passed_on_the_fallback(): + # fetch_native_types only reads top-level columns, so a nested path has no + # native type to compare against. The logicalType fallback would report a + # mismatched physical type as passed. + import ibis + from open_data_contract_standard.model import Server + + from datacontract.engines.ibis.ibis_check_execute import _run_physical_type + + schema = ibis.schema({"price": "float64", "orders": "array>"}) + native_types = {"price": "DOUBLE", "orders": "ARRAY>"} + server = Server(server="s", type="databricks") + + run = Run.create_run() + run.checks = [Check(type="field_physical_type", key="k")] + _run_physical_type(run, None, server, schema, native_types, _physical_type_spec("orders[].price")) + assert run.checks[0].result == ResultEnum.warning + assert "skipping the physical type check" in run.checks[0].reason + + # the same mismatch on a top-level column is still a real comparison + run = Run.create_run() + run.checks = [Check(type="field_physical_type", key="k")] + _run_physical_type(run, None, server, schema, native_types, _physical_type_spec("price")) + assert run.checks[0].result == ResultEnum.failed diff --git a/tests/test_nested_path_checks.py b/tests/test_nested_path_checks.py new file mode 100644 index 000000000..e1a9de896 --- /dev/null +++ b/tests/test_nested_path_checks.py @@ -0,0 +1,102 @@ +"""Array hops in a check's field path, and the predicates they compile to.""" + +import ibis +import pytest + +from datacontract.engines.ibis.ibis_check_execute import _missing_expr, _resolve_dtype, _row_predicate + +TABLE = ibis.table( + { + "order_id": "string", + "customer": "struct>>", + "items": "array>>>", + }, + name="orders", +) +COLUMNS = {c.lower(): c for c in TABLE.columns} + + +@pytest.mark.parametrize( + "path,expected", + [ + ("order_id", "string"), + ("customer.name", "string"), + ("items[].sku", "string"), + ("customer.tags[].tag_id", "string"), + ("items[].parts[].part_no", "string"), + ], +) +def test_resolve_dtype_steps_through_array_elements(path, expected): + assert str(_resolve_dtype(TABLE.schema(), path)) == expected + + +def test_resolve_dtype_is_case_insensitive_across_an_array_hop(): + assert str(_resolve_dtype(TABLE.schema(), "ITEMS[].SKU")) == "string" + + +def _sql(path): + predicate = _row_predicate(TABLE, COLUMNS, path, lambda c: _missing_expr(c, None)) + return " ".join(str(ibis.to_sql(TABLE.filter(predicate).count(), dialect="databricks")).split()) + + +def test_a_plain_path_compiles_to_a_column_predicate(): + assert "FILTER(" not in _sql("customer.name") + + +def test_an_array_hop_compiles_to_a_predicate_over_the_elements(): + # A predicate over the array keeps one row per parent, so counts stay in + # parent rows and an empty array is never a violation. + sql = _sql("items[].sku") + assert "SIZE(FILTER(`t0`.`items`" in sql + assert "`sku`)ISNULL" in sql.replace(" ", "") + + +def test_nested_array_hops_compile_to_nested_predicates(): + assert _sql("items[].parts[].part_no").count("FILTER(") == 2 + + +UPPERCASE = ibis.table( + {"ORDER_ID": "string", "CUSTOMER": "struct", "ITEMS": "array>"}, + name="ORDERS", +) +UPPERCASE_COLUMNS = {c.lower(): c for c in UPPERCASE.columns} + + +@pytest.mark.parametrize("path", ["customer.name", "items[].sku"]) +def test_a_path_resolves_to_a_value_when_the_backend_reports_uppercase_names(path): + # _resolve_dtype already folds case; the value side has to agree, or presence + # and type checks pass while the count checks raise KeyError. + predicate = _row_predicate(UPPERCASE, UPPERCASE_COLUMNS, path, lambda c: _missing_expr(c, None)) + assert "IS NULL" in str(ibis.to_sql(UPPERCASE.filter(predicate).count(), dialect="databricks")) + + +def test_item_duplicate_samples_use_the_same_predicate_as_the_check(): + # `unique` on an array item is not a column lookup, so the sample path has to + # take the array branch too or it silently reports no samples. + import pandas as pd + + from datacontract.engines.checks.check_spec import CheckSpec, MetricType + from datacontract.engines.ibis.ibis_check_execute import _samples_for + + t = ibis.memtable( + pd.DataFrame( + { + "order_id": ["clean", "repeats"], + "items": [[{"sku": "A"}, {"sku": "B"}], [{"sku": "C"}, {"sku": "C"}]], + } + ) + ) + spec = CheckSpec( + key="orders__items[].sku__field_unique", + category="schema", + type="field_unique", + name="unique", + model="orders", + field="items[].sku", + metric=MetricType.DUPLICATE_COUNT, + columns=["items[].sku"], + ) + samples = _samples_for(t, {c.lower(): c for c in t.columns}, t.schema(), spec, ["order_id"], set()) + + assert samples is not None + assert [row["order_id"] for row in samples] == ["repeats"] diff --git a/tests/test_snowflake_structured_types.py b/tests/test_snowflake_structured_types.py index 17e21081b..782b4cf89 100644 --- a/tests/test_snowflake_structured_types.py +++ b/tests/test_snowflake_structured_types.py @@ -166,9 +166,7 @@ def _physical_type_check(expected: SchemaProperty, structured_types): ) run = Run.create_run() run.checks = [Check(id="k", key="k", category="schema", type=spec.type, name=spec.name, model="m", field="s_obj")] - _run_physical_type( - run, None, None, {"S_OBJ": "map"}, {"s_obj": "S_OBJ"}, None, spec, structured_types - ) + _run_physical_type(run, None, None, {"S_OBJ": "map"}, None, spec, structured_types) return run.checks[0]