diff --git a/CHANGELOG.md b/CHANGELOG.md index f9b292c6..044e8b23 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,91 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +## [0.24.0] — 2026-10-27 + +### Fixed + +- **Exported lineage claimed a version the contract never pinned.** An ODPS `inputPorts[].version` + came from the consume's own `version` or the provider's `1.0.0` default, and a consume that states + a range (`versionConstraint: ^2.1`) has neither — so every such export asserted `version: 1.0.0`, + a pin the contract did not make and, for a product whose 1.x line is retired, a version that no + longer exists. The lowest version the range accepts is now resolved (`^2.1` → `2.1.0`, + `~1.4` → `1.4.0`, `>=3.0.0` → `3.0.0`). A range the resolver cannot read falls back to the old + constant rather than guessing: ODPS v1.0.0 has `version` in `InputPort`'s `required`, so the field + cannot simply be omitted, and a wrong exact version reads as a fact somebody established. The + refusals include a compound range, a wildcard, `>1.0.0` (whose true lower bound semver gives no + way to name), and anything whose components are not plain ASCII digits without a leading zero — + `^01.2` is a schema-valid constraint, but `01.2.0` matches neither `$defs.semver` nor any + resolver. `versionConstraint` still rides along in `customProperties`, so the range this product + accepts is not lost. + +- **An exported output port named none of its upstreams.** `outputPorts[].inputContracts` was + populated only from an imported ODPS document's pass-through, so a contract written in FLUID + exported output ports with no lineage at all and a catalog reading dependencies there saw a + product with none. The entries are now derived from the document's own resolved input ports — one + per `consumes[]`, carrying exactly the `id` and `version` Bitol's `$defs.InputContract` allows, + read from those ports rather than recomputed so the two halves of one document cannot disagree. An + imported document's own `inputContracts` still win, and `expects[]` — the carrier an imported + document's input ports arrive on — is excluded, so import → export stays a fixed point and never + adds a dependency the author did not write. A `builds[].properties.source` stream is excluded too: + its `:///` id is synthesized to populate the *required* input-port field + for a source that has no contract at all, and `inputContracts` is optional, so naming an id no + catalog can resolve is worse than naming nothing. Both Data Mesh Manager publish paths — the + registrar behind `fluid publish --target datamesh-manager` and `catalog.register`, and + `fluid datamesh-manager publish` — strip the derived lineage alongside the input ports it mirrors, + for the reason those were already stripped: Entropy draws product-to-product lineage from Access + resources, so naming the upstream twice would require mirroring it as a SourceSystem and render + duplicate graph nodes. They strip by different rules, so each passes the ports it dropped to one + shared helper. + +- **A quality expression exported as SQL no DQ engine could execute.** ODCS defines the `sql` rule's + `query` as an individual query returning a value that can be compared, and FLUID's + `exposes[].contract.quality[].expression` is a row predicate — it was emitted verbatim, so an + engine was handed `preferred_customer_flag IN ('Y', 'N')` with `mustBe: true` and nothing to + execute. A predicate now becomes a counting query compared to zero, against the `${table}` + placeholder from ODCS's own example. The negation is `(…) IS NOT TRUE`, not `NOT (…)`: under + three-valued logic a NULL row makes the naive form NULL and counts no failure, although that row's + predicate did not hold. An expression whose *first token* is `SELECT` or `WITH` passes through + unchanged; the token boundary matters, because matching a bare prefix left every predicate on a + column named `selected_at` or `with_tax` in the broken state this fixes. The predicate is wrapped + on its own line and a trailing `;` is dropped, so neither a line comment nor a stray semicolon can + swallow the closing paren. + +- **A renamed input port leaked a product consume into the Data Mesh Manager payload.** Found while + covering the lineage strip above. When two `consumes[]` share an `exposeId` the exporter renames + the second, and `fluid datamesh-manager publish` matched the renamed port by name — which misses + two of the three rename shapes: a `productId` tail containing `__` of its own + (`silver.crm__core` → `crm__core__customers`, split at the wrong `__`), and the numeric-suffix + fallback (`customers_2`, with no `__` to match on). Both published a port Entropy cannot resolve + without mirroring the upstream product as a SourceSystem. The match now also tests the contract id + the exporter actually stamps on the port, read from the exporter's own helper so the convention has + one definition; a source-system consume keeps its port, as before. This leak predates the release — + but deriving `outputPorts[].inputContracts` would have doubled it, so it is fixed here rather than + carried. + +### Added + +- `consumes[].contractVersion` (0.7.6 preview) — the exact upstream version this product was built + and tested against, within `versionConstraint`. It wins over the resolved lower bound as the + version an exported input port, and its matching `inputContracts` entry, claims. + +### Changed + +**Exported artifacts change shape in this release.** If you have tooling that reads forge's ODCS or +ODPS output, read this before upgrading. Each change is a fix for output that was wrong and is +described in full under *Fixed*; the summary is here because the old values may be baked into +consumers. + +- An ODPS export of a contract with `consumes[]` now carries `outputPorts[].inputContracts`, where + before it carried none, and an input port's `version` may differ from `1.0.0` where the contract + states a range. A consumer that compares exported documents byte-for-byte will see both. +- An ODCS `sql` quality rule derived from `exposes[].contract.quality[].expression` now carries a + counting query (`SELECT COUNT(*) FROM ${table} WHERE () IS NOT TRUE`, the predicate on + its own line) with `mustBe: 0`, where it previously carried the bare expression with + `mustBe: true`; its `description` changes with it. A DQ configuration or test that pins the old + `query` string or the boolean `mustBe` must be updated — ours was. An expression already written + as a `SELECT` or `WITH` query is unchanged. + ## [0.23.0] — 2026-10-27 ### Security diff --git a/fluid_build/build_runners/catalog_registrars/datamesh_manager.py b/fluid_build/build_runners/catalog_registrars/datamesh_manager.py index 564befa0..000c1e3e 100644 --- a/fluid_build/build_runners/catalog_registrars/datamesh_manager.py +++ b/fluid_build/build_runners/catalog_registrars/datamesh_manager.py @@ -421,15 +421,44 @@ def _render_dmm_data_product_body( parsed, payload ) if "inputPorts" in parsed and isinstance(parsed["inputPorts"], list): - parsed["inputPorts"] = [ - ip - for ip in parsed["inputPorts"] - if isinstance(ip, dict) - and "sourceSystemId" in ip - and ip["sourceSystemId"] - ] - if not parsed["inputPorts"]: + # Deferred: this module is imported purely for its + # ``register_catalog_backend`` side effect, so building + # the backend registry does not pull the ODPS provider + # in with it. + from fluid_build.providers.odps_standard.mappers.ports import ( + consume_input_port_contract_ids, + remove_output_port_input_contracts, + ) + + kept: List[Any] = [] + removed: List[Any] = [] + for ip in parsed["inputPorts"]: + keep = isinstance(ip, dict) and ip.get("sourceSystemId") + (kept if keep else removed).append(ip) + parsed["inputPorts"] = kept + if not kept: parsed.pop("inputPorts", None) + # The ODPS render derives + # ``outputPorts[].inputContracts`` from the same + # ``consumes[]``, so a stripped port stays named in the + # output port's lineage unless it goes with it — which + # would put back the unresolvable SourceSystem + # reference this strip exists to remove. + # + # Only the consume-derived ones, though. The rule above + # also drops an ``expects[]`` port, whose expose replays + # the upstream document's OWN ``inputContracts``; that + # is the only record of the dependency, since Access + # agreements are built from ``consumes[]``. + consume_ids = consume_input_port_contract_ids(payload.contract or {}) + remove_output_port_input_contracts( + parsed, + [ + ip + for ip in removed + if isinstance(ip, dict) and str(ip.get("contractId")) in consume_ids + ], + ) return parsed except Exception: # noqa: BLE001 — fall through to native LOG.debug( diff --git a/fluid_build/providers/datamesh_manager/_odps_helpers.py b/fluid_build/providers/datamesh_manager/_odps_helpers.py index eb03a562..adedb1d7 100644 --- a/fluid_build/providers/datamesh_manager/_odps_helpers.py +++ b/fluid_build/providers/datamesh_manager/_odps_helpers.py @@ -44,6 +44,10 @@ import logging from typing import Any, Dict, List, Mapping, Optional +from fluid_build.providers.odps_standard.mappers.ports import ( + input_port_contract_id, + remove_output_port_input_contracts, +) from fluid_build.util.contract import consumes_to_canonical_ports, kind_to_dmm_type LOG = logging.getLogger("fluid.providers.datamesh_manager.odps") @@ -180,6 +184,9 @@ def remove_odps_product_consume_input_ports( duplicate graph nodes. Explicit source-system consumes remain as input ports. + A removed port's ``outputPorts[].inputContracts`` entry goes with it — see + :func:`fluid_build.providers.odps_standard.mappers.ports.remove_output_port_input_contracts`. + Matching strategy: we identify product-consume ports by **both** ``(id, name)`` (the canonical exposeId — matches the inputPort before the dedup-rename in :func:`fluid_build.providers.odps_standard.mappers @@ -197,6 +204,12 @@ def remove_odps_product_consume_input_ports( product_port_names: set[str] = set() product_contract_ids: set[str] = set() + # The id the exporter actually stamps on a product-consume's input port, + # read from the same helper the exporter uses so the convention has one + # definition. Matching on it catches a renamed port whatever shape the + # rename took, including the numeric-suffix fallback that carries no + # ``__`` for the name checks below to find. + product_derived_contract_ids: set[str] = set() for canonical in consumes_to_canonical_ports(fluid, logger=LOG): reference = canonical.get("reference") if not reference or canonical.get("source_system_id"): @@ -206,6 +219,9 @@ def remove_odps_product_consume_input_ports( if value: product_port_names.add(str(value)) product_contract_ids.add(str(reference)) + derived = input_port_contract_id(canonical) + if derived: + product_derived_contract_ids.add(str(derived)) if not product_port_names and not product_contract_ids: return @@ -214,6 +230,19 @@ def _is_product_consume(port: Any) -> bool: if not isinstance(port, Mapping): return False name = str(port.get("name", "")) + # An exact match on the derived contract id is the most specific + # signal there is: it is ``{productId}.{exposeId}`` or an id the + # author pinned, not the bare productId the checks below fall back + # to, so it cannot catch a source-system port that merely shares an + # upstream. Being independent of the NAME is what matters — the name + # checks below miss two of the three rename shapes, a productId tail + # containing ``__`` of its own (``silver.crm__core`` → + # ``crm__core__customers``, split at the wrong ``__``) and the + # numeric-suffix fallback (``customers_2``, no ``__`` to find at + # all). Both leaked a port Entropy cannot resolve, and since the + # lineage key mirrors the ports, they leaked twice. + if str(port.get("contractId", "")) in product_derived_contract_ids: + return True # Fast path — original (pre-Gap-3) behavior. A port whose name is # exactly a canonical product-consume exposeId is a product-consume # regardless of contractId. This is the legacy path that @@ -243,11 +272,15 @@ def _is_product_consume(port: Any) -> bool: return True return False - retained = [port for port in input_ports if not _is_product_consume(port)] + retained: List[Any] = [] + removed: List[Any] = [] + for port in input_ports: + (removed if _is_product_consume(port) else retained).append(port) if retained: odps_payload["inputPorts"] = retained else: odps_payload.pop("inputPorts", None) + remove_output_port_input_contracts(odps_payload, removed) def ensure_odps_input_port_contract_ids( diff --git a/fluid_build/providers/odcs/mappers/quality.py b/fluid_build/providers/odcs/mappers/quality.py index 97e23b91..4b983647 100644 --- a/fluid_build/providers/odcs/mappers/quality.py +++ b/fluid_build/providers/odcs/mappers/quality.py @@ -446,20 +446,65 @@ def _dq_severity(check: Dict[str, Any], rule: Mapping[str, Any]) -> None: check["severity"] = _DQ_SEVERITY.get(severity, severity) +# ``\b`` so only a whole first token counts: a column named ``selected_at`` or +# ``with_tax`` begins with these letters and is a predicate, not a query. +_QUERY_FIRST_TOKEN_RE = re.compile(r"\s*(?:select|with)\b", re.IGNORECASE) + + def _expression_to_quality(expr: Mapping[str, Any]) -> Dict[str, Any]: """One ``exposes[].contract.quality`` entry as an ODCS ``sql`` rule. - The FLUID field is "SQL or boolean expression that must evaluate to true", - which is exactly ODCS ``$defs.DataQualitySql`` (``query`` plus one operator) - with ``mustBe: true``. ``mustBe`` is untyped in the spec, so a boolean is - as valid there as a row count. + The FLUID field is "SQL or boolean expression that must evaluate to true" — + a ROW PREDICATE, such as ``preferred_customer_flag IN ('Y', 'N')``. ODCS is + specific about what ``sql`` means: ``$defs.DataQuality.type`` says "'sql' is + an individual SQL query that returns a value that can be compared", and + ``$defs.DataQualitySql.query`` carries the example ``SELECT COUNT(*) FROM + ${table} WHERE ${column} IS NOT NULL``. Emitting the bare predicate as + ``query`` therefore handed an engine something it cannot execute: it was a + fragment, not a statement, and the ``mustBe: true`` beside it compared + against a value no query produced. + + A predicate is now exported as the count of rows that fail it, compared to + zero — the shape ODCS's own example takes, and one its + ``$defs.DataQualityOperators`` ``oneOf`` accepts. ``${table}`` is the + placeholder from that example. + + ``IS NOT TRUE`` rather than ``NOT (...)`` on purpose. Under SQL's + three-valued logic ``NOT (flag IN ('Y','N'))`` is NULL when ``flag`` is + NULL, so a naive negation counts no failure for the row — yet a row whose + predicate did not evaluate to TRUE has not satisfied "must evaluate to + true". ``IS NOT TRUE`` catches NULL and FALSE alike, and is supported by + DuckDB, PostgreSQL, MySQL, Snowflake and BigQuery. + + An expression that is already a query (its first token is ``SELECT`` or + ``WITH``) is passed through as written with the previous comparison: the + author wrote a statement and chose what it returns. The boundary matters — + ``startswith`` alone sent ``selected_at IS NOT NULL`` and ``with_tax > 0`` + down that branch and exported them unchanged, reinstating the very fragment + this is here to remove. + + A predicate still containing a ``;`` of its own is left as the author wrote + it: the field is specified as one predicate, and a second statement in it is + not something to guess at. """ - check: Dict[str, Any] = { - "type": "sql", - "query": expr["expression"], - "mustBe": True, - "description": f"Expression must hold for every row: {expr['expression']}", - } + expression = str(expr["expression"]).strip() + # A trailing ``;`` would end up inside the wrapper's parentheses below. + expression = expression.rstrip(";").rstrip() + is_query = bool(_QUERY_FIRST_TOKEN_RE.match(expression)) + + check: Dict[str, Any] = {"type": "sql"} + if is_query: + check["query"] = expression + check["mustBe"] = True + check["description"] = f"Expression must hold for every row: {expression}" + else: + # The predicate gets its own line: a trailing ``-- comment`` on one line + # with the wrapper swallowed ``) IS NOT TRUE`` and left a statement the + # engine could not parse. + check["query"] = f"SELECT COUNT(*) FROM ${{table}} WHERE (\n{expression}\n) IS NOT TRUE" + check["mustBe"] = 0 + check["description"] = f"Rows failing the expression must be 0: {expression}" + severity = expr.get("severity") if severity: check["severity"] = _DQ_SEVERITY.get(severity, severity) diff --git a/fluid_build/providers/odps_standard/mappers/ports.py b/fluid_build/providers/odps_standard/mappers/ports.py index d1fd6ea4..c83aaf82 100644 --- a/fluid_build/providers/odps_standard/mappers/ports.py +++ b/fluid_build/providers/odps_standard/mappers/ports.py @@ -36,8 +36,8 @@ from __future__ import annotations -from collections.abc import Mapping -from typing import Any, Dict, List, Optional +from collections.abc import Iterable, Mapping, MutableMapping +from typing import Any, Dict, List, Optional, Set, Tuple from fluid_build.util.contract import ( builds_to_canonical_input_ports, @@ -62,6 +62,56 @@ # open slot the closed object leaves, under a name that says whose it is. VERSION_CONSTRAINT_PROPERTY = "fluidVersionConstraint" + +def _range_lower_bound(constraint: Any) -> Optional[str]: + """The lowest version a semver range accepts, as an exact semver string. + + Only the forms ``$defs.semverRange`` documents: ``^1.2``, ``~1.4``, + ``>=1.0.0`` and an exact version. Anything else returns ``None`` so the + caller falls back rather than guesses — a wrong exact version in exported + lineage is worse than the constant it replaces, because it reads as a fact + somebody established. ``>1.0.0`` is among the refusals on purpose: it does + not accept 1.0.0, and its true lower bound is the next version, which semver + gives no way to name. + + Deliberately not a semver library: none of ``semver``, + ``semantic_version`` or ``packaging`` is a declared dependency of this + project, and the CLI's core is kept lean on purpose. ``schema_manager.VersionConstraint`` is also not reused: it rejects + the caret outright ("Invalid FLUID version format"), does not normalise to + three components, and is the FLUID *schema* version type — pointing it at an + upstream data product's version would overload it and put ``--min-version`` + at risk for no gain. + """ + if not isinstance(constraint, str): + return None + text = constraint.strip() + if not text or "," in text or "||" in text or " " in text: + return None + for prefix in ("^", "~", ">=", "="): + if text.startswith(prefix): + text = text[len(prefix) :].strip() + break + if not text or any(c in text for c in "<>=*xX^~"): + return None + parts = text.split(".") + # ``isascii()`` as well as ``isdigit()``: the latter is true for Arabic-Indic + # digits and superscripts, so ``^١.٢`` would otherwise be read as a range and + # its "lower bound" emitted verbatim into a version field. A leading zero is + # refused for the same reason — ``^01.2`` is a schema-valid constraint + # (``semverRange`` carries no pattern) but ``01.2.0`` matches neither + # ``$defs.semver`` nor any resolver, so it is worse than the fallback. + if not 1 <= len(parts) <= 3 or not all( + p.isascii() and p.isdigit() and (p == "0" or not p.startswith("0")) for p in parts + ): + return None + # ``^2.1`` and ``~1.4`` accept 2.1.0 and 1.4.0 as their lowest member; an + # omitted component is zero, which is how npm's range grammar — the one + # ``$defs.semverRange``'s examples come from — reads a partial version. + while len(parts) < 3: + parts.append("0") + return ".".join(parts) + + # ----- FLUID → ODPS -------------------------------------------------------- @@ -69,24 +119,186 @@ def to_odps(ctx: ExportCtx) -> None: fluid = ctx.fluid product_id = fluid_id(fluid) or "product" + # Input ports are resolved first because an output port's ``inputContracts`` + # names them: deriving that list from the already-resolved + # ``(contractId, version)`` pairs is what keeps the two halves of one + # document agreeing, instead of two sites each deciding a version. Only the + # authored subset feeds the derivation — see ``_build_input_ports``. The + # ``ctx.odps`` keys are still assigned outputPorts-first below, so the + # emitted key order is unchanged. + input_ports, authored_ports = _build_input_ports(fluid, ctx, product_id) + derived_input_contracts = _input_contracts_from_ports(authored_ports) + output_ports = [] for expose in fluid.get("exposes") or []: if not isinstance(expose, Mapping): continue - port = _expose_to_output_port(expose, product_id) + port = _expose_to_output_port(expose, product_id, derived_input_contracts) if port: output_ports.append(port) if output_ports: ctx.odps["outputPorts"] = output_ports - input_ports = _build_input_ports(fluid, ctx, product_id) if input_ports: ctx.odps["inputPorts"] = input_ports +def _input_contracts_from_ports( + input_ports: List[Dict[str, Any]], +) -> List[Dict[str, Any]]: + """The ``outputPorts[].inputContracts`` a product's own consumes imply. + + Takes the *authored* input ports only (``_build_input_ports``' second + return value), never the full list. + + Every output port receives the same entries. ODPS models the key per port, + but FLUID's ``consumes[]`` is product-level — ``$defs.consumeRef`` has no + expose selector — so there is no per-expose upstream mapping to narrow it + with, and claiming one would be invention. + + Bitol ``$defs.InputContract`` is ``additionalProperties: false`` with + ``required: [id, version]`` and no other properties, so an entry is exactly + the input port's resolved ``contractId`` and ``version`` — read from the + ports rather than recomputed, so the lineage cannot name a contract or a + version the document's own input ports do not. + + Until this existed, ``inputContracts`` came only from an imported ODPS + document's pass-through, so a contract written in FLUID exported output + ports that named none of their upstreams and a catalog reading lineage + there saw a product with no dependencies at all. + """ + contracts: List[Dict[str, Any]] = [] + seen: set = set() + for port in input_ports: + contract_id = port.get("contractId") + version = port.get("version") + if not contract_id or not version: + continue + key = (str(contract_id), str(version)) + if key in seen: + continue + seen.add(key) + contracts.append({"id": str(contract_id), "version": str(version)}) + return contracts + + +def _consume_to_input_port( + canonical: Mapping[str, Any], name: str, contract_id: Optional[str] +) -> Dict[str, Any]: + """One Bitol ODPS ``InputPort`` from one normalized ``consumes[]`` entry. + + The version is resolved in a chain: the exact version the author states + (``contractVersion``) wins; failing that, the lowest version the declared + range accepts; failing that, the consume's own ``version`` or the provider + default. The chain ENDS at that constant rather than omitting the field, + because ODPS v1.0.0 — the default emit target — has ``version`` in + ``InputPort``'s ``required``, so an omission would make every such export + invalid. Before the middle step existed, a consume stating only a range had + neither of the other two and every such export claimed ``version: 1.0.0`` — + a pin the contract never made. + + ``versionConstraint`` itself has no field in that closed object, so it rides + along as a ``customProperties`` entry. Dropping it left an export claiming + the upstream is pinned to exactly one version when the contract said + "any 1.x". + """ + constraint = canonical.get("version_constraint") + resolved_version = ( + canonical.get("contract_version") or _range_lower_bound(constraint) or canonical["version"] + ) + port: Dict[str, Any] = { + "name": name, + "version": str(resolved_version), + "contractId": contract_id or name, + } + if constraint: + port["customProperties"] = [ + { + "property": VERSION_CONSTRAINT_PROPERTY, + "value": str(constraint), + "description": ( + "FLUID consumes[].versionConstraint — the semver range " + "this product accepts from the upstream contract." + ), + } + ] + return port + + +def consume_input_port_contract_ids(fluid: Mapping[str, Any]) -> Set[str]: + """The ``contractId``s an export's input ports carry for ``consumes[]``. + + A publish target that strips input ports by a cruder rule than "is this a + product consume" needs this to tell which of the ports it dropped had + lineage *derived* from ``consumes[]`` — the only lineage + :func:`remove_output_port_input_contracts` may remove. + + An ``expects[]`` port is the case that makes the distinction matter. It is + dropped by a "no ``sourceSystemId``" rule just like a product consume, but + its expose replays the upstream document's own ``inputContracts`` verbatim + through the ODPS pass-through, and nothing else in the payload records that + dependency — Access agreements are built from ``consumes[]``. Removing it + would lose the upstream, not deduplicate it. + """ + ids: Set[str] = set() + for canonical in consumes_to_canonical_ports(fluid): + contract_id = input_port_contract_id(canonical) + if contract_id: + ids.add(str(contract_id)) + return ids + + +def remove_output_port_input_contracts( + odps_payload: MutableMapping[str, Any], removed_ports: Iterable[Any] +) -> None: + """Drop ``outputPorts[].inputContracts`` entries for removed input ports. + + The inverse of :func:`_input_contracts_from_ports`, and it lives beside it + so the two cannot drift. A publish target that removes an input port has to + remove the lineage entry that mirrors it: forge derives that key from the + product's own ``consumes[]``, so one dependency is stated twice in a + document, and removing only the port leaves the upstream still named. + + Both Data Mesh Manager publish paths need this — the + :class:`DataMeshManagerRegistrar` that backs ``fluid publish --target + datamesh-manager`` and ``catalog.register``, and the + ``DataMeshManagerProvider`` behind ``fluid datamesh-manager publish``. They + strip by different rules (a missing ``sourceSystemId`` vs. a canonical + product-consume match), so each passes the ports it dropped and this reads + the ``contractId``s off them. Entropy would otherwise have to mirror each + upstream product as a SourceSystem to resolve the reference, and render the + duplicate graph nodes the strip exists to prevent. + + A dropped port with no ``contractId`` needs no action: it never produced a + lineage entry, because :func:`_input_contracts_from_ports` skips it. + """ + contract_ids = { + str(port["contractId"]) + for port in removed_ports + if isinstance(port, Mapping) and port.get("contractId") + } + if not contract_ids: + return + for port in odps_payload.get("outputPorts") or []: + if not isinstance(port, MutableMapping): + continue + entries = port.get("inputContracts") + if not isinstance(entries, list): + continue + kept = [ + entry + for entry in entries + if not (isinstance(entry, Mapping) and str(entry.get("id")) in contract_ids) + ] + if kept: + port["inputContracts"] = kept + else: + port.pop("inputContracts", None) + + def _build_input_ports( fluid: Mapping[str, Any], ctx: ExportCtx, product_id: str -) -> List[Dict[str, Any]]: +) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]: """Assemble Bitol ODPS v1.0.0 ``InputPort`` list from any FLUID lineage. Three lineage sources are handled, in priority order: @@ -97,7 +309,7 @@ def _build_input_ports( since v1.0.0 ``InputPort`` is ``additionalProperties: false``, and carries ``versionConstraint`` — which that closed object has no field for — as a ``customProperties`` entry. ``contractId`` comes from - :func:`_input_port_contract_id`. + :func:`input_port_contract_id`. 2. ``builds[].properties.source`` (SDP convention) — normalized via :func:`builds_to_canonical_input_ports`. Emits per-source-stream ports with synthetic ``contractId`` and ``customProperties`` for @@ -113,8 +325,28 @@ def _build_input_ports( disambiguating name (``b1__data_analytics_platform`` / ``b2__…``). Before this dedup keyed by name alone, the second consume was silently dropped and downstream catalogs lost half the lineage. + + Returns ``(input_ports, authored_ports)``. ``authored_ports`` is the path-1 + subset — the upstream *contracts* this contract names, and the only lineage + ``outputPorts[].inputContracts`` may be derived from. + + Path 2 is excluded because its ``contractId`` is synthesized as + ``:///`` for a source that has no contract at all + (see :func:`builds_to_canonical_input_ports`). That synthesis exists to + populate ``inputPorts[].contractId``, which ODPS v1.0.0 requires; + ``inputContracts`` is optional, so naming an id no catalog can resolve is + worse than naming nothing — the failure + :func:`input_port_contract_id` describes: an id that resolves in no + catalog following the reference. + + Path 3 is excluded because ``expects[]`` is also the carrier an imported + ODPS document's input ports arrive on, and an imported document's + output-port lineage is whatever that document said — including nothing. + Deriving it there made import → export add a dependency the author never + wrote, which the round-trip tests caught. """ input_ports: List[Dict[str, Any]] = [] + authored_ports: List[Dict[str, Any]] = [] seen: set = set() # ``(name, contract_id)`` tuples — true dedup key. seen_names: set = set() # ``name`` only — drives the collision-rename path. logger = getattr(ctx, "logger", None) @@ -143,35 +375,16 @@ def _unique_name(base: str, ref: Optional[str]) -> str: for canonical in consumes_to_canonical_ports( fluid, default_version=default_version, logger=logger ): - contract_id = _input_port_contract_id(canonical) + contract_id = input_port_contract_id(canonical) dedup_key = (canonical["name"], contract_id) if dedup_key in seen: continue seen.add(dedup_key) name = _unique_name(canonical["name"], canonical.get("reference")) seen_names.add(name) - consume_port: Dict[str, Any] = { - "name": name, - "version": canonical["version"], - "contractId": contract_id or name, - } - constraint = canonical.get("version_constraint") - if constraint: - # ``version`` here is the consume's own ``version`` or the - # provider's default — never the range. Dropping the range left an - # export claiming the upstream is pinned to exactly that version - # when the contract said "any 1.x". - consume_port["customProperties"] = [ - { - "property": VERSION_CONSTRAINT_PROPERTY, - "value": str(constraint), - "description": ( - "FLUID consumes[].versionConstraint — the semver range " - "this product accepts from the upstream contract." - ), - } - ] + consume_port = _consume_to_input_port(canonical, name, contract_id) input_ports.append(consume_port) + authored_ports.append(consume_port) # Path 2 — builds[].properties.source (SDP source streams) for canonical in builds_to_canonical_input_ports( @@ -206,10 +419,10 @@ def _unique_name(base: str, ref: Optional[str]) -> str: seen_names.add(port_obj["name"]) input_ports.append(port_obj) - return input_ports + return input_ports, authored_ports -def _input_port_contract_id(canonical: Mapping[str, Any]) -> Optional[str]: +def input_port_contract_id(canonical: Mapping[str, Any]) -> Optional[str]: """The contract an upstream consume points at. An explicit ``consumes[].contractId`` wins: that is an upstream which has @@ -233,7 +446,11 @@ def _input_port_contract_id(canonical: Mapping[str, Any]) -> Optional[str]: return str(name) if name else None -def _expose_to_output_port(expose: Mapping[str, Any], product_id: str) -> Optional[Dict[str, Any]]: +def _expose_to_output_port( + expose: Mapping[str, Any], + product_id: str, + derived_input_contracts: Optional[List[Dict[str, Any]]] = None, +) -> Optional[Dict[str, Any]]: from fluid_build.providers.base import ProviderError # Bitol ODPS v1.0.0 OutputPort requires ``name``. Accept any of @@ -264,7 +481,15 @@ def _expose_to_output_port(expose: Mapping[str, Any], product_id: str) -> Option if "sbom" in pt: port["sbom"] = list(pt["sbom"]) if "input_contracts" in pt: + # An imported ODPS document's own lineage wins, so import → export stays + # a fixed point: the upstream stated these, and re-deriving them from our + # consumes could name different contracts. port["inputContracts"] = list(pt["input_contracts"]) + elif derived_input_contracts: + # Otherwise the product's own consumes ARE its lineage. A + # FLUID-authored contract used to export output ports naming no + # upstreams at all. + port["inputContracts"] = [dict(entry) for entry in derived_input_contracts] if expose.get("tags"): port["tags"] = list(expose["tags"]) diff --git a/fluid_build/schemas/fluid-schema-0.7.6.json b/fluid_build/schemas/fluid-schema-0.7.6.json index 7b1e9c3a..f6f338ea 100644 --- a/fluid_build/schemas/fluid-schema-0.7.6.json +++ b/fluid_build/schemas/fluid-schema-0.7.6.json @@ -2,7 +2,7 @@ "$schema": "https://json-schema.org/draft/2020-12/schema", "$id": "https://open-data-protocol.github.io/fluid/schema/fluid-schema-0.7.6.json", "title": "FLUID 0.7.6 — Declarative Packaging Modes (preview)", - "description": "FLUID contract schema 0.7.6 — the current PREVIEW version, opened when 0.7.5 was promoted to stable.\n\n🔥 NEW in v0.7.6:\n• Declarative packaging modes — a top-level `packaging` block (plus a per-exposure `binding.packaging` override) declares per-container ownership for the IaC emit: `isolated` (this product creates and owns the container) vs `shared` (a pre-existing, platform-owned pool referenced via an OpenTofu data source), with a per-kind `containers` override map for hybrid tiers and a `pool` id for tenancy/cost attribution (RFC-packaging-modes.md).\n• Optional Bitol-shaped fields the exporters already wanted, each one opt-in: `metadata.team` (a roster — name plus members, every member a `$defs.teamMember` whose only required field is `username`), the root `description` as `{purpose, usage, limitations}` as well as a plain string, `consumes[].contractId` (an upstream that has a contract id of its own), `exposes[].contract.id` (used verbatim as the ODCS `id` and the ODPS output port `contractId` instead of the derived `.`; two exposes may not share one) and `binding.location.host` plus `binding.serialization.{value, key}` for Kafka. A contract that sets none of them validates and exports exactly as it does today.\n• `exposes[].contract.schema[].quality` — column-level data quality rules in ODCS's own shape. The ODCS exporter always replayed this key verbatim and read it back on import, but `$defs.column` is closed and did not define it, so the one way to publish a hand-written column rule was to write a contract `fluid validate` refuses. Optional: a column without it exports exactly as it does today.\n• Backward compatible with v0.7.5 — every 0.7.5 contract validates as 0.7.6 unchanged; an ABSENT packaging block keeps the legacy byte-identical emit.", + "description": "FLUID contract schema 0.7.6 — the current PREVIEW version, opened when 0.7.5 was promoted to stable.\n\n🔥 NEW in v0.7.6:\n• Declarative packaging modes — a top-level `packaging` block (plus a per-exposure `binding.packaging` override) declares per-container ownership for the IaC emit: `isolated` (this product creates and owns the container) vs `shared` (a pre-existing, platform-owned pool referenced via an OpenTofu data source), with a per-kind `containers` override map for hybrid tiers and a `pool` id for tenancy/cost attribution (RFC-packaging-modes.md).\n• Optional Bitol-shaped fields the exporters already wanted, each one opt-in: `metadata.team` (a roster — name plus members, every member a `$defs.teamMember` whose only required field is `username`), the root `description` as `{purpose, usage, limitations}` as well as a plain string, `consumes[].contractId` (an upstream that has a contract id of its own), `consumes[].contractVersion` (the exact upstream version this product was built against, inside that range), `exposes[].contract.id` (used verbatim as the ODCS `id` and the ODPS output port `contractId` instead of the derived `.`; two exposes may not share one) and `binding.location.host` plus `binding.serialization.{value, key}` for Kafka. A contract that sets none of them validates and exports exactly as it does today.\n• `exposes[].contract.schema[].quality` — column-level data quality rules in ODCS's own shape. The ODCS exporter always replayed this key verbatim and read it back on import, but `$defs.column` is closed and did not define it, so the one way to publish a hand-written column rule was to write a contract `fluid validate` refuses. Optional: a column without it exports exactly as it does today.\n• Backward compatible with v0.7.5 — every 0.7.5 contract validates as 0.7.6 unchanged; an ABSENT packaging block keeps the legacy byte-identical emit.", "type": "object", "additionalProperties": false, "required": [ @@ -2967,6 +2967,13 @@ "versionConstraint": { "$ref": "#/$defs/semverRange" }, + "contractVersion": { + "$ref": "#/$defs/semver", + "description": "NEW in v0.7.6: The exact upstream contract version this product was built and tested against, within `versionConstraint`. Exported as the ODPS `inputPorts[].version`, and as the version of the matching `outputPorts[].inputContracts[]` entry. Without it the export derives the range's lower bound; for a range it cannot read it falls back to 1.0.0, because ODPS v1.0.0 requires a version on every input port and the field cannot simply be omitted. `versionConstraint` stays the range this product accepts; this is the single version its exported lineage claims. `fluid validate` does not check that it falls inside `versionConstraint`.", + "examples": [ + "2.1.0" + ] + }, "qosExpectations": { "type": "object", "additionalProperties": false, diff --git a/fluid_build/util/contract.py b/fluid_build/util/contract.py index 54956cd4..5c3294ce 100644 --- a/fluid_build/util/contract.py +++ b/fluid_build/util/contract.py @@ -608,7 +608,8 @@ def consumes_to_canonical_ports( # --- 0.7.x consumeRef canonical fields --- "reference": Optional[str], # productId "description": str, # purpose - "version_constraint": Optional[str], # semverRange + "version_constraint": Optional[str], # semverRange + "contract_version": Optional[str], # semver, explicit only "qos_expectations": Optional[Mapping], # freshnessMax / maxStaleness / ... "required_policies": Optional[list], "tags": Optional[list], @@ -683,6 +684,13 @@ def consumes_to_canonical_ports( "reference": get_consume_ref(consume), # 0.7.2 canonical fields "version_constraint": version_constraint if version_constraint else None, + # 0.7.6: the exact upstream version this product was built against, + # which is what an exported input port and its matching + # ``inputContracts`` entry should claim. ``version`` above is the + # consume's own ``version`` or the provider default, and a consume + # that states only a range has neither — that is how an export came + # to say ``version: 1.0.0`` for ``versionConstraint: ^2.1``. + "contract_version": consume.get("contractVersion") or consume.get("contract_version"), "qos_expectations": qos if isinstance(qos, Mapping) and qos else None, "required_policies": ( list(required_policies) diff --git a/tests/build_runners/test_catalog_datamesh_manager.py b/tests/build_runners/test_catalog_datamesh_manager.py index a1dfb28c..a8d2e1f8 100644 --- a/tests/build_runners/test_catalog_datamesh_manager.py +++ b/tests/build_runners/test_catalog_datamesh_manager.py @@ -87,6 +87,162 @@ def _acquisition_contract() -> Dict[str, Any]: } +def _consuming_contract() -> Dict[str, Any]: + """A Gold product whose only upstream is another FLUID product.""" + return { + "fluidVersion": "0.7.3", + "kind": "DataProduct", + "id": "gold.sales.orders", + "name": "Orders (Gold)", + "description": "Consumer-aligned orders built from the CRM core product", + "metadata": { + "layer": "Gold", + "owner": {"team": "data-platform", "email": "dp@co.example"}, + }, + "consumes": [{"productId": "silver.crm.core", "exposeId": "customers"}], + "exposes": [ + { + "exposeId": "orders", + "kind": "table", + "binding": { + "platform": "snowflake", + "format": "snowflake_table", + "location": { + "database": "GOLD", + "schema": "SALES", + "table": "ORDERS", + }, + }, + "contract": {"schema": [{"name": "id", "type": "string"}]}, + } + ], + } + + +class TestProductConsumeLineageIsNotPublishedTwice: + """Entropy draws product-to-product lineage from Access agreements, so the + registrar strips product-consume ``inputPorts`` — an upstream product would + otherwise have to be mirrored as a SourceSystem for the reference to + resolve, and the graph would render duplicate nodes. + + Since 0.24.0 the ODPS render also derives ``outputPorts[].inputContracts`` + from the same ``consumes[]``, so the dependency is stated twice in one + document. Stripping only the input port left the upstream named anyway, on + the canonical publish path. + """ + + @respx.mock + def test_the_dataproduct_put_names_no_upstream_product(self): + from fluid_build.api.catalog_publication import CatalogPublicationPayload + + respx.route(method="PUT").mock(return_value=httpx.Response(200, json={"ok": True})) + + payload = CatalogPublicationPayload.from_contract(_consuming_contract(), {}) + result = DataMeshManagerRegistrar(api_token="t").register_payload(payload) + assert result.succeeded is True + + bodies = [ + json.loads(call.request.content) + for call in respx.calls + if "/api/dataproducts/" in str(call.request.url) + ] + assert ( + bodies + ), f"no dataproduct PUT fired; urls were {[str(c.request.url) for c in respx.calls]}" + body = bodies[0] + assert "inputPorts" not in body, body.get("inputPorts") + for port in body.get("outputPorts") or []: + assert "inputContracts" not in port, port["inputContracts"] + + @respx.mock + def test_a_stated_lineage_survives_even_when_the_product_also_consumes(self): + """The cell that distinguishes "pass-through wins" from + "consume-derived": with both present the mapper emits the stated + entries, so there is no derived entry to strip and the stated one must + not be taken for it.""" + from fluid_build.api.catalog_publication import CatalogPublicationPayload + + respx.route(method="PUT").mock(return_value=httpx.Response(200, json={"ok": True})) + + contract = _consuming_contract() + contract["expects"] = [ + {"id": "customers_raw", "version": "1", "contractId": "vendor.customers_raw"} + ] + contract["exposes"][0]["odps_passthrough"] = { + "input_contracts": [{"id": "vendor.customers_raw", "version": "1"}] + } + + payload = CatalogPublicationPayload.from_contract(contract, {}) + assert DataMeshManagerRegistrar(api_token="t").register_payload(payload).succeeded is True + + bodies = [ + json.loads(call.request.content) + for call in respx.calls + if "/api/dataproducts/" in str(call.request.url) + ] + lineage = [port.get("inputContracts") for port in bodies[0].get("outputPorts") or []] + assert lineage == [[{"id": "vendor.customers_raw", "version": "1"}]], lineage + + @respx.mock + def test_a_stated_entry_naming_the_consumed_contract_is_still_removed(self): + """When the stated id IS the one the consume derives, the consume wins. + Keeping it would name an upstream product with no input port to resolve + it — the duplicate-graph-node leak this strip exists to prevent.""" + from fluid_build.api.catalog_publication import CatalogPublicationPayload + + respx.route(method="PUT").mock(return_value=httpx.Response(200, json={"ok": True})) + + contract = _consuming_contract() + contract["exposes"][0]["odps_passthrough"] = { + "input_contracts": [{"id": "silver.crm.core.customers", "version": "1.0.0"}] + } + + payload = CatalogPublicationPayload.from_contract(contract, {}) + assert DataMeshManagerRegistrar(api_token="t").register_payload(payload).succeeded is True + + bodies = [ + json.loads(call.request.content) + for call in respx.calls + if "/api/dataproducts/" in str(call.request.url) + ] + for port in bodies[0].get("outputPorts") or []: + assert "inputContracts" not in port, port["inputContracts"] + + @respx.mock + def test_an_imported_documents_own_lineage_is_not_stripped_with_it(self): + """The registrar drops an input port that carries no ``sourceSystemId``, + which catches an ``expects[]`` port as well as a product consume. Only + the consume-derived lineage may go with it: an ``expects[]`` port's + expose replays the upstream document's OWN ``inputContracts``, and + nothing else in the payload records that dependency, because Access + agreements are built from ``consumes[]``. + """ + from fluid_build.api.catalog_publication import CatalogPublicationPayload + + respx.route(method="PUT").mock(return_value=httpx.Response(200, json={"ok": True})) + + contract = _consuming_contract() + contract.pop("consumes") + contract["expects"] = [ + {"id": "customers_raw", "version": "1", "contractId": "vendor.customers_raw"} + ] + contract["exposes"][0]["odps_passthrough"] = { + "input_contracts": [{"id": "vendor.customers_raw", "version": "1"}] + } + + payload = CatalogPublicationPayload.from_contract(contract, {}) + assert DataMeshManagerRegistrar(api_token="t").register_payload(payload).succeeded is True + + bodies = [ + json.loads(call.request.content) + for call in respx.calls + if "/api/dataproducts/" in str(call.request.url) + ] + assert bodies, "no dataproduct PUT fired" + lineage = [port.get("inputContracts") for port in bodies[0].get("outputPorts") or []] + assert lineage == [[{"id": "vendor.customers_raw", "version": "1"}]], lineage + + # ── Registrar unit ─────────────────────────────────────────────────────── diff --git a/tests/providers/test_datamesh_manager_gaps.py b/tests/providers/test_datamesh_manager_gaps.py index 1324cf77..8d5f936b 100644 --- a/tests/providers/test_datamesh_manager_gaps.py +++ b/tests/providers/test_datamesh_manager_gaps.py @@ -713,6 +713,186 @@ def test_removes_product_consumes_and_retains_explicit_source_systems(self): {"name": "crm_source", "contractId": "bronze.telco.party_v1"} ] + @pytest.mark.parametrize( + ("label", "consumes"), + [ + ( + "productId tail contains __", + [ + {"productId": "silver.crm.core", "exposeId": "customers"}, + {"productId": "silver.crm__core", "exposeId": "customers"}, + ], + ), + ( + # Three consumes sharing an exposeId AND a productId tail: the + # first keeps ``customers``, the second takes + # ``core__customers``, and the third falls back to + # ``customers_2`` — the shape with no ``__`` to match on. + "numeric-suffix rename fallback", + [ + {"productId": "silver.a.core", "exposeId": "customers"}, + {"productId": "silver.b.core", "exposeId": "customers"}, + {"productId": "silver.c.core", "exposeId": "customers"}, + ], + ), + ( + "exposeId itself contains __", + [ + {"productId": "silver.crm.core", "exposeId": "cust__omers"}, + {"productId": "silver.party", "exposeId": "cust__omers"}, + ], + ), + ], + ) + def test_a_renamed_port_is_matched_whatever_shape_the_rename_took(self, label, consumes): + """Two consumes sharing an ``exposeId`` make the exporter rename the + second, and the rename has three shapes. Matching the name by splitting + at the FIRST ``__`` read ``silver.crm__core`` as the boundary and let + that port through; the numeric-suffix fallback carries no ``__`` at all. + A leaked port is an unresolvable SourceSystem reference, and since the + lineage key mirrors the ports, it leaked twice. + """ + from fluid_build.providers.datamesh_manager.datamesh_manager import ( + DataMeshManagerProvider, + ) + + contract = { + "fluidVersion": "0.7.3", + "kind": "DataProduct", + "id": "gold.sales.orders", + "name": "Orders", + "metadata": {"layer": "Gold", "owner": {"team": "dp", "email": "dp@co.example"}}, + "consumes": consumes, + "exposes": [ + { + "exposeId": "orders", + "kind": "table", + "binding": { + "platform": "snowflake", + "format": "snowflake_table", + "location": {"database": "G", "schema": "S", "table": "O"}, + }, + "contract": {"schema": [{"name": "id", "type": "string"}]}, + } + ], + } + payload = DataMeshManagerProvider(api_key="t").apply(contract, dry_run=True)["payload"] + assert not payload.get("inputPorts"), (label, payload.get("inputPorts")) + lineage = [port.get("inputContracts") for port in payload.get("outputPorts") or []] + assert not any(lineage), (label, lineage) + + def test_a_source_system_consume_keeps_its_port_and_its_lineage(self): + """The over-match guard. Entropy resolves a source-system port, so it + survives — and its lineage entry must survive with it.""" + from fluid_build.providers.datamesh_manager.datamesh_manager import ( + DataMeshManagerProvider, + ) + + contract = { + "fluidVersion": "0.7.3", + "kind": "DataProduct", + "id": "gold.sales.orders", + "name": "Orders", + "metadata": {"layer": "Gold", "owner": {"team": "dp", "email": "dp@co.example"}}, + "consumes": [ + {"productId": "bronze.telco.party", "exposeId": "account_source"}, + { + "productId": "bronze.telco.party", + "exposeId": "crm_source", + "sourceSystem": "bss-crm", + }, + ], + "exposes": [ + { + "exposeId": "orders", + "kind": "table", + "binding": { + "platform": "snowflake", + "format": "snowflake_table", + "location": {"database": "G", "schema": "S", "table": "O"}, + }, + "contract": {"schema": [{"name": "id", "type": "string"}]}, + } + ], + } + payload = DataMeshManagerProvider(api_key="t").apply(contract, dry_run=True)["payload"] + assert [ip["name"] for ip in payload["inputPorts"]] == ["crm_source"] + assert payload["outputPorts"][0]["inputContracts"] == [ + {"id": "bronze.telco.party.crm_source", "version": "1.0.0"} + ] + + def test_a_removed_ports_derived_lineage_goes_with_it(self): + """Since 0.24.0 the ODPS exporter derives ``outputPorts[].inputContracts`` + from ``consumes[]``, so one product-to-product edge is stated twice in a + document. Leaving the lineage behind puts back the duplicate graph nodes + this strip exists to prevent.""" + fluid = { + "consumes": [ + {"productId": "bronze.telco.party_v1", "exposeId": "account_source"}, + { + "productId": "bronze.telco.party_v1", + "exposeId": "crm_source", + "sourceSystem": "bss-crm", + }, + ], + } + odps_payload = { + "inputPorts": [ + { + "name": "account_source", + "contractId": "bronze.telco.party_v1.account_source", + }, + {"name": "crm_source", "contractId": "bronze.telco.party_v1.crm_source"}, + ], + "outputPorts": [ + { + "name": "accounts", + "inputContracts": [ + {"id": "bronze.telco.party_v1.account_source", "version": "1.0.0"}, + {"id": "bronze.telco.party_v1.crm_source", "version": "1.0.0"}, + ], + } + ], + } + + DataMeshManagerProvider._remove_odps_product_consume_input_ports(odps_payload, fluid) + + assert odps_payload["inputPorts"] == [ + {"name": "crm_source", "contractId": "bronze.telco.party_v1.crm_source"} + ] + # The source-system consume keeps its lineage, because its port is kept. + assert odps_payload["outputPorts"][0]["inputContracts"] == [ + {"id": "bronze.telco.party_v1.crm_source", "version": "1.0.0"} + ] + + def test_the_lineage_key_goes_away_when_every_entry_was_a_product_consume(self): + fluid = { + "consumes": [ + {"productId": "bronze.telco.party_v1", "exposeId": "account_source"}, + ], + } + odps_payload = { + "inputPorts": [ + { + "name": "account_source", + "contractId": "bronze.telco.party_v1.account_source", + } + ], + "outputPorts": [ + { + "name": "accounts", + "inputContracts": [ + {"id": "bronze.telco.party_v1.account_source", "version": "1.0.0"} + ], + } + ], + } + + DataMeshManagerProvider._remove_odps_product_consume_input_ports(odps_payload, fluid) + + assert "inputPorts" not in odps_payload + assert "inputContracts" not in odps_payload["outputPorts"][0] + def test_removes_input_ports_key_when_only_product_consumes_exist(self): fluid = { "consumes": [ diff --git a/tests/providers/test_input_port_lineage_and_quality_sql.py b/tests/providers/test_input_port_lineage_and_quality_sql.py new file mode 100644 index 00000000..b6b95789 --- /dev/null +++ b/tests/providers/test_input_port_lineage_and_quality_sql.py @@ -0,0 +1,448 @@ +# Copyright 2024-2026 Agentics Transformation Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Exported lineage says which version, and a quality expression is executable. + +Two defects found by feeding 0.23.0's exports to a resolver and a DQ engine. + +An input port claimed ``version: 1.0.0`` whatever the contract said, because +the only version in play was the consume's own or the provider's constant — a +consume stating ``versionConstraint: ^2.1`` has neither, so the export asserted +a pin the contract never made. And an output port named no upstreams at all +unless the contract had been imported from ODPS, so a catalog reading lineage +from ``outputPorts[].inputContracts`` saw a product with no dependencies. + +Separately, ``exposes[].contract.quality[].expression`` is a row predicate, and +it was emitted as the ODCS ``sql`` rule's ``query`` verbatim. ODCS defines that +field as "an individual SQL query that returns a value that can be compared", +so an engine executing it received a fragment. +""" + +from __future__ import annotations + +from typing import Any, Dict, List, Optional + +import pytest + +from fluid_build.providers.odcs import OdcsProvider +from fluid_build.providers.odcs.mappers.quality import _expression_to_quality +from fluid_build.providers.odps_standard.mappers.ports import ( + _input_contracts_from_ports, + _range_lower_bound, +) +from fluid_build.providers.odps_standard.validation import load_schema, validate +from fluid_build.schema_manager import FluidSchemaManager + +# The v1.1 type tests already define provider construction at a chosen +# apiVersion and the render-to-product step; reused rather than spelled a second +# time, the way tests/providers/test_duckdb_sandbox_workspace_root.py reuses a +# sibling module's contract fixtures. +from tests.providers.test_odps_v11_type import _product, _provider + +pytestmark = pytest.mark.unit + +PRODUCT_ID = "gold.sales.orders" +UPSTREAM_CONTRACT_ID = "silver.crm.core.customers" +EXPRESSION = "preferred_customer_flag IN ('Y', 'N')" + + +def _counting_query(expression: str) -> str: + """The wrapper `_expression_to_quality` builds for a row predicate. + + The predicate is on its own line so a trailing ``-- comment`` cannot eat the + closing paren and the comparison after it.""" + return f"SELECT COUNT(*) FROM ${{table}} WHERE (\n{expression}\n) IS NOT TRUE" + + +def _contract(consume: Optional[Dict[str, Any]] = None, **kw: Any) -> Dict[str, Any]: + """A 0.7.6 contract with one consume and one expose carrying a quality rule.""" + base_consume: Dict[str, Any] = {"productId": "silver.crm.core", "exposeId": "customers"} + if consume: + base_consume.update(consume) + contract: Dict[str, Any] = { + "fluidVersion": kw.pop("fluid_version", "0.7.6"), + "kind": "DataProduct", + "id": PRODUCT_ID, + "name": "orders", + "metadata": {"owner": {"team": "commerce"}}, + "consumes": [base_consume], + "exposes": [ + { + "exposeId": "orders", + "kind": "table", + "binding": { + "platform": "local", + "format": "parquet", + "location": {"path": "./orders"}, + }, + "contract": { + "schema": [{"name": "preferred_customer_flag", "type": "string"}], + "quality": [ + {"rule": "flag_domain", "expression": EXPRESSION, "severity": "error"} + ], + }, + } + ], + } + contract.update(kw) + return contract + + +def _odps(contract: Dict[str, Any], api_version: str = "v1.0.0") -> Dict[str, Any]: + return _product(_provider(api_version), contract) + + +def _sql_rules(odcs: Dict[str, Any]) -> List[Dict[str, Any]]: + found: List[Dict[str, Any]] = [] + + def walk(node: Any) -> None: + if isinstance(node, dict): + rules = node.get("quality") + if isinstance(rules, list): + found.extend(r for r in rules if isinstance(r, dict) and r.get("type") == "sql") + for value in node.values(): + walk(value) + elif isinstance(node, list): + for value in node: + walk(value) + + walk(odcs) + return found + + +# --------------------------------------------------------------------------- +# Defect 19a — the version an input port claims +# --------------------------------------------------------------------------- + + +class TestTheLowerBoundHelper: + @pytest.mark.parametrize( + ("constraint", "expected"), + [ + ("^2.1", "2.1.0"), + ("^2.1.3", "2.1.3"), + ("~1.4", "1.4.0"), + (">=3.0.0", "3.0.0"), + ("=2.0.0", "2.0.0"), + ("2.1.0", "2.1.0"), + ("2.1", "2.1.0"), + (" ^2.1 ", "2.1.0"), + ], + ) + def test_the_documented_forms_resolve(self, constraint, expected): + assert _range_lower_bound(constraint) == expected + + @pytest.mark.parametrize( + "constraint", + [ + ">1.0.0", # does NOT accept 1.0.0; its real lower bound is unnameable + ">=1.0.0,<2.0.0", + "^1.0 || ^2.0", + "1.x", + "1.*", + "<2.0.0", + "<=2.0.0", + "latest", + "^١.٢", # Arabic-Indic digits: str.isdigit() is true for these + "^2.²", # superscript two likewise + "^01.2", # a leading zero is not semver, and no resolver matches it + "^2.01", + "~00", + ">=0001.0.0", + "", + " ", + None, + 2.1, + ], + ) + def test_anything_else_refuses_rather_than_guesses(self, constraint): + """A wrong exact version in exported lineage reads as an established + fact, so refusing and falling back beats guessing.""" + assert _range_lower_bound(constraint) is None + + +class TestTheInputPortVersion: + @pytest.mark.parametrize( + ("consume", "expected"), + [ + pytest.param({"versionConstraint": "^2.1"}, "2.1.0", id="caret"), + pytest.param({"versionConstraint": "~1.4"}, "1.4.0", id="tilde"), + pytest.param({"versionConstraint": ">=3.0.0"}, "3.0.0", id="gte"), + pytest.param({"versionConstraint": "2.1.0"}, "2.1.0", id="exact-range"), + pytest.param( + {"versionConstraint": "^2.1", "contractVersion": "2.3.4"}, + "2.3.4", + id="contract-version-wins", + ), + pytest.param( + {"versionConstraint": ">=1.0.0,<2.0.0"}, "1.0.0", id="compound-falls-back" + ), + pytest.param({"versionConstraint": "1.x"}, "1.0.0", id="wildcard-falls-back"), + pytest.param({"versionConstraint": ">1.0.0"}, "1.0.0", id="exclusive-falls-back"), + pytest.param({}, "1.0.0", id="nothing-to-go-on"), + ], + ) + def test_the_version_an_input_port_claims(self, consume, expected): + """The acceptance case is ``^2.1``, which used to export ``version: + 1.0.0`` — a pin the contract never made. + + A range the resolver cannot read still falls back to that constant rather + than guessing, and the document stays valid either way, which is why the + chain ends at a constant: ODPS v1.0.0 has ``version`` in ``InputPort``'s + ``required``, so the field cannot simply be omitted. + """ + odps = _odps(_contract(consume)) + assert odps["inputPorts"][0]["version"] == expected + validate(odps, load_schema("v1.0.0")) + + def test_the_range_still_rides_along_in_its_custom_property(self): + """Resolving a version must not lose the range the contract accepts.""" + odps = _odps(_contract({"versionConstraint": "^2.1"})) + props = odps["inputPorts"][0]["customProperties"] + assert [p["value"] for p in props if p["property"] == "fluidVersionConstraint"] == ["^2.1"] + + +# --------------------------------------------------------------------------- +# Defect 19b — the lineage an output port carries +# --------------------------------------------------------------------------- + + +class TestOutputPortLineage: + def test_an_output_port_names_each_consume(self): + odps = _odps(_contract({"versionConstraint": "^2.1"})) + assert odps["outputPorts"][0]["inputContracts"] == [ + {"id": UPSTREAM_CONTRACT_ID, "version": "2.1.0"} + ] + + def test_the_lineage_version_equals_the_input_port_version(self): + """One resolution, two places — they cannot disagree.""" + odps = _odps(_contract({"versionConstraint": "~1.4"})) + port = odps["inputPorts"][0] + assert odps["outputPorts"][0]["inputContracts"] == [ + {"id": port["contractId"], "version": port["version"]} + ] + + def test_the_entries_carry_only_what_bitol_allows(self): + """``$defs.InputContract`` is additionalProperties:false with exactly + ``id`` and ``version``.""" + odps = _odps(_contract({"versionConstraint": "^2.1"})) + for entry in odps["outputPorts"][0]["inputContracts"]: + assert set(entry) == {"id", "version"} + + @pytest.mark.parametrize("api_version", ["v1.0.0", "v1.1.0"]) + def test_the_document_still_validates(self, api_version): + odps = _odps(_contract({"versionConstraint": "^2.1"}), api_version=api_version) + validate(odps, load_schema(api_version)) + + def test_duplicate_upstreams_collapse_to_one_entry(self): + ports = [ + {"name": "a", "contractId": "x.y", "version": "1.0.0"}, + {"name": "b", "contractId": "x.y", "version": "1.0.0"}, + {"name": "c", "contractId": "x.z", "version": "2.0.0"}, + ] + assert _input_contracts_from_ports(ports) == [ + {"id": "x.y", "version": "1.0.0"}, + {"id": "x.z", "version": "2.0.0"}, + ] + + def test_a_port_missing_either_half_is_skipped(self): + assert _input_contracts_from_ports([{"name": "a"}]) == [] + assert _input_contracts_from_ports([{"name": "a", "contractId": "x.y"}]) == [] + + def test_an_imported_documents_own_lineage_wins(self): + """Import then export must stay a fixed point: the upstream stated these, + and re-deriving from our consumes could name different contracts.""" + contract = _contract({"versionConstraint": "^2.1"}) + contract["exposes"][0]["odps_passthrough"] = { + "input_contracts": [{"id": "vendor.supplied", "version": "9.9.9"}] + } + odps = _odps(contract) + assert odps["outputPorts"][0]["inputContracts"] == [ + {"id": "vendor.supplied", "version": "9.9.9"} + ] + + def test_an_imported_documents_input_ports_are_not_treated_as_lineage(self): + """``expects[]`` is the carrier an imported ODPS document's input ports + arrive on, so deriving lineage from it would make import then export add + a dependency the source document never stated.""" + contract = _contract() + contract.pop("consumes") + contract["expects"] = [ + {"id": "customers_raw", "version": "1", "contractId": "vendor.customers_raw"} + ] + odps = _odps(contract) + assert odps["inputPorts"][0]["contractId"] == "vendor.customers_raw" + assert "inputContracts" not in odps["outputPorts"][0] + + def test_a_source_stream_is_an_input_port_but_not_a_contract(self): + """``builds[].properties.source`` synthesizes ``:///`` + as a contractId so ODPS v1.0.0's required ``inputPorts[].contractId`` is + populated for a source that has no contract at all. ``inputContracts`` + is optional, so naming an id no catalog can resolve is worse than + naming nothing.""" + contract = _contract() + contract.pop("consumes") + contract["builds"] = [ + { + "buildId": "b1", + "engine": "duckdb", + "properties": { + "source": { + "kind": "postgres", + "database": "crmdb", + "streams": ["customers"], + } + }, + } + ] + odps = _odps(contract) + assert odps["inputPorts"][0]["contractId"].startswith("postgres://") + assert "inputContracts" not in odps["outputPorts"][0] + + def test_a_contract_with_no_consumes_gets_no_lineage_key(self): + contract = _contract() + contract.pop("consumes") + odps = _odps(contract) + assert "inputContracts" not in odps["outputPorts"][0] + + +# --------------------------------------------------------------------------- +# Defect 19c — the new field is preview-gated +# --------------------------------------------------------------------------- + + +class TestTheFieldIsPreviewGated: + def test_it_validates_at_0_7_6(self): + contract = _contract({"versionConstraint": "^2.1", "contractVersion": "2.3.4"}) + result = FluidSchemaManager().validate_contract(contract, offline_only=True) + assert result.is_valid, [str(e) for e in result.errors] + + def test_0_7_5_rejects_it(self): + contract = _contract( + {"versionConstraint": "^2.1", "contractVersion": "2.3.4"}, fluid_version="0.7.5" + ) + result = FluidSchemaManager().validate_contract(contract, offline_only=True) + assert not result.is_valid + assert any("contractVersion" in str(e) for e in result.errors) + + def test_a_non_semver_value_is_rejected_with_the_pattern(self): + contract = _contract({"contractVersion": "2.1"}) + result = FluidSchemaManager().validate_contract(contract, offline_only=True) + assert not result.is_valid + assert any("contractVersion" in str(e) for e in result.errors) + + +# --------------------------------------------------------------------------- +# Defect 20 — a quality expression an engine can execute +# --------------------------------------------------------------------------- + + +class TestTheQualityExpression: + def test_a_predicate_becomes_a_counting_query_compared_to_zero(self): + rule = _expression_to_quality({"expression": EXPRESSION}) + assert rule["type"] == "sql" + assert rule["query"] == _counting_query(EXPRESSION) + assert rule["mustBe"] == 0 + + def test_it_uses_the_placeholder_from_the_odcs_example(self): + rule = _expression_to_quality({"expression": EXPRESSION}) + assert "${table}" in rule["query"] + + def test_it_negates_with_is_not_true_not_a_bare_not(self): + """Pins the three-valued-logic choice. ``NOT (p)`` is NULL when ``p`` is + NULL, so a naive negation counts no failure for a NULL row — and a row + whose predicate did not evaluate TRUE has not held.""" + rule = _expression_to_quality({"expression": EXPRESSION}) + assert "IS NOT TRUE" in rule["query"] + assert f"NOT ({EXPRESSION})" not in rule["query"] + + @pytest.mark.parametrize("prefix", ["SELECT", "select", "WITH", " SELECT", "SELECT\n"]) + def test_an_expression_that_is_already_a_query_passes_through(self, prefix): + expression = f"{prefix} COUNT(*) FROM orders" + rule = _expression_to_quality({"expression": expression}) + assert rule["query"] == expression.strip() + assert rule["mustBe"] is True + + @pytest.mark.parametrize( + "expression", + [ + "selected_at IS NOT NULL", + "selection_count > 0", + "select_count = 1", + "with_tax > 0", + "withholding_amount >= 0", + "withdrawn_flag IS NOT NULL", + ], + ) + def test_a_column_that_merely_starts_with_select_or_with_is_a_predicate(self, expression): + """``startswith`` with no word boundary sent these down the pass-through + branch and exported the bare fragment with ``mustBe: true`` — the exact + thing this is here to remove, reinstated for a whole class of columns.""" + rule = _expression_to_quality({"expression": expression}) + assert rule["query"] == _counting_query(expression) + assert rule["mustBe"] == 0 + + def test_a_trailing_semicolon_does_not_land_inside_the_parentheses(self): + rule = _expression_to_quality({"expression": "amount > 0;"}) + assert rule["query"] == _counting_query("amount > 0") + + def test_the_rule_reaches_the_exported_document(self): + odcs = OdcsProvider().render(_contract(), expose_id="orders") + queries = [r["query"] for r in _sql_rules(odcs)] + assert any("SELECT COUNT(*)" in q and "IS NOT TRUE" in q for q in queries) + + +class TestTheQueryActuallyRuns: + """The point of the fix is that an engine can execute what we emit, so run it.""" + + @staticmethod + def _count(con: Any, query: str) -> int: + return con.execute(query.replace("${table}", "t")).fetchone()[0] + + def test_duckdb_executes_it_and_counts_the_failing_rows(self): + duckdb = pytest.importorskip("duckdb") + query = _expression_to_quality({"expression": EXPRESSION})["query"] + + con = duckdb.connect() + con.execute("CREATE TABLE t (preferred_customer_flag VARCHAR)") + con.execute("INSERT INTO t VALUES ('Y'), ('N')") + assert self._count(con, query) == 0 + + con.execute("INSERT INTO t VALUES ('maybe')") + assert self._count(con, query) == 1 + + con.execute("INSERT INTO t VALUES (NULL)") + assert self._count(con, query) == 2, "a NULL row did not evaluate TRUE, so it failed" + + def test_the_bare_not_form_would_have_missed_the_null_row(self): + """Why ``IS NOT TRUE`` rather than ``NOT (...)``, measured rather than + argued. If this ever stops holding, the choice above is free to change.""" + duckdb = pytest.importorskip("duckdb") + con = duckdb.connect() + con.execute("CREATE TABLE t (preferred_customer_flag VARCHAR)") + con.execute("INSERT INTO t VALUES ('Y'), ('maybe'), (NULL)") + + emitted = _expression_to_quality({"expression": EXPRESSION})["query"] + naive = f"SELECT COUNT(*) FROM ${{table}} WHERE NOT ({EXPRESSION})" + + assert self._count(con, emitted) == 2 + assert self._count(con, naive) == 1 + + +class TestTheNoOpCases: + def test_a_contract_without_quality_exports_no_sql_rule(self): + contract = _contract() + contract["exposes"][0]["contract"].pop("quality") + odcs = OdcsProvider().render(contract, expose_id="orders") + assert [r for r in _sql_rules(odcs) if "IS NOT TRUE" in r.get("query", "")] == [] diff --git a/tests/providers/test_odcs_quality_export.py b/tests/providers/test_odcs_quality_export.py index a0cb666e..4a4041f6 100644 --- a/tests/providers/test_odcs_quality_export.py +++ b/tests/providers/test_odcs_quality_export.py @@ -228,11 +228,17 @@ def test_dq_threshold_and_operator_become_the_matching_odcs_operator() -> None: def test_quality_expression_lands_on_the_object_as_a_sql_rule() -> None: + """ODCS defines ``sql`` as a query returning a comparable value, so a FLUID + row predicate becomes one. It used to be emitted verbatim, which handed a DQ + engine a fragment it could not execute.""" rules = {r["id"]: r for r in _object_quality(_render(_RICH))} assert "amount_positive" in rules, f"expression not exported: {sorted(rules)}" assert rules["amount_positive"]["type"] == "sql" - assert rules["amount_positive"]["query"] == "amount > 0" - assert rules["amount_positive"]["mustBe"] is True + assert ( + rules["amount_positive"]["query"] + == "SELECT COUNT(*) FROM ${table} WHERE (\namount > 0\n) IS NOT TRUE" + ) + assert rules["amount_positive"]["mustBe"] == 0 def test_freshness_rule_exports_as_an_odcs_custom_rule() -> None: