fix: exported lineage named no upstreams, claimed an unset version, and shipped unexecutable quality SQL

CHANGELOG.md

@@ -7,6 +7,91 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [0.24.0] — 2026-10-27
+
+### Fixed
+
+- **Exported lineage claimed a version the contract never pinned.** An ODPS `inputPorts[].version`
+  came from the consume's own `version` or the provider's `1.0.0` default, and a consume that states
+  a range (`versionConstraint: ^2.1`) has neither — so every such export asserted `version: 1.0.0`,
+  a pin the contract did not make and, for a product whose 1.x line is retired, a version that no
+  longer exists. The lowest version the range accepts is now resolved (`^2.1` → `2.1.0`,
+  `~1.4` → `1.4.0`, `>=3.0.0` → `3.0.0`). A range the resolver cannot read falls back to the old
+  constant rather than guessing: ODPS v1.0.0 has `version` in `InputPort`'s `required`, so the field
+  cannot simply be omitted, and a wrong exact version reads as a fact somebody established. The
+  refusals include a compound range, a wildcard, `>1.0.0` (whose true lower bound semver gives no
+  way to name), and anything whose components are not plain ASCII digits without a leading zero —
+  `^01.2` is a schema-valid constraint, but `01.2.0` matches neither `$defs.semver` nor any
+  resolver. `versionConstraint` still rides along in `customProperties`, so the range this product
+  accepts is not lost.
+
+- **An exported output port named none of its upstreams.** `outputPorts[].inputContracts` was
+  populated only from an imported ODPS document's pass-through, so a contract written in FLUID
+  exported output ports with no lineage at all and a catalog reading dependencies there saw a
+  product with none. The entries are now derived from the document's own resolved input ports — one
+  per `consumes[]`, carrying exactly the `id` and `version` Bitol's `$defs.InputContract` allows,
+  read from those ports rather than recomputed so the two halves of one document cannot disagree. An
+  imported document's own `inputContracts` still win, and `expects[]` — the carrier an imported
+  document's input ports arrive on — is excluded, so import → export stays a fixed point and never
+  adds a dependency the author did not write. A `builds[].properties.source` stream is excluded too:
+  its `<kind>://<database>/<stream>` id is synthesized to populate the *required* input-port field
+  for a source that has no contract at all, and `inputContracts` is optional, so naming an id no
+  catalog can resolve is worse than naming nothing. Both Data Mesh Manager publish paths — the
+  registrar behind `fluid publish --target datamesh-manager` and `catalog.register`, and
+  `fluid datamesh-manager publish` — strip the derived lineage alongside the input ports it mirrors,
+  for the reason those were already stripped: Entropy draws product-to-product lineage from Access
+  resources, so naming the upstream twice would require mirroring it as a SourceSystem and render
+  duplicate graph nodes. They strip by different rules, so each passes the ports it dropped to one
+  shared helper.
+
+- **A quality expression exported as SQL no DQ engine could execute.** ODCS defines the `sql` rule's
+  `query` as an individual query returning a value that can be compared, and FLUID's
+  `exposes[].contract.quality[].expression` is a row predicate — it was emitted verbatim, so an
+  engine was handed `preferred_customer_flag IN ('Y', 'N')` with `mustBe: true` and nothing to
+  execute. A predicate now becomes a counting query compared to zero, against the `${table}`
+  placeholder from ODCS's own example. The negation is `(…) IS NOT TRUE`, not `NOT (…)`: under
+  three-valued logic a NULL row makes the naive form NULL and counts no failure, although that row's
+  predicate did not hold. An expression whose *first token* is `SELECT` or `WITH` passes through
+  unchanged; the token boundary matters, because matching a bare prefix left every predicate on a
+  column named `selected_at` or `with_tax` in the broken state this fixes. The predicate is wrapped
+  on its own line and a trailing `;` is dropped, so neither a line comment nor a stray semicolon can
+  swallow the closing paren.
+
+- **A renamed input port leaked a product consume into the Data Mesh Manager payload.** Found while
+  covering the lineage strip above. When two `consumes[]` share an `exposeId` the exporter renames
+  the second, and `fluid datamesh-manager publish` matched the renamed port by name — which misses
+  two of the three rename shapes: a `productId` tail containing `__` of its own
+  (`silver.crm__core` → `crm__core__customers`, split at the wrong `__`), and the numeric-suffix
+  fallback (`customers_2`, with no `__` to match on). Both published a port Entropy cannot resolve
+  without mirroring the upstream product as a SourceSystem. The match now also tests the contract id
+  the exporter actually stamps on the port, read from the exporter's own helper so the convention has
+  one definition; a source-system consume keeps its port, as before. This leak predates the release —
+  but deriving `outputPorts[].inputContracts` would have doubled it, so it is fixed here rather than
+  carried.
+
+### Added
+
+- `consumes[].contractVersion` (0.7.6 preview) — the exact upstream version this product was built
+  and tested against, within `versionConstraint`. It wins over the resolved lower bound as the
+  version an exported input port, and its matching `inputContracts` entry, claims.
+
+### Changed
+
+**Exported artifacts change shape in this release.** If you have tooling that reads forge's ODCS or
+ODPS output, read this before upgrading. Each change is a fix for output that was wrong and is
+described in full under *Fixed*; the summary is here because the old values may be baked into
+consumers.
+
+- An ODPS export of a contract with `consumes[]` now carries `outputPorts[].inputContracts`, where
+  before it carried none, and an input port's `version` may differ from `1.0.0` where the contract
+  states a range. A consumer that compares exported documents byte-for-byte will see both.
+- An ODCS `sql` quality rule derived from `exposes[].contract.quality[].expression` now carries a
+  counting query (`SELECT COUNT(*) FROM ${table} WHERE (<expression>) IS NOT TRUE`, the predicate on
+  its own line) with `mustBe: 0`, where it previously carried the bare expression with
+  `mustBe: true`; its `description` changes with it. A DQ configuration or test that pins the old
+  `query` string or the boolean `mustBe` must be updated — ours was. An expression already written
+  as a `SELECT` or `WITH` query is unchanged.
+
 ## [0.23.0] — 2026-10-27
 
 ### Security

fluid_build/build_runners/catalog_registrars/datamesh_manager.py

@@ -421,15 +421,44 @@ def _render_dmm_data_product_body(
                         parsed, payload
                     )
                     if "inputPorts" in parsed and isinstance(parsed["inputPorts"], list):
-                        parsed["inputPorts"] = [
-                            ip
-                            for ip in parsed["inputPorts"]
-                            if isinstance(ip, dict)
-                            and "sourceSystemId" in ip
-                            and ip["sourceSystemId"]
-                        ]
-                        if not parsed["inputPorts"]:
+                        # Deferred: this module is imported purely for its
+                        # ``register_catalog_backend`` side effect, so building
+                        # the backend registry does not pull the ODPS provider
+                        # in with it.
+                        from fluid_build.providers.odps_standard.mappers.ports import (
+                            consume_input_port_contract_ids,
+                            remove_output_port_input_contracts,
+                        )
+
+                        kept: List[Any] = []
+                        removed: List[Any] = []
+                        for ip in parsed["inputPorts"]:
+                            keep = isinstance(ip, dict) and ip.get("sourceSystemId")
+                            (kept if keep else removed).append(ip)
+                        parsed["inputPorts"] = kept
+                        if not kept:
                             parsed.pop("inputPorts", None)
+                        # The ODPS render derives
+                        # ``outputPorts[].inputContracts`` from the same
+                        # ``consumes[]``, so a stripped port stays named in the
+                        # output port's lineage unless it goes with it — which
+                        # would put back the unresolvable SourceSystem
+                        # reference this strip exists to remove.
+                        #
+                        # Only the consume-derived ones, though. The rule above
+                        # also drops an ``expects[]`` port, whose expose replays
+                        # the upstream document's OWN ``inputContracts``; that
+                        # is the only record of the dependency, since Access
+                        # agreements are built from ``consumes[]``.
+                        consume_ids = consume_input_port_contract_ids(payload.contract or {})
+                        remove_output_port_input_contracts(
+                            parsed,
+                            [
+                                ip
+                                for ip in removed
+                                if isinstance(ip, dict) and str(ip.get("contractId")) in consume_ids
+                            ],
+                        )
                     return parsed
             except Exception:  # noqa: BLE001 — fall through to native
                 LOG.debug(

fluid_build/providers/datamesh_manager/_odps_helpers.py

@@ -44,6 +44,10 @@
 import logging
 from typing import Any, Dict, List, Mapping, Optional
 
+from fluid_build.providers.odps_standard.mappers.ports import (
+    input_port_contract_id,
+    remove_output_port_input_contracts,
+)
 from fluid_build.util.contract import consumes_to_canonical_ports, kind_to_dmm_type
 
 LOG = logging.getLogger("fluid.providers.datamesh_manager.odps")
@@ -180,6 +184,9 @@ def remove_odps_product_consume_input_ports(
     duplicate graph nodes. Explicit source-system consumes remain as input
     ports.
 
+    A removed port's ``outputPorts[].inputContracts`` entry goes with it — see
+    :func:`fluid_build.providers.odps_standard.mappers.ports.remove_output_port_input_contracts`.
+
     Matching strategy: we identify product-consume ports by **both**
     ``(id, name)`` (the canonical exposeId — matches the inputPort before
     the dedup-rename in :func:`fluid_build.providers.odps_standard.mappers
@@ -197,6 +204,12 @@ def remove_odps_product_consume_input_ports(
 
     product_port_names: set[str] = set()
     product_contract_ids: set[str] = set()
+    # The id the exporter actually stamps on a product-consume's input port,
+    # read from the same helper the exporter uses so the convention has one
+    # definition. Matching on it catches a renamed port whatever shape the
+    # rename took, including the numeric-suffix fallback that carries no
+    # ``__`` for the name checks below to find.
+    product_derived_contract_ids: set[str] = set()
     for canonical in consumes_to_canonical_ports(fluid, logger=LOG):
         reference = canonical.get("reference")
         if not reference or canonical.get("source_system_id"):
@@ -206,6 +219,9 @@ def remove_odps_product_consume_input_ports(
             if value:
                 product_port_names.add(str(value))
         product_contract_ids.add(str(reference))
+        derived = input_port_contract_id(canonical)
+        if derived:
+            product_derived_contract_ids.add(str(derived))
 
     if not product_port_names and not product_contract_ids:
         return
@@ -214,6 +230,19 @@ def _is_product_consume(port: Any) -> bool:
         if not isinstance(port, Mapping):
             return False
         name = str(port.get("name", ""))
+        # An exact match on the derived contract id is the most specific
+        # signal there is: it is ``{productId}.{exposeId}`` or an id the
+        # author pinned, not the bare productId the checks below fall back
+        # to, so it cannot catch a source-system port that merely shares an
+        # upstream. Being independent of the NAME is what matters — the name
+        # checks below miss two of the three rename shapes, a productId tail
+        # containing ``__`` of its own (``silver.crm__core`` →
+        # ``crm__core__customers``, split at the wrong ``__``) and the
+        # numeric-suffix fallback (``customers_2``, no ``__`` to find at
+        # all). Both leaked a port Entropy cannot resolve, and since the
+        # lineage key mirrors the ports, they leaked twice.
+        if str(port.get("contractId", "")) in product_derived_contract_ids:
+            return True
         # Fast path — original (pre-Gap-3) behavior. A port whose name is
         # exactly a canonical product-consume exposeId is a product-consume
         # regardless of contractId. This is the legacy path that
@@ -243,11 +272,15 @@ def _is_product_consume(port: Any) -> bool:
                     return True
         return False
 
-    retained = [port for port in input_ports if not _is_product_consume(port)]
+    retained: List[Any] = []
+    removed: List[Any] = []
+    for port in input_ports:
+        (removed if _is_product_consume(port) else retained).append(port)
     if retained:
         odps_payload["inputPorts"] = retained
     else:
         odps_payload.pop("inputPorts", None)
+    remove_output_port_input_contracts(odps_payload, removed)
 
 
 def ensure_odps_input_port_contract_ids(

fluid_build/providers/odcs/mappers/quality.py

@@ -446,20 +446,65 @@ def _dq_severity(check: Dict[str, Any], rule: Mapping[str, Any]) -> None:
         check["severity"] = _DQ_SEVERITY.get(severity, severity)
 
 
+# ``\b`` so only a whole first token counts: a column named ``selected_at`` or
+# ``with_tax`` begins with these letters and is a predicate, not a query.
+_QUERY_FIRST_TOKEN_RE = re.compile(r"\s*(?:select|with)\b", re.IGNORECASE)
+
+
 def _expression_to_quality(expr: Mapping[str, Any]) -> Dict[str, Any]:
     """One ``exposes[].contract.quality`` entry as an ODCS ``sql`` rule.
 
-    The FLUID field is "SQL or boolean expression that must evaluate to true",
-    which is exactly ODCS ``$defs.DataQualitySql`` (``query`` plus one operator)
-    with ``mustBe: true``. ``mustBe`` is untyped in the spec, so a boolean is
-    as valid there as a row count.
+    The FLUID field is "SQL or boolean expression that must evaluate to true" —
+    a ROW PREDICATE, such as ``preferred_customer_flag IN ('Y', 'N')``. ODCS is
+    specific about what ``sql`` means: ``$defs.DataQuality.type`` says "'sql' is
+    an individual SQL query that returns a value that can be compared", and
+    ``$defs.DataQualitySql.query`` carries the example ``SELECT COUNT(*) FROM
+    ${table} WHERE ${column} IS NOT NULL``. Emitting the bare predicate as
+    ``query`` therefore handed an engine something it cannot execute: it was a
+    fragment, not a statement, and the ``mustBe: true`` beside it compared
+    against a value no query produced.
+
+    A predicate is now exported as the count of rows that fail it, compared to
+    zero — the shape ODCS's own example takes, and one its
+    ``$defs.DataQualityOperators`` ``oneOf`` accepts. ``${table}`` is the
+    placeholder from that example.
+
+    ``IS NOT TRUE`` rather than ``NOT (...)`` on purpose. Under SQL's
+    three-valued logic ``NOT (flag IN ('Y','N'))`` is NULL when ``flag`` is
+    NULL, so a naive negation counts no failure for the row — yet a row whose
+    predicate did not evaluate to TRUE has not satisfied "must evaluate to
+    true". ``IS NOT TRUE`` catches NULL and FALSE alike, and is supported by
+    DuckDB, PostgreSQL, MySQL, Snowflake and BigQuery.
+
+    An expression that is already a query (its first token is ``SELECT`` or
+    ``WITH``) is passed through as written with the previous comparison: the
+    author wrote a statement and chose what it returns. The boundary matters —
+    ``startswith`` alone sent ``selected_at IS NOT NULL`` and ``with_tax > 0``
+    down that branch and exported them unchanged, reinstating the very fragment
+    this is here to remove.
+
+    A predicate still containing a ``;`` of its own is left as the author wrote
+    it: the field is specified as one predicate, and a second statement in it is
+    not something to guess at.
     """
-    check: Dict[str, Any] = {
-        "type": "sql",
-        "query": expr["expression"],
-        "mustBe": True,
-        "description": f"Expression must hold for every row: {expr['expression']}",
-    }
+    expression = str(expr["expression"]).strip()
+    # A trailing ``;`` would end up inside the wrapper's parentheses below.
+    expression = expression.rstrip(";").rstrip()
+    is_query = bool(_QUERY_FIRST_TOKEN_RE.match(expression))
+
+    check: Dict[str, Any] = {"type": "sql"}
+    if is_query:
+        check["query"] = expression
+        check["mustBe"] = True
+        check["description"] = f"Expression must hold for every row: {expression}"
+    else:
+        # The predicate gets its own line: a trailing ``-- comment`` on one line
+        # with the wrapper swallowed ``) IS NOT TRUE`` and left a statement the
+        # engine could not parse.
+        check["query"] = f"SELECT COUNT(*) FROM ${{table}} WHERE (\n{expression}\n) IS NOT TRUE"
+        check["mustBe"] = 0
+        check["description"] = f"Rows failing the expression must be 0: {expression}"
+
     severity = expr.get("severity")
     if severity:
         check["severity"] = _DQ_SEVERITY.get(severity, severity)