fix(iceberg): close the gaps the post-merge inspection of #707 found

CHANGELOG.md

@@ -16,6 +16,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   `iceberg_catalog_move_blocked` and prints the `tofu state rm` commands that release
   them, instead of planning to destroy them: destroying a Glue database deletes every
   table in it. The resources stay in AWS; run the commands, then apply again.
+- The same guard covers Snowflake. An Iceberg expose on `platform: snowflake` with
+  `catalog: lakekeeper`, `bigquery`, the `iceberg-rest` spelling, `hive`, `jdbc`,
+  `hadoop` or `dynamodb` got a Snowflake EXTERNAL VOLUME from 0.19.0 and earlier and no
+  longer does (`rest`, `iceberg_rest`, `polaris`, `unity`, `nessie` and `glue` never
+  had one). `fluid apply` stops with `iceberg_catalog_move_blocked` and the `tofu state
+  rm` command for the volume instead of planning to drop it.
+- A GCP Iceberg expose that a streaming sink writes to must now name its catalog in
+  `binding.location.catalog`. Without one, the sink wrote through whatever catalog
+  reached the worker (REST by default, Glue for `sink.catalog: glue` or a hand-written
+  `catalog-impl`) while dbt-bigquery and the GCP IaC created a BigLake table, so one
+  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.
 
 ### Changed
 
@@ -47,6 +60,44 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- **Command output keeps its square brackets.** Rich read text in brackets as console
+  markup and dropped it, so `fluid validate` printed `exposes declares
+  governance.lakeFormation` instead of `exposes[orders] declares ...`, bundle findings
+  lost their `[error]`/`[warning]` severity, and a suggestion such as `pip install
+  'data-product-forge[gcp]'` lost its extra. Validation messages, error context and
+  suggestions now print literally, and a printed `tofu state rm <address>` command stays
+  on one line, so a copied line runs with its address.
+- **The streaming runners check the sink they push.** The Kafka Connect and embedded
+  Debezium Server runners ran the Iceberg sink checks only for a config they derived; a
+  hand-written `sink_connector_config` or `server.sink.config` that `fluid validate`
+  refuses was still deployed. A hand-written config in a build that declares
+  `sink.format: iceberg` is now checked the same way, before anything is created. In a
+  contract with several builds, each runner also reads the build it is running instead
+  of the first one, which the checks already did. A `sink_connector_config`,
+  `server.sink.config` or `iceberg_catalog_overrides` whose `type` or `catalog-impl`
+  selects a different catalog than the expose's is now an error on every platform: the
+  sink would write one catalog while dbt and the IaC read another. Declare the catalog
+  the sink writes to in `binding.location.catalog`; a REST endpoint that fronts Glue
+  (Glue's Iceberg REST endpoint) is `catalog: rest`.
+- **`catalog: bigquery` on a Kafka Connect sink warns.** It sets
+  `iceberg.catalog.type=bigquery`, which the published Apache Iceberg sink (1.9.2) cannot
+  load; Iceberg adds the type in 1.10. The Nessie and BigQuery runtime warnings follow the
+  catalog type that actually reaches the worker, so a hand-written config that sets
+  `type: rest` no longer draws them.
+- **A typo in `location.catalog` is reported once, as a typo.** The native AWS planner now
+  refuses an unknown catalog value as the IaC does (it used to plan the S3 buckets and no
+  Glue table), without first logging it as a `plan_failed` JSON line, and a typo no longer
+  also draws a Lake Formation refusal advising to "govern access in the <typo> catalog".
+- **The Iceberg catalog-move guard no longer blocks a legitimate apply.** It flagged a
+  Glue database that a since-removed parquet expose had created as a catalog move, and
+  nothing could get past it. A database is now flagged only when the moved expose's own
+  Glue table is in state, or when the expose names no table: its database is then all
+  an earlier release created for it, and the guard used to let its destroy through. On
+  Snowflake the guard finds the EXTERNAL VOLUME by its contract-derived name, so it also
+  stops an upgrade that changed `location.warehouse` to a catalog name or dropped
+  `iam_role_arn` in the same edit. When the guard cannot read state or its probe fails,
+  it says so at WARNING level instead of DEBUG.
+
 - **A Kafka Connect Iceberg sink on AWS Glue starts.** The derived connector config set
   both `iceberg.catalog.type` and `iceberg.catalog.catalog-impl`, and Apache Iceberg's
   `CatalogUtil` refuses that ("both type and catalog-impl are set"), so the sink failed

fluid_build/_error_catalog.py

@@ -438,6 +438,21 @@ def slug_for(event: str) -> str:
         ["Pass the product id to roll back: fluid rollback <product-id>"],
         None,
     ),
+    # ── apply safety gates ─────────────────────────────────────────────────
+    "iceberg_catalog_move_blocked": (
+        [
+            (
+                "Run the printed `tofu state rm` commands: they release the resources from "
+                + "this contract's OpenTofu state and touch nothing in the cloud"
+            ),
+            "Then re-run fluid apply",
+            (
+                "The released Glue database/table or Snowflake EXTERNAL VOLUME stays in place; "
+                + "delete it by hand only if nothing else uses it"
+            ),
+        ],
+        "iceberg-catalog-move",
+    ),
 }
 
 

fluid_build/_errors.py

@@ -81,6 +81,8 @@
     "dlq": "advanced/typed-cli-errors.html#pipeline-operations",
     "replay": "advanced/typed-cli-errors.html#pipeline-operations",
     "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",
 }
 _DOC_FALLBACK = "advanced/production-troubleshooting.html"
 

fluid_build/build_runners/debezium/runner.py

@@ -34,7 +34,7 @@
 import time
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Any, ClassVar, Dict, FrozenSet, List, Optional
+from typing import Any, ClassVar, Dict, FrozenSet, List, Mapping, Optional
 
 from fluid_build.api.runner import (
     RunContext,
@@ -300,7 +300,22 @@ def _execute(ctx: RunContext, runner: DebeziumRunner) -> RunResult:
 
     started_at, t_start = begin_acquisition_run(ctx, runner)
 
-    props = ctx.contract.get("builds", [{}])[0].get("properties", {})
+    from ..kafka_connect.iceberg_sink_validation import executing_build
+
+    # The build this run executes, the SAME one the Iceberg preflight checks.
+    # Reading ``builds[0]`` here ran a later build with the first build's
+    # Debezium settings while the preflight vetted the later build's.
+    build = executing_build(ctx.contract, ctx.build_id)
+    if build is None:
+        return _failed(
+            ctx,
+            started_at,
+            t_start,
+            f"debezium: the run context's build {ctx.build_id!r} is not in the "
+            "contract's builds, so its properties cannot be read; run a build the "
+            "contract declares",
+        )
+    props = build.get("properties") or {}
     dbz_props = props.get("debezium", {}) or {}
     deployment = dbz_props.get("deployment", {}) or {}
     mode = deployment.get("mode", "bring-your-own")
@@ -316,7 +331,7 @@ def _execute(ctx: RunContext, runner: DebeziumRunner) -> RunResult:
         )
     if mode == "embedded":
         return _execute_debezium_server(
-            ctx, deployment, dbz_props, snapshot_mode, started_at, t_start
+            ctx, build, deployment, dbz_props, snapshot_mode, started_at, t_start
         )
     return _failed(ctx, started_at, t_start, f"debezium: unknown deployment.mode '{mode}'")
 
@@ -477,6 +492,7 @@ def _escape_properties_value(value: str) -> str:
 
 def _execute_debezium_server(
     ctx: RunContext,
+    build: Mapping[str, Any],
     deployment: Dict[str, Any],
     dbz_props: Dict[str, Any],
     snapshot_mode: str,
@@ -509,28 +525,34 @@ def _execute_debezium_server(
 
     # Iceberg sink: derive the bare-key config from the contract's iceberg expose
     # binding so the embedded server lands in the SAME table the static Glue
-    # table / KC sink resolve to (RFC zero-drift spine). Mirrors the KC gate:
-    # derivation is OFF whenever a hand-written ``config`` block is present, so
-    # those contracts stay byte-for-byte identical (an explicit empty ``config:
-    # {}`` counts as present, matching the KC ``sink_connector_config`` default);
+    # table / KC sink resolve to (RFC zero-drift spine). Gated by
+    # ``iceberg_sink_plan``, the one gate `fluid validate` and the KC runner
+    # read too: derivation is OFF whenever a hand-written ``config`` block is
+    # present, so nothing is derived for it (an explicit empty ``config: {}``
+    # counts as present, matching the KC ``sink_connector_config`` default);
     # opt back in with ``server.sink.iceberg_sink_enabled: true``. Derived keys
     # go UNDER the hand-written ones, so an operator key always wins.
+    from ..kafka_connect.iceberg_sink_validation import (
+        iceberg_sink_plan,
+        iceberg_sink_preflight,
+    )
+
     sink_config = handwritten
-    iceberg_on = sink_block.get("iceberg_sink_enabled", "config" not in sink_block)
-    if sink_type == "iceberg" and iceberg_on:
+    sink_plan = iceberg_sink_plan(build)
+    if sink_plan is not None:
+        # Fail closed BEFORE application.properties is written, for a derived
+        # AND a hand-written Iceberg sink: the same checks `fluid validate`
+        # runs, so a sink it rejects never boots.
+        preflight_error = iceberg_sink_preflight(ctx.contract, ctx.build_id, log=LOG)
+        if preflight_error:
+            return _failed(ctx, started_at, t_start, preflight_error)
+    if sink_plan is not None and sink_plan.derives:
         from ...providers._iceberg_catalog import (
             find_iceberg_expose_binding,
             resolve_iceberg_catalog,
         )
-        from ..kafka_connect.iceberg_sink_validation import iceberg_sink_preflight
         from .iceberg_sink import emit_debezium_iceberg_sink_config
 
-        # Fail closed BEFORE application.properties is written: the same
-        # checks `fluid validate` runs, so a sink it rejects never boots.
-        preflight_error = iceberg_sink_preflight(ctx.contract, ctx.build_id, log=LOG)
-        if preflight_error:
-            return _failed(ctx, started_at, t_start, preflight_error)
-
         binding = find_iceberg_expose_binding(ctx.contract)
         if binding is not None:
             resolved = resolve_iceberg_catalog(