feat(databricks): a dbt job runs its contract's DQ rules as dbt tests

AGENTS.md

@@ -55,6 +55,8 @@ The `replace` rows describe the native engine. On the OpenTofu engine (aws, gcp,
 
 On Databricks a dbt build with `execution.runtime.platform: databricks` runs as the job `fluid apply` provisions. `amend-and-build` and `replace-and-build` start a run of the job this apply's OpenTofu state holds (found by its `fluid_product` / `fluid_build` tags), wait for it, and fail with the task's error and the end of dbt's output; `replace-and-build` adds `--full-refresh` for that run only. The job's settings, schedule and pause state are never written; `amend` and `create-only` only provision it, and a continuous job is never started. The wait is capped by `FLUID_DATABRICKS_RUN_TIMEOUT_SECONDS` (an hour, or the job's own limits plus 15 minutes); at the cap, on Ctrl-C and on SIGTERM or SIGHUP (unless ignored, as under `nohup`) the run is cancelled, while a run that cannot be read five times in a row is left alone. To build an estate in lineage order, provision every product in parallel with `--mode amend`, then run `--mode amend-and-build` level by level; the second apply plans no change. Code: `build_runners/_databricks_job_run.py`.
 
+The job also runs the contract's DQ rules (`exposes[].contract.dq.rules` of each expose the build writes) as dbt tests: one singular test per check, `<test path>/fluid_dq/<model>/<rule id>.sql`, uploaded beside the project, so `dbt build` runs each after its model and a failing `error`/`critical` rule fails the run; the run report prints each rule's result. `util/dq_checks.py` reads the rules once for every engine (`read_rules` → `DqCheck` / `DqRefusal`), a `valid_values` rule's allowed values through `quality_engine.value_domain` as `fluid test` reads them; `iac/providers/databricks_dbt_tests.py` renders them. A rule no test can run, a Git source with rules, a missing model or source, a test name the project uses, and a selection that skips the model are refused at validate and plan.
+
 ### Plan-Binding (Terraform-Style "Apply Consumes Exact Plan")
 
 Stage 6 `fluid plan` emits two cryptographic fields in `plan.json`:

CHANGELOG.md

@@ -251,6 +251,38 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   strict bound stays unmapped there. On the tower-management industry pack `fluid test` ran
   38 of its 168 rules before; it now runs every rule of every deployed product.
 
+- **A Databricks dbt job runs its contract's data-quality rules as dbt tests.** Each rule in
+  `exposes[].contract.dq.rules` of an expose a Databricks dbt build writes becomes a singular
+  dbt test, `tests/fluid_dq/<model>/<rule id>.sql`, uploaded beside the project (your own files
+  upload byte for byte). The job's `dbt build` runs each test right after its model, so the run
+  page lists every rule by its id, passed, warned or failed; a rule of severity `critical` or
+  `error` that fails skips the models below it and fails the run, `warn` and `info` are reported
+  and do not. uniqueness runs dbt's `unique`, completeness `not_null` on each selected column (a
+  comma-separated selector is one test per column). valid_values allows what `fluid test` allows
+  (`quality_engine.value_domain`): the values the rule's description lists (`<column> valid
+  values: A, B.`), as dbt's `accepted_values`, or else every declaration of the column, which
+  must all hold: `accepted_values` for its `enum` validationRule (`A | B` or `A,B`), a regex
+  test for its `regex`, a range test for its `range` (`>= 0`, `>= 0 and <= 100`, `between 0 and
+  100`), and dbt's `relationships` for its `labels.references: <product id>.<column>` against
+  the dbt source the project declares for that consumed expose; accuracy is a bound on every
+  value, freshness the newest value's age against `window`, schema each contract column with its
+  declared type. A threshold below 1.0 is the share of rows that must pass: the non-null share
+  for completeness and the distinct share for uniqueness, as `fluid test` computes them, and the
+  share of non-null values that are allowed for valid_values. No dbt package is installed: the
+  regex, range, freshness, schema and share tests are forge-cli's own SQL. Under `--mode
+  amend-and-build`, `fluid apply` prints what dbt found for each rule after the run. An
+  `anomaly_detection` rule is left to Databricks data quality monitoring, which learns its
+  bounds (it stays refused where the expose turns monitoring off). The `fluid_keys` task that
+  declares the table's Unity Catalog keys again now runs once dbt has ended, whatever its
+  outcome: a failing rule fails dbt after its model replaced the table, and the table keeps its
+  keys. Refused at `fluid validate` and `fluid plan`, with the fix: a rule no test can run as
+  written (drift detection, valid_values with no allowed values, an unknown column, an
+  unparseable window, a threshold no number holds, a range no value meets), a Git-sourced build
+  with rules (no file can be added to a checkout), a model or a referenced source the project
+  does not hold, a build whose dbt selection leaves the model out, and a rule id whose test name
+  the project's own tests already use. `fluid_build/util/dq_checks.py` reads the rules once, for
+  the engines that run them; this job is its first.
+
 ### Fixed
 
 - **Quality checks measured a constant on BigQuery (and would have on Databricks).** The

fluid_build/build_runners/_databricks_job_run.py

@@ -281,9 +281,23 @@ def run_job_build(
         env=os.environ if env is None else env,
         sleep=sleep,
         clock=clock,
+        dq_checks=_dq_checks(contract, build),
     )
 
 
+def _dq_checks(contract: Mapping[str, Any], build: Mapping[str, Any]) -> Tuple[Any, ...]:
+    """The contract rules the job runs as dbt tests (``iac/providers/databricks_dbt_tests.py``),
+    which the report names one by one; none when they cannot be read, since the report is a
+    courtesy and the apply already refused a rule it could not run."""
+    try:
+        from fluid_build.iac.providers.databricks_dbt_tests import expected_tests
+
+        return tuple(expected_tests(contract, build))
+    except Exception:  # noqa: BLE001 — never fail a build over its report
+        LOG.debug("databricks_job_run_dq_checks_unread", exc_info=True)
+        return ()
+
+
 def start_and_wait(
     client: Any,
     job: JobRef,
@@ -295,8 +309,12 @@ def start_and_wait(
     say: Optional[Say] = None,
     sleep: Callable[[float], None] = time.sleep,
     clock: Callable[[], float] = time.monotonic,
+    dq_checks: Sequence[Any] = (),
 ) -> int:
-    """Run ``job`` now through ``client`` and wait for the run to end. Returns 0 or 1."""
+    """Run ``job`` now through ``client`` and wait for the run to end. Returns 0 or 1.
+
+    ``dq_checks`` are the contract rules its dbt task runs as tests: the report says what
+    dbt found for each (:func:`fluid_build.iac.providers.databricks_dbt_tests.report`)."""
     say = say or _say
     api = client.api_client
     label = f"Databricks job '{name}' ({job.job_id})"
@@ -343,7 +361,9 @@ def start_and_wait(
     if ended is None:
         return 1
     # The run has ended: an interrupt from here on stops the report, never a run.
-    return _report_end(api, ended.run, run_id, build_id, ended.elapsed, ended.url, say)
+    return _report_end(
+        api, ended.run, run_id, build_id, ended.elapsed, ended.url, say, dq_checks=dq_checks
+    )
 
 
 @dataclass(frozen=True)
@@ -506,24 +526,40 @@ def _report_end(
     elapsed: float,
     url: str,
     say: Say,
+    *,
+    dq_checks: Sequence[Any] = (),
 ) -> int:
+    from fluid_build.iac.providers.databricks_jobs import DBT_TASK_KEY
+
     code, message = _outcome(run)
     say(f"   {_clock(elapsed)} TERMINATED {code}")
     LOG.debug("databricks_job_run_ended run_id=%s result=%s", run_id, code)
     if code == "SUCCESS":
         for task_key, task in _latest_attempts(run):
-            summary = _dbt_summary(api, task["run_id"])
+            output, _unread = _task_output(api, task["run_id"])
+            summary = _dbt_summary(output)
             if summary:
                 say(f"   task '{task_key}': {summary}")
+            if task_key == DBT_TASK_KEY:
+                for line in _rule_lines(output, dq_checks):
+                    say(line)
         say(f"✅ Build '{build_id}' - Databricks run {run_id} succeeded in {_duration(elapsed)}")
         return 0
     console_error(
         f"Build '{build_id}' - Databricks run {run_id} ended {code} after {_duration(elapsed)}"
         + (f": {message}" if message else "")
     )
     for task_key, task in _latest_attempts(run):
-        if _outcome(task)[0] != "SUCCESS":
-            _report_task(api, task_key, task["run_id"], say)
+        failed = _outcome(task)[0] != "SUCCESS"
+        if not failed and task_key != DBT_TASK_KEY:
+            continue
+        output, unread = _task_output(api, task["run_id"])
+        rules = _rule_lines(output, dq_checks) if task_key == DBT_TASK_KEY else []
+        if failed:
+            _report_task(task_key, output, unread, say, rules)
+        else:  # dbt passed and a later task failed: its tests still ran
+            for line in rules:
+                say(line)
     if url:
         say(f"   run page: {url}")
     return 1
@@ -558,26 +594,47 @@ def _queued_behind(api: Any, job_id: int, run_id: int) -> str:
     )
 
 
-def _dbt_summary(api: Any, task_run: Any) -> str:
-    """dbt's own last ``Done. PASS=...`` line of a task's output, or ``""``: what a run that
-    succeeded built, at the cost of one read that may fail without failing the build."""
+def _task_output(api: Any, task_run: Any) -> Tuple[Mapping[str, Any], Optional[BaseException]]:
+    """A task's ``runs/get-output``, read once for every line about it: ``({}, error)`` when
+    it cannot be read."""
     try:
-        output = api.do("GET", f"{_JOBS}/runs/get-output", query={"run_id": task_run}) or {}
-    except Exception:  # noqa: BLE001 — the run succeeded; its summary is a courtesy
-        return ""
+        return api.do("GET", f"{_JOBS}/runs/get-output", query={"run_id": task_run}) or {}, None
+    except Exception as exc:  # noqa: BLE001 — what is said about a task never fails the build
+        return {}, exc
+
+
+def _dbt_summary(output: Mapping[str, Any]) -> str:
+    """dbt's own last ``Done. PASS=...`` line of a task's output, or ``""``: what a run that
+    succeeded built."""
     done = [line for line in clean(output.get("logs") or "").splitlines() if "Done. PASS=" in line]
     return " ".join(_DBT_TIMESTAMP_RE.sub("", done[-1]).split()) if done else ""
 
 
-def _report_task(api: Any, task_key: str, task_run: Any, say: Say) -> None:
-    try:
-        output = api.do("GET", f"{_JOBS}/runs/get-output", query={"run_id": task_run}) or {}
-    except Exception as exc:  # noqa: BLE001
-        say(f"   task '{task_key}': its output could not be read ({_failure(exc)})")
+def _rule_lines(output: Mapping[str, Any], dq_checks: Sequence[Any]) -> List[str]:
+    """What dbt found for each contract rule the dbt task ran as a test."""
+    if not dq_checks:
+        return []
+    from fluid_build.iac.providers.databricks_dbt_tests import report
+
+    logs = clean(output.get("logs") or "")
+    return report(logs, dq_checks, truncated=bool(output.get("logs_truncated")))
+
+
+def _report_task(
+    task_key: str,
+    output: Mapping[str, Any],
+    unread: Optional[BaseException],
+    say: Say,
+    rules: Sequence[str] = (),
+) -> None:
+    if unread is not None:
+        say(f"   task '{task_key}': its output could not be read ({_failure(unread)})")
         return
     # One line: Databricks quotes the failed shell command with its newlines escaped.
     error = " ".join(clean(output.get("error") or "").replace("\\n", " ").split())
     say(f"   task '{task_key}'" + (f": {error}" if error else " failed"))
+    for line in rules:
+        say(line)
     logs = clean(output.get("logs") or output.get("error_trace") or "")
     lines = [line.rstrip() for line in logs.splitlines() if line.strip()]
     if not lines:

fluid_build/iac/providers/databricks.py

@@ -2807,16 +2807,18 @@ def _job_actions(jobs: Sequence[DbtJob]) -> List[Dict[str, Any]]:
     """Plan actions for the dbt jobs and the project files they upload."""
     actions: List[Dict[str, Any]] = []
     for job in jobs:
-        actions.append(
-            {
-                "op": "databricks.job.ensure",
-                "resource_type": "databricks_job",
-                "resource_id": job.name,
-                "build": job.build_id,
-                "source": job.source,
-                "trigger": job.trigger,
-            }
-        )
+        action: Dict[str, Any] = {
+            "op": "databricks.job.ensure",
+            "resource_type": "databricks_job",
+            "resource_id": job.name,
+            "build": job.build_id,
+            "source": job.source,
+            "trigger": job.trigger,
+        }
+        if job.tests:
+            # The contract's DQ rules, each a dbt test the job runs (databricks_dbt_tests.py).
+            action["dq_tests"] = list(job.tests)
+        actions.append(action)
         actions.extend(
             {
                 "op": "databricks.workspace_file.ensure",