From 5f7805471e987390ff5919a033b45af662cb0c60 Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Thu, 12 Nov 2026 22:13:57 +0100 Subject: [PATCH 1/6] feat(databricks): data quality monitoring from a contract's freshness and completeness SLOs An expose bound to Databricks that states a freshness or completeness SLO (qos.freshnessSLO, qos.completenessTarget, an observability.defaultSLIs freshness or completeness block, or a contract.dq.rules entry of type freshness, completeness or anomaly_detection) turns on Databricks data quality monitoring (formerly anomaly detection) for the schema the product owns: one databricks_data_quality_monitor per schema (object_type "schema", object_id = the schema's schema_id), created after the schema's tables so the first scan finds them, with the product's other tables there in excluded_table_full_names. contract.dq.monitoring.enabled: false opts an expose out. No new schema field and no new CLI flag. The monitor learns each table's bounds and takes no threshold (provider 1.140.0: excluded_table_full_names is its only setting), so the SLOs pick the tables; the apply notice says so, and fluid validate warns about a dq.monitoring.window other than one day. A monitor turned on outside forge-cli is adopted, not created over (a second create is 400 RESOURCE_ALREADY_EXISTS). Its import id is "schema,", a UUID Databricks assigns, so ImportBlock gains id_from: the engine completes the id from state after the schema's own import and skips it while the schema is not there. Removing a monitor deletes its hidden scan job and what it learned, never table data: it joins RUN_DEFINITION_RESOURCE_TYPES (not gated, printed). Proven live on Databricks Free Edition 2026-11-12 (scratch catalog fluid_iactest_dqm, dropped after): apply +14, an empty second plan, a table_results row (Training) for the watched table and none for the excluded one, Catalog Explorer's Quality indicator, adoption of a REST-made monitor, and an ungated opt-out. Borrowed: the provider's databricks_data_quality_monitor resource (no REST code); learned-vs-stated thresholds as complements follows Elementary's freshness_anomalies beside data_freshness_sla. --- fluid_build/cli/_apply_opentofu_engine.py | 36 +- fluid_build/iac/importer.py | 13 +- fluid_build/iac/providers/databricks.py | 74 +++- .../iac/providers/databricks_quality.py | 191 +++++++++ tests/iac/_fake_databricks.py | 84 +++- .../iac/test_iac_databricks_engine_phase2.py | 1 + .../iac/test_iac_databricks_jobs_fake_e2e.py | 5 +- tests/iac/test_iac_databricks_quality.py | 381 ++++++++++++++++++ .../test_iac_databricks_quality_fake_e2e.py | 156 +++++++ tests/iac/test_iac_tofu_validate.py | 32 ++ 10 files changed, 960 insertions(+), 13 deletions(-) create mode 100644 fluid_build/iac/providers/databricks_quality.py create mode 100644 tests/iac/test_iac_databricks_quality.py create mode 100644 tests/iac/test_iac_databricks_quality_fake_e2e.py diff --git a/fluid_build/cli/_apply_opentofu_engine.py b/fluid_build/cli/_apply_opentofu_engine.py index 22f94aee..47e5fdd3 100644 --- a/fluid_build/cli/_apply_opentofu_engine.py +++ b/fluid_build/cli/_apply_opentofu_engine.py @@ -270,8 +270,9 @@ def _apply_via_opentofu(args, logger: logging.Logger, contract: Optional[Mapping retired = _retired_definitions(changes, removals) if retired: cprint( - f" {len(retired)} job(s), workspace file(s) or folder(s) removed (no table data " - "lost; not gated; a job's run history goes with it): " + ", ".join(retired) + f" {len(retired)} job(s), workspace file(s), folder(s) or data quality " + "monitor(s) removed (no table data lost; not gated; a job's run history and " + "what a monitor has learned go with them): " + ", ".join(retired) ) _report_plan_notices(plugin, contract, logger) # In-place updates that delete data — a dropped column — which the destroy @@ -916,7 +917,10 @@ def _adopt_existing( for block in candidates: if block.to in in_state: continue - result = runner.tofu_import(workdir, block.to, block.id, env=env) + import_id = _import_id(block, workdir, env) + if import_id is None: + continue + result = runner.tofu_import(workdir, block.to, import_id, env=env) if result.ok: adopted += 1 if held is not None: @@ -934,6 +938,21 @@ def _adopt_existing( cprint(f" brownfield: adopted {adopted} pre-existing resource(s) into state") +def _import_id(block: Any, workdir: str, env: Mapping[str, str]) -> Optional[str]: + """``block.id``, or for a block whose id the cloud assigns (``ImportBlock.id_from``) that id + completed from state, which the imports before it have just filled. ``None`` when the + resource it derives from is not in state: then neither exists yet, and there is nothing + to adopt.""" + if getattr(block, "id_from", None) is None: + return block.id + address, attribute = block.id_from + for resource in runner.tofu_state_resources(workdir, env=env): + if resource.get("address") == address: + value = (resource.get("values") or {}).get(attribute) + return block.id.replace("{}", value) if isinstance(value, str) and value else None + return None + + @dataclass class _Imports: """What one run's brownfield adoption imported (:func:`_adopt_existing`).""" @@ -1472,9 +1491,16 @@ def _report_suppressed_drift( #: no table data: a Databricks job (its run history goes with it) and a workspace #: file of a dbt project uploaded from the build's repository. Removing a build #: from the contract removes them, ungated, with a line saying so (founder -#: decision for the Databricks provider, 2026-10-29). +#: decision for the Databricks provider, 2026-10-29). A schema's data quality +#: monitor is one too: deleting it deletes its hidden scan job and what it has +#: learned, never a row (``iac/providers/databricks_quality.py``). RUN_DEFINITION_RESOURCE_TYPES = frozenset( - {"databricks_job", "databricks_workspace_file", "databricks_directory"} + { + "databricks_job", + "databricks_workspace_file", + "databricks_directory", + "databricks_data_quality_monitor", + } ) diff --git a/fluid_build/iac/importer.py b/fluid_build/iac/importer.py index 0beba929..887e3255 100644 --- a/fluid_build/iac/importer.py +++ b/fluid_build/iac/importer.py @@ -17,7 +17,7 @@ from __future__ import annotations from dataclasses import dataclass -from typing import Any, Dict, List +from typing import Any, Dict, List, Optional, Tuple @dataclass(frozen=True) @@ -36,6 +36,12 @@ class ImportBlock: #: adopted (a Databricks grant is authoritative for its principal), so a #: failed import costs an honest plan, never a silent replace. best_effort: bool = False + #: ``(address, attribute)`` of the resource whose value in state completes ``id`` + #: (its ``{}``): an import id the cloud assigns, such as a Databricks schema's UUID, + #: known only once that resource is in state. The import is skipped while it is not, + #: and the block is never rendered into a module (:func:`import_section`): an import + #: id there must be known before the plan. + id_from: Optional[Tuple[str, str]] = None def import_section(blocks: List[ImportBlock]) -> Dict[str, Any]: @@ -44,6 +50,7 @@ def import_section(blocks: List[ImportBlock]) -> Dict[str, Any]: Returns an empty dict when there is nothing to import (so callers can unconditionally merge the result). """ - if not blocks: + literal = [block for block in blocks if block.id_from is None] + if not literal: return {} - return {"import": [{"to": block.to, "id": block.id} for block in blocks]} + return {"import": [{"to": block.to, "id": block.id} for block in literal]} diff --git a/fluid_build/iac/providers/databricks.py b/fluid_build/iac/providers/databricks.py index d1c17fea..0b22ae5f 100644 --- a/fluid_build/iac/providers/databricks.py +++ b/fluid_build/iac/providers/databricks.py @@ -87,6 +87,16 @@ A predicate that is not one boolean expression over the table's columns is ``databricks-row-filter-invalid``; ``access: allow``, a principal with no SQL test and a text mask on a non-string column are refused too. ``fluid verify`` checks the bindings. +* An expose that states a freshness or completeness SLO (``qos.freshnessSLO``, + ``qos.completenessTarget``, ``observability.defaultSLIs``, a ``contract.dq.rules`` + entry of type ``freshness``, ``completeness`` or ``anomaly_detection``) turns on + Databricks data quality monitoring for its schema: one + ``databricks_data_quality_monitor`` per owned schema, the module's other tables there + excluded, ``contract.dq.monitoring.enabled: false`` to opt out + (``databricks_quality.py``). Databricks learns each table's bounds itself: the SLO + values pick the tables and are not thresholds the monitor checks. Its results are the + health indicator on the table in Catalog Explorer and + ``system.data_quality_monitoring.table_results``; removing it is not gated. * 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 +215,8 @@ reconcile, ) from .databricks_keys import lookup as lookup_key_tables +from .databricks_quality import RESOURCE_TYPE as QUALITY_MONITOR +from .databricks_quality import SchemaMonitor, monitor_notices, plan_monitors, window_warnings from .databricks_retention import Retention, check_expire _LOG = logging.getLogger(__name__) @@ -349,6 +361,8 @@ class _Compiled: warnings: List[str] = field(default_factory=list) #: ``(entity type, full name, tag key, tag value)`` per tag assignment emitted. tags: List[Tuple[str, str, str, str]] = field(default_factory=list) + #: The data quality monitor emitted per schema (``databricks_quality.py``). + monitors: List[SchemaMonitor] = field(default_factory=list) #: Unity Catalog allows 50 tags on a securable ("Apply tags to Unity Catalog securable @@ -1382,6 +1396,7 @@ def _compile( _emit_grants(compiled, table, cid, privileges.get(table.full_name, {})) _emit_tags(compiled, table, cid, contract) _check_own_keys(compiled, contract) + _emit_quality_monitors(compiled, cid) _merge_jobs( compiled, compile_dbt_jobs( @@ -1601,15 +1616,60 @@ def _refuse_duplicate_tables( bound.setdefault(table.full_name, table.where) +def _schema_key(cid: str, catalog: str, schema: str) -> str: + """The resource key of an owned schema, and of its data quality monitor.""" + return safe_ident(f"{cid}_{catalog}_{schema}") + + def _keys(cid: str, table: _Table) -> Tuple[str, str, str]: """``(catalog, schema, table)`` resource keys of one table.""" return ( safe_ident(f"{cid}_{table.catalog}"), - safe_ident(f"{cid}_{table.catalog}_{table.schema}"), + _schema_key(cid, table.catalog, table.schema), safe_ident(f"{cid}_{table.catalog}_{table.schema}_{table.name}"), ) +def _emit_quality_monitors(compiled: _Compiled, cid: str) -> None: + """Databricks data quality monitoring on each owned schema one of its tables asks it for + (``databricks_quality.py``), keyed like the schema it watches. + + A monitor someone turned on outside forge-cli is adopted, not created over: a second + create fails (``RESOURCE_ALREADY_EXISTS``, measured on Free Edition 2026-11-12). Its + import id is ``schema,``, a UUID Databricks assigns, so the import takes it + from the schema's state after the schema's own import (``ImportBlock.id_from``). + + It is created after the schema's tables (``depends_on``): enabling it starts a scan, and + a table that scan misses may wait up to two weeks for its health indicator (Databricks + docs, "intelligent scanning"). + """ + compiled.warnings.extend(window_warnings(compiled.tables)) + for monitor in plan_monitors(compiled.tables): + key = _schema_key(cid, monitor.catalog, monitor.schema) + body = monitor.body(tofu_ref(f"databricks_schema.{key}.schema_id")) + body["depends_on"] = [ + f"databricks_sql_table.{_keys(cid, table)[2]}" + for table in compiled.tables + if (table.catalog, table.schema) == (monitor.catalog, monitor.schema) + ] + _claim( + compiled, + QUALITY_MONITOR, + key, + f"data quality monitor on schema {monitor.full_name}", + body, + ) + compiled.imports.append( + ImportBlock( + f"{QUALITY_MONITOR}.{key}", + "schema,{}", + best_effort=True, + id_from=(f"databricks_schema.{key}", "schema_id"), + ) + ) + compiled.monitors.append(monitor) + + def _emit_table( compiled: _Compiled, table: _Table, cid: str, warehouse: str, region: str, *, dbt_owned: bool ) -> None: @@ -2413,7 +2473,7 @@ def plan_notices(self, contract: Mapping[str, Any]) -> List[str]: """What the apply cannot do itself and someone must: ``USE CATALOG``; and the dbt jobs it creates paused until the tables they write are protected.""" compiled = _compile(contract, os.environ) - notices = _use_catalog_notices(compiled) + notices = _use_catalog_notices(compiled) + monitor_notices(compiled.monitors) certified = sorted( { f"{full_name} ({value})" @@ -2718,6 +2778,16 @@ def planned_objects(contract: Mapping[str, Any]) -> List[Dict[str, Any]]: "tag_value": tag_value, } ) + for monitor in compiled.monitors: + actions.append( + { + "op": "databricks.quality_monitor", + "resource_type": QUALITY_MONITOR, + "resource_id": monitor.full_name, + "monitored": list(monitor.monitored), + "excluded": list(monitor.excluded), + } + ) actions.extend(_job_actions(compiled.jobs)) unique: Dict[Tuple[str, str], Dict[str, Any]] = {} for action in actions: diff --git a/fluid_build/iac/providers/databricks_quality.py b/fluid_build/iac/providers/databricks_quality.py new file mode 100644 index 00000000..7cddabf0 --- /dev/null +++ b/fluid_build/iac/providers/databricks_quality.py @@ -0,0 +1,191 @@ +# Copyright 2025 Agentics Transformation Limited +# SPDX-License-Identifier: Apache-2.0 + +"""Databricks data quality monitoring on the schemas a product owns. + +Databricks data quality monitoring (formerly anomaly detection) is enabled per Unity +Catalog schema. On serverless compute it scans the schema's tables for freshness (a +commit unusually late for the table) and completeness (fewer rows in the last 24 hours +than the table's history predicts), writes one row per table and scan to +``system.data_quality_monitoring.table_results`` and shows the result as the health +indicator on each table's Overview in Catalog Explorer +(https://docs.databricks.com/aws/en/data-quality-monitoring/anomaly-detection/). + +An expose asks for it with fields it already has, no new one: a freshness or completeness +SLO (``qos.freshnessSLO``, ``qos.completenessTarget``, an ``observability.defaultSLIs`` +``freshness`` or ``completeness`` block) or a ``contract.dq.rules`` entry of type +``freshness``, ``completeness`` or ``anomaly_detection``. ``contract.dq.monitoring.enabled: +false`` turns it off for that expose. A schema gets one ``databricks_data_quality_monitor`` +(``object_type = "schema"``; the pinned provider 1.140.0 has the resource) when at least one +of the module's tables in it asks, and lists the module's other tables there in +``anomaly_detection_config.excluded_table_full_names``. A table the contract does not +declare is scanned too: exclusions are by name. + +The monitor learns each table's bounds from that table's own history. It takes no +threshold: on provider 1.140.0 ``excluded_table_full_names`` is its only setting, and the +job parameters that took threshold overrides are closed to new customers since 2025-07-21. +So the contract's values (``PT15M``, ``0.98``) decide which tables are watched, and are +not the bounds the monitor checks. Learned and stated thresholds are complements, not +substitutes (Elementary's ``freshness_anomalies`` beside its ``data_freshness_sla``). + +The monitor's ``object_id`` is the owned schema's ``schema_id``, so OpenTofu deletes it +before the schema whenever both go. Removing it (an apply whose contract no longer asks, +or the schema's removal) deletes its hidden scan job and what it has learned, never table +data, so the data-loss gate does not count it (``RUN_DEFINITION_RESOURCE_TYPES``). +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, Dict, List, Mapping, Protocol, Sequence, Tuple + +from ..naming import TofuExpr +from .databricks_retention import DAY, iso_duration_seconds + +#: The provider resource: on a schema, anomaly detection. +RESOURCE_TYPE = "databricks_data_quality_monitor" + +#: The ``contract.dq.rules`` types that ask for it: the two checks it runs, and the +#: schema's own name for the feature (``dqRule.type``). +_RULE_TYPES = ("freshness", "completeness", "anomaly_detection") + + +class MonitoredTable(Protocol): + """What this module reads of a resolved table (``databricks.py::_Table``).""" + + catalog: str + schema: str + exposure: Mapping[str, Any] + + @property + def full_name(self) -> str: ... + + @property + def where(self) -> str: ... + + +def _dq(exposure: Mapping[str, Any]) -> Mapping[str, Any]: + contract = exposure.get("contract") + dq = contract.get("dq") if isinstance(contract, Mapping) else None + return dq if isinstance(dq, Mapping) else {} + + +def _monitoring(exposure: Mapping[str, Any]) -> Mapping[str, Any]: + monitoring = _dq(exposure).get("monitoring") + return monitoring if isinstance(monitoring, Mapping) else {} + + +def monitoring_request(exposure: Mapping[str, Any]) -> Tuple[str, ...]: + """The fields of ``exposure`` that ask for data quality monitoring, in contract order; + ``()`` when none does, or when ``contract.dq.monitoring.enabled`` is ``false``.""" + if _monitoring(exposure).get("enabled") is False: + return () + asks: List[str] = [] + qos = exposure.get("qos") + if isinstance(qos, Mapping): + if qos.get("freshnessSLO"): + asks.append("qos.freshnessSLO") + if qos.get("completenessTarget") is not None: + asks.append("qos.completenessTarget") + observability = exposure.get("observability") + slis = observability.get("defaultSLIs") if isinstance(observability, Mapping) else None + if isinstance(slis, Mapping) and slis.get("enabled") is not False: + for name in ("freshness", "completeness"): + sli = slis.get(name) + if isinstance(sli, Mapping) and sli.get("enabled") is not False: + asks.append(f"observability.defaultSLIs.{name}") + rules = _dq(exposure).get("rules") + for index, rule in enumerate(rules if isinstance(rules, list) else []): + if isinstance(rule, Mapping) and rule.get("type") in _RULE_TYPES: + asks.append(f"contract.dq.rules[{index}] ({rule['type']})") + return tuple(asks) + + +@dataclass(frozen=True) +class SchemaMonitor: + """One schema's monitor: the module's tables in it that asked, and the ones it skips.""" + + catalog: str + schema: str + #: Full names, sorted. + monitored: Tuple[str, ...] + excluded: Tuple[str, ...] + + @property + def full_name(self) -> str: + return f"{self.catalog}.{self.schema}" + + def body(self, schema_id: TofuExpr) -> Dict[str, Any]: + """The resource body; with nothing excluded the config is ``{}``, no empty list.""" + config: Dict[str, Any] = {} + if self.excluded: + config["excluded_table_full_names"] = list(self.excluded) + return { + "object_type": "schema", + "object_id": schema_id, + "anomaly_detection_config": config, + } + + +def plan_monitors(tables: Sequence[MonitoredTable]) -> List[SchemaMonitor]: + """One monitor per schema in which at least one table asks (:func:`monitoring_request`), + in the order the schemas first appear.""" + by_schema: Dict[Tuple[str, str], Tuple[List[str], List[str]]] = {} + for table in tables: + asked, skipped = by_schema.setdefault((table.catalog, table.schema), ([], [])) + (asked if monitoring_request(table.exposure) else skipped).append(table.full_name) + return [ + SchemaMonitor(catalog, schema, tuple(sorted(set(asked))), tuple(sorted(set(skipped)))) + for (catalog, schema), (asked, skipped) in by_schema.items() + if asked + ] + + +def window_warnings(tables: Sequence[MonitoredTable]) -> List[str]: + """A ``fluid validate`` warning for each monitored table whose ``contract.dq.monitoring + .window`` is not the one day the monitor counts rows over: the window is not applied.""" + warnings: List[str] = [] + for table in tables: + window = _monitoring(table.exposure).get("window") + if window is None or not monitoring_request(table.exposure): + continue + try: + if iso_duration_seconds(window) == DAY: + continue + except ValueError: + continue # not a duration: the schema reports it + warnings.append( + f"{table.where}: contract.dq.monitoring.window {window} is not applied. Databricks " + "data quality monitoring counts a table's rows over the last 24 hours and judges " + "its freshness against the table's own commit history." + ) + return warnings + + +def monitor_notices(monitors: Sequence[SchemaMonitor]) -> List[str]: + """What the apply says about each monitor: where the results are, who can read them, + and that the bounds and the cost are Databricks'.""" + notices: List[str] = [] + for monitor in monitors: + skipped = f"; excluded: {', '.join(monitor.excluded)}" if monitor.excluded else "" + notices.append( + f"data quality monitoring is on for schema {monitor.full_name} (watched: " + f"{', '.join(monitor.monitored)}{skipped}). Databricks learns each table's freshness " + "and completeness bounds from its history, not from the contract's SLO values, " + "scans on serverless compute billed as DATA_QUALITY_MONITORING, and shows the " + "result as the table's health indicator in Catalog Explorer. " + "system.data_quality_monitoring.table_results holds every scan; only account " + "admins can read it unless they grant SELECT." + ) + return notices + + +__all__ = [ + "RESOURCE_TYPE", + "MonitoredTable", + "SchemaMonitor", + "monitor_notices", + "monitoring_request", + "plan_monitors", + "window_warnings", +] diff --git a/tests/iac/_fake_databricks.py b/tests/iac/_fake_databricks.py index a8139499..f6b6762c 100644 --- a/tests/iac/_fake_databricks.py +++ b/tests/iac/_fake_databricks.py @@ -6,9 +6,10 @@ What it proves: what the real ``databricks/databricks`` provider (1.140.0) and ``tofu`` plan, apply and import for a live object. It serves ``data.databricks_catalog``, ``databricks_schema``, ``databricks_sql_table`` (whose DDL goes through the SQL Statement -Execution API), ``databricks_grant`` and ``databricks_entity_tag_assignment`` (create, get, +Execution API), ``databricks_grant``, ``databricks_entity_tag_assignment`` (create, get, update, delete and the paginated list ``fluid verify`` reads, on a catalog, schema, table or -column; ``tags``, ``tag_page_size``). A +column; ``tags``, ``tag_page_size``) and a schema's ``databricks_data_quality_monitor`` +(``quality_monitors``). A small SQL interpreter understands exactly the statements ``databricks_sql_table`` sends (CREATE [OR REPLACE] TABLE, ALTER TABLE ADD / DROP / RENAME / ALTER COLUMN, SET / UNSET TBLPROPERTIES, CLUSTER BY, COMMENT ON TABLE, DROP TABLE), and the statements forge-cli sends for a contract's @@ -981,6 +982,9 @@ def __init__( self.statements: Dict[str, Dict[str, Any]] = {} # job_id -> {job_id, created_time, creator_user_name, settings}; settings as the provider sent self.jobs: Dict[int, Dict[str, Any]] = {} + # (object_type, object_id) -> the data quality monitor as the API serves it + # (``/api/data-quality/v1/monitors``); object_id is a schema's schema_id + self.quality_monitors: Dict[Tuple[str, str], Dict[str, Any]] = {} self.workspace_files: Dict[str, bytes] = {} # workspace path -> content self.workspace_dirs: set = set() # directories a workspace import / mkdirs created self.missing_parent_uploads: List[str] = [] # imports that met no parent folder @@ -1294,8 +1298,84 @@ def _register(self) -> None: a("POST", r"/api/2\.0/sql/statements/?", self._exec_sql) a("GET", r"/api/2\.0/sql/statements/([^/]+)", self._get_stmt) a("GET", r"/api/2\.0/sql/warehouses/([^/]+)", self._get_wh) + # -- data quality monitoring (anomaly detection on a schema) + a("POST", r"/api/data-quality/v1/monitors", self._dqm_create) + a("GET", r"/api/data-quality/v1/monitors/(schema|table)/([^/]+)", self._dqm_get) + a("PATCH", r"/api/data-quality/v1/monitors/(schema|table)/([^/]+)", self._dqm_update) + a("DELETE", r"/api/data-quality/v1/monitors/(schema|table)/([^/]+)", self._dqm_delete) # ------------------------------------------------------------------ handlers + # -- data quality monitoring. Shapes as Free Edition answered them on 2026-11-12: a second + # create is 400 RESOURCE_ALREADY_EXISTS, a missing monitor 404 RESOURCE_DOES_NOT_EXIST + # ("Cannot find Schema with ID"), a create starts a run (PENDING), and the response carries + # read-only fields (workflow id, job type, run id and status) the provider does not model. + def _dqm_missing(self, object_id: str) -> Tuple[int, Any]: + return _err( + 404, + "RESOURCE_DOES_NOT_EXIST", + f"Cannot find Schema with ID: /metastores:ms-fake/schemas:{object_id}", + ) + + def _dqm_doc(self, key: Tuple[str, str]) -> Dict[str, Any]: + return copy.deepcopy(self.quality_monitors[key]) + + def _dqm_create(self, m: Any, q: Any, b: Any) -> Tuple[int, Any]: + object_type, object_id = b.get("object_type"), b.get("object_id") + if object_type != "schema": + return _err(400, "INVALID_PARAMETER_VALUE", "fake: only schema monitors are served") + if not any(s["schema_id"] == object_id for s in self.schemas.values()): + return _err(404, "SCHEMA_DOES_NOT_EXIST", f"Schema {object_id} does not exist") + key = (object_type, object_id) + if key in self.quality_monitors: + return _err( + 400, + "RESOURCE_ALREADY_EXISTS", + "Already exists AnomalyDetectionConfig with ID: " + f"/metastores:ms-fake/schemas:{object_id}/", + ) + config = dict(b.get("anomaly_detection_config") or {}) + with self._lock: + self._job_seq += 1 + run = self._job_seq + config.update( + { + "anomaly_detection_workflow_id": 500000 + run, + "job_type": "ANOMALY_DETECTION_JOB_TYPE_INTERNAL_HIDDEN", + "last_run_id": str(900000 + run), + "latest_run_status": "ANOMALY_DETECTION_RUN_STATUS_PENDING", + } + ) + self.quality_monitors[key] = { + "object_type": object_type, + "object_id": object_id, + "anomaly_detection_config": config, + } + return 200, self._dqm_doc(key) + + def _dqm_get(self, m: Any, q: Any, b: Any) -> Tuple[int, Any]: + key = (m.group(1), m.group(2)) + if key not in self.quality_monitors: + return self._dqm_missing(m.group(2)) + return 200, self._dqm_doc(key) + + def _dqm_update(self, m: Any, q: Any, b: Any) -> Tuple[int, Any]: + key = (m.group(1), m.group(2)) + if key not in self.quality_monitors: + return self._dqm_missing(m.group(2)) + config = self.quality_monitors[key]["anomaly_detection_config"] + sent = b.get("anomaly_detection_config") or {} + if "excluded_table_full_names" in sent: + config["excluded_table_full_names"] = list(sent["excluded_table_full_names"]) + else: + config.pop("excluded_table_full_names", None) + return 200, self._dqm_doc(key) + + def _dqm_delete(self, m: Any, q: Any, b: Any) -> Tuple[int, Any]: + key = (m.group(1), m.group(2)) + if self.quality_monitors.pop(key, None) is None: + return self._dqm_missing(m.group(2)) + return 200, {} + def _me(self, m: Any, q: Any, b: Any) -> Tuple[int, Any]: return 200, { "id": "1", diff --git a/tests/iac/test_iac_databricks_engine_phase2.py b/tests/iac/test_iac_databricks_engine_phase2.py index d72d86aa..5ba83c2e 100644 --- a/tests/iac/test_iac_databricks_engine_phase2.py +++ b/tests/iac/test_iac_databricks_engine_phase2.py @@ -375,6 +375,7 @@ def test_run_definitions_are_the_job_its_files_and_their_folders(self): "databricks_job", "databricks_workspace_file", "databricks_directory", + "databricks_data_quality_monitor", } def test_removed_jobs_and_files_are_not_data_bearing(self): diff --git a/tests/iac/test_iac_databricks_jobs_fake_e2e.py b/tests/iac/test_iac_databricks_jobs_fake_e2e.py index d7050592..6397f873 100644 --- a/tests/iac/test_iac_databricks_jobs_fake_e2e.py +++ b/tests/iac/test_iac_databricks_jobs_fake_e2e.py @@ -202,7 +202,10 @@ def test_removing_the_job_is_not_gated_and_is_said(workspace, fake, capfd): # The build now runs locally: its job and uploaded project go; its table stays. rc, out = apply_contract(capfd, write_contract(workspace, job_contract(platform=None))) assert rc == 0, out - assert "job(s), workspace file(s) or folder(s) removed (no table data lost; not gated" in out + assert ( + "job(s), workspace file(s), folder(s) or data quality monitor(s) removed (no table data lost; not gated" + in out + ) assert fake.jobs == {} and fake.workspace_files == {} assert not [folder for folder in fake.workspace_dirs if folder.startswith(ROOT)] assert TABLE in fake.tables diff --git a/tests/iac/test_iac_databricks_quality.py b/tests/iac/test_iac_databricks_quality.py new file mode 100644 index 00000000..68125eb4 --- /dev/null +++ b/tests/iac/test_iac_databricks_quality.py @@ -0,0 +1,381 @@ +# Copyright 2025 Agentics Transformation Limited +# SPDX-License-Identifier: Apache-2.0 + +"""Databricks data quality monitoring, Stage 1 unit: what asks for it, the monitor per schema, +its adoption by a derived import id, the apply notice and the ungated removal. + +Pure functions only: no ``tofu``, no network. ``test_iac_databricks_quality_fake_e2e.py`` pins +what real ``tofu`` and the provider do with the module against the stand-in workspace. +""" + +from __future__ import annotations + +import copy +import json +import logging +from typing import Any, Dict, List + +import pytest + +from fluid_build.cli import _apply_opentofu_engine as engine +from fluid_build.iac import IAC_PLUGINS, build_module +from fluid_build.iac.importer import ImportBlock, import_section +from fluid_build.iac.providers import databricks as dbx +from fluid_build.iac.providers import databricks_quality as dq +from fluid_build.iac.runner import TofuResult + +pytestmark = [pytest.mark.unit, pytest.mark.databricks] + +_LOG = logging.getLogger("test.iac.databricks_quality") +MONITOR = "databricks_data_quality_monitor" +KEY = "iot_meters_main_iot" + + +@pytest.fixture(autouse=True) +def _warehouse(monkeypatch): + monkeypatch.setenv(dbx.WAREHOUSE_ENV, "wh-0123456789abcdef") + + +def _expose(table: str, schema: str = "iot", **fields: Any) -> Dict[str, Any]: + return { + "exposeId": table, + "kind": "table", + "binding": { + "platform": "databricks", + "format": "delta_table", + "location": {"database": "main", "schema": schema, "table": table}, + }, + "contract": {"schema": [{"name": "id", "type": "bigint"}]}, + **fields, + } + + +def _contract(*exposes: Dict[str, Any]) -> Dict[str, Any]: + return { + "fluidVersion": "0.7.6", + "kind": "DataProduct", + "id": "iot.meters", + "name": "Meters", + "exposes": list(exposes) or [_expose("readings", qos={"freshnessSLO": "PT15M"})], + } + + +def _module(contract: Dict[str, Any]) -> Dict[str, Any]: + return json.loads(build_module(IAC_PLUGINS["databricks"], contract)) + + +def _dq(**dq_fields: Any) -> Dict[str, Any]: + return {"schema": [{"name": "id", "type": "bigint"}], "dq": dq_fields} + + +# ── what asks for it ──────────────────────────────────────────────────────── +class TestMonitoringRequest: + @pytest.mark.parametrize( + "fields, asked", + [ + ({"qos": {"freshnessSLO": "PT15M"}}, ("qos.freshnessSLO",)), + ({"qos": {"completenessTarget": 0.98}}, ("qos.completenessTarget",)), + ( + {"observability": {"defaultSLIs": {"freshness": {"threshold": "PT1H"}}}}, + ("observability.defaultSLIs.freshness",), + ), + ( + {"observability": {"defaultSLIs": {"completeness": {"threshold": 0.9}}}}, + ("observability.defaultSLIs.completeness",), + ), + ( + {"contract": _dq(rules=[{"id": "f", "type": "freshness", "severity": "warn"}])}, + ("contract.dq.rules[0] (freshness)",), + ), + ( + { + "contract": _dq( + rules=[ + {"id": "u", "type": "uniqueness", "severity": "error"}, + {"id": "c", "type": "completeness", "severity": "error"}, + {"id": "a", "type": "anomaly_detection", "severity": "info"}, + ] + ) + }, + ("contract.dq.rules[1] (completeness)", "contract.dq.rules[2] (anomaly_detection)"), + ), + ], + ) + def test_each_freshness_or_completeness_slo_asks(self, fields, asked): + assert dq.monitoring_request(_expose("t", **fields)) == asked + + @pytest.mark.parametrize( + "fields", + [ + {}, + {"qos": {"availability": "99.9%", "latencyP95": "PT5S"}}, + {"observability": {"defaultSLIs": {"latency": {"threshold": "PT5S"}}}}, + {"observability": {"defaultSLIs": {"freshness": {"enabled": False}}}}, + { + "observability": { + "defaultSLIs": {"enabled": False, "freshness": {"threshold": "PT1H"}} + } + }, + {"contract": _dq(rules=[{"id": "u", "type": "uniqueness", "severity": "error"}])}, + # malformed shapes the schema refuses never crash the emitter + {"qos": ["PT15M"], "observability": "x", "contract": {"dq": {"rules": "x"}}}, + ], + ) + def test_nothing_else_asks(self, fields): + assert dq.monitoring_request(_expose("t", **fields)) == () + + def test_monitoring_disabled_opts_out_whatever_the_slos(self): + expose = _expose( + "t", + qos={"freshnessSLO": "PT15M", "completenessTarget": 0.98}, + contract=_dq( + monitoring={"enabled": False}, + rules=[{"id": "f", "type": "freshness", "severity": "warn"}], + ), + ) + assert dq.monitoring_request(expose) == () + + def test_monitoring_enabled_alone_asks_for_nothing(self): + # the schema's default; an SLO is what says what to watch + assert dq.monitoring_request(_expose("t", contract=_dq(monitoring={"enabled": True}))) == () + + +# ── one monitor per schema ────────────────────────────────────────────────── +class _T: + def __init__(self, full_name: str, exposure: Dict[str, Any]): + self.catalog, self.schema, _ = full_name.split(".") + self.full_name = full_name + self.where = full_name + self.exposure = exposure + + +ASKS = {"qos": {"freshnessSLO": "PT15M"}} + + +class TestPlanMonitors: + def test_one_per_schema_with_its_other_tables_excluded_sorted(self): + monitors = dq.plan_monitors( + [ + _T("main.iot.z_notes", {}), + _T("main.iot.readings", ASKS), + _T("main.ops.alarms", ASKS), + _T("main.iot.a_notes", {}), + _T("main.ref.codes", {}), + ] + ) + assert monitors == [ + dq.SchemaMonitor( + "main", "iot", ("main.iot.readings",), ("main.iot.a_notes", "main.iot.z_notes") + ), + dq.SchemaMonitor("main", "ops", ("main.ops.alarms",), ()), + ] + + def test_no_table_asks_no_monitor(self): + assert dq.plan_monitors([_T("main.iot.readings", {})]) == [] + + def test_the_body_watches_the_schema_by_its_id(self): + ref = dbx.tofu_ref("databricks_schema.k.schema_id") + plain = dq.SchemaMonitor("main", "iot", ("main.iot.r",), ()) + assert plain.body(ref) == { + "object_type": "schema", + "object_id": ref, + "anomaly_detection_config": {}, + } + skipping = dq.SchemaMonitor("main", "iot", ("main.iot.r",), ("main.iot.n",)) + assert skipping.body(ref)["anomaly_detection_config"] == { + "excluded_table_full_names": ["main.iot.n"] + } + + +class TestWindowWarnings: + @pytest.mark.parametrize("window", ["P1D", "PT24H"]) + def test_one_day_is_what_it_counts_over(self, window): + expose = {**ASKS, "contract": _dq(monitoring={"window": window})} + assert dq.window_warnings([_T("main.iot.r", expose)]) == [] + + def test_another_window_is_not_applied(self): + expose = {**ASKS, "contract": _dq(monitoring={"window": "PT1H"})} + (warning,) = dq.window_warnings([_T("main.iot.r", expose)]) + assert "contract.dq.monitoring.window PT1H is not applied" in warning + + @pytest.mark.parametrize( + "expose", + [ + {"contract": _dq(monitoring={"window": "PT1H"})}, # not monitored + {**ASKS, "contract": _dq(monitoring={"window": "soon"})}, # the schema refuses it + ], + ) + def test_silent_otherwise(self, expose): + assert dq.window_warnings([_T("main.iot.r", expose)]) == [] + + +# ── in the module ─────────────────────────────────────────────────────────── +class TestEmitter: + def test_an_slo_puts_a_monitor_on_the_owned_schema(self): + module = _module( + _contract( + _expose("readings", qos={"freshnessSLO": "PT15M", "completenessTarget": 0.98}), + _expose("notes"), + ) + ) + assert module["resource"][MONITOR] == { + KEY: { + "object_type": "schema", + "object_id": f"${{databricks_schema.{KEY}.schema_id}}", + "anomaly_detection_config": {"excluded_table_full_names": ["main.iot.notes"]}, + # after the tables: the scan enabling it starts must find them + "depends_on": [ + f"databricks_sql_table.{KEY}_readings", + f"databricks_sql_table.{KEY}_notes", + ], + } + } + assert KEY in module["resource"]["databricks_schema"] + assert set(module["resource"]["databricks_sql_table"]) == { + f"{KEY}_readings", + f"{KEY}_notes", + } + + def test_no_slo_leaves_the_module_as_it_was(self): + assert MONITOR not in _module(_contract(_expose("readings")))["resource"] + + def test_opting_out_removes_only_the_monitor(self): + on = _module(_contract()) + opted_out = _contract( + _expose( + "readings", + qos={"freshnessSLO": "PT15M"}, + contract=_dq(monitoring={"enabled": False}), + ) + ) + off = _module(opted_out) + assert MONITOR not in off["resource"] + del on["resource"][MONITOR] + assert on == off + + def test_each_schema_gets_its_own(self): + module = _module( + _contract( + _expose("readings", qos={"freshnessSLO": "PT15M"}), + _expose("alarms", schema="ops", qos={"completenessTarget": 0.99}), + ) + ) + assert sorted(module["resource"][MONITOR]) == [KEY, "iot_meters_main_ops"] + + def test_the_monitor_is_adopted_by_the_schema_id_in_state(self): + imports = IAC_PLUGINS["databricks"].discover_imports(_contract()) + schema, monitor = ( + next(b for b in imports if b.to == f"databricks_schema.{KEY}"), + next(b for b in imports if b.to == f"{MONITOR}.{KEY}"), + ) + assert imports.index(schema) < imports.index(monitor) + assert monitor == ImportBlock( + f"{MONITOR}.{KEY}", + "schema,{}", + best_effort=True, + id_from=(f"databricks_schema.{KEY}", "schema_id"), + ) + # never rendered into a module: its id is known only once the schema is in state + assert import_section([monitor]) == {} + assert import_section([schema, monitor]) == {"import": [{"to": schema.to, "id": schema.id}]} + + def test_plan_names_the_monitor_and_the_apply_says_what_it_does(self): + contract = _contract(_expose("readings", qos={"freshnessSLO": "PT15M"}), _expose("notes")) + (action,) = [a for a in dbx.planned_objects(contract) if a["resource_type"] == MONITOR] + assert action == { + "op": "databricks.quality_monitor", + "resource_type": MONITOR, + "resource_id": "main.iot", + "monitored": ["main.iot.readings"], + "excluded": ["main.iot.notes"], + } + (notice,) = [n for n in IAC_PLUGINS["databricks"].plan_notices(contract) if "quality" in n] + assert notice.startswith( + "data quality monitoring is on for schema main.iot (watched: main.iot.readings; " + "excluded: main.iot.notes)." + ) + assert "not from the contract's SLO values" in notice + assert "DATA_QUALITY_MONITORING" in notice + assert "system.data_quality_monitoring.table_results" in notice + + def test_validate_warns_about_a_window_it_cannot_apply(self): + contract = _contract( + _expose( + "readings", + qos={"freshnessSLO": "PT15M"}, + contract=_dq(monitoring={"window": "PT1H"}), + ) + ) + errors, warnings = dbx.validate_databricks_binding(contract) + assert errors == [] + assert any("window PT1H is not applied" in w for w in warnings) + + +# ── the engine ────────────────────────────────────────────────────────────── +class TestEngine: + def test_removing_a_monitor_is_not_data_loss(self): + removals = [(f"{MONITOR}.{KEY}", MONITOR)] + changes = {"add": 0, "change": 0, "remove": 1} + counted, revoked = engine._data_bearing_changes(changes, removals) + assert counted["remove"] == 0 and revoked == [] + assert engine._retired_definitions(changes, removals) == [f"{MONITOR}.{KEY}"] + + def _state(self, schema_id: Any) -> List[Dict[str, Any]]: + return [ + { + "address": f"databricks_schema.{KEY}", + "type": "databricks_schema", + "values": {"name": "iot", "schema_id": schema_id}, + } + ] + + def test_a_derived_import_id_comes_from_state(self, monkeypatch): + block = ImportBlock( + f"{MONITOR}.{KEY}", "schema,{}", id_from=(f"databricks_schema.{KEY}", "schema_id") + ) + monkeypatch.setattr( + engine.runner, "tofu_state_resources", lambda *a, **k: self._state("u-1") + ) + assert engine._import_id(block, "/w", {}) == "schema,u-1" + for missing in ([], self._state(None), self._state("")): + monkeypatch.setattr( + engine.runner, "tofu_state_resources", lambda *a, _m=missing, **k: _m + ) + assert engine._import_id(block, "/w", {}) is None + literal = ImportBlock("databricks_schema.x", "main.iot") + assert engine._import_id(literal, "/w", {}) == "main.iot" + + def test_adoption_imports_the_monitor_after_its_schema(self, monkeypatch): + state: List[Dict[str, Any]] = [] + calls: List[tuple] = [] + monkeypatch.setattr(engine.runner, "tofu_state_list", lambda *a, **k: []) + monkeypatch.setattr(engine.runner, "tofu_state_resources", lambda *a, **k: list(state)) + + def fake_import(workdir, address, rid, env=None): + calls.append((address, rid)) + if address == f"databricks_schema.{KEY}": + state.extend(self._state("u-42")) + return TofuResult("import", 0, "", "", []) + + monkeypatch.setattr(engine.runner, "tofu_import", fake_import) + contract = _contract() + engine._adopt_existing(IAC_PLUGINS["databricks"], contract, [], "/w", {}, _LOG) + assert (f"{MONITOR}.{KEY}", "schema,u-42") in calls + addresses = [address for address, _ in calls] + assert addresses.index(f"databricks_schema.{KEY}") < addresses.index(f"{MONITOR}.{KEY}") + + def test_no_schema_in_state_no_monitor_import(self, monkeypatch): + calls: List[str] = [] + monkeypatch.setattr(engine.runner, "tofu_state_list", lambda *a, **k: []) + monkeypatch.setattr(engine.runner, "tofu_state_resources", lambda *a, **k: []) + + def fake_import(workdir, address, rid, env=None): + calls.append(address) + return TofuResult("import", 1, "", "Cannot import non-existent remote object", []) + + monkeypatch.setattr(engine.runner, "tofu_import", fake_import) + engine._adopt_existing( + IAC_PLUGINS["databricks"], copy.deepcopy(_contract()), [], "/w", {}, _LOG + ) + assert f"{MONITOR}.{KEY}" not in calls + assert f"databricks_schema.{KEY}" in calls diff --git a/tests/iac/test_iac_databricks_quality_fake_e2e.py b/tests/iac/test_iac_databricks_quality_fake_e2e.py new file mode 100644 index 00000000..f98fcf92 --- /dev/null +++ b/tests/iac/test_iac_databricks_quality_fake_e2e.py @@ -0,0 +1,156 @@ +# Copyright 2025 Agentics Transformation Limited +# SPDX-License-Identifier: Apache-2.0 + +"""Databricks data quality monitoring through the real ``fluid`` CLI, ``tofu`` and provider, +against a stand-in. + +Pinned (Stage 1: real ``tofu`` and ``databricks/databricks`` 1.140.0, the stand-in workspace of +``_fake_databricks.py``, its data quality endpoints shaped as Free Edition answered them): an +expose with a freshness SLO turns monitoring on for its schema, watching it by the schema's +``schema_id`` and excluding the product's other table there; a second plan is empty; a monitor +someone turned on outside forge-cli is adopted, not created over; opting out deletes only the +monitor, not gated; and OpenTofu deletes the monitor before its schema. + +Not proven here: what real Databricks scans and reports (the live Free Edition run, +HONESTLY_TESTED.md). +""" + +from __future__ import annotations + +import copy +from pathlib import Path +from typing import Any, Dict, Iterator + +import pytest + +from fluid_build.iac import runner + +from ._databricks_cli import ( # noqa: F401 — plugin_cache is a fixture + apply_contract, + isolate_workspace, + make_contract, + plugin_cache, + write_contract, +) +from ._fake_databricks import FakeDatabricks + +pytestmark = [ + pytest.mark.integration, + pytest.mark.provider, + pytest.mark.databricks, + pytest.mark.skipif(runner.tofu_path() is None, reason="tofu binary not installed"), +] + +CREATE = ("POST", "/api/data-quality/v1/monitors") + + +@pytest.fixture +def fake() -> Iterator[FakeDatabricks]: + server = FakeDatabricks().start() + try: + yield server + finally: + server.stop() + + +@pytest.fixture +def workspace(monkeypatch, tmp_path: Path, fake: FakeDatabricks, plugin_cache: str) -> Path: + isolate_workspace( + monkeypatch, + tmp_path, + host=fake.endpoint, + warehouse_id=fake.warehouse_id, + plugin_cache=plugin_cache, + ) + return tmp_path + + +def _monitored(enabled: bool = True) -> Dict[str, Any]: + """``make_contract``'s orders with a freshness SLO, and a notes table without one.""" + contract = make_contract() + orders = contract["exposes"][0] + orders["qos"] = {"freshnessSLO": "PT15M", "completenessTarget": 0.98} + if not enabled: + orders["contract"]["dq"] = {"monitoring": {"enabled": False}} + notes = copy.deepcopy(orders) + notes.pop("qos") + notes["contract"].pop("dq", None) + notes["exposeId"] = "notes" + notes["binding"]["location"]["table"] = "notes" + contract["exposes"].append(notes) + return contract + + +def _monitor(fake: FakeDatabricks) -> Dict[str, Any]: + return fake.quality_monitors[("schema", fake.schemas["main.sales"]["schema_id"])] + + +def test_an_slo_turns_monitoring_on_for_the_schema_and_a_second_plan_is_empty( + workspace, fake, capfd +): + path = write_contract(workspace, _monitored()) + rc, out = apply_contract(capfd, path) + assert rc == 0, out + assert list(fake.quality_monitors) == [("schema", "sid-main.sales")] + config = _monitor(fake)["anomaly_detection_config"] + assert config["excluded_table_full_names"] == ["main.sales.notes"] + assert ( + "data quality monitoring is on for schema main.sales (watched: main.sales.orders; " + "excluded: main.sales.notes)" in out + ) + assert fake.unknown == [] + + rc, out = apply_contract(capfd, path, "--mode", "dry-run") + assert rc == 0, out + assert "tofu plan: +0 ~0 -0" in out + + +def test_a_monitor_turned_on_outside_forge_cli_is_adopted_not_created_over(workspace, fake, capfd): + fake.create_table( + "main.sales.orders", + [("id", "bigint", False), ("amount", "double", True), ("region", "string", True)], + properties={"delta.columnMapping.mode": "name"}, + ) + key = ("schema", fake.schemas["main.sales"]["schema_id"]) + fake.quality_monitors[key] = { + "object_type": "schema", + "object_id": key[1], + "anomaly_detection_config": {"anomaly_detection_workflow_id": 7}, + } + path = write_contract(workspace, _monitored()) + + rc, out = apply_contract(capfd, path) + assert rc == 0, out + assert "adopted" in out + assert CREATE not in fake.requests + assert _monitor(fake)["anomaly_detection_config"]["excluded_table_full_names"] == [ + "main.sales.notes" + ] + assert "tofu plan: +0 ~0 -0" in apply_contract(capfd, path, "--mode", "dry-run")[1] + + +def test_opting_out_deletes_only_the_monitor_and_is_not_gated(workspace, fake, capfd): + assert apply_contract(capfd, write_contract(workspace, _monitored()))[0] == 0 + marker = fake.tables["main.sales.orders"]["marker"] + + rc, out = apply_contract(capfd, write_contract(workspace, _monitored(enabled=False))) + assert rc == 0, out + assert fake.quality_monitors == {} + assert "data quality monitor(s) removed (no table data lost; not gated" in out + assert fake.tables["main.sales.orders"]["marker"] == marker and fake.replaced == [] + assert "data quality monitoring is on" not in out + + +def test_a_destroy_deletes_the_monitor_before_its_schema(workspace, fake, capfd): + assert apply_contract(capfd, write_contract(workspace, _monitored()))[0] == 0 + (workdir,) = (workspace / ".fluid" / "iac" / "databricks").iterdir() + sid = fake.schemas["main.sales"]["schema_id"] + fake.requests.clear() + + result = runner.tofu_destroy(str(workdir)) + assert result.ok, result.stderr or result.stdout + deletes = [path for method, path in fake.requests if method == "DELETE"] + monitor = deletes.index(f"/api/data-quality/v1/monitors/schema/{sid}") + schema = deletes.index("/api/2.1/unity-catalog/schemas/main.sales") + assert monitor < schema + assert fake.quality_monitors == {} and "main.sales" not in fake.schemas diff --git a/tests/iac/test_iac_tofu_validate.py b/tests/iac/test_iac_tofu_validate.py index 834c2a9b..5cac9794 100644 --- a/tests/iac/test_iac_tofu_validate.py +++ b/tests/iac/test_iac_tofu_validate.py @@ -503,6 +503,38 @@ def test_a_databricks_dbt_job_passes_tofu_validate(source, tmp_path, monkeypatch assert validate.returncode == 0, validate.stderr or validate.stdout +@pytest.mark.skipif(_TOFU is None, reason="tofu binary not installed") +def test_a_databricks_data_quality_monitor_passes_tofu_validate(tmp_path, monkeypatch): + """An SLO puts a ``databricks_data_quality_monitor`` on the schema, excluding the + product's other table there: the attributes are the pinned provider's own.""" + monkeypatch.setenv("DATABRICKS_WAREHOUSE_ID", "wh-validate-only") + contract = json.loads(json.dumps(_SAMPLE_CONTRACTS["databricks"])) + orders = contract["exposes"][0] + orders["qos"] = {"freshnessSLO": "PT15M", "completenessTarget": 0.98} + notes = json.loads(json.dumps(orders)) + notes.pop("qos") + notes["exposeId"] = "notes" + notes["binding"]["location"]["table"] = "notes" + contract["exposes"].append(notes) + plugin = IAC_PLUGINS["databricks"] + text = build_module(plugin, contract) + monitors = json.loads(text)["resource"]["databricks_data_quality_monitor"] + assert [m["anomaly_detection_config"] for m in monitors.values()] == [ + {"excluded_table_full_names": ["main.sales.notes"]} + ] + + (tmp_path / "main.tf.json").write_text(text) + write_dependency_lock(plugin, tmp_path) + _tofu_init_or_skip(tmp_path) + validate = subprocess.run( + [_TOFU, "validate", "-no-color"], + cwd=tmp_path, + capture_output=True, + text=True, + ) + assert validate.returncode == 0, validate.stderr or validate.stdout + + # Per-cloud cross-account / cross-project shapes. Stage 1 unit tests # pin the dict; this pins the *.tf.json against the real ``tofu`` # binary — catches any provider-schema mismatch in the new From 6427a44ea6729f146f91b1415bc05410f2e75653 Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Thu, 12 Nov 2026 22:13:57 +0100 Subject: [PATCH 2/6] docs(databricks): changelog and test record for data quality monitoring --- CHANGELOG.md | 19 +++++++++++++++++++ HONESTLY_TESTED.md | 2 ++ 2 files changed, 21 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index ab0be845..7ae4866a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -29,6 +29,25 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 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. +- **Databricks data quality monitoring from a contract's freshness and completeness SLOs.** An + expose bound to Databricks that states a freshness or completeness SLO (`qos.freshnessSLO`, + `qos.completenessTarget`, an `observability.defaultSLIs` freshness or completeness block, or a + `contract.dq.rules` entry of type `freshness`, `completeness` or `anomaly_detection`) now turns + on Databricks data quality monitoring (formerly anomaly detection) for the schema the product + owns: one `databricks_data_quality_monitor` per schema, created after its tables, with the + product's other tables there excluded by name. Catalog Explorer then shows a Quality indicator + on the table, and every scan lands in `system.data_quality_monitoring.table_results`. No new + contract field: `contract.dq.monitoring.enabled: false` opts an expose out. Databricks learns + each table's bounds from its history and takes no threshold, so the SLO values choose the tables + and are not the bounds it checks; `fluid apply` says so, and `fluid validate` warns about a + `dq.monitoring.window` other than one day. A monitor turned on outside forge-cli is adopted, not + created over. Removing one (opting out, dropping the SLOs, removing the expose) deletes its + hidden scan job and what it has learned, never table data, so it is not gated, and is printed. + **Cost:** scans run on serverless compute billed as `DATA_QUALITY_MONITORING`, once when + monitoring is turned on and then as often as each table is updated. Databricks Free Edition runs + it (proven 2026-11-12), and its daily compute quota covers these scans too: a workspace over the + quota stops all compute for the rest of the day, so on Free Edition opt out the products you do + not need watched. ### Fixed diff --git a/HONESTLY_TESTED.md b/HONESTLY_TESTED.md index 09b3e457..31b6b515 100644 --- a/HONESTLY_TESTED.md +++ b/HONESTLY_TESTED.md @@ -91,9 +91,11 @@ Markers declared in `pyproject.toml`. | Stage 1 real tofu, stand-in workspace: dbt jobs | `tests/iac/test_iac_databricks_jobs_fake_e2e.py` | ✅ 6. A local dbt build applies as one serverless job plus the workspace files dbt reads, each in a folder created before it (no upload meets a missing parent) (never `profiles.yml`, `target/` or a hidden file), and a second plan is empty. A changed model updates its file in place. A GIT build runs its tag on a continuous trigger and uploads nothing. Turning a job build local removes the job, its files and their folders without `--allow-data-loss`, and says so; a folder holding a file the apply did not put there is not deleted. The table a dbt build writes keeps what dbt's CREATE OR REPLACE made of it, and the apply reports the column drift instead of failing on a type change. | Whether Databricks accepts and runs the job: serverless environment version 2, the `dbt-databricks>=1.8.0,<2.0.0` range, the commands, a WORKSPACE `project_directory` under `/Workspace`, `run_as` permissions, a Git credential for a private repository, `dbt deps` with a local package (not uploaded). The manual live row proves the first four. Whether the jobs API echoes a server default the stand-in does not (the fake's builder found `performance_target` and an all-false `notification_settings` would each plan an update), which would make a second plan non-empty (live, it was empty). What dbt's CREATE OR REPLACE does to a table with column mapping (the module sets none on a dbt-written table). Not bound by a plan: an apply uploads the dbt project as it is when it runs (`planDigest` and `bundleDigest` cover the contract, not the project's files, as for a local dbt run), and a bundle does not carry the project. Known gap, by design: a job is not adopted. Its import id is a job id, which neither the contract nor `tofu` can look up by name, so an apply that lost its state creates a second job. `fluid verify` flags the duplicate (a stubbed client proves it). Workspace files need no import: the provider's create overwrites. | | Stage 1 unit: Databricks retention (`lifecycle.expire`) | `tests/iac/test_iac_databricks_retention.py`, `tests/cli/test_verify_databricks_retention.py` | ✅ 195 + 18. ISO 8601 durations (years and months at their shortest, P1Y = 365 and P1M = 28 days; a fraction, a sign, a lower-case letter, `PT`, `P1DT`, out-of-order and over-long numbers refused), Delta `interval` properties (rounded up, months refused) and the history as the larger of the two properties with Delta's defaults. A dbt model's `materialized` resolved from `config()` in the SQL, the model's YAML `config:` and `dbt_project.yml` (the `+` prefix or none, root, project, folder and model-name levels, the most specific wins, SQL over YAML over project), the default `view`, the model found by alias or file name, commented-out and `raw` configs ignored, a `config()` inside a Jinja block, `var()`, `**`, a custom schema or database, a Python model, a versioned or twin model, a linked file and a Git repository refused as undeterminable. The build's `--select` must include the resolved model (name, `fqn`, `path:`, `*`, unions and intersections; a graph operator on another node, a tag, a config or run state is refused as not evaluable). Accepted: a dbt `table` model and a DuckDB acquisition with P30D, P365D, P6Y; refused: P7D, P29D, P4W, P1M, a history set longer than the retention, `incremental`, `snapshot`, `view`, another acquisition engine, no writer, a mixed set of writers. `fluid validate` and the emitter agree on kind and message. `fluid verify`'s retention dimension against a stubbed client and, over HTTP, the stand-in workspace (pass, fail with the `ALTER TABLE`, the runtime property over the declared one, skip, predictive optimization off as a warning). | Live behaviour: no real Databricks endpoint was called. Whether Unity Catalog's GetTable serves a table's `delta.logRetentionDuration` and `delta.deletedFileRetentionDuration` in `properties` or `delta_runtime_properties_kvpairs` (read from databricks-sdk-go v0.190.0 `model.go`, not observed), whether `effective_predictive_optimization_flag` is filled for a managed table, that predictive optimization runs `VACUUM` within the retention (a Databricks guarantee, not tested), and that dbt-databricks `table` replaces the table each run. Delta history is only half the promise: the acquisition acceptance rests on the landing being an `INSERT OVERWRITE` of the whole table in the one mode it runs (`full_refresh`, read from the landing's own `LOAD_MODES`; another source mode is refused as `incremental`, none as `unmapped`), which the bronze landing row covers against the stand-in, not live. A GetTable answer with no properties is a verify warning (Delta's defaults assumed). Embedded-logic DuckDB builds, `dlt`, `airbyte` and the other acquisition engines are refused. Selection is matched on names and paths only, not on dbt's graph: `+other+` is refused even if `other` is upstream of the model. | | Stage 1: Databricks Unity Catalog metadata (comments, product tags, column tags; live: the 2026-11-04 row) | `tests/iac/test_iac_databricks_metadata.py` (the emitter), `tests/iac/test_iac_databricks_metadata_fake_e2e.py` (real `fluid` + `tofu` + databricks/databricks 1.140.0 against `tests/iac/_fake_databricks.py`), `tests/cli/test_verify_databricks.py` (the comment and column-tag cases) | ✅ 43 + 5 + 7. The table's `comment` is the expose's description, else the product's (its `purpose` when structured), each described column's `comment` its schema description, stripped; none on a table a dbt build writes (dbt's CREATE OR REPLACE owns them; `fluid validate` warns). The product's identity and owner team are tags on the table only (`fluid_product_id`, `fluid_product_name`, `fluid_domain`, `fluid_layer`, `fluid_product_type`, `fluid_product_version`, `fluid_version`, `fluid_owner_team`), an unstated field makes none, and they count with the governance labels toward 50; `lifecycle.state: deprecated` (the expose's, else the product's) sets `system.certification_status = deprecated`, else the label `certification: certified` (the expose's, else the product's) sets `certified` and is no plain tag (a product-level label certifies every expose's table, as intended; an expose-level one only its own); another label value is `databricks-certification-invalid`; a plan notice says the identity needs ASSIGN on that governed tag. Against the stand-in: certified, then deprecated in place, then removed. A classified column (`sensitivity`) carries `fluid_sensitivity` as a `columns` tag named `catalog.schema.table.column`, imported best-effort by the provider's id. Refused through `validate` and `emit` alike: a backslash in any comment (`databricks-comment-unsafe`; the provider's `parseComment` escapes only quotes) and a `${` (Spark variable substitution), a label meeting a product key (`databricks-tag-collision`), a value over 256 characters (`databricks-tag-value-too-long`), over 1,000 column tags (`databricks-tags-too-many`); a `${...}` in a tag value is rendered as a literal. Against the stand-in: apply writes the comments (an apostrophe included, through the provider's escaping), the product tags and the column tag, never an `OWNER TO`; a second plan is empty; a changed description is an in-place `COMMENT ON TABLE`, never a replace; removing a classification destroys only its column tag; a table made before this is adopted and gains its comments in place (`+5 ~1` for the product tags and the table update). `fluid verify` passes after the apply and fails a comment or column tag changed by hand; against a stubbed client it reads each column's tag list to its last page, ignores a tag or comment someone else added, and checks no comment on a dbt-written table. | **Not run live.** What real Unity Catalog stores and returns for a table or column comment (whitespace, newlines, length limits, which the docs do not state), whether it accepts a tag value with `, . : / = -` (the docs restrict only keys), whether `entity-tag-assignments/columns/{catalog.schema.table.column}` resolves as the SDK documents, and whether a dbt CREATE OR REPLACE keeps column tags (undocumented; `fluid verify` is the gate), and whether the Entity Tag Assignments API sets the system tag `system.certification_status` (its docs show the UI and `SET TAG`; the SDK documents governed tags through the API, needing ASSIGN; the stand-in checks no permission). A contract that adds a column while comments change for the first time hits the provider's mixed-change refusal (its fix-it says to apply in two steps). | +| Stage 1: Databricks data quality monitoring (live: the 2026-11-12 row) | `tests/iac/test_iac_databricks_quality.py` (what asks for it, the monitor per schema, the derived import, the notice, the engine), `tests/iac/test_iac_databricks_quality_fake_e2e.py` (real `fluid` + `tofu` + databricks/databricks 1.140.0 against `tests/iac/_fake_databricks.py`), `tests/iac/test_iac_tofu_validate.py::test_a_databricks_data_quality_monitor_passes_tofu_validate` | ✅ 34 + 4 + 1. A freshness or completeness SLO (`qos.freshnessSLO`, `qos.completenessTarget`, an `observability.defaultSLIs` freshness or completeness block, a `contract.dq.rules` entry of type `freshness`, `completeness` or `anomaly_detection`) puts one `databricks_data_quality_monitor` on the owned schema, watching it by `schema_id`, created after the schema's tables, the product's other tables there in `excluded_table_full_names`; `contract.dq.monitoring.enabled: false` removes only the monitor; a `dq.monitoring.window` other than one day is a `fluid validate` warning. Against the stand-in: a second plan is empty; a monitor turned on outside forge-cli is adopted by `schema,` taken from state after the schema's import (no create); opting out deletes it, not gated; `tofu destroy` deletes it before the schema. | What Databricks scans and reports (the live row); the stand-in serves schema monitors only. | | Manual live, Free Edition (founder-approved) | the real `fluid` CLI, no test file: run 2026-11-01 from main dfd6f145, databricks-sdk 0.158.0, every `fluid_iactest_*` object swept after (1 schema, 1 table, 1 folder; 0 left) | ✅ validate, `generate iac --validate`, plan, dry-run and apply `+9`; a second plan is EMPTY against the real Jobs API; Databricks accepts the Quartz schedule, serverless environment 2 with `dbt-databricks>=1.8.0,<2.0.0` (dbt 1.12.6, adapter 1.13.0), the WORKSPACE `project_directory` and the commands as emitted; one `run-now` ends SUCCESS and dbt builds the table; `fluid verify` with the real SDK (columns, types, one job); `fluid diff`'s state check (its live check has no Databricks inspector); a model change `~1`; the build made local removes job, files and folders `-5` without `--allow-data-loss` (the empty `.fluid//` folders above `files/` stay: only `files/` and below are managed). It found the USE CATALOG check reading a group-conveyed privilege as none. After the fix, `fluid verify` on a fresh one-table product (`+4`) passes every dimension, catalog access included (1 schema and 1 table swept, 0 left). | A GIT source, profile or OAuth auth, a run-as service principal, table adoption and a larger project, live; the region residency check (the metastore precondition and `fluid verify`'s residency dimension), which was added after this run and not run live; dbt's CREATE OR REPLACE drops a contract's NOT NULL, so verify flags a `required` column a dbt model does not constrain itself. | | Manual live, Free Edition (founder-approved), masks and row filters | the real `fluid` CLI, no test file: run 2026-11-03 at the tip of #785, a real workspace with an AWS us-west-2 metastore; every object swept after (0 left) | ✅ 10 steps passed. Functions are created and read back (owner is the user's email; `routine_definition` is the body without `RETURN`, backticks kept; `full_data_type` is STRING or BOOLEAN; a zero-parameter function's FunctionInfo has no `input_params`). The row filter and masks are bound and enforced on the restricted user's own SELECT: only the consented rows, `msisdn` as `XXXXX4567`, a 3-character value hashed, the denied column NULL. `SET ROW FILTER ... ON ()` is accepted and a `SET` replaces a live filter. Tags are present, `fluid verify` passes every dimension, a second apply is `+0 ~0 -0`, the residency precondition fails the plan and names both regions, `generate iac` refuses, and the sweep left 0. | Still open, as this run showed: superseded functions were left behind (orphans; the follow-up that drops them after a successful apply is not run live); `tofu destroy` of the module fails at a schema that still holds functions, so a teardown must drop them first; and the residency failure text was forge-cli's 800-character tail of the tofu JSON, which cut off the sentence naming the live region (the one-line explanation is not run live). A metastore in another cloud or region, a service principal as the run identity, and a group-conveyed grant on the live workspace are still not run. | | Manual live, Free Edition (founder-approved), 2026-11-04 | the real `fluid` CLI, no test file: #801 + #831 at a7169202, run by the demo session, every object swept back to the baseline after | ✅ Unity Catalog metadata read back live: the table and column comments, the product tags (`fluid_product_id` and the rest), the owner-team tag, the `fluid_sensitivity` column tags and `system.certification_status = certified` (accepted for an admin personal access token). The bronze landing: 5 rows loaded, the email column hashed, the staged files deleted from the volume; a second run overwrote the table (the Delta history shows `numRemovedFiles=1`). The column masks, the row filter and a scoped grant were enforced after the `INSERT OVERWRITE`. `lifecycle.retention: P30D` with `expire: true` was accepted, `fluid verify` passed, and predictive optimization was on. A second apply planned `+0 ~0 -0`. A changed row filter bound the new `fluid_rf_` function and dropped the superseded one. Findings fixed in the follow-up PR: the landing left the empty `///` directories and the local masked copy under `.fluid/staging//`, and a dry run did not list the superseded function drops. | Not tested live: `system.certification_status` set by a non-admin identity holding only ASSIGN on the governed tag (only an admin token was used). Open: `facets.databricks_load.rows` counts the rows loaded, which a row-filtered `SELECT COUNT(*)` by a restricted reader does not match (by design of the filter, not reconciled in the run record). | +| Manual live, Free Edition (founder-approved), data quality monitoring, 2026-11-12 | the real `fluid` CLI, no test file: this branch, provider 1.140.0, scratch catalog `fluid_iactest_dqm` (dropped CASCADE after, 0 left) | ✅ Free Edition runs the feature: `POST /api/data-quality/v1/monitors` creates a hidden anomaly-detection workflow whose first run ended SUCCESS after about 8 minutes. `fluid apply` of a two-table product with the pack's smart-meter SLOs: `+14`, a second plan empty; `system.data_quality_monitoring.table_results` then held `readings` as `Training` (freshness and completeness `Unknown`, `RECENTLY_CREATED`) and nothing for the excluded `notes`; Catalog Explorer showed Quality `Training` on `readings` and `Excluded` (Reason: User Configured) on `notes`. A schema whose monitor was made over REST: `adopted 2 pre-existing resource(s)`, then an empty plan. `dq.monitoring.enabled: false`: `-1`, printed as not gated, and the monitor GET is 404. A second create on one schema is `400 RESOURCE_ALREADY_EXISTS`; a missing monitor `404 RESOURCE_DOES_NOT_EXIST`. | A scan beyond `Training` (the model needs days of history); the serverless cost (no `DATA_QUALITY_MONITORING` row in `system.billing.usage` within the session); alerts (`notification_settings` is not in provider 1.140.0). | | Stage 2 emulator (nightly, non-gating) | `tests/iac/test_iac_databricks_minilake.py` (job `nightly-databricks-minilake`) | ⚠️ 1 passed, 4 strict xfails (manual dispatch, run 42267640334, 2026-11-06, minilake v1.9.0). Schema, table and grant creates work, the replace gate holds, adding a column plans no replace, and a grant revoke is classified not gated. The four tests about those resources apply the module without its tag assignments (the `untagged` fixture). An xfail is satisfied only by its own gap's failure (`MinilakeGap`), and a pass fails the run, so an image that closes a gap turns the job red until the mark comes off. Earlier: 1/4 from the first run (2026-10-30) to the 2026-11-04 nightly, then 0/4 after #831, every apply stopping at the first tag. | A second plan is not empty and no in-place table update applies: minilake drops a CREATE TABLE's `TBLPROPERTIES` and `COMMENT`, and DuckDB cannot parse the `ALTER TABLE ... SET TBLPROPERTIES` the provider sends on every update of a table that sets properties. No tag is applied: minilake has no entity-tag-assignments API (it answers 501). v1.9.3 (2026-11-05) has both gaps, so the pin stays; an upstream issue for each is drafted, not filed. Jobs and workspace files are not in this flow yet. | | Stage 3 live | — | ❌ not automated: manual maintainer runs only, never CI. Two have run: the manual rows above (2026-11-01 and 2026-11-03). | Everything the rows above leave to the live tier. | From 2370672b3fba9d07a0588ff62192c2503ae5aa75 Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Thu, 12 Nov 2026 22:15:16 +0100 Subject: [PATCH 3/6] refactor(databricks): monitor plan actions in their own helper, as the job actions are --- fluid_build/iac/providers/databricks.py | 25 +++++++++++++++---------- 1 file changed, 15 insertions(+), 10 deletions(-) diff --git a/fluid_build/iac/providers/databricks.py b/fluid_build/iac/providers/databricks.py index 0b22ae5f..bf323ac0 100644 --- a/fluid_build/iac/providers/databricks.py +++ b/fluid_build/iac/providers/databricks.py @@ -2715,6 +2715,20 @@ def _job_actions(jobs: Sequence[DbtJob]) -> List[Dict[str, Any]]: return actions +def _monitor_actions(monitors: Sequence[SchemaMonitor]) -> List[Dict[str, Any]]: + """One plan action per schema's data quality monitor, keyed by the schema it watches.""" + return [ + { + "op": "databricks.quality_monitor", + "resource_type": QUALITY_MONITOR, + "resource_id": monitor.full_name, + "monitored": list(monitor.monitored), + "excluded": list(monitor.excluded), + } + for monitor in monitors + ] + + def planned_objects(contract: Mapping[str, Any]) -> List[Dict[str, Any]]: """The objects the module manages, as plan actions, for the native planner. @@ -2778,16 +2792,7 @@ def planned_objects(contract: Mapping[str, Any]) -> List[Dict[str, Any]]: "tag_value": tag_value, } ) - for monitor in compiled.monitors: - actions.append( - { - "op": "databricks.quality_monitor", - "resource_type": QUALITY_MONITOR, - "resource_id": monitor.full_name, - "monitored": list(monitor.monitored), - "excluded": list(monitor.excluded), - } - ) + actions.extend(_monitor_actions(compiled.monitors)) actions.extend(_job_actions(compiled.jobs)) unique: Dict[Tuple[str, str], Dict[str, Any]] = {} for action in actions: From d502ebb0ddd25374b8cca84f3428cf8a3acce14c Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Thu, 12 Nov 2026 22:30:09 +0100 Subject: [PATCH 4/6] fix(databricks): a refused monitor names the opt-out; the last exclusion leaving converges Review round 1 of the data quality monitoring change. - When Databricks refuses the monitor (no serverless compute, no MANAGE on the schema), tofu apply fails after the schema's tables exist. The engine's _apply_failure now asks the plugin for hints on the failed resource (optional apply_failure_hints(address, detail), documented in iac/base.py); the Databricks plugin answers for a databricks_data_quality_monitor with what it needs and the opt-out, contract.dq.monitoring.enabled: false. The event stays opentofu_apply_failed. - The last excluded table starting to ask shrinks the exclusion list to nothing. Provider 1.140.0 PATCHes update_mask=anomaly_detection_config, data_profiling_config with an empty config; measured live on Free Edition 2026-11-12, the list is gone and the next plan is empty. The stand-in now honours update_mask, and a test pins the request shape. - The apply notice says Quality indicator, Catalog Explorer's own label, and names the opt-out; the window warning says what to do. - The destroy-order test retries one tofu destroy on a transport error: the stand-in closes every connection (HTTP/1.0), and under load the provider can write to one it closed; state keeps the order. --- CHANGELOG.md | 3 +- HONESTLY_TESTED.md | 4 +- fluid_build/cli/_apply_opentofu_engine.py | 12 ++++-- fluid_build/iac/base.py | 3 ++ fluid_build/iac/providers/databricks.py | 15 ++++++- .../iac/providers/databricks_quality.py | 35 +++++++++++----- tests/iac/_fake_databricks.py | 16 ++++++-- tests/iac/test_iac_databricks_quality.py | 40 +++++++++++++++++++ .../test_iac_databricks_quality_fake_e2e.py | 40 ++++++++++++++++++- 9 files changed, 145 insertions(+), 23 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 7ae4866a..97c47ff4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -41,7 +41,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 each table's bounds from its history and takes no threshold, so the SLO values choose the tables and are not the bounds it checks; `fluid apply` says so, and `fluid validate` warns about a `dq.monitoring.window` other than one day. A monitor turned on outside forge-cli is adopted, not - created over. Removing one (opting out, dropping the SLOs, removing the expose) deletes its + created over. If Databricks refuses the monitor (no serverless compute, or no MANAGE on the + schema), the apply fails after the tables are created and names the opt-out first. Removing one (opting out, dropping the SLOs, removing the expose) deletes its hidden scan job and what it has learned, never table data, so it is not gated, and is printed. **Cost:** scans run on serverless compute billed as `DATA_QUALITY_MONITORING`, once when monitoring is turned on and then as often as each table is updated. Databricks Free Edition runs diff --git a/HONESTLY_TESTED.md b/HONESTLY_TESTED.md index 31b6b515..d63ede54 100644 --- a/HONESTLY_TESTED.md +++ b/HONESTLY_TESTED.md @@ -91,11 +91,11 @@ Markers declared in `pyproject.toml`. | Stage 1 real tofu, stand-in workspace: dbt jobs | `tests/iac/test_iac_databricks_jobs_fake_e2e.py` | ✅ 6. A local dbt build applies as one serverless job plus the workspace files dbt reads, each in a folder created before it (no upload meets a missing parent) (never `profiles.yml`, `target/` or a hidden file), and a second plan is empty. A changed model updates its file in place. A GIT build runs its tag on a continuous trigger and uploads nothing. Turning a job build local removes the job, its files and their folders without `--allow-data-loss`, and says so; a folder holding a file the apply did not put there is not deleted. The table a dbt build writes keeps what dbt's CREATE OR REPLACE made of it, and the apply reports the column drift instead of failing on a type change. | Whether Databricks accepts and runs the job: serverless environment version 2, the `dbt-databricks>=1.8.0,<2.0.0` range, the commands, a WORKSPACE `project_directory` under `/Workspace`, `run_as` permissions, a Git credential for a private repository, `dbt deps` with a local package (not uploaded). The manual live row proves the first four. Whether the jobs API echoes a server default the stand-in does not (the fake's builder found `performance_target` and an all-false `notification_settings` would each plan an update), which would make a second plan non-empty (live, it was empty). What dbt's CREATE OR REPLACE does to a table with column mapping (the module sets none on a dbt-written table). Not bound by a plan: an apply uploads the dbt project as it is when it runs (`planDigest` and `bundleDigest` cover the contract, not the project's files, as for a local dbt run), and a bundle does not carry the project. Known gap, by design: a job is not adopted. Its import id is a job id, which neither the contract nor `tofu` can look up by name, so an apply that lost its state creates a second job. `fluid verify` flags the duplicate (a stubbed client proves it). Workspace files need no import: the provider's create overwrites. | | Stage 1 unit: Databricks retention (`lifecycle.expire`) | `tests/iac/test_iac_databricks_retention.py`, `tests/cli/test_verify_databricks_retention.py` | ✅ 195 + 18. ISO 8601 durations (years and months at their shortest, P1Y = 365 and P1M = 28 days; a fraction, a sign, a lower-case letter, `PT`, `P1DT`, out-of-order and over-long numbers refused), Delta `interval` properties (rounded up, months refused) and the history as the larger of the two properties with Delta's defaults. A dbt model's `materialized` resolved from `config()` in the SQL, the model's YAML `config:` and `dbt_project.yml` (the `+` prefix or none, root, project, folder and model-name levels, the most specific wins, SQL over YAML over project), the default `view`, the model found by alias or file name, commented-out and `raw` configs ignored, a `config()` inside a Jinja block, `var()`, `**`, a custom schema or database, a Python model, a versioned or twin model, a linked file and a Git repository refused as undeterminable. The build's `--select` must include the resolved model (name, `fqn`, `path:`, `*`, unions and intersections; a graph operator on another node, a tag, a config or run state is refused as not evaluable). Accepted: a dbt `table` model and a DuckDB acquisition with P30D, P365D, P6Y; refused: P7D, P29D, P4W, P1M, a history set longer than the retention, `incremental`, `snapshot`, `view`, another acquisition engine, no writer, a mixed set of writers. `fluid validate` and the emitter agree on kind and message. `fluid verify`'s retention dimension against a stubbed client and, over HTTP, the stand-in workspace (pass, fail with the `ALTER TABLE`, the runtime property over the declared one, skip, predictive optimization off as a warning). | Live behaviour: no real Databricks endpoint was called. Whether Unity Catalog's GetTable serves a table's `delta.logRetentionDuration` and `delta.deletedFileRetentionDuration` in `properties` or `delta_runtime_properties_kvpairs` (read from databricks-sdk-go v0.190.0 `model.go`, not observed), whether `effective_predictive_optimization_flag` is filled for a managed table, that predictive optimization runs `VACUUM` within the retention (a Databricks guarantee, not tested), and that dbt-databricks `table` replaces the table each run. Delta history is only half the promise: the acquisition acceptance rests on the landing being an `INSERT OVERWRITE` of the whole table in the one mode it runs (`full_refresh`, read from the landing's own `LOAD_MODES`; another source mode is refused as `incremental`, none as `unmapped`), which the bronze landing row covers against the stand-in, not live. A GetTable answer with no properties is a verify warning (Delta's defaults assumed). Embedded-logic DuckDB builds, `dlt`, `airbyte` and the other acquisition engines are refused. Selection is matched on names and paths only, not on dbt's graph: `+other+` is refused even if `other` is upstream of the model. | | Stage 1: Databricks Unity Catalog metadata (comments, product tags, column tags; live: the 2026-11-04 row) | `tests/iac/test_iac_databricks_metadata.py` (the emitter), `tests/iac/test_iac_databricks_metadata_fake_e2e.py` (real `fluid` + `tofu` + databricks/databricks 1.140.0 against `tests/iac/_fake_databricks.py`), `tests/cli/test_verify_databricks.py` (the comment and column-tag cases) | ✅ 43 + 5 + 7. The table's `comment` is the expose's description, else the product's (its `purpose` when structured), each described column's `comment` its schema description, stripped; none on a table a dbt build writes (dbt's CREATE OR REPLACE owns them; `fluid validate` warns). The product's identity and owner team are tags on the table only (`fluid_product_id`, `fluid_product_name`, `fluid_domain`, `fluid_layer`, `fluid_product_type`, `fluid_product_version`, `fluid_version`, `fluid_owner_team`), an unstated field makes none, and they count with the governance labels toward 50; `lifecycle.state: deprecated` (the expose's, else the product's) sets `system.certification_status = deprecated`, else the label `certification: certified` (the expose's, else the product's) sets `certified` and is no plain tag (a product-level label certifies every expose's table, as intended; an expose-level one only its own); another label value is `databricks-certification-invalid`; a plan notice says the identity needs ASSIGN on that governed tag. Against the stand-in: certified, then deprecated in place, then removed. A classified column (`sensitivity`) carries `fluid_sensitivity` as a `columns` tag named `catalog.schema.table.column`, imported best-effort by the provider's id. Refused through `validate` and `emit` alike: a backslash in any comment (`databricks-comment-unsafe`; the provider's `parseComment` escapes only quotes) and a `${` (Spark variable substitution), a label meeting a product key (`databricks-tag-collision`), a value over 256 characters (`databricks-tag-value-too-long`), over 1,000 column tags (`databricks-tags-too-many`); a `${...}` in a tag value is rendered as a literal. Against the stand-in: apply writes the comments (an apostrophe included, through the provider's escaping), the product tags and the column tag, never an `OWNER TO`; a second plan is empty; a changed description is an in-place `COMMENT ON TABLE`, never a replace; removing a classification destroys only its column tag; a table made before this is adopted and gains its comments in place (`+5 ~1` for the product tags and the table update). `fluid verify` passes after the apply and fails a comment or column tag changed by hand; against a stubbed client it reads each column's tag list to its last page, ignores a tag or comment someone else added, and checks no comment on a dbt-written table. | **Not run live.** What real Unity Catalog stores and returns for a table or column comment (whitespace, newlines, length limits, which the docs do not state), whether it accepts a tag value with `, . : / = -` (the docs restrict only keys), whether `entity-tag-assignments/columns/{catalog.schema.table.column}` resolves as the SDK documents, and whether a dbt CREATE OR REPLACE keeps column tags (undocumented; `fluid verify` is the gate), and whether the Entity Tag Assignments API sets the system tag `system.certification_status` (its docs show the UI and `SET TAG`; the SDK documents governed tags through the API, needing ASSIGN; the stand-in checks no permission). A contract that adds a column while comments change for the first time hits the provider's mixed-change refusal (its fix-it says to apply in two steps). | -| Stage 1: Databricks data quality monitoring (live: the 2026-11-12 row) | `tests/iac/test_iac_databricks_quality.py` (what asks for it, the monitor per schema, the derived import, the notice, the engine), `tests/iac/test_iac_databricks_quality_fake_e2e.py` (real `fluid` + `tofu` + databricks/databricks 1.140.0 against `tests/iac/_fake_databricks.py`), `tests/iac/test_iac_tofu_validate.py::test_a_databricks_data_quality_monitor_passes_tofu_validate` | ✅ 34 + 4 + 1. A freshness or completeness SLO (`qos.freshnessSLO`, `qos.completenessTarget`, an `observability.defaultSLIs` freshness or completeness block, a `contract.dq.rules` entry of type `freshness`, `completeness` or `anomaly_detection`) puts one `databricks_data_quality_monitor` on the owned schema, watching it by `schema_id`, created after the schema's tables, the product's other tables there in `excluded_table_full_names`; `contract.dq.monitoring.enabled: false` removes only the monitor; a `dq.monitoring.window` other than one day is a `fluid validate` warning. Against the stand-in: a second plan is empty; a monitor turned on outside forge-cli is adopted by `schema,` taken from state after the schema's import (no create); opting out deletes it, not gated; `tofu destroy` deletes it before the schema. | What Databricks scans and reports (the live row); the stand-in serves schema monitors only. | +| Stage 1: Databricks data quality monitoring (live: the 2026-11-12 row) | `tests/iac/test_iac_databricks_quality.py` (what asks for it, the monitor per schema, the derived import, the notice, the engine), `tests/iac/test_iac_databricks_quality_fake_e2e.py` (real `fluid` + `tofu` + databricks/databricks 1.140.0 against `tests/iac/_fake_databricks.py`), `tests/iac/test_iac_tofu_validate.py::test_a_databricks_data_quality_monitor_passes_tofu_validate` | ✅ 37 + 6 + 1. A freshness or completeness SLO (`qos.freshnessSLO`, `qos.completenessTarget`, an `observability.defaultSLIs` freshness or completeness block, a `contract.dq.rules` entry of type `freshness`, `completeness` or `anomaly_detection`) puts one `databricks_data_quality_monitor` on the owned schema, watching it by `schema_id`, created after the schema's tables, the product's other tables there in `excluded_table_full_names`; `contract.dq.monitoring.enabled: false` removes only the monitor; a `dq.monitoring.window` other than one day is a `fluid validate` warning. Against the stand-in: a second plan is empty; a monitor turned on outside forge-cli is adopted by `schema,` taken from state after the schema's import (no create); the last exclusion leaving is a PATCH whose `update_mask` names the whole config (the stand-in honours the mask), and the plan converges; a monitor Databricks refuses fails the apply after the tables, with the opt-out first among the suggestions (`apply_failure_hints`); opting out deletes it, not gated; `tofu destroy` deletes it before the schema. | What Databricks scans and reports (the live row); the stand-in serves schema monitors only. | | Manual live, Free Edition (founder-approved) | the real `fluid` CLI, no test file: run 2026-11-01 from main dfd6f145, databricks-sdk 0.158.0, every `fluid_iactest_*` object swept after (1 schema, 1 table, 1 folder; 0 left) | ✅ validate, `generate iac --validate`, plan, dry-run and apply `+9`; a second plan is EMPTY against the real Jobs API; Databricks accepts the Quartz schedule, serverless environment 2 with `dbt-databricks>=1.8.0,<2.0.0` (dbt 1.12.6, adapter 1.13.0), the WORKSPACE `project_directory` and the commands as emitted; one `run-now` ends SUCCESS and dbt builds the table; `fluid verify` with the real SDK (columns, types, one job); `fluid diff`'s state check (its live check has no Databricks inspector); a model change `~1`; the build made local removes job, files and folders `-5` without `--allow-data-loss` (the empty `.fluid//` folders above `files/` stay: only `files/` and below are managed). It found the USE CATALOG check reading a group-conveyed privilege as none. After the fix, `fluid verify` on a fresh one-table product (`+4`) passes every dimension, catalog access included (1 schema and 1 table swept, 0 left). | A GIT source, profile or OAuth auth, a run-as service principal, table adoption and a larger project, live; the region residency check (the metastore precondition and `fluid verify`'s residency dimension), which was added after this run and not run live; dbt's CREATE OR REPLACE drops a contract's NOT NULL, so verify flags a `required` column a dbt model does not constrain itself. | | Manual live, Free Edition (founder-approved), masks and row filters | the real `fluid` CLI, no test file: run 2026-11-03 at the tip of #785, a real workspace with an AWS us-west-2 metastore; every object swept after (0 left) | ✅ 10 steps passed. Functions are created and read back (owner is the user's email; `routine_definition` is the body without `RETURN`, backticks kept; `full_data_type` is STRING or BOOLEAN; a zero-parameter function's FunctionInfo has no `input_params`). The row filter and masks are bound and enforced on the restricted user's own SELECT: only the consented rows, `msisdn` as `XXXXX4567`, a 3-character value hashed, the denied column NULL. `SET ROW FILTER ... ON ()` is accepted and a `SET` replaces a live filter. Tags are present, `fluid verify` passes every dimension, a second apply is `+0 ~0 -0`, the residency precondition fails the plan and names both regions, `generate iac` refuses, and the sweep left 0. | Still open, as this run showed: superseded functions were left behind (orphans; the follow-up that drops them after a successful apply is not run live); `tofu destroy` of the module fails at a schema that still holds functions, so a teardown must drop them first; and the residency failure text was forge-cli's 800-character tail of the tofu JSON, which cut off the sentence naming the live region (the one-line explanation is not run live). A metastore in another cloud or region, a service principal as the run identity, and a group-conveyed grant on the live workspace are still not run. | | Manual live, Free Edition (founder-approved), 2026-11-04 | the real `fluid` CLI, no test file: #801 + #831 at a7169202, run by the demo session, every object swept back to the baseline after | ✅ Unity Catalog metadata read back live: the table and column comments, the product tags (`fluid_product_id` and the rest), the owner-team tag, the `fluid_sensitivity` column tags and `system.certification_status = certified` (accepted for an admin personal access token). The bronze landing: 5 rows loaded, the email column hashed, the staged files deleted from the volume; a second run overwrote the table (the Delta history shows `numRemovedFiles=1`). The column masks, the row filter and a scoped grant were enforced after the `INSERT OVERWRITE`. `lifecycle.retention: P30D` with `expire: true` was accepted, `fluid verify` passed, and predictive optimization was on. A second apply planned `+0 ~0 -0`. A changed row filter bound the new `fluid_rf_` function and dropped the superseded one. Findings fixed in the follow-up PR: the landing left the empty `///` directories and the local masked copy under `.fluid/staging//`, and a dry run did not list the superseded function drops. | Not tested live: `system.certification_status` set by a non-admin identity holding only ASSIGN on the governed tag (only an admin token was used). Open: `facets.databricks_load.rows` counts the rows loaded, which a row-filtered `SELECT COUNT(*)` by a restricted reader does not match (by design of the filter, not reconciled in the run record). | -| Manual live, Free Edition (founder-approved), data quality monitoring, 2026-11-12 | the real `fluid` CLI, no test file: this branch, provider 1.140.0, scratch catalog `fluid_iactest_dqm` (dropped CASCADE after, 0 left) | ✅ Free Edition runs the feature: `POST /api/data-quality/v1/monitors` creates a hidden anomaly-detection workflow whose first run ended SUCCESS after about 8 minutes. `fluid apply` of a two-table product with the pack's smart-meter SLOs: `+14`, a second plan empty; `system.data_quality_monitoring.table_results` then held `readings` as `Training` (freshness and completeness `Unknown`, `RECENTLY_CREATED`) and nothing for the excluded `notes`; Catalog Explorer showed Quality `Training` on `readings` and `Excluded` (Reason: User Configured) on `notes`. A schema whose monitor was made over REST: `adopted 2 pre-existing resource(s)`, then an empty plan. `dq.monitoring.enabled: false`: `-1`, printed as not gated, and the monitor GET is 404. A second create on one schema is `400 RESOURCE_ALREADY_EXISTS`; a missing monitor `404 RESOURCE_DOES_NOT_EXIST`. | A scan beyond `Training` (the model needs days of history); the serverless cost (no `DATA_QUALITY_MONITORING` row in `system.billing.usage` within the session); alerts (`notification_settings` is not in provider 1.140.0). | +| Manual live, Free Edition (founder-approved), data quality monitoring, 2026-11-12 | the real `fluid` CLI, no test file: this branch, provider 1.140.0, scratch catalog `fluid_iactest_dqm` (dropped CASCADE after, 0 left) | ✅ Free Edition runs the feature: `POST /api/data-quality/v1/monitors` creates a hidden anomaly-detection workflow whose first run ended SUCCESS after about 8 minutes. `fluid apply` of a two-table product with the pack's smart-meter SLOs: `+14`, a second plan empty; `system.data_quality_monitoring.table_results` then held `readings` as `Training` (freshness and completeness `Unknown`, `RECENTLY_CREATED`) and nothing for the excluded `notes`; Catalog Explorer showed Quality `Training` on `readings` and `Excluded` (Reason: User Configured) on `notes`. A schema whose monitor was made over REST: `adopted 2 pre-existing resource(s)`, then an empty plan. `dq.monitoring.enabled: false`: `-1`, printed as not gated, and the monitor GET is 404. The last excluded table starting to ask: `~1` (provider 1.140.0 PATCHes with `update_mask=anomaly_detection_config,data_profiling_config` and an empty config), the live monitor's exclusion list is gone, and the next plan is empty. A second create on one schema is `400 RESOURCE_ALREADY_EXISTS`; a missing monitor `404 RESOURCE_DOES_NOT_EXIST`. | A scan beyond `Training` (the model needs days of history); the serverless cost (no `DATA_QUALITY_MONITORING` row in `system.billing.usage` within the session: its rows lagged the scans by hours); a workspace or identity Databricks refuses (the stand-in row); alerts (`notification_settings` is not in provider 1.140.0). | | Stage 2 emulator (nightly, non-gating) | `tests/iac/test_iac_databricks_minilake.py` (job `nightly-databricks-minilake`) | ⚠️ 1 passed, 4 strict xfails (manual dispatch, run 42267640334, 2026-11-06, minilake v1.9.0). Schema, table and grant creates work, the replace gate holds, adding a column plans no replace, and a grant revoke is classified not gated. The four tests about those resources apply the module without its tag assignments (the `untagged` fixture). An xfail is satisfied only by its own gap's failure (`MinilakeGap`), and a pass fails the run, so an image that closes a gap turns the job red until the mark comes off. Earlier: 1/4 from the first run (2026-10-30) to the 2026-11-04 nightly, then 0/4 after #831, every apply stopping at the first tag. | A second plan is not empty and no in-place table update applies: minilake drops a CREATE TABLE's `TBLPROPERTIES` and `COMMENT`, and DuckDB cannot parse the `ALTER TABLE ... SET TBLPROPERTIES` the provider sends on every update of a table that sets properties. No tag is applied: minilake has no entity-tag-assignments API (it answers 501). v1.9.3 (2026-11-05) has both gaps, so the pin stays; an upstream issue for each is drafted, not filed. Jobs and workspace files are not in this flow yet. | | Stage 3 live | — | ❌ not automated: manual maintainer runs only, never CI. Two have run: the manual rows above (2026-11-01 and 2026-11-03). | Everything the rows above leave to the live tier. | diff --git a/fluid_build/cli/_apply_opentofu_engine.py b/fluid_build/cli/_apply_opentofu_engine.py index 47e5fdd3..c5f5411b 100644 --- a/fluid_build/cli/_apply_opentofu_engine.py +++ b/fluid_build/cli/_apply_opentofu_engine.py @@ -316,7 +316,7 @@ def _apply_via_opentofu(args, logger: logging.Logger, contract: Optional[Mapping "opentofu apply raw output tail: %s", _tail(apply_result.stderr or apply_result.stdout), ) - raise _apply_failure(apply_result, str(workdir)) + raise _apply_failure(apply_result, str(workdir), plugin) applied = runner.change_summary(apply_result) record(applied) @@ -1630,7 +1630,7 @@ def _mentions_state_lock(text: str) -> bool: return "state lock" in low or "locking state" in low -def _apply_failure(result: Any, workdir: str) -> CLIError: +def _apply_failure(result: Any, workdir: str, plugin: Any = None) -> CLIError: """The error for a failed ``tofu apply``: what failed, on which resource, and the way on. ``-json`` puts every diagnostic on stdout as one JSON line, so the raw tail @@ -1638,11 +1638,14 @@ def _apply_failure(result: Any, workdir: str) -> CLIError: a summary, the resource address and the provider's detail; output that holds none (a crash, a timeout) falls back to the first ``Error:``, else the exit code. Tofu records state as it goes, so what was created before the failure - is normally tracked and a re-run converges: the suggestions say so. + is normally tracked and a re-run converges: the suggestions say so. A plugin's + optional ``apply_failure_hints(address, detail)`` puts its own way on first, for the + resource that failed. """ diagnostics = _diagnostics(getattr(result, "events", None)) context: Dict[str, Any] = {} locked = False + address = detail = "" if diagnostics: first = diagnostics[0] # Redacted before it is cut: a credential split at the limit (a URL that lost @@ -1668,6 +1671,9 @@ def _apply_failure(result: Any, workdir: str) -> CLIError: locked = _mentions_state_lock(output) err = CLIError(1, "opentofu_apply_failed", context) hints = [f"To see what exists, run 'tofu state list' in {shlex.quote(workdir)}"] + plugin_hints = getattr(plugin, "apply_failure_hints", None) + if address and callable(plugin_hints): + hints = list(plugin_hints(address, detail)) + hints if locked: hints.append("The state is locked: wait for the other run, or 'tofu force-unlock '") err.suggestions = list(err.suggestions) + hints diff --git a/fluid_build/iac/base.py b/fluid_build/iac/base.py index a693a9a9..70a9744a 100644 --- a/fluid_build/iac/base.py +++ b/fluid_build/iac/base.py @@ -66,6 +66,9 @@ class IacProviderPlugin(Protocol): * ``explain_plan_failure(output, contract, state)``: an :class:`UnsupportedBindingError` naming the fix for a failed ``tofu`` run, or ``None``. + * ``apply_failure_hints(address, detail)``: suggestions, first in line, for + a ``tofu apply`` that failed on the resource at ``address``; ``[]`` when + the plugin has none. * ``recorded_regions(state)``: the regions the state's resources live in, so an apply pinned to another region is refused, not re-created there. * ``suppressed_drift(contract, prior_state)``: drift the module hides with diff --git a/fluid_build/iac/providers/databricks.py b/fluid_build/iac/providers/databricks.py index bf323ac0..5cbd45b3 100644 --- a/fluid_build/iac/providers/databricks.py +++ b/fluid_build/iac/providers/databricks.py @@ -95,7 +95,7 @@ excluded, ``contract.dq.monitoring.enabled: false`` to opt out (``databricks_quality.py``). Databricks learns each table's bounds itself: the SLO values pick the tables and are not thresholds the monitor checks. Its results are the - health indicator on the table in Catalog Explorer and + Quality indicator on the table in Catalog Explorer and ``system.data_quality_monitoring.table_results``; removing it is not gated. * A dbt build that runs on Databricks (``execution.runtime.platform: databricks``) becomes a serverless ``databricks_job`` with a ``dbt_task``, and @@ -216,7 +216,13 @@ ) from .databricks_keys import lookup as lookup_key_tables from .databricks_quality import RESOURCE_TYPE as QUALITY_MONITOR -from .databricks_quality import SchemaMonitor, monitor_notices, plan_monitors, window_warnings +from .databricks_quality import ( + SchemaMonitor, + failure_hints, + monitor_notices, + plan_monitors, + window_warnings, +) from .databricks_retention import Retention, check_expire _LOG = logging.getLogger(__name__) @@ -2512,6 +2518,11 @@ def plan_notices(self, contract: Mapping[str, Any]) -> List[str]: ) return notices + def apply_failure_hints(self, address: str, detail: str = "") -> List[str]: + """The way on when ``tofu apply`` failed on ``address``: a data quality monitor + Databricks refused (``databricks_quality.failure_hints``); ``[]`` otherwise.""" + return failure_hints(address) + def explain_plan_failure( self, output: str, diff --git a/fluid_build/iac/providers/databricks_quality.py b/fluid_build/iac/providers/databricks_quality.py index 7cddabf0..73754acf 100644 --- a/fluid_build/iac/providers/databricks_quality.py +++ b/fluid_build/iac/providers/databricks_quality.py @@ -7,9 +7,9 @@ Catalog schema. On serverless compute it scans the schema's tables for freshness (a commit unusually late for the table) and completeness (fewer rows in the last 24 hours than the table's history predicts), writes one row per table and scan to -``system.data_quality_monitoring.table_results`` and shows the result as the health -indicator on each table's Overview in Catalog Explorer -(https://docs.databricks.com/aws/en/data-quality-monitoring/anomaly-detection/). +``system.data_quality_monitoring.table_results`` and shows the result as the Quality +indicator on each table's Overview in Catalog Explorer (the docs' "health indicator", +https://docs.databricks.com/aws/en/data-quality-monitoring/anomaly-detection/). An expose asks for it with fields it already has, no new one: a freshness or completeness SLO (``qos.freshnessSLO``, ``qos.completenessTarget``, an ``observability.defaultSLIs`` @@ -157,7 +157,7 @@ def window_warnings(tables: Sequence[MonitoredTable]) -> List[str]: warnings.append( f"{table.where}: contract.dq.monitoring.window {window} is not applied. Databricks " "data quality monitoring counts a table's rows over the last 24 hours and judges " - "its freshness against the table's own commit history." + "its freshness against the table's own commit history. Set it to P1D, or remove it." ) return warnings @@ -171,19 +171,36 @@ def monitor_notices(monitors: Sequence[SchemaMonitor]) -> List[str]: notices.append( f"data quality monitoring is on for schema {monitor.full_name} (watched: " f"{', '.join(monitor.monitored)}{skipped}). Databricks learns each table's freshness " - "and completeness bounds from its history, not from the contract's SLO values, " - "scans on serverless compute billed as DATA_QUALITY_MONITORING, and shows the " - "result as the table's health indicator in Catalog Explorer. " - "system.data_quality_monitoring.table_results holds every scan; only account " - "admins can read it unless they grant SELECT." + "and completeness bounds from its history, not from the contract's SLO values; it " + "scans on serverless compute billed as DATA_QUALITY_MONITORING and shows the result " + "as the table's Quality indicator in Catalog Explorer. " + "system.data_quality_monitoring.table_results holds every scan (account admins " + "only, unless they grant SELECT). contract.dq.monitoring.enabled: false turns it " + "off for an expose." ) return notices +def failure_hints(address: str) -> List[str]: + """The way on when ``tofu apply`` failed creating or changing the monitor at + ``address``; ``[]`` for any other resource. The monitor is created after the schema's + tables (``depends_on``), so they exist by then.""" + if not address.startswith(f"{RESOURCE_TYPE}."): + return [] + return [ + "Databricks did not turn on data quality monitoring for this schema (its tables were " + "created first). It needs a Unity Catalog workspace with serverless compute, and " + "MANAGE on the schema for this identity (the schema's owner has it).", + "Or turn it off for the expose: set contract.dq.monitoring.enabled: false, then " + "re-run fluid apply.", + ] + + __all__ = [ "RESOURCE_TYPE", "MonitoredTable", "SchemaMonitor", + "failure_hints", "monitor_notices", "monitoring_request", "plan_monitors", diff --git a/tests/iac/_fake_databricks.py b/tests/iac/_fake_databricks.py index f6b6762c..dfc83164 100644 --- a/tests/iac/_fake_databricks.py +++ b/tests/iac/_fake_databricks.py @@ -1362,12 +1362,20 @@ def _dqm_update(self, m: Any, q: Any, b: Any) -> Tuple[int, Any]: key = (m.group(1), m.group(2)) if key not in self.quality_monitors: return self._dqm_missing(m.group(2)) + # update_mask semantics: each masked field is replaced by the body's value, and one + # the body leaves out is cleared; what the mask does not name is kept. Provider 1.140.0 + # sends update_mask=anomaly_detection_config,data_profiling_config, the whole config. config = self.quality_monitors[key]["anomaly_detection_config"] sent = b.get("anomaly_detection_config") or {} - if "excluded_table_full_names" in sent: - config["excluded_table_full_names"] = list(sent["excluded_table_full_names"]) - else: - config.pop("excluded_table_full_names", None) + for path in ",".join(q.get("update_mask") or []).split(","): + head, _, leaf = path.strip().partition(".") + if head != "anomaly_detection_config": + continue + for name in [leaf] if leaf else ["excluded_table_full_names"]: + if name in sent: + config[name] = list(sent[name]) + else: + config.pop(name, None) return 200, self._dqm_doc(key) def _dqm_delete(self, m: Any, q: Any, b: Any) -> Tuple[int, Any]: diff --git a/tests/iac/test_iac_databricks_quality.py b/tests/iac/test_iac_databricks_quality.py index 68125eb4..8f16523a 100644 --- a/tests/iac/test_iac_databricks_quality.py +++ b/tests/iac/test_iac_databricks_quality.py @@ -197,6 +197,7 @@ def test_another_window_is_not_applied(self): expose = {**ASKS, "contract": _dq(monitoring={"window": "PT1H"})} (warning,) = dq.window_warnings([_T("main.iot.r", expose)]) assert "contract.dq.monitoring.window PT1H is not applied" in warning + assert warning.endswith("Set it to P1D, or remove it.") @pytest.mark.parametrize( "expose", @@ -297,6 +298,9 @@ def test_plan_names_the_monitor_and_the_apply_says_what_it_does(self): assert "not from the contract's SLO values" in notice assert "DATA_QUALITY_MONITORING" in notice assert "system.data_quality_monitoring.table_results" in notice + # Catalog Explorer's own label, and the way out of the cost + assert "the table's Quality indicator in Catalog Explorer" in notice + assert notice.endswith("contract.dq.monitoring.enabled: false turns it off for an expose.") def test_validate_warns_about_a_window_it_cannot_apply(self): contract = _contract( @@ -379,3 +383,39 @@ def fake_import(workdir, address, rid, env=None): ) assert f"{MONITOR}.{KEY}" not in calls assert f"databricks_schema.{KEY}" in calls + + +# ── a monitor Databricks refuses ──────────────────────────────────────────── +class TestApplyFailure: + @staticmethod + def _failed(address: str) -> TofuResult: + events = [ + { + "diagnostic": { + "severity": "error", + "summary": "failed to create data_quality_monitor", + "detail": "PERMISSION_DENIED: User does not have MANAGE on Schema", + "address": address, + } + } + ] + return TofuResult("apply", 1, "", "", events) + + def test_a_refused_monitor_names_the_opt_out_first(self): + err = engine._apply_failure( + self._failed(f"{MONITOR}.{KEY}"), "/w", IAC_PLUGINS["databricks"] + ) + assert err.event == "opentofu_apply_failed" + assert err.context["resource"] == f"{MONITOR}.{KEY}" + first, second, last = err.suggestions[-3:] + assert first.startswith("Databricks did not turn on data quality monitoring") + assert "MANAGE on the schema" in first + assert "contract.dq.monitoring.enabled: false" in second + assert last.startswith("To see what exists") + + @pytest.mark.parametrize("plugin", [None, IAC_PLUGINS["databricks"]]) + def test_any_other_failure_is_reported_as_before(self, plugin): + address = f"databricks_sql_table.{KEY}_readings" + err = engine._apply_failure(self._failed(address), "/w", plugin) + assert not any("data quality" in hint for hint in err.suggestions) + assert dq.failure_hints(address) == [] diff --git a/tests/iac/test_iac_databricks_quality_fake_e2e.py b/tests/iac/test_iac_databricks_quality_fake_e2e.py index f98fcf92..3781943d 100644 --- a/tests/iac/test_iac_databricks_quality_fake_e2e.py +++ b/tests/iac/test_iac_databricks_quality_fake_e2e.py @@ -8,8 +8,10 @@ ``_fake_databricks.py``, its data quality endpoints shaped as Free Edition answered them): an expose with a freshness SLO turns monitoring on for its schema, watching it by the schema's ``schema_id`` and excluding the product's other table there; a second plan is empty; a monitor -someone turned on outside forge-cli is adopted, not created over; opting out deletes only the -monitor, not gated; and OpenTofu deletes the monitor before its schema. +someone turned on outside forge-cli is adopted, not created over; the last exclusion leaving +clears the list and the plan converges; a monitor Databricks refuses fails the apply with the +opt-out named, its tables kept; opting out deletes only the monitor, not gated; and OpenTofu +deletes the monitor before its schema. Not proven here: what real Databricks scans and reports (the live Free Edition run, HONESTLY_TESTED.md). @@ -18,6 +20,7 @@ from __future__ import annotations import copy +import re from pathlib import Path from typing import Any, Dict, Iterator @@ -42,6 +45,7 @@ ] CREATE = ("POST", "/api/data-quality/v1/monitors") +_TRANSPORT = re.compile(r"broken pipe|connection reset by peer") @pytest.fixture @@ -129,6 +133,34 @@ def test_a_monitor_turned_on_outside_forge_cli_is_adopted_not_created_over(works assert "tofu plan: +0 ~0 -0" in apply_contract(capfd, path, "--mode", "dry-run")[1] +def test_the_last_exclusion_leaving_clears_the_list_and_the_plan_converges(workspace, fake, capfd): + assert apply_contract(capfd, write_contract(workspace, _monitored()))[0] == 0 + both = _monitored() + both["exposes"][1]["qos"] = {"freshnessSLO": "PT1H"} + path = write_contract(workspace, both) + fake.queries.clear() + + rc, out = apply_contract(capfd, path) + assert rc == 0, out + assert "tofu plan: +0 ~1 -0" in out + # provider 1.140.0 replaces the whole config: the mask names it, the body has no list + (patch,) = [query for method, _path, query in fake.queries if method == "PATCH"] + assert "anomaly_detection_config" in patch + assert "excluded_table_full_names" not in _monitor(fake)["anomaly_detection_config"] + assert "tofu plan: +0 ~0 -0" in apply_contract(capfd, path, "--mode", "dry-run")[1] + + +def test_a_refused_monitor_fails_the_apply_with_the_way_out(workspace, fake, capfd): + fake.fail_route("POST", r"^/api/data-quality/v1/monitors$", 400, "feature not enabled") + + rc, out = apply_contract(capfd, write_contract(workspace, _monitored())) + assert rc != 0 + assert "Databricks did not turn on data quality monitoring for this schema" in out + assert "set contract.dq.monitoring.enabled: false, then re-run fluid apply" in out + # created first, and kept: a re-run converges + assert {"main.sales.orders", "main.sales.notes"} <= set(fake.tables) + + def test_opting_out_deletes_only_the_monitor_and_is_not_gated(workspace, fake, capfd): assert apply_contract(capfd, write_contract(workspace, _monitored()))[0] == 0 marker = fake.tables["main.sales.orders"]["marker"] @@ -148,6 +180,10 @@ def test_a_destroy_deletes_the_monitor_before_its_schema(workspace, fake, capfd) fake.requests.clear() result = runner.tofu_destroy(str(workdir)) + if not result.ok and _TRANSPORT.search(result.stdout + result.stderr): + # The stand-in closes every connection (HTTP/1.0); under load the provider can + # write to one it closed. State keeps what went, so the order holds across runs. + result = runner.tofu_destroy(str(workdir)) assert result.ok, result.stderr or result.stdout deletes = [path for method, path in fake.requests if method == "DELETE"] monitor = deletes.index(f"/api/data-quality/v1/monitors/schema/{sid}") From ebb8c343eb1634e173ec6c969ee90c2b02879bb2 Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Thu, 12 Nov 2026 22:49:49 +0100 Subject: [PATCH 5/6] fix(databricks): the refused-monitor hint reads right for an update too; changelog rewrapped --- CHANGELOG.md | 34 +++++++++---------- .../iac/providers/databricks_quality.py | 6 ++-- tests/iac/test_iac_databricks_quality.py | 2 +- .../test_iac_databricks_quality_fake_e2e.py | 2 +- 4 files changed, 22 insertions(+), 22 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 97c47ff4..fb9fa13d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -32,23 +32,23 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - **Databricks data quality monitoring from a contract's freshness and completeness SLOs.** An expose bound to Databricks that states a freshness or completeness SLO (`qos.freshnessSLO`, `qos.completenessTarget`, an `observability.defaultSLIs` freshness or completeness block, or a - `contract.dq.rules` entry of type `freshness`, `completeness` or `anomaly_detection`) now turns - on Databricks data quality monitoring (formerly anomaly detection) for the schema the product - owns: one `databricks_data_quality_monitor` per schema, created after its tables, with the - product's other tables there excluded by name. Catalog Explorer then shows a Quality indicator - on the table, and every scan lands in `system.data_quality_monitoring.table_results`. No new - contract field: `contract.dq.monitoring.enabled: false` opts an expose out. Databricks learns - each table's bounds from its history and takes no threshold, so the SLO values choose the tables - and are not the bounds it checks; `fluid apply` says so, and `fluid validate` warns about a - `dq.monitoring.window` other than one day. A monitor turned on outside forge-cli is adopted, not - created over. If Databricks refuses the monitor (no serverless compute, or no MANAGE on the - schema), the apply fails after the tables are created and names the opt-out first. Removing one (opting out, dropping the SLOs, removing the expose) deletes its - hidden scan job and what it has learned, never table data, so it is not gated, and is printed. - **Cost:** scans run on serverless compute billed as `DATA_QUALITY_MONITORING`, once when - monitoring is turned on and then as often as each table is updated. Databricks Free Edition runs - it (proven 2026-11-12), and its daily compute quota covers these scans too: a workspace over the - quota stops all compute for the rest of the day, so on Free Edition opt out the products you do - not need watched. + `contract.dq.rules` entry of type `freshness`, `completeness` or `anomaly_detection`) now turns on + Databricks data quality monitoring (formerly anomaly detection) for the schema the product owns: + one `databricks_data_quality_monitor` per schema, created after its tables, with the product's + other tables there excluded by name. Catalog Explorer then shows a Quality indicator on the table, + and every scan lands in `system.data_quality_monitoring.table_results`. No new contract field: + `contract.dq.monitoring.enabled: false` opts an expose out. Databricks learns each table's bounds + from its history and takes no threshold, so the SLO values choose the tables and are not the + bounds it checks; `fluid apply` says so, and `fluid validate` warns about a `dq.monitoring.window` + other than one day. A monitor turned on outside forge-cli is adopted, not created over. If + Databricks refuses the monitor (no serverless compute, or no MANAGE on the schema), the apply + fails after the tables are created and names the opt-out first. Removing one (opting out, dropping + the SLOs, removing the expose) deletes its hidden scan job and what it has learned, never table + data, so it is not gated, and is printed. **Cost:** scans run on serverless compute billed as + `DATA_QUALITY_MONITORING`, once when monitoring is turned on and then as often as each table is + updated. Databricks Free Edition runs it (proven 2026-11-12), and its daily compute quota covers + these scans too: a workspace over the quota stops all compute for the rest of the day, so on Free + Edition opt out the products you do not need watched. ### Fixed diff --git a/fluid_build/iac/providers/databricks_quality.py b/fluid_build/iac/providers/databricks_quality.py index 73754acf..9886881e 100644 --- a/fluid_build/iac/providers/databricks_quality.py +++ b/fluid_build/iac/providers/databricks_quality.py @@ -188,9 +188,9 @@ def failure_hints(address: str) -> List[str]: if not address.startswith(f"{RESOURCE_TYPE}."): return [] return [ - "Databricks did not turn on data quality monitoring for this schema (its tables were " - "created first). It needs a Unity Catalog workspace with serverless compute, and " - "MANAGE on the schema for this identity (the schema's owner has it).", + "Databricks refused the data quality monitor on this schema; the schema's tables are " + "in place (they come first). Monitoring needs a Unity Catalog workspace with " + "serverless compute, and MANAGE on the schema for this identity (its owner has it).", "Or turn it off for the expose: set contract.dq.monitoring.enabled: false, then " "re-run fluid apply.", ] diff --git a/tests/iac/test_iac_databricks_quality.py b/tests/iac/test_iac_databricks_quality.py index 8f16523a..acccea32 100644 --- a/tests/iac/test_iac_databricks_quality.py +++ b/tests/iac/test_iac_databricks_quality.py @@ -408,7 +408,7 @@ def test_a_refused_monitor_names_the_opt_out_first(self): assert err.event == "opentofu_apply_failed" assert err.context["resource"] == f"{MONITOR}.{KEY}" first, second, last = err.suggestions[-3:] - assert first.startswith("Databricks did not turn on data quality monitoring") + assert first.startswith("Databricks refused the data quality monitor on this schema") assert "MANAGE on the schema" in first assert "contract.dq.monitoring.enabled: false" in second assert last.startswith("To see what exists") diff --git a/tests/iac/test_iac_databricks_quality_fake_e2e.py b/tests/iac/test_iac_databricks_quality_fake_e2e.py index 3781943d..a7f1b842 100644 --- a/tests/iac/test_iac_databricks_quality_fake_e2e.py +++ b/tests/iac/test_iac_databricks_quality_fake_e2e.py @@ -155,7 +155,7 @@ def test_a_refused_monitor_fails_the_apply_with_the_way_out(workspace, fake, cap rc, out = apply_contract(capfd, write_contract(workspace, _monitored())) assert rc != 0 - assert "Databricks did not turn on data quality monitoring for this schema" in out + assert "Databricks refused the data quality monitor on this schema" in out assert "set contract.dq.monitoring.enabled: false, then re-run fluid apply" in out # created first, and kept: a re-run converges assert {"main.sales.orders", "main.sales.notes"} <= set(fake.tables) From befe92eac56e3a6710d7ca3e6a3061609b39288c Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Thu, 12 Nov 2026 23:31:38 +0100 Subject: [PATCH 6/6] style(databricks): explicit hint strings and documented Protocol members (code-quality review) --- .../iac/providers/databricks_quality.py | 20 ++++++++++++------- 1 file changed, 13 insertions(+), 7 deletions(-) diff --git a/fluid_build/iac/providers/databricks_quality.py b/fluid_build/iac/providers/databricks_quality.py index 9886881e..5de906fc 100644 --- a/fluid_build/iac/providers/databricks_quality.py +++ b/fluid_build/iac/providers/databricks_quality.py @@ -58,10 +58,12 @@ class MonitoredTable(Protocol): exposure: Mapping[str, Any] @property - def full_name(self) -> str: ... + def full_name(self) -> str: + """``catalog.schema.table``, lower-cased as Unity Catalog stores it.""" @property - def where(self) -> str: ... + def where(self) -> str: + """Where the contract declares it, for messages: ``exposes[i] (exposeId)``.""" def _dq(exposure: Mapping[str, Any]) -> Mapping[str, Any]: @@ -188,11 +190,15 @@ def failure_hints(address: str) -> List[str]: if not address.startswith(f"{RESOURCE_TYPE}."): return [] return [ - "Databricks refused the data quality monitor on this schema; the schema's tables are " - "in place (they come first). Monitoring needs a Unity Catalog workspace with " - "serverless compute, and MANAGE on the schema for this identity (its owner has it).", - "Or turn it off for the expose: set contract.dq.monitoring.enabled: false, then " - "re-run fluid apply.", + ( + "Databricks refused the data quality monitor on this schema; the schema's tables " + "are in place (they come first). Monitoring needs a Unity Catalog workspace with " + "serverless compute, and MANAGE on the schema for this identity (its owner has it)." + ), + ( + "Or turn it off for the expose: set contract.dq.monitoring.enabled: false, then " + "re-run fluid apply." + ), ]