From 04436db1547080786433f8cd113f56e1abac7372 Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Tue, 27 Oct 2026 06:57:15 +0100 Subject: [PATCH 01/10] fix(odcs): a quality expression exported as SQL no engine could execute MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 a DQ engine reading an exported contract was handed `preferred_customer_flag IN ('Y', 'N')` with `mustBe: true` and nothing it could execute. A predicate now becomes a counting query compared to zero, against the `${table}` placeholder ODCS's own example uses. 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 — verified against DuckDB, where the naive form misses the NULL row and this one does not. The whole-number comparison follows OpenMetadata's convention for the same shape of check. An expression already written as a `SELECT` or `WITH` query passes through unchanged. --- fluid_build/providers/odcs/mappers/quality.py | 48 +++++++++++++++---- tests/providers/test_odcs_quality_export.py | 10 +++- 2 files changed, 46 insertions(+), 12 deletions(-) diff --git a/fluid_build/providers/odcs/mappers/quality.py b/fluid_build/providers/odcs/mappers/quality.py index 97e23b91..07aa42d3 100644 --- a/fluid_build/providers/odcs/mappers/quality.py +++ b/fluid_build/providers/odcs/mappers/quality.py @@ -449,17 +449,45 @@ def _dq_severity(check: Dict[str, Any], rule: Mapping[str, Any]) -> None: 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 the one OpenMetadata + documents for ``sql`` rules (exactly one comparison against a whole number). + ``${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 (it starts with ``SELECT`` or + ``WITH``) is passed through as written with the previous comparison: the + author wrote a statement and chose what it returns. """ - 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() + is_query = expression.upper().startswith(("SELECT", "WITH")) + + 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: + check["query"] = f"SELECT COUNT(*) FROM ${{table}} WHERE ({expression}) 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/tests/providers/test_odcs_quality_export.py b/tests/providers/test_odcs_quality_export.py index a0cb666e..9c14aa80 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 (amount > 0) IS NOT TRUE" + ) + assert rules["amount_positive"]["mustBe"] == 0 def test_freshness_rule_exports_as_an_odcs_custom_rule() -> None: From 5a929b18ce0057814086a0f5bdc9fe6481f70d64 Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Tue, 27 Oct 2026 06:57:29 +0100 Subject: [PATCH 02/10] fix(odps): exported lineage named no upstreams and claimed an unset version MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two defects in the same export, both found by feeding a 0.23.0 export to a lineage resolver. `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 never made 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`), reading only the forms `$defs.semverRange` documents. A range the resolver cannot read — a compound, a wildcard, or `>1.0.0`, whose true lower bound semver gives no way to name — falls back to the old constant rather than guessing: ODPS v1.0.0 has `version` in `InputPort`'s `required` so the field cannot be omitted, and a wrong exact version in exported lineage reads as a fact somebody established. Deliberately not a semver library: `semver`, `semantic_version` and `packaging` are all in the resolved environment but none is a declared dependency and the core is kept lean, while `schema_manager.VersionConstraint` rejects the caret outright and is the FLUID *schema* version type. `outputPorts[].inputContracts` was populated only from an imported 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, 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 then export stays a fixed point and never adds a dependency the author did not write. Adds `consumes[].contractVersion` to the 0.7.6 preview schema: the exact upstream version this product was built and tested against, within `versionConstraint`. It wins over the resolved lower bound. `_consume_input_port` is extracted so `_build_input_ports` does not grow. --- .../providers/odps_standard/mappers/ports.py | 187 +++++++-- fluid_build/schemas/fluid-schema-0.7.6.json | 7 + fluid_build/util/contract.py | 7 + ...test_input_port_lineage_and_quality_sql.py | 395 ++++++++++++++++++ 4 files changed, 569 insertions(+), 27 deletions(-) create mode 100644 tests/providers/test_input_port_lineage_and_quality_sql.py diff --git a/fluid_build/providers/odps_standard/mappers/ports.py b/fluid_build/providers/odps_standard/mappers/ports.py index d1fd6ea4..a8b55cad 100644 --- a/fluid_build/providers/odps_standard/mappers/ports.py +++ b/fluid_build/providers/odps_standard/mappers/ports.py @@ -37,7 +37,7 @@ from __future__ import annotations from collections.abc import Mapping -from typing import Any, Dict, List, Optional +from typing import Any, Dict, List, Optional, Tuple from fluid_build.util.contract import ( builds_to_canonical_input_ports, @@ -62,6 +62,49 @@ # 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. ``semver``, ``semantic_version`` and + ``packaging`` are all present in the resolved environment but none 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(".") + if not 1 <= len(parts) <= 3 or not all(p.isdigit() 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 also what semver.org says a range's + # lower bound is. + while len(parts) < 3: + parts.append("0") + return ".".join(parts) + + # ----- FLUID → ODPS -------------------------------------------------------- @@ -69,24 +112,110 @@ 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. + + 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_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 _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: @@ -113,8 +242,18 @@ 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 subset + from paths 1 and 2 — the lineage this contract states itself, and the only + lineage ``outputPorts[].inputContracts`` may be derived from. 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) @@ -150,28 +289,9 @@ def _unique_name(base: str, ref: Optional[str]) -> str: 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_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( @@ -196,6 +316,7 @@ def _unique_name(base: str, ref: Optional[str]) -> str: if custom_props: port["customProperties"] = custom_props input_ports.append(port) + authored_ports.append(port) # Path 3 — expects[] (FLUID 0.7.1 legacy) for expect in fluid.get("expects") or []: @@ -206,7 +327,7 @@ 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]: @@ -233,7 +354,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 +389,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..5674d192 100644 --- a/fluid_build/schemas/fluid-schema-0.7.6.json +++ b/fluid_build/schemas/fluid-schema-0.7.6.json @@ -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.", + "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..47be7e51 100644 --- a/fluid_build/util/contract.py +++ b/fluid_build/util/contract.py @@ -683,6 +683,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/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..a62e4c91 --- /dev/null +++ b/tests/providers/test_input_port_lineage_and_quality_sql.py @@ -0,0 +1,395 @@ +# 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 _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 + + +# --------------------------------------------------------------------------- +# 19. 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", + "", + " ", + 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"] + + +# --------------------------------------------------------------------------- +# 19. 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_contract_with_no_consumes_gets_no_lineage_key(self): + contract = _contract() + contract.pop("consumes") + odps = _odps(contract) + assert "inputContracts" not in odps["outputPorts"][0] + + +# --------------------------------------------------------------------------- +# 19. 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) + + +# --------------------------------------------------------------------------- +# 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"] == ( + f"SELECT COUNT(*) FROM ${{table}} WHERE ({EXPRESSION}) IS NOT TRUE" + ) + 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"]) + 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 + + def test_the_exported_odcs_validates(self): + from fluid_build.providers.odcs.validation import collect_errors + from fluid_build.providers.odcs.validation import load_schema as odcs_schema + + odcs = OdcsProvider().render(_contract({"versionConstraint": "^2.1"}), expose_id="orders") + assert collect_errors(odcs, odcs_schema()) == [] + + 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", "")] == [] From e25aa77fe7ace3748dfa091e23ca0303750f29da Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Tue, 27 Oct 2026 06:58:41 +0100 Subject: [PATCH 03/10] fix(odps): the range reader accepted digits no version field should carry MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `str.isdigit()` is true for Arabic-Indic digits and for superscripts, so `versionConstraint: ^١.٢` was read as a range and its "lower bound" emitted verbatim into an exported `inputPorts[].version`. The check now requires `isascii()` as well, so an unreadable range falls back to the constant the way every other unreadable form does. --- fluid_build/providers/odps_standard/mappers/ports.py | 5 ++++- tests/providers/test_input_port_lineage_and_quality_sql.py | 2 ++ 2 files changed, 6 insertions(+), 1 deletion(-) diff --git a/fluid_build/providers/odps_standard/mappers/ports.py b/fluid_build/providers/odps_standard/mappers/ports.py index a8b55cad..883338be 100644 --- a/fluid_build/providers/odps_standard/mappers/ports.py +++ b/fluid_build/providers/odps_standard/mappers/ports.py @@ -95,7 +95,10 @@ def _range_lower_bound(constraint: Any) -> Optional[str]: if not text or any(c in text for c in "<>=*xX^~"): return None parts = text.split(".") - if not 1 <= len(parts) <= 3 or not all(p.isdigit() for p in parts): + # ``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. + if not 1 <= len(parts) <= 3 or not all(p.isascii() and p.isdigit() 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 also what semver.org says a range's diff --git a/tests/providers/test_input_port_lineage_and_quality_sql.py b/tests/providers/test_input_port_lineage_and_quality_sql.py index a62e4c91..709aaee9 100644 --- a/tests/providers/test_input_port_lineage_and_quality_sql.py +++ b/tests/providers/test_input_port_lineage_and_quality_sql.py @@ -146,6 +146,8 @@ def test_the_documented_forms_resolve(self, constraint, expected): "<2.0.0", "<=2.0.0", "latest", + "^١.٢", # Arabic-Indic digits: str.isdigit() is true for these + "^2.²", # superscript two likewise "", " ", None, From 06a5b707fe42d979d549c695c814ff89a3a9d199 Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Tue, 27 Oct 2026 06:58:41 +0100 Subject: [PATCH 04/10] =?UTF-8?q?docs(changelog):=200.24.0=20=E2=80=94=20e?= =?UTF-8?q?xported=20lineage=20and=20executable=20quality=20SQL?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- CHANGELOG.md | 47 +++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 47 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index f9b292c6..13186a56 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,53 @@ 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. + `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 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 already written as a `SELECT` or `WITH` query passes through + unchanged. + +### 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 + +- 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. + ## [0.23.0] — 2026-10-27 ### Security From 81228286055a9642c06dbee0621911515bce439a Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Tue, 27 Oct 2026 07:24:26 +0100 Subject: [PATCH 05/10] fix(odps,odcs): three defects a review round found in the new export paths MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit All three were reproduced by execution before the fix and are pinned by tests that fail when it is reverted. `is_query` tested `expression.upper().startswith(("SELECT", "WITH"))`, which has no word boundary, so a predicate on any column whose name merely begins with those letters — `selected_at IS NOT NULL`, `with_tax > 0`, `withholding_amount >= 0` — was classified as an already-written query and exported verbatim with `mustBe: true`. For a whole class of columns that silently reinstated the unexecutable fragment this change exists to remove. The first token is now matched with `\b`. The wrapper was also built on one line, so a predicate ending in `;` produced `WHERE (amount > 0;) IS NOT TRUE` and one ending in a `-- comment` swallowed the closing paren and the comparison after it. Both are parse errors in DuckDB. A trailing `;` is now dropped and the predicate sits on its own line. The emitted query survives the YAML round-trip exactly and executes. `_range_lower_bound` accepted numeric components with a leading zero, so `versionConstraint: ^01.2` — schema-valid, since `$defs.semverRange` carries no pattern — resolved to `01.2.0`, which matches neither `$defs.semver` nor any resolver. Pre-change these exported `1.0.0`: wrong, but resolvable. They now fall back like every other unreadable range. `outputPorts[].inputContracts` is no longer derived from `builds[].properties.source`. That path synthesizes its `contractId` as `:///` precisely because the source has no contract at all, to populate the input-port field ODPS v1.0.0 requires. `inputContracts` is optional, so naming an id no catalog can resolve is worse than naming nothing — the failure `_input_port_contract_id` already warns about. An SDP exported two such entries per product and nothing covered the branch; it is covered now. Also corrects two docstring claims that did not hold: `semver` and `semantic_version` resolve in some environments and not others, so the sentence now rests only on what is true of the project (neither is a declared dependency), and semver.org defines no ranges, so the partial-version rule is credited to npm's range grammar, which is where `$defs.semverRange`'s examples come from. `_consume_input_port` is renamed `_consume_to_input_port`, the shape this file spells for a one-to-one mapper without exception. --- fluid_build/providers/odcs/mappers/quality.py | 31 ++++++-- .../providers/odps_standard/mappers/ports.py | 55 ++++++++----- ...test_input_port_lineage_and_quality_sql.py | 77 +++++++++++++++---- tests/providers/test_odcs_quality_export.py | 2 +- 4 files changed, 125 insertions(+), 40 deletions(-) diff --git a/fluid_build/providers/odcs/mappers/quality.py b/fluid_build/providers/odcs/mappers/quality.py index 07aa42d3..4b983647 100644 --- a/fluid_build/providers/odcs/mappers/quality.py +++ b/fluid_build/providers/odcs/mappers/quality.py @@ -446,6 +446,11 @@ 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. @@ -460,9 +465,9 @@ def _expression_to_quality(expr: Mapping[str, Any]) -> Dict[str, Any]: 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 the one OpenMetadata - documents for ``sql`` rules (exactly one comparison against a whole number). - ``${table}`` is the placeholder from that example. + 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 @@ -471,12 +476,21 @@ def _expression_to_quality(expr: Mapping[str, Any]) -> Dict[str, Any]: 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 (it starts with ``SELECT`` or + 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. + 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. """ expression = str(expr["expression"]).strip() - is_query = expression.upper().startswith(("SELECT", "WITH")) + # 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: @@ -484,7 +498,10 @@ def _expression_to_quality(expr: Mapping[str, Any]) -> Dict[str, Any]: check["mustBe"] = True check["description"] = f"Expression must hold for every row: {expression}" else: - check["query"] = f"SELECT COUNT(*) FROM ${{table}} WHERE ({expression}) IS NOT TRUE" + # 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}" diff --git a/fluid_build/providers/odps_standard/mappers/ports.py b/fluid_build/providers/odps_standard/mappers/ports.py index 883338be..a6bef938 100644 --- a/fluid_build/providers/odps_standard/mappers/ports.py +++ b/fluid_build/providers/odps_standard/mappers/ports.py @@ -74,10 +74,9 @@ def _range_lower_bound(constraint: Any) -> Optional[str]: 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. ``semver``, ``semantic_version`` and - ``packaging`` are all present in the resolved environment but none 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 + 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`` @@ -97,12 +96,17 @@ def _range_lower_bound(constraint: Any) -> Optional[str]: 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. - if not 1 <= len(parts) <= 3 or not all(p.isascii() and p.isdigit() for p in parts): + # 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 also what semver.org says a range's - # lower bound is. + # 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) @@ -147,6 +151,11 @@ def _input_contracts_from_ports( 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 @@ -173,7 +182,7 @@ def _input_contracts_from_ports( return contracts -def _consume_input_port( +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. @@ -246,14 +255,23 @@ def _build_input_ports( 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 subset - from paths 1 and 2 — the lineage this contract states itself, and the only - lineage ``outputPorts[].inputContracts`` may be derived from. 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. + 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` already warns about. + + 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]] = [] @@ -292,7 +310,7 @@ def _unique_name(base: str, ref: Optional[str]) -> str: seen.add(dedup_key) name = _unique_name(canonical["name"], canonical.get("reference")) seen_names.add(name) - consume_port = _consume_input_port(canonical, name, contract_id) + consume_port = _consume_to_input_port(canonical, name, contract_id) input_ports.append(consume_port) authored_ports.append(consume_port) @@ -319,7 +337,6 @@ def _unique_name(base: str, ref: Optional[str]) -> str: if custom_props: port["customProperties"] = custom_props input_ports.append(port) - authored_ports.append(port) # Path 3 — expects[] (FLUID 0.7.1 legacy) for expect in fluid.get("expects") or []: diff --git a/tests/providers/test_input_port_lineage_and_quality_sql.py b/tests/providers/test_input_port_lineage_and_quality_sql.py index 709aaee9..b6b95789 100644 --- a/tests/providers/test_input_port_lineage_and_quality_sql.py +++ b/tests/providers/test_input_port_lineage_and_quality_sql.py @@ -57,6 +57,14 @@ 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"} @@ -114,7 +122,7 @@ def walk(node: Any) -> None: # --------------------------------------------------------------------------- -# 19. the version an input port claims +# Defect 19a — the version an input port claims # --------------------------------------------------------------------------- @@ -148,6 +156,10 @@ def test_the_documented_forms_resolve(self, constraint, expected): "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, @@ -202,7 +214,7 @@ def test_the_range_still_rides_along_in_its_custom_property(self): # --------------------------------------------------------------------------- -# 19. the lineage an output port carries +# Defect 19b — the lineage an output port carries # --------------------------------------------------------------------------- @@ -273,6 +285,31 @@ def test_an_imported_documents_input_ports_are_not_treated_as_lineage(self): 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") @@ -281,7 +318,7 @@ def test_a_contract_with_no_consumes_gets_no_lineage_key(self): # --------------------------------------------------------------------------- -# 19. the new field is preview-gated +# Defect 19c — the new field is preview-gated # --------------------------------------------------------------------------- @@ -307,7 +344,7 @@ def test_a_non_semver_value_is_rejected_with_the_pattern(self): # --------------------------------------------------------------------------- -# 20. a quality expression an engine can execute +# Defect 20 — a quality expression an engine can execute # --------------------------------------------------------------------------- @@ -315,9 +352,7 @@ 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"] == ( - f"SELECT COUNT(*) FROM ${{table}} WHERE ({EXPRESSION}) IS NOT TRUE" - ) + assert rule["query"] == _counting_query(EXPRESSION) assert rule["mustBe"] == 0 def test_it_uses_the_placeholder_from_the_odcs_example(self): @@ -332,19 +367,35 @@ def test_it_negates_with_is_not_true_not_a_bare_not(self): assert "IS NOT TRUE" in rule["query"] assert f"NOT ({EXPRESSION})" not in rule["query"] - @pytest.mark.parametrize("prefix", ["SELECT", "select", "WITH", " SELECT"]) + @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 - def test_the_exported_odcs_validates(self): - from fluid_build.providers.odcs.validation import collect_errors - from fluid_build.providers.odcs.validation import load_schema as odcs_schema + @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 - odcs = OdcsProvider().render(_contract({"versionConstraint": "^2.1"}), expose_id="orders") - assert collect_errors(odcs, odcs_schema()) == [] + 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") diff --git a/tests/providers/test_odcs_quality_export.py b/tests/providers/test_odcs_quality_export.py index 9c14aa80..4a4041f6 100644 --- a/tests/providers/test_odcs_quality_export.py +++ b/tests/providers/test_odcs_quality_export.py @@ -236,7 +236,7 @@ def test_quality_expression_lands_on_the_object_as_a_sql_rule() -> None: assert rules["amount_positive"]["type"] == "sql" assert ( rules["amount_positive"]["query"] - == "SELECT COUNT(*) FROM ${table} WHERE (amount > 0) IS NOT TRUE" + == "SELECT COUNT(*) FROM ${table} WHERE (\namount > 0\n) IS NOT TRUE" ) assert rules["amount_positive"]["mustBe"] == 0 From 185fd51ce4156c6a7644fe245117cc35bedf5875 Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Tue, 27 Oct 2026 07:24:40 +0100 Subject: [PATCH 06/10] fix(datamesh-manager): a removed input port's derived lineage went with it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `remove_odps_product_consume_input_ports` strips product-to-product consumes from the ODPS payload before publish, because Entropy draws that lineage from Access resources — keeping the consumes as input ports would require mirroring each upstream as a SourceSystem and render duplicate graph nodes. Deriving `outputPorts[].inputContracts` states the same dependency a second time in the same document, and nothing stripped it, so `fluid publish --target datamesh-manager` began POSTing product-to-product lineage on the one payload it had been deliberately removed from. A removed port's lineage entry now goes with it, and the key is dropped when nothing is left. An explicit source-system consume keeps its entry, because its input port is kept too. --- .../datamesh_manager/_odps_helpers.py | 49 ++++++++++++- tests/providers/test_datamesh_manager_gaps.py | 72 +++++++++++++++++++ 2 files changed, 119 insertions(+), 2 deletions(-) diff --git a/fluid_build/providers/datamesh_manager/_odps_helpers.py b/fluid_build/providers/datamesh_manager/_odps_helpers.py index eb03a562..31a0a945 100644 --- a/fluid_build/providers/datamesh_manager/_odps_helpers.py +++ b/fluid_build/providers/datamesh_manager/_odps_helpers.py @@ -42,7 +42,7 @@ from __future__ import annotations import logging -from typing import Any, Dict, List, Mapping, Optional +from typing import Any, Dict, List, Mapping, MutableMapping, Optional, Set from fluid_build.util.contract import consumes_to_canonical_ports, kind_to_dmm_type @@ -180,6 +180,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:`_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 @@ -243,11 +246,53 @@ 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 _remove_output_port_input_contracts( + odps_payload: Dict[str, Any], removed_ports: List[Any] +) -> None: + """Drop ``outputPorts[].inputContracts`` entries for a removed input port. + + Since 0.24.0 the ODPS exporter derives that key from a product's own + ``consumes[]``, so one product-to-product dependency is stated twice in a + document. Removing only the input port left the same upstream named in the + output port's lineage, which puts back exactly the duplicate graph nodes + :func:`remove_odps_product_consume_input_ports` exists to prevent — the + product would have to be mirrored as a SourceSystem to resolve the + reference. Explicit source-system consumes keep their lineage, because + their input ports are kept too. + """ + contract_ids: Set[str] = { + 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 ensure_odps_input_port_contract_ids( diff --git a/tests/providers/test_datamesh_manager_gaps.py b/tests/providers/test_datamesh_manager_gaps.py index 1324cf77..f9fa52a5 100644 --- a/tests/providers/test_datamesh_manager_gaps.py +++ b/tests/providers/test_datamesh_manager_gaps.py @@ -713,6 +713,78 @@ def test_removes_product_consumes_and_retains_explicit_source_systems(self): {"name": "crm_source", "contractId": "bronze.telco.party_v1"} ] + 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": [ From 394957da286a69c79f49755ddfeaa0cd8eb0b793 Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Tue, 27 Oct 2026 07:24:40 +0100 Subject: [PATCH 07/10] docs: name contractVersion in the preview summary, and every changed export MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The 0.7.6 root description's "NEW in v0.7.6" list named its six sibling opt-in fields and not `contractVersion`, so an author reading the preview summary would not learn the field exists. It also now says what `fluid validate` does not check: that the value falls inside `versionConstraint`. `consumes_to_canonical_ports`' docstring claims to describe every field it returns and no longer did — `contract_version` is in the dict. Listing it also disambiguates it from `get_contract_version` in the same module, which returns the FLUID *schema* version. The changelog's `### Changed` listed only the ODPS keys and omitted the ODCS `sql` rule reshaping — `query`, `mustBe: true → 0` and `description` all change for every expression rule, which is the value most likely already baked into a consumer; our own test had to be rewritten for it. The section now opens with the upgrade warning 0.23.0 uses and tells the reader what to do. --- CHANGELOG.md | 35 +++++++++++++++++---- fluid_build/schemas/fluid-schema-0.7.6.json | 4 +-- fluid_build/util/contract.py | 3 +- 3 files changed, 33 insertions(+), 9 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 13186a56..557184bf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -18,9 +18,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 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. - `versionConstraint` still rides along in `customProperties`, so the range this product accepts is - not lost. + 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 @@ -30,7 +33,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 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. + 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. A `fluid publish --target datamesh-manager` + payload has the derived lineage stripped alongside the input ports it mirrors, for the reason + those were already stripped — Entropy draws product-to-product lineage from Access resources, and + naming the upstream twice would render duplicate graph nodes. - **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 @@ -39,8 +48,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 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 already written as a `SELECT` or `WITH` query passes through - unchanged. + 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. ### Added @@ -50,9 +62,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### 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 diff --git a/fluid_build/schemas/fluid-schema-0.7.6.json b/fluid_build/schemas/fluid-schema-0.7.6.json index 5674d192..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": [ @@ -2969,7 +2969,7 @@ }, "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.", + "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" ] diff --git a/fluid_build/util/contract.py b/fluid_build/util/contract.py index 47be7e51..c2aee8a9 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], + "contract_version": Optional[str], # upstream contract version, explicit only # semverRange "qos_expectations": Optional[Mapping], # freshnessMax / maxStaleness / ... "required_policies": Optional[list], "tags": Optional[list], From 0c2513c537bd1edbe8647c5db0771a6323b433a9 Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Tue, 27 Oct 2026 08:14:45 +0100 Subject: [PATCH 08/10] fix(datamesh-manager): the lineage strip missed the canonical publish path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The previous commit fixed `DataMeshManagerProvider`, which backs `fluid datamesh-manager publish`. That is the secondary path. The canonical one — `fluid publish --target datamesh-manager` and `catalog.register: [datamesh_manager]` — goes through `DataMeshManagerRegistrar`, whose own module notes DMM has "exactly one publish path". It filters `inputPorts` to those carrying a `sourceSystemId`, dropping product-to-product ports, and never touched `inputContracts` — so the derived lineage still named those upstreams on the payload PUT to `/api/dataproducts/{id}`, and the changelog's claim about that command was false as written. The strip now lives once, as `remove_output_port_input_contracts` in the ODPS mapper that owns the derivation, so the two cannot drift. The two publish paths strip by different rules — a missing `sourceSystemId` in the registrar, a canonical product-consume match in the provider — so each passes the ports it dropped and the helper reads the `contractId`s off them. The registrar imports it inside the function, because that module is imported for its `register_catalog_backend` side effect and must stay off the light CLI's cold path; `tests/perf/test_startup_budget.py` still passes. A test drives the registrar end to end through `register_payload` with the HTTP layer mocked and asserts the dataproduct body names no upstream product. It fails when either the registrar call or the shared helper is reverted. Also corrects two docstrings: the canonical-shape block had `# semverRange` displaced onto the new `contract_version` line, which documented an exact `$defs.semver` field as a range, and `_build_input_ports` said `_input_port_contract_id` "warns about" a failure it only describes — it holds no logger. --- CHANGELOG.md | 11 +-- .../catalog_registrars/datamesh_manager.py | 29 +++++--- .../datamesh_manager/_odps_helpers.py | 47 ++----------- .../providers/odps_standard/mappers/ports.py | 53 ++++++++++++++- fluid_build/util/contract.py | 4 +- .../test_catalog_datamesh_manager.py | 68 +++++++++++++++++++ 6 files changed, 155 insertions(+), 57 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 557184bf..e11813a9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -36,10 +36,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 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. A `fluid publish --target datamesh-manager` - payload has the derived lineage stripped alongside the input ports it mirrors, for the reason - those were already stripped — Entropy draws product-to-product lineage from Access resources, and - naming the upstream twice would render duplicate graph nodes. + 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 diff --git a/fluid_build/build_runners/catalog_registrars/datamesh_manager.py b/fluid_build/build_runners/catalog_registrars/datamesh_manager.py index 564befa0..8144feac 100644 --- a/fluid_build/build_runners/catalog_registrars/datamesh_manager.py +++ b/fluid_build/build_runners/catalog_registrars/datamesh_manager.py @@ -421,15 +421,28 @@ 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 for its + # ``register_catalog_backend`` side effect, so it stays + # off the light CLI's cold path (tests/perf). + from fluid_build.providers.odps_standard.mappers.ports import ( + 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. + remove_output_port_input_contracts(parsed, removed) 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 31a0a945..ab7e53fb 100644 --- a/fluid_build/providers/datamesh_manager/_odps_helpers.py +++ b/fluid_build/providers/datamesh_manager/_odps_helpers.py @@ -42,8 +42,11 @@ from __future__ import annotations import logging -from typing import Any, Dict, List, Mapping, MutableMapping, Optional, Set +from typing import Any, Dict, List, Mapping, Optional +from fluid_build.providers.odps_standard.mappers.ports import ( + 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") @@ -181,7 +184,7 @@ def remove_odps_product_consume_input_ports( ports. A removed port's ``outputPorts[].inputContracts`` entry goes with it — see - :func:`_remove_output_port_input_contracts`. + :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 @@ -254,45 +257,7 @@ def _is_product_consume(port: Any) -> bool: odps_payload["inputPorts"] = retained else: odps_payload.pop("inputPorts", None) - _remove_output_port_input_contracts(odps_payload, removed) - - -def _remove_output_port_input_contracts( - odps_payload: Dict[str, Any], removed_ports: List[Any] -) -> None: - """Drop ``outputPorts[].inputContracts`` entries for a removed input port. - - Since 0.24.0 the ODPS exporter derives that key from a product's own - ``consumes[]``, so one product-to-product dependency is stated twice in a - document. Removing only the input port left the same upstream named in the - output port's lineage, which puts back exactly the duplicate graph nodes - :func:`remove_odps_product_consume_input_ports` exists to prevent — the - product would have to be mirrored as a SourceSystem to resolve the - reference. Explicit source-system consumes keep their lineage, because - their input ports are kept too. - """ - contract_ids: Set[str] = { - 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) + remove_output_port_input_contracts(odps_payload, removed) def ensure_odps_input_port_contract_ids( diff --git a/fluid_build/providers/odps_standard/mappers/ports.py b/fluid_build/providers/odps_standard/mappers/ports.py index a6bef938..156e78ac 100644 --- a/fluid_build/providers/odps_standard/mappers/ports.py +++ b/fluid_build/providers/odps_standard/mappers/ports.py @@ -36,7 +36,7 @@ from __future__ import annotations -from collections.abc import Mapping +from collections.abc import Iterable, Mapping, MutableMapping from typing import Any, Dict, List, Optional, Tuple from fluid_build.util.contract import ( @@ -225,6 +225,54 @@ def _consume_to_input_port( return port +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 ) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]: @@ -265,7 +313,8 @@ def _build_input_ports( 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` already warns about. + :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 diff --git a/fluid_build/util/contract.py b/fluid_build/util/contract.py index c2aee8a9..5c3294ce 100644 --- a/fluid_build/util/contract.py +++ b/fluid_build/util/contract.py @@ -608,8 +608,8 @@ def consumes_to_canonical_ports( # --- 0.7.x consumeRef canonical fields --- "reference": Optional[str], # productId "description": str, # purpose - "version_constraint": Optional[str], - "contract_version": Optional[str], # upstream contract version, explicit only # 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], diff --git a/tests/build_runners/test_catalog_datamesh_manager.py b/tests/build_runners/test_catalog_datamesh_manager.py index a1dfb28c..5fb2a5b9 100644 --- a/tests/build_runners/test_catalog_datamesh_manager.py +++ b/tests/build_runners/test_catalog_datamesh_manager.py @@ -87,6 +87,74 @@ 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"] + + # ── Registrar unit ─────────────────────────────────────────────────────── From 95902b3e81bc5c7dbe20d717afe74ec5b7e5f066 Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Tue, 27 Oct 2026 08:33:54 +0100 Subject: [PATCH 09/10] fix(datamesh-manager): the lineage strip reached past the consumes it was for MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The previous commit drove the strip from every input port the registrar's filter dropped. That filter drops anything without a `sourceSystemId`, which catches an `expects[]` port as well as a product consume — and an `expects[]` port's expose replays the upstream document's OWN `inputContracts` verbatim through the ODPS pass-through. So publishing an imported product deleted the only record of its upstream: Access agreements are built from `consumes[]`, and that document has none, so nothing replaced it. Reproduced on the real wire path through `register_payload` with the HTTP layer mocked: with the strip as written the `vendor.customers_raw` entry was gone, and with it disabled the entry survived. `consume_input_port_contract_ids` now names the ids a `consumes[]` export actually produces, and only ports matching those are passed to the strip. Three cases are pinned: a consume's derived lineage is dropped, an `expects[]` port's own lineage is kept, and a contract with both keeps the pass-through, since that wins in the mapper. The provider path never had this reach — its `_is_product_consume` matches canonical product consumes only. Also corrects the deferred import's comment. The claim that this module is on the light CLI's cold path is false in both directions: `fluid --help` already loads `providers.odps_standard`, and `catalog_registrars` is not imported on that path at all. The import stays deferred for the reason that does hold — building the backend registry should not pull the ODPS provider in with it. --- .../catalog_registrars/datamesh_manager.py | 24 ++++++++++--- .../providers/odps_standard/mappers/ports.py | 25 +++++++++++++- .../test_catalog_datamesh_manager.py | 34 +++++++++++++++++++ 3 files changed, 78 insertions(+), 5 deletions(-) diff --git a/fluid_build/build_runners/catalog_registrars/datamesh_manager.py b/fluid_build/build_runners/catalog_registrars/datamesh_manager.py index 8144feac..000c1e3e 100644 --- a/fluid_build/build_runners/catalog_registrars/datamesh_manager.py +++ b/fluid_build/build_runners/catalog_registrars/datamesh_manager.py @@ -421,10 +421,12 @@ def _render_dmm_data_product_body( parsed, payload ) if "inputPorts" in parsed and isinstance(parsed["inputPorts"], list): - # Deferred: this module is imported for its - # ``register_catalog_backend`` side effect, so it stays - # off the light CLI's cold path (tests/perf). + # 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, ) @@ -442,7 +444,21 @@ def _render_dmm_data_product_body( # output port's lineage unless it goes with it — which # would put back the unresolvable SourceSystem # reference this strip exists to remove. - remove_output_port_input_contracts(parsed, removed) + # + # 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/odps_standard/mappers/ports.py b/fluid_build/providers/odps_standard/mappers/ports.py index 156e78ac..b5e3c4fa 100644 --- a/fluid_build/providers/odps_standard/mappers/ports.py +++ b/fluid_build/providers/odps_standard/mappers/ports.py @@ -37,7 +37,7 @@ from __future__ import annotations from collections.abc import Iterable, Mapping, MutableMapping -from typing import Any, Dict, List, Optional, Tuple +from typing import Any, Dict, List, Optional, Set, Tuple from fluid_build.util.contract import ( builds_to_canonical_input_ports, @@ -225,6 +225,29 @@ def _consume_to_input_port( 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: diff --git a/tests/build_runners/test_catalog_datamesh_manager.py b/tests/build_runners/test_catalog_datamesh_manager.py index 5fb2a5b9..18ac357c 100644 --- a/tests/build_runners/test_catalog_datamesh_manager.py +++ b/tests/build_runners/test_catalog_datamesh_manager.py @@ -154,6 +154,40 @@ def test_the_dataproduct_put_names_no_upstream_product(self): for port in body.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 ─────────────────────────────────────────────────────── From 063a4e1d7f2db8b883c9409fe2c147a072393eff Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Tue, 27 Oct 2026 08:53:19 +0100 Subject: [PATCH 10/10] fix(datamesh-manager): a renamed port leaked a product consume on publish MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Found while covering the lineage strip, and fixed here rather than carried because deriving `outputPorts[].inputContracts` would have doubled it: the leaked port and its mirrored lineage entry. When two `consumes[]` share an `exposeId` the exporter renames the second, and `_is_product_consume` matched the renamed port by name. That misses two of the three rename shapes: a `productId` tail containing `__` of its own (`silver.crm__core` → `crm__core__customers`, where splitting at the first `__` reads the boundary wrong) and the numeric-suffix fallback (`customers_2`, which carries no `__` at all). Either one published an input port Entropy cannot resolve without mirroring the upstream product as a SourceSystem — the duplicate graph node the strip exists to prevent. Reproduced through `apply(..., dry_run=True)` on contracts `fluid validate` accepts. The match now also tests the contract id the exporter stamps on the port, via `input_port_contract_id` — the exporter's own helper, made public rather than copied, since the id convention already carries a comment about not keeping two copies in step. That id is `{productId}.{exposeId}` or one the author pinned, so it cannot catch a source-system port that merely shares an upstream, and it is independent of the name the rename disturbs. Measured before settling on it: the predicate alone closes all three rename shapes, so the name checks are left exactly as they were instead of also being rewritten. An earlier draft changed the `__` split to a suffix search as well; that turned out to be redundant and was dropped, and the test that appeared to cover it was vacuous — its three consumes shared a `contractId`, so the exporter's dedup collapsed them to one port and no rename happened. The case now uses three consumes sharing an `exposeId` and a `productId` tail, which genuinely produces `customers_2`, and both leak shapes fail when the predicate is reverted. Four more cases the audit found uncovered are pinned too: a stated pass-through lineage surviving when the product also consumes, a stated entry naming the consumed contract still being removed, and a source-system consume keeping both its port and its lineage. --- CHANGELOG.md | 12 ++ .../datamesh_manager/_odps_helpers.py | 23 ++++ .../providers/odps_standard/mappers/ports.py | 10 +- .../test_catalog_datamesh_manager.py | 54 +++++++++ tests/providers/test_datamesh_manager_gaps.py | 108 ++++++++++++++++++ 5 files changed, 202 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e11813a9..044e8b23 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -57,6 +57,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 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 diff --git a/fluid_build/providers/datamesh_manager/_odps_helpers.py b/fluid_build/providers/datamesh_manager/_odps_helpers.py index ab7e53fb..adedb1d7 100644 --- a/fluid_build/providers/datamesh_manager/_odps_helpers.py +++ b/fluid_build/providers/datamesh_manager/_odps_helpers.py @@ -45,6 +45,7 @@ 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 @@ -203,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"): @@ -212,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 @@ -220,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 diff --git a/fluid_build/providers/odps_standard/mappers/ports.py b/fluid_build/providers/odps_standard/mappers/ports.py index b5e3c4fa..c83aaf82 100644 --- a/fluid_build/providers/odps_standard/mappers/ports.py +++ b/fluid_build/providers/odps_standard/mappers/ports.py @@ -242,7 +242,7 @@ def consume_input_port_contract_ids(fluid: Mapping[str, Any]) -> Set[str]: """ ids: Set[str] = set() for canonical in consumes_to_canonical_ports(fluid): - contract_id = _input_port_contract_id(canonical) + contract_id = input_port_contract_id(canonical) if contract_id: ids.add(str(contract_id)) return ids @@ -309,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 @@ -336,7 +336,7 @@ def _build_input_ports( 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 + :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 @@ -375,7 +375,7 @@ 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 @@ -422,7 +422,7 @@ def _unique_name(base: str, ref: Optional[str]) -> str: 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 diff --git a/tests/build_runners/test_catalog_datamesh_manager.py b/tests/build_runners/test_catalog_datamesh_manager.py index 18ac357c..a8d2e1f8 100644 --- a/tests/build_runners/test_catalog_datamesh_manager.py +++ b/tests/build_runners/test_catalog_datamesh_manager.py @@ -154,6 +154,60 @@ def test_the_dataproduct_put_names_no_upstream_product(self): 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``, diff --git a/tests/providers/test_datamesh_manager_gaps.py b/tests/providers/test_datamesh_manager_gaps.py index f9fa52a5..8d5f936b 100644 --- a/tests/providers/test_datamesh_manager_gaps.py +++ b/tests/providers/test_datamesh_manager_gaps.py @@ -713,6 +713,114 @@ 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