fix(databricks): a schema the contract stops using is destroyed, not left failing the apply

CHANGELOG.md

@@ -135,7 +135,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   `tofu apply`, like the column masks: `CREATE VIEW` when it is missing, `ALTER VIEW ... AS`
   (which keeps its grants) when its definition changed, nothing when it did not, and
   `DROP VIEW`, before the apply, once the contract no longer declares it (in the schemas its
-  tables are in, so another environment of the product keeps its own). `--mode dry-run` prints
+  tables are in, or one the plan destroys, so another environment of the product keeps its
+  own). `--mode dry-run` prints
   every statement. What metric view YAML cannot express is refused at `fluid validate`, not
   approximated: derived metrics, a ratio with a filter, a non-additive dimension that is no time
   dimension, `hour` or `minute` on a `DATE`, colliding names, a name that is no column, and
@@ -285,6 +286,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- **Databricks: a schema the contract stops using is destroyed, not left failing the apply.**
+  When a table moves to another schema, or a schema's last expose is removed, the plan destroys
+  the old schema, and Unity Catalog refuses to delete a schema that still holds anything. The
+  product's metric views and the functions of its column masks and row filters are no OpenTofu
+  resource, so `fluid apply` failed there ("Schema is not empty") until someone dropped them by
+  hand. Now, in each schema the saved plan destroys and the contract no longer names, it drops
+  this product's metric views over the tables the plan destroys before `tofu apply`, and the
+  mask and row filter functions once `tofu apply` has destroyed the tables that used them, then
+  finishes the schema's destroy with a second, destroy-only plan. A function a table still uses
+  is never dropped (Unity Catalog binds a mask by the function's name). Only the deploy
+  identity's objects go, a view only with this product's `fluid.product_id`, and only in schemas
+  this environment's own OpenTofu state holds, so another environment of the product keeps its
+  objects. Anything else the schema holds is named, with what to do. `--mode dry-run` prints the
+  drops and runs none.
+
 - **Quality checks measured a constant on BigQuery (and would have on Databricks).** The
   completeness, uniqueness, accuracy and valid_values checks quoted their column the ANSI way,
   `"col"`, which BigQuery and Spark SQL read as a string literal, and cast ratios to

fluid_build/cli/_apply_opentofu_engine.py

@@ -302,8 +302,10 @@ def _apply_via_opentofu(
 
         _enforce_data_loss_gate(args, logger, provider, changes, data_changes, lost_in_place)
 
+        destroyed = _planned_destroys(plugin, changes, str(workdir), env)
         if dry_run:
             _enforce_access(plugin, "pre_apply", contract, dry_run=True)
+            _prepare_destroys(plugin, contract, destroyed, dry_run=True)
             _converge_output_ports(plugin, contract, dry_run=True)
             cprint("\ndry-run: plan only — not applying.")
             info(logger, "opentofu_apply_dry_run", provider=provider, **changes)
@@ -313,6 +315,7 @@ def _apply_via_opentofu(
         # grant the apply adds lands on a protected table. A failure stops the run here:
         # nothing has been applied.
         _enforce_access(plugin, "pre_apply", contract)
+        _prepare_destroys(plugin, contract, destroyed)
         imported.keep()
 
     # ``fluid apply`` is the headline state-changing command, so a receiver
@@ -322,7 +325,7 @@ def _apply_via_opentofu(
     # (``resolve_lineage_emitter`` returns the null emitter), and every
     # emit call soft-fails — lineage never breaks an apply.
     with _apply_lineage(contract, provider=provider, planned=changes) as record:
-        apply_result = runner.tofu_apply(str(workdir), env=env)
+        apply_result, waited = _apply_saved_plan(plugin, contract, destroyed, str(workdir), env)
         if not apply_result.ok:
             # A table the failed apply created, its grants and the job that fills it can all
             # exist with no mask or filter: bind what exists (never drop) before failing.
@@ -332,7 +335,7 @@ def _apply_via_opentofu(
                 _tail(apply_result.stderr or apply_result.stdout),
             )
             raise _apply_failure(apply_result, str(workdir), plugin)
-        applied = runner.change_summary(apply_result)
+        applied = _applied(changes, apply_result, waited)
         record(applied)
 
     cprint(f"\n  tofu apply complete: +{applied['add']} ~{applied['change']} -{applied['remove']}")
@@ -376,19 +379,25 @@ def _converge_output_ports(
     and the run exits non-zero; the tables the apply made stay, and the next apply converges
     the space. With ``dry_run`` the plugin prints what it would do and writes nothing.
     """
-    run = getattr(plugin, "converge_output_ports", None)
+    # Its lines quote titles (anyone's, from the Genie list) and the ``[fluid-genie:...]``
+    # marker, which Rich would read as markup.
+    _run_hook(
+        plugin, "converge_output_ports", "opentofu_output_port_failed", contract, dry_run=dry_run
+    )
+
+
+def _run_hook(plugin: Any, hook: str, event: str, *args: Any, **kwargs: Any) -> None:
+    """Run the plugin's optional ``hook``, if it has one, with what it prints going out
+    literally: it quotes the contract and the server (statements, names, a server's error text
+    with its ``[ERROR_CLASS]``), data, not markup, so a "[word]" in it neither vanishes nor
+    raises. A typed refusal is the run's error (:func:`_binding_error`, under ``event``)."""
+    run = getattr(plugin, hook, None)
     if not callable(run):
         return
     try:
-        # Literally: its lines quote titles (anyone's, from the Genie list) and the
-        # ``[fluid-genie:...]`` marker, which Rich would read as markup.
-        run(contract, say=cprint_literal, dry_run=dry_run)
+        run(*args, say=cprint_literal, **kwargs)
     except UnsupportedBindingError as exc:
-        raise CLIError(
-            1,
-            "opentofu_output_port_failed",
-            {"kind": exc.kind, "error": str(exc), "remediation": list(exc.remediation)},
-        ) from exc
+        raise _binding_error(exc, event) from exc
 
 
 def _enforce_access(
@@ -403,22 +412,140 @@ def _enforce_access(
     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
-    # What the hooks print quotes the contract and the server (statements, descriptions, errors):
-    # data, not markup, so a "[word]" in it neither vanishes nor raises.
+    # ``post_apply`` takes no ``dry_run``: it never runs in one.
+    flags = {"dry_run": True} if dry_run else {}
+    _run_hook(plugin, hook, "opentofu_access_enforcement_failed", contract, **flags)
+
+
+def _planned_destroys(
+    plugin: Any, changes: Mapping[str, int], workdir: str, env: Mapping[str, str]
+) -> List[Dict[str, Any]]:
+    """What the saved plan only deletes (``runner.planned_deletes``), for the plugin's
+    ``prepare_destroys`` and ``unblock_destroys``. ``[]`` for a plugin without
+    ``prepare_destroys`` or a plan that removes nothing; a plan that cannot be read back is
+    said, and nothing is cleared before the apply."""
+    if not callable(getattr(plugin, "prepare_destroys", None)) or not changes.get("remove"):
+        return []
+    shown = runner.tofu_show_plan(workdir, plan_file="tfplan", env=env)
+    if shown is None:
+        cprint(
+            "  note: the saved plan could not be read back, so what the provider plugin made "
+            "outside OpenTofu in the resources it destroys is not cleared first."
+        )
+        return []
+    return runner.planned_deletes(shown)
+
+
+def _prepare_destroys(
+    plugin: Any,
+    contract: Mapping[str, Any],
+    destroyed: List[Dict[str, Any]],
+    *,
+    dry_run: bool = False,
+) -> None:
+    """Run the plugin's ``prepare_destroys`` hook, if it has one, before ``tofu apply``.
+
+    Optional plugin capability, discovered like ``pre_apply``: the Databricks plugin drops what
+    it made outside OpenTofu (metric views, column mask and row filter functions) in the
+    schemas the plan destroys, which Unity Catalog would otherwise refuse to delete
+    (``iac/providers/databricks_vacated.py``). It gets the plan's deletes
+    (:func:`_planned_destroys`). With ``dry_run`` it prints the statements and runs none. What
+    it prints quotes live objects, so it goes out literally; a refusal stops the run before the
+    apply.
+    """
+    if destroyed:
+        _run_hook(
+            plugin,
+            "prepare_destroys",
+            "opentofu_destroy_preparation_failed",
+            contract,
+            destroyed,
+            dry_run=dry_run,
+        )
+
+
+def _apply_saved_plan(
+    plugin: Any,
+    contract: Mapping[str, Any],
+    destroyed: List[Dict[str, Any]],
+    workdir: str,
+    env: Mapping[str, str],
+) -> Tuple[Any, int]:
+    """``tofu apply`` of the saved plan, with the destroys it could not make yet resumed
+    (:func:`_resume_blocked_destroys`): the result that stands, and how many destroys waited
+    for a second apply (0 when none did)."""
+    result = runner.tofu_apply(workdir, env=env)
+    if result.ok:
+        return result, 0
+    return _resume_blocked_destroys(plugin, contract, result, destroyed, workdir, env) or (
+        result,
+        0,
+    )
+
+
+def _applied(changes: Mapping[str, int], result: Any, waited: int) -> Dict[str, int]:
+    """What the apply made: its summary, or after a resume the plan's own changes (the first
+    apply made all but the destroys that waited, the second plan held nothing else), with the
+    second apply's destroys in place of those that waited."""
+    applied = runner.change_summary(result)
+    if not waited:
+        return applied
+    return {**changes, "remove": int(changes["remove"]) - waited + applied["remove"]}
+
+
+def _resume_blocked_destroys(
+    plugin: Any,
+    contract: Mapping[str, Any],
+    result: Any,
+    destroyed: List[Dict[str, Any]],
+    workdir: str,
+    env: Mapping[str, str],
+) -> Optional[Tuple[Any, int]]:
+    """Finish the destroys a ``tofu apply`` could not make yet: ``(the second apply's result,
+    how many waited)``, or ``None`` and the apply's own failure stands.
+
+    OpenTofu destroys a resource and then what holds it, inside one apply. Unity Catalog keeps
+    the functions of a table's column masks and row filter until the table is gone, and refuses
+    to delete a schema that holds them, so a schema the contract stopped using fails to destroy
+    in the very apply that destroyed its tables. Optional plugin capability
+    ``unblock_destroys(contract, blocked, say)``: the addresses of the destroys it cleared (the
+    Databricks plugin drops those functions once no table uses them, and says what it dropped,
+    ``iac/providers/databricks_vacated.py``). Fails closed: it resumes only when every
+    resource the apply failed on is a delete of the saved plan that the plugin cleared, no
+    error is left unaccounted for, and the plan made after it holds nothing but some of those
+    deletes, which the data-loss gate already passed.
+    """
+    unblock = getattr(plugin, "unblock_destroys", None)
+    errored = runner.errored_resources(result)
+    planned = {entry["address"]: entry for entry in destroyed}
+    if not callable(unblock) or not errored or len(_diagnostics(result.events)) != len(errored):
+        return None  # nothing to resume, or an error no failed resource accounts for
+    if any(action != "delete" or address not in planned for address, action in errored):
+        return None
+    waited = sorted({address for address, _action in errored})
     try:
-        if dry_run:
-            run(contract, say=cprint_literal, dry_run=True)
-        else:
-            run(contract, say=cprint_literal)
-    except UnsupportedBindingError as exc:
-        raise CLIError(
-            1,
-            getattr(exc, "event", None) or "opentofu_access_enforcement_failed",
-            {"kind": exc.kind, "error": str(exc), "remediation": list(exc.remediation)},
-        ) from exc
+        blocked = [planned[address] for address in waited]
+        cleared = set(unblock(contract, blocked, say=cprint_literal) or ())
+    except Exception as exc:  # noqa: BLE001 — the apply's own failure is what the run raises
+        cprint_literal(
+            f"  what holds {', '.join(waited)} could not be cleared: {type(exc).__name__}"
+        )
+        return None
+    if cleared != set(waited):
+        return None  # a failure the plugin did not clear: the apply's own error stands
+    plan = runner.tofu_plan(workdir, env=env)
+    removals = {address for address, _kind in runner.planned_removals(plan)}
+    only_those = {"add": 0, "change": 0, "remove": len(removals)}
+    if not (plan.ok and removals and removals <= set(waited)) or (
+        runner.change_summary(plan) != only_those
+    ):
+        cprint_literal(
+            "  not destroyed again: the plan made after the apply holds more than "
+            + ", ".join(waited)
+        )
+        return None
+    cprint(f"  tofu plan (the destroys that waited): +0 ~0 -{len(removals)}")
+    return runner.tofu_apply(workdir, env=env), len(waited)
 
 
 def build_after_apply(
@@ -740,10 +867,12 @@ def emit_module(
     return module, actions
 
 
-def _binding_error(exc: UnsupportedBindingError) -> CLIError:
+def _binding_error(exc: UnsupportedBindingError, event: str = "unsupported_binding") -> CLIError:
+    """A typed refusal as the run's error, under ``event``, or the refusal's own when its class
+    names one (a Genie space's ``opentofu_output_port_failed``)."""
     return CLIError(
         1,
-        "unsupported_binding",
+        getattr(exc, "event", None) or event,
         {"kind": exc.kind, "error": str(exc), "remediation": list(exc.remediation)},
     )
 

fluid_build/iac/base.py

@@ -77,6 +77,14 @@ class IacProviderPlugin(Protocol):
       the current state before the plan.
     * ``catalog_move_spec()``: the ``CatalogMoveSpec`` of an out-of-tree cloud
       whose emitter moved its catalog resources (``iac/catalog_moves.py``).
+    * ``prepare_destroys(contract, destroyed, say, dry_run)``: before ``tofu
+      apply``, clears what the plugin made outside OpenTofu in the resources the
+      saved plan only deletes (``destroyed``: ``{address, type, before}`` each),
+      so the cloud does not refuse their destroy.
+    * ``unblock_destroys(contract, blocked, say)``: after an apply that failed
+      at destroying ``blocked`` (some of those), clears what the apply's own
+      destroys freed, and returns the addresses it cleared; the engine plans and
+      applies those destroys again only when they are every resource that failed.
     """
 
     #: Short cloud name, e.g. "gcp" / "aws" / "snowflake".

fluid_build/iac/providers/databricks.py

@@ -106,6 +106,13 @@
   values pick the tables and are not thresholds the monitor checks. Its results are the
   Quality indicator on the table in Catalog Explorer and
   ``system.data_quality_monitoring.table_results``; removing it is not gated.
+* A schema the contract stops using (its table moved, its last expose removed) is destroyed by
+  the plan. The metric views and the mask and row filter functions this product made there are
+  no OpenTofu resource, and Unity Catalog does not delete a schema that holds them, so
+  :meth:`DatabricksIacPlugin.prepare_destroys` drops them before ``tofu apply`` and
+  :meth:`~DatabricksIacPlugin.unblock_destroys` drops the functions the apply's destroyed tables
+  used, once they are gone, before the engine resumes that schema's destroy
+  (``databricks_vacated.py``).
 * A dbt build that runs on Databricks (``execution.runtime.platform:
   databricks``) becomes a serverless ``databricks_job`` with a ``dbt_task``, and
   a local project its ``databricks_workspace_file`` uploads
@@ -205,6 +212,7 @@
 from ..provider_match import canonical_cloud, is_cloud
 from ..versions import required_providers
 from . import databricks_metrics as _metric_views
+from . import databricks_vacated as _vacated
 from .databricks_access import TableEnforcement, compile_enforcement
 from .databricks_auth import tofu_env_overlay
 from .databricks_enforce import (
@@ -2537,6 +2545,31 @@ def post_apply(
             _declare_keys(contract, say, client_factory)
             _metric_views.post_apply(views, say, client_factory)
 
+    def prepare_destroys(
+        self,
+        contract: Mapping[str, Any],
+        destroyed: Iterable[Mapping[str, Any]],
+        say: Optional[Callable[[str], None]] = None,
+        dry_run: bool = False,
+        client_factory: Optional[Callable[[], Any]] = None,
+    ) -> None:
+        """Before ``tofu apply``: clear what this product made outside OpenTofu in the schemas
+        the plan destroys (``databricks_vacated.prepare``); ``dry_run`` prints, runs none."""
+        _vacated.prepare(
+            _vacated_of(contract, destroyed), say or print, client_factory, dry_run=dry_run
+        )
+
+    def unblock_destroys(
+        self,
+        contract: Mapping[str, Any],
+        blocked: Iterable[Mapping[str, Any]],
+        say: Optional[Callable[[str], None]] = None,
+        client_factory: Optional[Callable[[], Any]] = None,
+    ) -> List[str]:
+        """After an apply that failed at destroying ``blocked``: drop what its destroyed tables
+        used there (``databricks_vacated.unblock``); the addresses of those it cleared."""
+        return list(_vacated.unblock(_vacated_of(contract, blocked), say or print, client_factory))
+
     def post_build(
         self,
         contract: Mapping[str, Any],
@@ -2948,6 +2981,19 @@ def validate_databricks_binding(
     return errors, warnings
 
 
+def _vacated_of(
+    contract: Mapping[str, Any], destroyed: Iterable[Mapping[str, Any]]
+) -> _vacated.Vacated:
+    """The schemas among a plan's deletes that ``contract`` no longer uses
+    (``databricks_vacated.vacated``); a contract that does not compile raises its refusal."""
+    compiled = _compile(contract, os.environ)
+    if compiled.refusals:
+        raise compiled.refusals[0]
+    declared = {f"{table.catalog}.{table.schema}" for table in compiled.tables}
+    # From the contract, not the compiled views: one with no Databricks expose left compiles none.
+    return _vacated.vacated(destroyed, declared, _metric_views.product_id_of(contract))
+
+
 def _genie(contract: Mapping[str, Any], compiled: _Compiled, env: Mapping[str, str]) -> Any:
     """The Genie space ``contract`` asks for, from this compile's tables and grants
     (``databricks_genie.compile_space``); imported here, the Light CLI's ``--help`` never

fluid_build/iac/providers/databricks_enforce.py

@@ -17,7 +17,10 @@
   function carries this module's name prefix. A filter an admin set by hand is left alone.
   Dropping widens access, so it waits for the apply to succeed and is printed as such. After an
   apply that succeeded it also drops this module's superseded functions that no table in the
-  schema uses and this run's identity owns (:func:`drop_orphan_functions`).
+  schema uses and this run's identity owns (:func:`drop_orphan_functions`). Those are the
+  contract's schemas; in a schema the contract stops using, which the plan destroys, the
+  functions go by the same rules, once OpenTofu has destroyed the tables that used them
+  (``databricks_vacated.py``).
 
 Both read the live bindings first (``GET /api/2.1/unity-catalog/tables/{name}``:
 ``row_filter.function_name`` / ``input_column_names`` and ``columns[].mask.function_name``,