fix(iceberg): sinks write the expose their build names; dynamodb/jdbc/bigquery sinks start; Confluent reads as Glue; policy compile speaks up

.github/workflows/ci.yml

@@ -408,12 +408,12 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        # "1.20"      = lower-bound pin from pyproject.toml `mcp>=1.20`
+        # "1.22"      = lower-bound pin from pyproject.toml `mcp>=1.22`
         #               (see comment there for the FastMCP API features we
         #               need: deferred-annotation handling, ClientSession
         #               ``sampling_capabilities`` kwarg, etc.)
         # "range-max" = newest release inside the pyproject constraint
-        #               (`mcp>=1.20,<3.0`) — what a fresh
+        #               (`mcp>=1.22,<3.0`) — what a fresh
         #               `pip install data-product-forge` resolves today.
         #               Since Phase 2 widened the ceiling this resolves to
         #               2.x, and this leg is NOT warn-wrapped, so 2.x is
@@ -425,7 +425,7 @@ jobs:
         #               warn-only, for the same reason it was warn-only
         #               for 2.x: an upstream release must not be able to
         #               paint every open PR red.
-        mcp-version: ["1.20", "range-max", "latest"]
+        mcp-version: ["1.22", "range-max", "latest"]
     steps:
       - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1  # v7.0.1
       - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97  # v7.0.0
@@ -451,7 +451,7 @@ jobs:
           done
           case "${{ matrix.mcp-version }}" in
             # Keep the range-max constraint in sync with pyproject.toml.
-            range-max) pip install --upgrade "mcp>=1.20,<3.0" ;;
+            range-max) pip install --upgrade "mcp>=1.22,<3.0" ;;
             latest)    pip install --upgrade mcp ;;
             *)         pip install --force-reinstall "mcp==${{ matrix.mcp-version }}" ;;
           esac
@@ -470,7 +470,7 @@ jobs:
           version = md.version("mcp")
           major = int(version.split(".")[0])
           print(f"leg={leg} resolved mcp=={version} (major {major})")
-          expected = {"1.20": 1, "range-max": 2, "latest": 2}[leg]
+          expected = {"1.22": 1, "range-max": 2, "latest": 2}[leg]
           if major != expected:
               sys.exit(
                   f"::error::leg '{leg}' resolved mcp {version}, expected major {expected}. "

CHANGELOG.md

@@ -9,6 +9,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Upgrade notes
 
+- The minimum `mcp` is now 1.22 (was 1.20). mcp 1.20 and 1.21 import
+  `eval_type_backport` from pydantic's private `_typing_extra` module, which pydantic 2.14
+  removed, so `mcp.server.fastmcp`, and every forge-cli MCP surface with it, failed to
+  import on a fresh install that resolved those releases with current pydantic. 1.22 is
+  the first 1.x that imports with pydantic 2.14. The CI drift leg that pins the floor moves
+  from 1.20 to 1.22.
 - An AWS contract that declared `location.catalog` other than `glue` (for example
   `rest`, `iceberg_rest`, `polaris`, `unity`, `nessie` or `lakekeeper`) on an Iceberg
   expose holds a Glue database and table in its OpenTofu state that earlier releases
@@ -29,6 +35,47 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   table lived in two catalogs; `fluid validate` and the runners now refuse it unless the
   sink writes BigLake. Set `catalog: bigquery`, or the REST kind your catalog is. A
   hand-written sink config must then select that same catalog.
+- An Iceberg expose with `catalog: dynamodb` or `catalog: jdbc` that a streaming sink
+  writes to now needs a warehouse: both catalogs refuse to start without one. Set
+  `location.warehouse`, or an explicit `location.bucket` on `platform: aws` or `gcp`, from
+  which forge-cli derives `s3://` or `gs://` `<bucket>/<path>` (`path` defaults to
+  `<database>/<table>/`), or set the sink's `warehouse` property in an override. With none
+  of these, `fluid validate` and the runner preflight refuse the build.
+- An Iceberg expose with `catalog: bigquery` that a streaming sink writes to now needs
+  `location.project`, which becomes the catalog's `gcp.bigquery.project-id`, or that
+  property in an override. A blank value counts as missing.
+- The derived BigQuery sink config no longer sets `client.region`: `location.region` goes
+  to `gcp.bigquery.location`. Its warehouse is now the `gs://` storage URI dbt-bigquery and
+  the GCP IaC use (a `gs://` `location.warehouse`, else `gs://<bucket>` plus
+  `location.path` when set, with the bucket's `{{ env.* }}` templates rendered), not
+  `location.warehouse` as written; a warehouse with another scheme, or none and no bucket,
+  gives none, and `fluid validate` warns.
+- A `platform: confluent` Iceberg expose with no `location.catalog` now reads as Glue, the
+  catalog Tableflow publishes it to, and dbt-snowflake `catalogs.yml` writes
+  `catalog_linked_database_type: glue` for it. `fluid policy compile` grants a Confluent
+  Tableflow expose its S3 bucket and the Glue table Tableflow publishes: the topic's table
+  (`topic`, else `location.table`, else the expose id, as the Tableflow IaC names it) in
+  `location.database`, or, with none, in the Kafka cluster id database Tableflow defaults
+  to. With neither database nor `kafka_cluster_id` it warns and grants no Glue table.
+- A Kafka Connect build whose Iceberg sink config forge-cli derives now writes exactly one
+  Iceberg expose, picked by its `outputs`. `fluid validate` and the run preflight refuse it
+  when the outputs name no Iceberg sink expose or name two or more, and when it declares
+  no outputs and the contract has two or more. Outputs that named no Iceberg sink expose
+  used to draw only the warning "the join is implicit". Split the build into one build per
+  expose, or hand-write `properties.kafka-connect.sink_connector_config`.
+- An embedded Debezium Server build whose Iceberg sink config forge-cli derives now writes
+  the Iceberg sink exposes its `outputs` name, or all of them with no outputs. It is
+  refused when those exposes sit in more than one database, resolve to different catalogs,
+  or when the outputs name none of them.
+- `fluid policy compile` now exits 1 with `policy_compiler_crashed` and writes no bindings
+  file when the policy compiler crashes. It used to write an empty bindings list and exit
+  0, so the next stage read the crash as "no grants to enforce".
+- A Kafka Connect or embedded Debezium build whose sink config forge-cli derives is now
+  refused when its expose names no table namespace (`location.database`, or on GCP
+  `location.dataset`), and a Kafka Connect one when its expose names no `location.table`.
+  The derived config used to name the table `None.<table>`, or leave Debezium Server on its
+  default namespace. An override that sets `iceberg.tables` or `table-namespace` itself is
+  not affected.
 
 ### Changed
 
@@ -57,6 +104,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   it accepted. They ran beside `publish-pypi` and ignored its result, so a refused upload
   (for example "file already exists" on a re-dispatched tag) still cut a release and
   moved the GHCR `latest` tag. The release takes the quality gate's validated tag name.
+- A derived Kafka Connect or embedded Debezium Server Iceberg sink resolves the exposes it
+  writes from the build's `outputs`, through one resolver that `fluid validate`, the run
+  preflight, both runners and the Kafka Connect late-arrival target read. The Debezium
+  Server sink writes every captured table under one `table-namespace` through one catalog,
+  so the exposes must share one `binding.location.database` and one catalog. A Glue
+  warehouse is a per-table prefix: it is not compared, and the sink takes the first
+  expose's. A DynamoDB or JDBC catalog creates every missing table under the one warehouse
+  the sink is given, so warehouses derived from `location.bucket` that differ (each is that
+  expose's own table prefix) are refused; set one `location.warehouse` on those exposes, or
+  split the build. A hand-written config, and an override that sets
+  `iceberg.tables` (Kafka Connect) or `table-namespace` (Debezium Server), is still checked
+  against the first Iceberg sink expose.
+- The Kafka Connect deriver now forwards the catalog's own properties (for BigQuery,
+  `gcp.bigquery.project-id` and `gcp.bigquery.location`) under the `iceberg.catalog.`
+  prefix, as the Debezium Server deriver already did.
 
 ### Fixed
 
@@ -206,6 +268,61 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   a name its SQL cannot quote all passed stage 2 and failed first at
   `fluid generate iac` or `fluid apply`. Both now run the same derivation
   (`lf_governance`) and refuse with the same message.
+- **A second CDC build no longer streams into the first build's table.** A Kafka Connect
+  or embedded Debezium build whose sink config forge-cli derives always wrote the first
+  Iceberg expose, so a build for `refunds` pushed `iceberg.tables=sales.orders` while
+  `fluid validate`, the preflight and the run all passed. It now writes the expose its
+  `outputs` name (see Upgrade notes).
+- **A GCP streaming sink writes into the expose's BigQuery dataset.** The sink named the
+  table's namespace from `location.database` only, but a GCP binding names its dataset in
+  `location.dataset`, so a BigQuery or other GCP Iceberg sink pushed
+  `iceberg.tables=None.<table>` and `fluid validate` passed. The namespace is now
+  `location.database`, else `location.dataset` (the BigQuery metastore catalog maps a
+  namespace to a dataset), and a sink with neither is refused (see Upgrade notes). The
+  Kafka Connect late-events table and the Debezium shared-database check read it the
+  same way.
+- **DynamoDB, JDBC and BigQuery sink configs carry what the catalog needs to start.**
+  `DynamoDbCatalog` and `JdbcCatalog` refuse to initialise without a warehouse, and
+  `BigQueryMetastoreCatalog` without `gcp.bigquery.project-id`; forge-cli derived neither
+  and `fluid validate` asked for neither, so the connector failed at start (see Upgrade
+  notes). A warehouse or project an override supplies counts, and a whitespace-only
+  `location.warehouse` or `location.project` counts as unset.
+- **A bucket written as a `{{ env.* }}` template is read as written.** When its variable is
+  unset or empty where `fluid validate` runs, validate warns and names the variable instead
+  of asking for a bucket the contract already sets. The runner preflight, where the sink
+  config is derived, refuses the build instead of pushing a warehouse in a bucket such as
+  `acme--lake` that the contract does not name; this includes a run from a plan, whose
+  embedded contract is read as written. A Kafka Connect BigQuery sink reads the warehouse
+  only to auto-create tables, so it is refused only with auto-create on; otherwise it
+  warns and pushes no warehouse. An override warehouse is kept, and a hand-written sink
+  config that does not derive is not checked against the bucket's variables.
+- **The BigQuery no-warehouse warning states what fails.** With auto-create on, the Kafka
+  Connect sink calls `createNamespace` for each table it creates, and
+  `BigQueryMetastoreCatalog` refuses that without a warehouse, so auto-create fails even in
+  a dataset that exists. For Debezium Server it says the server does not boot without
+  `debezium.sink.iceberg.warehouse`, which has no default.
+- **`fluid policy compile` grants a Confluent Tableflow table with no catalog.** A
+  `platform: confluent` Iceberg expose with no catalog read as `rest`, so `fluid policy
+  compile` dropped its grants with a false "cataloged in 'rest'" warning (see Upgrade
+  notes).
+- **`fluid policy compile` shows its warnings.** Each compiler warning, for example a grant
+  on an Iceberg table in a catalog it cannot grant on, which compiles to no table grant, is
+  now printed at WARNING level, and the exit code stays 0; the warnings are still written to
+  `bindings.json`. `fluid policy apply` echoes the warnings stored in `bindings.json`. A
+  contract with no grants prints no warning.
+- **A policy compiler crash fails the command.** `fluid policy compile` and the policy step
+  of `fluid generate artifacts` now fail with exit 1 and `policy_compiler_crashed` (see
+  Upgrade notes). The error names the schema type errors at the values the compiler reads
+  (`accessPolicy`, its grants, each grant's permissions, each expose's binding and
+  location) by path and expected type, never by value (a value of the wrong type can be a
+  connection URL with a password in it), or the compiler's own exception when the schema
+  finds none there.
+- **The Snowflake EXTERNAL VOLUME catalog-move guard names both causes.** It blocks the
+  same applies, but its message no longer claims forge-cli stopped
+  creating the volume or that the moved exposes held it. It names both possible causes (an
+  upgrade from a release that gave a non-Snowflake-managed catalog a volume, or a
+  Snowflake-managed expose removed or re-cataloged in this change) and keeps the `tofu
+  state rm` remediation.
 
 ### Documentation
 

fluid_build/_error_catalog.py

@@ -341,6 +341,19 @@ def slug_for(event: str) -> str:
         ["Check the agent-policy block in the contract; run 'fluid policy check <contract>'"],
         "sovereignty",
     ),
+    "policy_compiler_crashed": (
+        [
+            (
+                "Run 'fluid validate <contract>': policy compile reads accessPolicy and "
+                + "exposes without validating them against the contract schema"
+            ),
+            (
+                "If the contract validates, re-run with 'fluid --log-level DEBUG policy compile "
+                + "<contract>' to see the compiler's traceback"
+            ),
+        ],
+        "policy-compile#errors",
+    ),
     "policy_apply_failed": (
         ["Run 'fluid policy check <contract>' to surface the offending rule before apply"],
         "sovereignty",

fluid_build/_errors.py

@@ -83,6 +83,8 @@
     "infra#drift": "advanced/typed-cli-errors.html#governance",
     # the safety gate's own section, which shows its message and remediation
     "iceberg-catalog-move": "cli/apply.html#iceberg-catalog-move-guard",
+    # the command's own Errors section
+    "policy-compile#errors": "cli/policy-compile.html#errors",
 }
 _DOC_FALLBACK = "advanced/production-troubleshooting.html"
 

fluid_build/_mcp_compat.py

@@ -35,9 +35,9 @@
   ``open_inmemory_session``).
 
 This module is the ONLY place allowed to branch on the SDK generation.
-Runtime code keeps the ``mcp>=1.20,<2.0`` install pin (pyproject.toml)
-until the 2.x drift leg has a green track record — the dual support here
-is what turns that leg green in the first place.
+The install pin (pyproject.toml) is ``mcp>=1.22,<3.0``, so both
+generations install — the dual support here is what keeps both drift legs
+green.
 
 Tier-0 leaf rules: no module-scope ``mcp`` import (the startup budget in
 ``tests/perf/test_startup_budget.py`` forbids ``mcp`` on ``fluid --help``);

fluid_build/build_runners/base.py

@@ -203,13 +203,18 @@ def _execute_acquisition_build(
     *,
     dry_run: bool,
     sample_rows: Any = None,
+    as_written: Optional[Dict[str, Any]] = None,
 ) -> int:
     """Dispatch an acquisition build to its runner.
 
     Looks up the engine in ``_ACQUISITION_RUNNER_REGISTRY`` and invokes
     the registered callable. Capability negotiation runs first so a
     contract asking for an unsupported capability fails with the rich
     ``CapabilityMismatchError`` instead of running the wrong shape.
+
+    ``as_written`` is ``contract`` before any ``{{ env.* }}`` resolution, when
+    the caller resolved some already (a plan-embedded contract); ``None``
+    means ``contract`` is as written.
     """
     engine = (build.get("engine") or "").strip().lower()
 
@@ -252,21 +257,32 @@ def _execute_acquisition_build(
     # everything (including secret-shaped vars). Catalog-export paths use
     # ``cli/_common.py::resolve_contract_env_templates`` instead, which
     # leaves sensitive placeholders literal.
+    unresolved_contract = contract if as_written is None else as_written
     build = _resolve_env_placeholders(build)
     contract = _resolve_env_placeholders(contract)
 
     module_path, function_name = entry
+    import contextlib
     import importlib
 
     module = importlib.import_module(module_path, package="fluid_build.build_runners")
     runner_fn = getattr(module, function_name)
-    return runner_fn(
-        build,
-        contract,
-        contract_dir,
-        dry_run=dry_run,
-        sample_rows=sample_rows,
-    )
+    # The Iceberg sink preflight reads a bucket's ``{{ env.X }}`` template as
+    # written: the resolution above turned an unset ``X`` into "", which would
+    # point the sink at a bucket the contract does not name.
+    scope: contextlib.AbstractContextManager[None] = contextlib.nullcontext()
+    if engine in ("kafka-connect", "debezium"):
+        from .kafka_connect.iceberg_sink_validation import contract_as_written
+
+        scope = contract_as_written(unresolved_contract)
+    with scope:
+        return runner_fn(
+            build,
+            contract,
+            contract_dir,
+            dry_run=dry_run,
+            sample_rows=sample_rows,
+        )
 
 
 def _execute_embedded_sql_build_snowflake(
@@ -989,6 +1005,7 @@ def run_builds_from_args(
                     1, "contract_load_failed", {"path": str(contract_path), "error": str(exc)}
                 )
         contract = plan_data.get("contract") or {}
+        contract_as_written = contract
         if not contract:
             raise CLIError(
                 1,
@@ -1061,6 +1078,7 @@ def run_builds_from_args(
             raise
         except Exception as e:
             raise CLIError(1, "contract_load_failed", {"path": str(contract_path), "error": str(e)})
+        contract_as_written = contract
         if str(contract_path).lower().endswith((".tgz", ".tar.gz")):
             # A bundle's own directory is not where its contract lives:
             # anchor at the source contract its MANIFEST records.
@@ -1151,12 +1169,16 @@ def run_builds_from_args(
 
         if is_acquisition_build(build):
             sample_rows = getattr(args, "sample_rows", None)
+            # A plan's contract had its ``{{ env.* }}`` templates resolved
+            # above, with an empty variable rendered to "": the Iceberg sink
+            # preflight reads the bucket template from the plan as written.
             result = _execute_acquisition_build(
                 build,
                 contract,
                 contract_path.parent,
                 dry_run=args.dry_run,
                 sample_rows=sample_rows,
+                as_written=contract_as_written,
             )
             _report_build(report, contract_path.parent, product_id, build_id, result, runs_before)
             if result == 0:

fluid_build/build_runners/debezium/iceberg_sink.py

@@ -92,8 +92,11 @@ def emit_debezium_iceberg_sink_config(
         # GLOBAL (applies to all derived tables), not a per-table default.
         cfg["partition-by"] = ",".join(resolved.partition_by)
 
-    # Native Iceberg props (s3.* / gcs.* / adls.* / jdbc.* ...) pass straight
-    # through with the same prefix the runner adds.
+    # Native Iceberg props (s3.* / gcs.* / adls.* / jdbc.* / gcp.bigquery.* ...)
+    # pass straight through with the same prefix the runner adds: the server
+    # maps every ``debezium.sink.iceberg.*`` key into the catalog properties
+    # (memiiso/debezium-server-iceberg 1.2.0.Final IcebergConfig.java:18-21,
+    # IcebergChangeConsumer.java:100-102).
     for k, v in (resolved.extra_catalog_props or {}).items():
         cfg[str(k)] = str(v)