diff --git a/CHANGELOG.md b/CHANGELOG.md index 78a0c6c38..9446e20b6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed - `datacontract test` for Databricks no longer fails all checks of a model with a `GEOGRAPHY` or `GEOMETRY` column (#1483) +- `datacontract test` and `datacontract export sodacl` freshness and retention checks now honor the schema object's and property's `physicalName` (#1488) ## [1.1.0] - 2026-08-04 diff --git a/datacontract/engines/checks/create_checks.py b/datacontract/engines/checks/create_checks.py index cd9a67368..0af0e347b 100644 --- a/datacontract/engines/checks/create_checks.py +++ b/datacontract/engines/checks/create_checks.py @@ -778,11 +778,11 @@ def _to_servicelevel_checks(data_contract: OpenDataContractStandard, server: Opt return checks for sla in data_contract.slaProperties: if sla.property == "freshness": - check = _freshness_check(data_contract, sla) + check = _freshness_check(data_contract, sla, server) if check is not None: checks.append(check) elif sla.property == "retention": - check = _retention_check(data_contract, sla) + check = _retention_check(data_contract, sla, server) if check is not None: checks.append(check) return checks @@ -795,16 +795,34 @@ def _split_element(element: Optional[str]) -> Optional[tuple[str, str]]: return model, field -def _freshness_check(data_contract: OpenDataContractStandard, sla) -> Optional[CheckSpec]: - if sla.element is None or sla.value is None: - return None - parts = _split_element(sla.element) +def _resolve_sla_element( + data_contract: OpenDataContractStandard, element: Optional[str], server: Optional[Server] +) -> Optional[tuple[str, str]]: + """The (model, field) a contract-language sla element points at, in warehouse terms.""" + parts = _split_element(element) if parts is None: - logger.info("freshness element is not a single model.field, skipping") + logger.info(f"sla element {element!r} is not a single model.field, skipping") return None model, field = parts - if _get_schema_by_name(data_contract, model) is None: + schema_object = _get_schema_by_name(data_contract, model) + if schema_object is None: return None + server_type = server.type if server and server.type else None + prop = next((p for p in schema_object.properties or [] if p.name == field), None) + if prop is not None and prop.physicalName: + field = prop.physicalName + return to_schema_name(schema_object, server_type), field + + +def _freshness_check( + data_contract: OpenDataContractStandard, sla, server: Optional[Server] = None +) -> Optional[CheckSpec]: + if sla.element is None or sla.value is None: + return None + resolved = _resolve_sla_element(data_contract, sla.element, server) + if resolved is None: + return None + model, field = resolved unit = (sla.unit or "d").lower() if unit in ("d", "day", "days"): @@ -829,16 +847,15 @@ def _freshness_check(data_contract: OpenDataContractStandard, sla) -> Optional[C ) -def _retention_check(data_contract: OpenDataContractStandard, sla) -> Optional[CheckSpec]: +def _retention_check( + data_contract: OpenDataContractStandard, sla, server: Optional[Server] = None +) -> Optional[CheckSpec]: if sla.element is None or sla.value is None: return None - parts = _split_element(sla.element) - if parts is None: - logger.info("retention element is not a single model.field, skipping") - return None - model, field = parts - if _get_schema_by_name(data_contract, model) is None: + resolved = _resolve_sla_element(data_contract, sla.element, server) + if resolved is None: return None + model, field = resolved seconds = _retention_value_to_seconds(sla.value, sla.unit) if seconds is None: return None diff --git a/datacontract/export/sodacl_check_builder.py b/datacontract/export/sodacl_check_builder.py index c45551ecb..2f02d56a7 100644 --- a/datacontract/export/sodacl_check_builder.py +++ b/datacontract/export/sodacl_check_builder.py @@ -1006,6 +1006,15 @@ def _get_schema_by_name(data_contract: OpenDataContractStandard, name: str) -> O return next((s for s in data_contract.schema_ if s.name == name), None) +def _resolve_physical_names(schema_object: SchemaObject, field_name: str, server: Optional[Server]) -> tuple[str, str]: + """The (dataset, column) a servicelevel check must scan, in warehouse terms.""" + server_type = server.type if server and server.type else None + prop = next((p for p in schema_object.properties or [] if p.name == field_name), None) + if prop is not None and prop.physicalName: + field_name = prop.physicalName + return to_schema_name(schema_object, server_type), field_name + + def to_servicelevel_checks(data_contract: OpenDataContractStandard, server: Optional[Server] = None) -> List[Check]: checks: List[Check] = [] if data_contract.slaProperties is None: @@ -1017,7 +1026,7 @@ def to_servicelevel_checks(data_contract: OpenDataContractStandard, server: Opti if check is not None: checks.append(check) elif sla.property == "retention": - check = to_servicelevel_retention_check(data_contract, sla) + check = to_servicelevel_retention_check(data_contract, sla, server) if check is not None: checks.append(check) @@ -1054,6 +1063,7 @@ def to_sla_freshness_check( if schema is None: logger.info(f"Model {model_name} not found in schema, skipping freshness check") return None + model_name, field_name = _resolve_physical_names(schema, field_name, server) # Build threshold from value and unit unit = sla.unit.lower() if sla.unit else "d" @@ -1096,7 +1106,9 @@ def to_sla_freshness_check( ) -def to_servicelevel_retention_check(data_contract: OpenDataContractStandard, sla) -> Check | None: +def to_servicelevel_retention_check( + data_contract: OpenDataContractStandard, sla, server: Optional[Server] = None +) -> Check | None: """Create a retention check from an ODCS retention SLA property.""" if sla.element is None: logger.info("slaProperties.retention.element is not defined, skipping retention check") @@ -1124,6 +1136,7 @@ def to_servicelevel_retention_check(data_contract: OpenDataContractStandard, sla if schema is None: logger.info(f"Model {model_name} not found in schema, skipping retention check") return None + model_name, field_name = _resolve_physical_names(schema, field_name, server) # Convert retention value to seconds # Supports both numeric value + unit (ODCS style: value=3, unit=y) diff --git a/tests/fixtures/postgres/servicelevels.odcs.yaml b/tests/fixtures/postgres/servicelevels.odcs.yaml index e3bcd2add..69db3827d 100644 --- a/tests/fixtures/postgres/servicelevels.odcs.yaml +++ b/tests/fixtures/postgres/servicelevels.odcs.yaml @@ -6,7 +6,7 @@ version: 0.0.1 domain: my-domain-team status: active schema: -- name: my_table +- name: my_logical_table physicalName: my_table logicalType: object physicalType: table @@ -41,4 +41,4 @@ slaProperties: - property: freshness value: 1 unit: h - element: my_table.field_three + element: my_logical_table.field_three diff --git a/tests/test_create_checks_servicelevel.py b/tests/test_create_checks_servicelevel.py new file mode 100644 index 000000000..2c3e2a818 --- /dev/null +++ b/tests/test_create_checks_servicelevel.py @@ -0,0 +1,87 @@ +"""Freshness/retention checks target physicalName, like every other check. + +The sla element speaks contract language; the engine reads the warehouse, so an +object or property named differently there must be read by its physicalName. +""" + +from open_data_contract_standard.model import ( + OpenDataContractStandard, + SchemaObject, + SchemaProperty, + Server, + ServiceLevelAgreementProperty, +) + +from datacontract.engines.checks.create_checks import create_checks + + +def _contract(schema_object: SchemaObject, *slas: ServiceLevelAgreementProperty): + return OpenDataContractStandard( + version="1", + kind="DataContract", + apiVersion="v3.1.0", + id="x", + schema=[schema_object], + slaProperties=list(slas), + ) + + +def _of_type(checks, check_type): + return [c for c in checks if c.type == check_type] + + +def test_servicelevel_checks_resolve_physical_names(): + """An object and property with physicalName are measured under those names.""" + schema = SchemaObject( + name="events", + physicalName="events_v1", + properties=[SchemaProperty(name="ts", physicalName="TS", logicalType="timestamp")], + ) + contract = _contract( + schema, + ServiceLevelAgreementProperty(property="freshness", element="events.ts", value=24, unit="h"), + ServiceLevelAgreementProperty(property="retention", element="events.ts", value=1, unit="y"), + ) + + checks = create_checks(contract, Server(server="s", type="snowflake")) + + for check_type in ("servicelevel_freshness", "servicelevel_retention"): + (check,) = _of_type(checks, check_type) + assert check.model == "events_v1" + assert check.field == "TS" + + +def test_servicelevel_checks_fall_back_to_names_without_physical_names(): + """Without physicalName, service level checks use the logical names (unchanged).""" + schema = SchemaObject( + name="events", + properties=[SchemaProperty(name="ts", logicalType="timestamp")], + ) + contract = _contract( + schema, + ServiceLevelAgreementProperty(property="freshness", element="events.ts", value=24, unit="h"), + ) + + checks = create_checks(contract, Server(server="s", type="snowflake")) + + (check,) = _of_type(checks, "servicelevel_freshness") + assert check.model == "events" + assert check.field == "ts" + + +def test_servicelevel_checks_keep_kafka_logical_name(): + """to_schema_name reads the Spark SQL view (logical name) on kafka, not the topic.""" + schema = SchemaObject( + name="events", + physicalName="events-topic", + properties=[SchemaProperty(name="ts", logicalType="timestamp")], + ) + contract = _contract( + schema, + ServiceLevelAgreementProperty(property="freshness", element="events.ts", value=24, unit="h"), + ) + + checks = create_checks(contract, Server(server="s", type="kafka")) + + (check,) = _of_type(checks, "servicelevel_freshness") + assert check.model == "events" diff --git a/tests/test_data_contract_checks.py b/tests/test_data_contract_checks.py index e169d245d..254770394 100644 --- a/tests/test_data_contract_checks.py +++ b/tests/test_data_contract_checks.py @@ -23,6 +23,7 @@ create_checks, prepare_query, to_schema_checks, + to_servicelevel_checks, to_sla_freshness_check, ) @@ -572,6 +573,69 @@ def test_field_checks_fall_back_to_name_without_physical_name(): assert present.field == "sku" +def test_servicelevel_checks_use_physical_names_when_set(): + """Freshness/retention SodaCL must target physicalName, like the schema checks. + + The `checks for X` block names the dataset Soda scans; the sla element + speaks contract language. Without resolution, a contract whose object + name differs from the relation name emits checks against a dataset that + does not exist. + """ + contract = OpenDataContractStandard( + kind="DataContract", + apiVersion="v3.1.0", + id="freshness-test", + schema=[ + SchemaObject( + name="events", + physicalName="events_v1", + properties=[SchemaProperty(name="ts", physicalName="TS", logicalType="timestamp")], + ) + ], + ) + contract.slaProperties = [ + ServiceLevelAgreementProperty(property="freshness", element="events.ts", value=24, unit="h"), + ServiceLevelAgreementProperty(property="retention", element="events.ts", value=1, unit="y"), + ] + + freshness, retention = to_servicelevel_checks(contract, Server(type="snowflake")) + + assert freshness.model == "events_v1" + fresh_impl = yaml.safe_load(freshness.implementation) + assert list(fresh_impl["checks for events_v1"][0].keys()) == ["freshness(TS) < 24h"] + + assert retention.model == "events_v1" + ret_impl = yaml.safe_load(retention.implementation) + (entry,) = ret_impl["checks for events_v1"] + ((_, config),) = entry.items() + assert "MIN(TS)" in config["events_v1_servicelevel_retention expression"] + + +def test_servicelevel_checks_keep_kafka_logical_name(): + """to_schema_name reads the Spark SQL view (logical name) on kafka, not the topic.""" + contract = OpenDataContractStandard( + kind="DataContract", + apiVersion="v3.1.0", + id="freshness-test", + schema=[ + SchemaObject( + name="events", + physicalName="events-topic", + properties=[SchemaProperty(name="ts", logicalType="timestamp")], + ) + ], + ) + contract.slaProperties = [ + ServiceLevelAgreementProperty(property="freshness", element="events.ts", value=24, unit="h"), + ] + + (freshness,) = to_servicelevel_checks(contract, Server(type="kafka")) + + assert freshness.model == "events" + fresh_impl = yaml.safe_load(freshness.implementation) + assert "checks for events" in fresh_impl + + # --------------------------------------------------------------------------- # Databricks varchar/map type-check skip (#1245, #1219) # --------------------------------------------------------------------------- diff --git a/tests/test_test_postgres.py b/tests/test_test_postgres.py index d5dc573ca..746b491d4 100644 --- a/tests/test_test_postgres.py +++ b/tests/test_test_postgres.py @@ -113,6 +113,12 @@ def test_test_postgres_servicelevels_freshness_should_fail_odcs(postgres_contain print(run.pretty()) assert run.result == "failed" + # Must fail because the data is stale (a measurement happened), not because + # the logical schema object name was read as the relation name. + freshness = next(c for c in run.checks if c.type == "servicelevel_freshness") + assert freshness.result == "failed" + assert freshness.diagnostics is not None + assert freshness.diagnostics["age_seconds"] > 3600 def _setup_datacontract(file):