diff --git a/CHANGELOG.md b/CHANGELOG.md index d3bd9927..44ab1520 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -50,6 +50,26 @@ 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. 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/HONESTLY_TESTED.md b/HONESTLY_TESTED.md index 55ce77ba..51ff5cba 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` | ✅ 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. 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 1 unit: Databricks Genie space (`extensions.databricksGenie`) | `tests/iac/test_iac_databricks_genie.py` (the compile, the plugin's `converge_output_ports` against the stand-in's Genie and Permissions APIs, the engine hook) | ✅ 90. What the space holds (sample questions split on `\|`, 32-hex ids sorted, column descriptions with allowed values, range, example and synonyms, one instruction with the metrics' SQL, `agentPolicy` and the access statement), prompt matching never on, a sensitive column's example left out, the refusals (`fluid validate` and before `tofu`), create + share then no change, an update with the `etag`, a 409 not overwritten, a hand edit set back, removal trashes, a stranger's space untouched, a space planted with the marker in another folder never adopted, shared or trashed, a marker quoted inside a description no match, two markers refused, an apply straight after a create finding the space (the list lags), a title another space has kept as the service suffixed it and a rename onto one said as such, a transient failure to read a space's folder failing the apply instead of making a second space, a new space outside the folder the next apply looks in trashed and said, a policy that allows `analysis` but not `qa` accepted and one that denies `analysis` refused, its lines printed literally (Rich read the marker and a `[/x]` title as markup), a principal granted `manage` (`ALL_PRIVILEGES`) counted as a reader, metric views counted toward the 50-entry cap, remediation by HTTP status, a missing SDK stopping the apply before `tofu`, the native plan refusing what `emit` refuses, one snippet per metric name, a principal the service keeps in another case not shared again, a dry run writes nothing, a metric view used once it exists and sent where the service keeps it, readers' `CAN_RUN` with editors kept, the real SDK's error classes read as 403/404/409, no credential in a message, the stand-in refusing what the service refuses, and measure SQL identical to the governed MCP query path's | Genie's answers (only the live row below asks one) | | Stage 1 real tofu, stand-in workspace: Genie space | `tests/iac/test_iac_databricks_genie_fake_e2e.py` | ✅ 3. The real `fluid` CLI and `tofu`: validate, dry-run, apply (the space after the tables, shared with the readers), a second apply that changes nothing, the block removed (trashed); a Genie failure fails the apply after the tables with its fix and the next apply converges; an invalid block refused before `tofu` | The stand-in is modelled on the measured API, not the service itself | | Manual live, Free Edition, Genie space, 2026-11-12 | the real `fluid` CLI, no test file: branch `feat/databricks-genie-space`, databricks-sdk 0.162.0, a copy of industry-models tower-management `dp019_site_360` in the scratch catalog `fluid_iactest_genie` (no dbt build, no MENA sovereignty block: the workspace is in us-west-2), 6 synthetic rows | ✅ dry-run (would create, would share), apply `+73` then the space created and shared `CAN_RUN` with two persona service principals, a second apply `+0 ~0 -0` and "no change", the question "Which on-air sites in Amman are colocation-ready with hybrid power and no standing risk?" answered through the Genie Conversation API with the 2 expected sites (its SQL took the latest `snapshot_date` and the inverse of the `at_risk_site_count` filter from the instruction), a metric view created by hand picked up ("updated: metric views", then "no change"), the block removed (trashed). Two defects found and fixed in the run: the metric-view read failed on a schema not created yet, and the service keeps a metric view among the tables. A re-check after the security review (the plugin's own hook over `samples.nyctaxi.trips`, nothing created in Unity Catalog) found the list shows a new space only ~9 s later (a run straight after made a second space) and that a taken title is suffixed on create and 409 on PATCH; fixed, then create, an immediate second run ("no change", same space), and the block removed (trashed, via the folder check) passed live. Swept: the space trashed, `DROP CATALOG fluid_iactest_genie CASCADE`, 0 left | A reader without `SELECT` asking (no persona credential here; the access model is Genie's documented one); a service principal as the deploying identity (its folder is `/Users/`, the one Databricks Asset Bundles build from `current_user.userName`; a create that lands anywhere else is trashed and stops the apply); a space over several products | diff --git a/fluid_build/cli/_apply_opentofu_engine.py b/fluid_build/cli/_apply_opentofu_engine.py index 122ace51..3b49fbb0 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 @@ -316,7 +317,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) @@ -947,7 +948,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: @@ -965,6 +969,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`).""" @@ -1503,9 +1522,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", + } ) @@ -1635,7 +1661,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 @@ -1643,11 +1669,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 @@ -1673,6 +1702,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/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 34f1d1e0..7da9aa14 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 + 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 a local project its ``databricks_workspace_file`` uploads @@ -211,6 +221,14 @@ reconcile, ) from .databricks_keys import lookup as lookup_key_tables +from .databricks_quality import RESOURCE_TYPE as QUALITY_MONITOR +from .databricks_quality import ( + SchemaMonitor, + failure_hints, + monitor_notices, + plan_monitors, + window_warnings, +) from .databricks_retention import Retention, check_expire _LOG = logging.getLogger(__name__) @@ -355,6 +373,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 @@ -1388,6 +1408,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( @@ -1607,15 +1628,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: @@ -2450,7 +2516,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})" @@ -2489,6 +2555,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, @@ -2692,6 +2763,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. @@ -2758,6 +2843,7 @@ def planned_objects(contract: Mapping[str, Any]) -> List[Dict[str, Any]]: "tag_value": tag_value, } ) + actions.extend(_monitor_actions(compiled.monitors)) 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..5de906fc --- /dev/null +++ b/fluid_build/iac/providers/databricks_quality.py @@ -0,0 +1,214 @@ +# 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 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`` +``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: + """``catalog.schema.table``, lower-cased as Unity Catalog stores it.""" + + @property + def where(self) -> str: + """Where the contract declares it, for messages: ``exposes[i] (exposeId)``.""" + + +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. Set it to P1D, or remove it." + ) + 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; 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 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." + ), + ] + + +__all__ = [ + "RESOURCE_TYPE", + "MonitoredTable", + "SchemaMonitor", + "failure_hints", + "monitor_notices", + "monitoring_request", + "plan_monitors", + "window_warnings", +] diff --git a/tests/iac/_fake_databricks.py b/tests/iac/_fake_databricks.py index 186c0de9..f47eb32c 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 @@ -1024,6 +1025,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 @@ -1353,6 +1357,11 @@ 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) # ---- Genie spaces and their permissions (databricks-sdk-py service/dashboards.py, iam.py) a("GET", r"/api/2\.0/genie/spaces", self._genie_list) a("POST", r"/api/2\.0/genie/spaces", self._genie_create) @@ -1363,6 +1372,85 @@ def _register(self) -> None: a("PUT", r"/api/2\.0/permissions/genie/([^/]+)", self._genie_put_acl) # ------------------------------------------------------------------ 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)) + # 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 {} + 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]: + 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]: name = getattr(self, "me_user_name", _USER) return 200, { 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..acccea32 --- /dev/null +++ b/tests/iac/test_iac_databricks_quality.py @@ -0,0 +1,421 @@ +# 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 + assert warning.endswith("Set it to P1D, or remove it.") + + @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 + # 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( + _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 + + +# ── 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 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") + + @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 new file mode 100644 index 00000000..a7f1b842 --- /dev/null +++ b/tests/iac/test_iac_databricks_quality_fake_e2e.py @@ -0,0 +1,192 @@ +# 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; 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). +""" + +from __future__ import annotations + +import copy +import re +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") +_TRANSPORT = re.compile(r"broken pipe|connection reset by peer") + + +@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_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 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) + + +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)) + 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}") + 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