Resolve physicalName in servicelevel freshness and retention checks

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
 

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

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)

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

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"

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)
 # ---------------------------------------------------------------------------

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):