feat(databricks): Unity Catalog primary and foreign keys from the contract

CHANGELOG.md

@@ -7,6 +7,29 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- **Databricks: Unity Catalog primary and foreign keys from the contract.** A column tagged
+  `primary-key` (all of them, in schema order, for a composite key; each must be
+  `required: true`) becomes the table's PRIMARY KEY, and one tagged `foreign-key` with
+  `labels.references: <product id>.<column>` becomes a FOREIGN KEY to that product's table,
+  so Catalog Explorer draws the entity-relationship diagram and Genie can join the products.
+  The parent may live in another contract: it is found by the `fluid_product_id` tag every
+  forge table carries (`system.information_schema.table_tags`), wherever an overlay put it.
+  `fluid apply` declares the keys on the SQL warehouse after `tofu apply` (the provider has no
+  constraint argument), re-declares any that are missing, replaces or drops the ones it
+  declared (named `fluid_pk_<table>` / `fluid_fk_<table>__<columns>`) when the contract changes,
+  and leaves any other key in place. A key that cannot be declared yet (its parent not
+  deployed, say) is skipped with the reason and the fix, never failing the run;
+  `--mode dry-run` prints the statements. Because a dbt `table` build's `CREATE OR REPLACE`
+  drops a table's keys and every foreign key that references it (measured on a warehouse),
+  each Databricks dbt job that writes a keyed table gets a second task, `fluid_keys`, that
+  declares them again after dbt, with the contract's own foreign keys into those tables, in a
+  script that passes over a key it cannot add instead of failing the run. Such a key (a key
+  column the dbt model leaves nullable, say) is reported by the next `fluid apply`, and the
+  script needs a SQL warehouse that runs SQL scripting. A dbt build that
+  `--mode amend-and-build` runs locally is followed by the same re-declaration.
+
 ### Fixed
 
 - **`fluid ai models --provider` works, and takes the names `fluid forge --llm-provider`

fluid_build/cli/_apply_opentofu_engine.py

@@ -345,16 +345,17 @@ def _enforce_access(
     binds Unity Catalog column masks and row filters, which the provider cannot manage
     (``iac/providers/databricks_enforce.py``). Unlike the advisory hooks this one gates: a
     refusal or failure is a typed error and the run exits non-zero. With ``dry_run`` the
-    plugin prints the statements and runs none.
+    plugin prints the statements and runs none. What a hook prints is data (table names,
+    statements, a server's error text with its ``[ERROR_CLASS]``), so it goes out literally.
     """
     run = getattr(plugin, hook, None)
     if not callable(run):
         return
     try:
         if dry_run:
-            run(contract, say=cprint, dry_run=True)
+            run(contract, say=cprint_literal, dry_run=True)
         else:
-            run(contract, say=cprint)
+            run(contract, say=cprint_literal)
     except UnsupportedBindingError as exc:
         raise CLIError(
             1,
@@ -363,6 +364,47 @@ def _enforce_access(
         ) from exc
 
 
+def build_after_apply(
+    args, logger: logging.Logger, contract: Mapping[str, Any], build_kwargs: Mapping[str, Any]
+) -> int:
+    """The build phase of a build-augmented mode, after a ``tofu apply`` that succeeded, then
+    the plugin's ``post_build`` hook (:func:`after_builds`), whatever the builds returned: one
+    that failed part way may already have rebuilt a table. A dry run changes nothing, so it
+    runs no hook (its plan already printed what the apply would declare)."""
+    from fluid_build.build_runners import run_builds_from_args
+
+    rc = run_builds_from_args(args, logger, force_run=True, **build_kwargs)
+    if not getattr(args, "dry_run", False):
+        after_builds(args, logger, contract)
+    return rc
+
+
+def after_builds(args, logger: logging.Logger, contract: Mapping[str, Any]) -> None:
+    """The plugin's ``post_build`` hook, if it has one, after the build phase.
+
+    Optional plugin capability, discovered like ``plan_notices``, and advisory: a dbt build run
+    here may have rebuilt a table with CREATE OR REPLACE, which drops what the apply declared on
+    it that dbt does not (the Databricks plugin's Unity Catalog keys), and the plugin declares
+    it again. It gets the contract as the apply did, its ``{{ env.* }}`` templates resolved
+    (:func:`_load_contract`), and never changes the run's exit code.
+    """
+    try:
+        resolved = _load_contract(args, logger, contract)
+        plugin = get_iac_plugin(
+            _resolve_provider(resolved, getattr(args, "provider", None) or "auto")
+        )
+    except Exception:  # noqa: BLE001 — the apply resolved it already; nothing to add here
+        return
+    run = getattr(plugin, "post_build", None)
+    if not callable(run):
+        return
+    try:
+        with contract_source(source_contract_file(getattr(args, "contract", None))):
+            run(resolved, say=cprint_literal)
+    except Exception as exc:  # noqa: BLE001 — advisory: the build's own result stands
+        cprint_literal(f"  after the build: {type(exc).__name__}: {exc}")
+
+
 def _enforce_after_failed_apply(plugin: Any, contract: Mapping[str, Any]) -> None:
     """Best effort, bind-only ``post_apply`` for a ``tofu apply`` that failed part way.
 
@@ -378,7 +420,7 @@ def _enforce_after_failed_apply(plugin: Any, contract: Mapping[str, Any]) -> Non
         "exist before the run stops (no mask or filter is dropped)."
     )
     try:
-        run(contract, say=cprint, after_failed_apply=True)
+        run(contract, say=cprint_literal, after_failed_apply=True)
     except UnsupportedBindingError as exc:
         cprint_literal(
             f"  access enforcement after the failed apply: {exc} {' '.join(exc.remediation)}"

fluid_build/cli/apply.py

@@ -2086,9 +2086,9 @@ def run(args, logger: logging.Logger) -> int:
         # not silently dropped on the cloud-provider path.
         if rc == 0 and needs_build(resolved_mode):
             args.build_id = resolved_build_id
-            from fluid_build.build_runners import run_builds_from_args
+            from fluid_build.cli._apply_opentofu_engine import build_after_apply
 
-            return run_builds_from_args(args, logger, force_run=True, **build_kwargs)
+            return build_after_apply(args, logger, source_contract, build_kwargs)
         return rc
 
     try:

fluid_build/iac/naming.py

@@ -48,6 +48,14 @@ def digest_ident(value: str) -> str:
     return f"{safe_ident(value)}_{_digest(value, 32)}"
 
 
+def bounded_ident(text: str, limit: int, sep: str = "_") -> str:
+    """``text``, or, when it is longer than ``limit``, its first characters, ``sep`` and an
+    8-digit digest of the whole: a name a service caps, whose cut forms stay distinct."""
+    if len(text) <= limit:
+        return text
+    return f"{text[: limit - 9]}{sep}{_digest(text, 8)}"
+
+
 def _digest(value: str, length: int) -> str:
     return hashlib.sha256(value.encode("utf-8")).hexdigest()[:length]