diff --git a/.secrets.baseline b/.secrets.baseline index c7698974..0ec2d290 100644 --- a/.secrets.baseline +++ b/.secrets.baseline @@ -157,7 +157,7 @@ "filename": "AGENTS.md", "hashed_secret": "9d4e1e23bd5b727046a9e3b4b7db57bd8d6ee684", "is_verified": false, - "line_number": 675 + "line_number": 677 } ], "Jenkinsfile": [ @@ -2317,5 +2317,5 @@ } ] }, - "generated_at": "2026-11-12T23:57:29Z" + "generated_at": "2026-11-13T00:16:47Z" } diff --git a/AGENTS.md b/AGENTS.md index 9abd0b20..16979406 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -55,6 +55,8 @@ The `replace` rows describe the native engine. On the OpenTofu engine (aws, gcp, On Databricks a dbt build with `execution.runtime.platform: databricks` runs as the job `fluid apply` provisions. `amend-and-build` and `replace-and-build` start a run of the job this apply's OpenTofu state holds (found by its `fluid_product` / `fluid_build` tags), wait for it, and fail with the task's error and the end of dbt's output; `replace-and-build` adds `--full-refresh` for that run only. The job's settings, schedule and pause state are never written; `amend` and `create-only` only provision it, and a continuous job is never started. The wait is capped by `FLUID_DATABRICKS_RUN_TIMEOUT_SECONDS` (an hour, or the job's own limits plus 15 minutes); at the cap, on Ctrl-C and on SIGTERM or SIGHUP (unless ignored, as under `nohup`) the run is cancelled, while a run that cannot be read five times in a row is left alone. To build an estate in lineage order, provision every product in parallel with `--mode amend`, then run `--mode amend-and-build` level by level; the second apply plans no change. Code: `build_runners/_databricks_job_run.py`. +The job also runs the contract's DQ rules (`exposes[].contract.dq.rules` of each expose the build writes) as dbt tests: one singular test per check, `/fluid_dq//.sql`, uploaded beside the project, so `dbt build` runs each after its model and a failing `error`/`critical` rule fails the run; the run report prints each rule's result. `util/dq_checks.py` reads the rules once for every engine (`read_rules` → `DqCheck` / `DqRefusal`), a `valid_values` rule's allowed values through `quality_engine.value_domain` as `fluid test` reads them; `iac/providers/databricks_dbt_tests.py` renders them. A rule no test can run, a Git source with rules, a missing model or source, a test name the project uses, and a selection that skips the model are refused at validate and plan. + ### Plan-Binding (Terraform-Style "Apply Consumes Exact Plan") Stage 6 `fluid plan` emits two cryptographic fields in `plan.json`: diff --git a/CHANGELOG.md b/CHANGELOG.md index fb454f39..5b14efe3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -251,6 +251,38 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 strict bound stays unmapped there. On the tower-management industry pack `fluid test` ran 38 of its 168 rules before; it now runs every rule of every deployed product. +- **A Databricks dbt job runs its contract's data-quality rules as dbt tests.** Each rule in + `exposes[].contract.dq.rules` of an expose a Databricks dbt build writes becomes a singular + dbt test, `tests/fluid_dq//.sql`, uploaded beside the project (your own files + upload byte for byte). The job's `dbt build` runs each test right after its model, so the run + page lists every rule by its id, passed, warned or failed; a rule of severity `critical` or + `error` that fails skips the models below it and fails the run, `warn` and `info` are reported + and do not. uniqueness runs dbt's `unique`, completeness `not_null` on each selected column (a + comma-separated selector is one test per column). valid_values allows what `fluid test` allows + (`quality_engine.value_domain`): the values the rule's description lists (` valid + values: A, B.`), as dbt's `accepted_values`, or else every declaration of the column, which + must all hold: `accepted_values` for its `enum` validationRule (`A | B` or `A,B`), a regex + test for its `regex`, a range test for its `range` (`>= 0`, `>= 0 and <= 100`, `between 0 and + 100`), and dbt's `relationships` for its `labels.references: .` against + the dbt source the project declares for that consumed expose; accuracy is a bound on every + value, freshness the newest value's age against `window`, schema each contract column with its + declared type. A threshold below 1.0 is the share of rows that must pass: the non-null share + for completeness and the distinct share for uniqueness, as `fluid test` computes them, and the + share of non-null values that are allowed for valid_values. No dbt package is installed: the + regex, range, freshness, schema and share tests are forge-cli's own SQL. Under `--mode + amend-and-build`, `fluid apply` prints what dbt found for each rule after the run. An + `anomaly_detection` rule is left to Databricks data quality monitoring, which learns its + bounds (it stays refused where the expose turns monitoring off). The `fluid_keys` task that + declares the table's Unity Catalog keys again now runs once dbt has ended, whatever its + outcome: a failing rule fails dbt after its model replaced the table, and the table keeps its + keys. Refused at `fluid validate` and `fluid plan`, with the fix: a rule no test can run as + written (drift detection, valid_values with no allowed values, an unknown column, an + unparseable window, a threshold no number holds, a range no value meets), a Git-sourced build + with rules (no file can be added to a checkout), a model or a referenced source the project + does not hold, a build whose dbt selection leaves the model out, and a rule id whose test name + the project's own tests already use. `fluid_build/util/dq_checks.py` reads the rules once, for + the engines that run them; this job is its first. + ### Fixed - **Quality checks measured a constant on BigQuery (and would have on Databricks).** The diff --git a/HONESTLY_TESTED.md b/HONESTLY_TESTED.md index c3f29fa2..90ee80e3 100644 --- a/HONESTLY_TESTED.md +++ b/HONESTLY_TESTED.md @@ -89,6 +89,7 @@ Markers declared in `pyproject.toml`. | Stage 1 real tofu, stand-in workspace: masks and row filters | `tests/iac/test_iac_databricks_fake_e2e.py` (17 of its 45; needs databricks-sdk, the `[databricks]` extra, and skips without it) | ✅ 17. The real `fluid` CLI and `tofu` apply bind the filter and masks (after the apply, on a table it created); a second apply executes no statement; a table that already exists is bound before the apply's first grant on it; `--mode dry-run` prints the statements and executes none; a changed predicate binds a new function and the apply drops the old one; a removed restriction is dropped and said to widen access without `--allow-data-loss`; a binding that fails revokes SELECT from the restricted principal only, fails the run (`opentofu_access_enforcement_failed`), and the next apply binds before it grants again; a failure before the apply stops it with nothing applied; an invalid predicate is refused before anything runs; an apply that fails after creating the table (the grant on it refused) still binds the filter and masks and the run exits with `opentofu_apply_failed`, an apply that fails before the table exists binds nothing and revokes nothing, and a failed apply drops no binding the contract no longer asks for (it is dropped by the next apply that succeeds); a function taken under the hashed name stops the apply before it runs; a row filter set by hand stops it and is left in place; a binding that fails with no deny-all possible revokes every grantee but the owner and this run's identity and the next apply binds first and grants again; a failed mask leaves the deny-all filter and every grant until the next apply; a filter needing a column the apply adds is held by the deny-all filter, then replaced, in that order around the added column; a dbt job created with its restricted table is paused and the next apply unpauses it in place; `generate iac` refuses the module. | Live behaviour: that `CREATE FUNCTION IF NOT EXISTS`, `SET ROW FILTER` and `SET MASK` run on a real warehouse as modelled, that a second `SET` replaces a binding already on the table (if it does not, a changed predicate fails closed with the old binding in place), that `current_user()` of a service principal is its application ID, that a Free Edition or classic warehouse can run them, and whether the SDK's statement API answers as the stand-in does; whether `SET ROW FILTER ... ON ()` (a zero-column function) is accepted; whether FunctionInfo's `routine_definition`, `input_params` and `full_data_type` come back in the form compared (whitespace-normalised, lower-case types), whose mismatch would refuse a good function; whether `scim/v2/Me` `userName` is the owner `CREATE FUNCTION` records for a service principal; that `SET` replaces a bound filter or mask (the stand-in does; none of the fixes uses a DROP first, which would leave a window with no filter); the permissions and effective-permissions fields used to lock a table; and the pause behaviour of a job on a live workspace. Unproven for ALL_PRIVILEGES grantees on a live workspace: a revoke of ALL PRIVILEGES removes more than read access, by design of the lock. A principal that is a metastore admin or the table's owner is never restricted by these bindings. | | Stage 1: Databricks bronze landing and masking at landing (live: the 2026-11-04 row) | `tests/build_runners/test_duckdb_databricks_load.py` (the runner and the loader), `tests/iac/test_iac_databricks_masking_at_landing.py` (the emitter), `tests/iac/test_iac_databricks_landing_fake_e2e.py` (real `fluid` + `tofu` + databricks-sdk; skips without the `[databricks]` extra), `tests/iac/_fake_databricks.py` | ✅ A `platform: databricks` binding no longer writes a local file: the run stages Parquet under `.fluid/staging`, uploads it to `DATABRICKS_STAGING_VOLUME` through the Files API, runs one `INSERT OVERWRITE ... read_files` over the one file it uploaded (a file someone else puts in the run's directory is not loaded) and deletes the staged files (also after a failed upload, a failed or short load and a Ctrl-C, which cancels the statement first; one that cannot be deleted fails the run naming it first), then removes the build's and the product's staging directories when each is listed empty (never one another run or build still uses; a failure to remove one is logged, not fatal), and deletes the local masked copy under `.fluid/staging//` after the load attempt, success or failure (`facets.databricks_load.local_staging_deleted`; one that cannot be deleted fails the run naming it); each load first deletes an earlier run's staging of the same build once all of it is over an hour old, and leaves a recent one. The statement's text, quoting and column list are pinned, as are the volume's validation, the staging path (no contract id climbs out), the refusals before any work (unset volume or warehouse, mode, sink, stream count), a missing table (names `fluid apply`, uploads nothing), credential scrubbing and the matrix of when the emitter accepts `policy.privacy.masking` (every writer a DuckDB acquisition build) against when it refuses it, through both `validate` and `emit`. End to end against the stand-in: `fluid apply` creates the table with its mask and row filter, `--mode amend-and-build` loads the rows with the masked column hashed and the cleartext absent, the volume is empty, nothing but the load runs on the warehouse, the table is not replaced, a second plan is empty, and a second run overwrites the rows. | **Not run live.** The stand-in is not Databricks: it reads the uploaded Parquet with DuckDB, so what `read_files` accepts, that `INSERT OVERWRITE` keeps a managed table's row filter and column masks, implicit casts from the Parquet types to the table's, the Files API's limits and a real warehouse's permissions are unproven. The run-time schema gate reads a declared `string` as a type change against DuckDB's `VARCHAR`, so a contract declares `varchar` for a text column. | | 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 + manual live: contract DQ rules as dbt tests in a Databricks dbt job | `tests/util/test_dq_checks.py` (the reader), `tests/iac/test_iac_databricks_dbt_tests.py` (the tests, each rendered in a dbt-shaped Jinja context, at parse and at execute), `tests/build_runners/test_databricks_job_run.py` (the run's report) | ✅ 51 + 65 + 4. Every rule type read once (`util/dq_checks.py`): selectors split on commas; a `valid_values` rule read as `fluid test`'s `value_domain` reads it (the rule's own list, else the column's first `enum`, split on `|` and `,`, `regex` and `range`, and `labels.references` through `consumes[]`); shares below 1.0, the severity map, unique names; each refusal with its fix. Each check rendered: dbt's own `unique`, `not_null`, `accepted_values`, `relationships` through `adapter.dispatch`, forge-cli's regex, bounds, freshness, schema and share SQL; hostile values (`'`, `\\`, `}}`, `{%`, `#}`, a character outside the BMP) stay inside their Jinja string and SQL literal; every test refs its model while dbt parses, and a real dbt (dbt-core with dbt-duckdb) parses the upload and `dbt ls --select +orders+` lists every rule's test. Refused at plan: a test name the project's own tests use, a selection that leaves the model out, a threshold no float holds, an empty range, an unpaired surrogate (in a rule, a value or the test path; one from the project's YAML is named escaped in a refusal); an `anomaly_detection` rule is left to data quality monitoring (#864) unless the expose turns it off, and one monitoring could not read either (a bad severity) is still refused. The `fluid_keys` task (#860) runs with `run_if: ALL_DONE`, so a failing rule does not leave the table without its keys. **Live, Databricks Free Edition, 2026-11-12** (scratch catalog `fluid_iactest_dq`, swept after: job, upload folder, catalog; 0 left): one dbt product, 9 rules as 11 tests, `fluid apply --mode amend-and-build` (`tofu` +27). Run 1, with a seeded bad row, ended `RUN_EXECUTION_ERROR`: completeness on `status` and the `amount` range `FAIL 1` (error), the country regex and the customer reference `WARN 1`, six `PASS`, each printed by `fluid apply`; it also showed the schema test was never selected (its `ref` sat behind `if execute`: dbt found 11 tests, ran 10), fixed. Run 2, clean: `SUCCESS`, the model and all 11 tests `PASS`; the next plan `+0 ~0 -0`. Offline: the demo's 14 dbt products compile to 230 tests with no refusal, and `dbt parse` / `dbt ls --select +model+` (dbt-core 1.12.5) select all 20 of DP005. | A share below 1.0 that fails, a reference share, and a regex or bounds share were not run live (only a passing completeness share). A Git-sourced build with rules is refused by design (no file can be added to a checkout). Not handled: a versioned model, and a `dbt_project.yml` that sets `+where` or `+limit` on tests, which would change the generated tests too. dbt's result lines are read from the task's output: lines Databricks cut from a long output are reported as not in it. The run page was read through the Jobs API (`runs/get-output`), not in a browser. | | 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 metric views (`exposes[].semantics`) | `tests/iac/test_iac_databricks_metrics.py` (the YAML, the refusals and the hooks against `tests/iac/_fake_databricks.py`), `tests/iac/test_iac_databricks_metrics_fake_e2e.py` (real `fluid` + `tofu` + databricks/databricks 1.140.0 against the stand-in) | ✅ 91 + 4. A site-360-shaped semantics block compiles to the metric view YAML it should (the entity as its key column and categorical and time dimensions as dimensions, with comments, display names and synonyms from the columns; non-additive measures as `semiadditive: last` windows; filtered simple metrics as `FILTER (WHERE ...)`; a ratio as `MEASURE(a) / NULLIF(MEASURE(b), 0)`, after its inputs); the view is `_semantic_model` whatever the model is called; every column reference is `source.` (`
.` too), a unit or type keyword (`EXTRACT(YEAR ...)`, `INTERVAL 1 DAY`, `::date`) stays one; the time grains (`TRUNC` on a `DATE`, `DATE_TRUNC` otherwise); 42 refusal cases (names, collisions, expressions, quoting, calls that would lend the owner's rights, text limits). The hooks: a first apply creates the view with its `fluid.*` properties and grants it to the table's readers; a second runs nothing; a changed or hand-edited definition is an `ALTER VIEW` that keeps the grants; `SELECT` and `ALL PRIVILEGES` are revoked from anyone else, before the apply and first in it, so a failed grant or update leaves no removed reader; a view the contract no longer declares, in a schema the product's tables are in, is dropped before the apply, another product's is left and another environment's (same id, another schema of the catalog) is kept; a table, a view without this product's property or one another identity owns at the name is refused before the apply; a failed statement is `databricks-metric-view-failed` naming the view; without databricks-sdk a declared view is refused; when a restricted table's binding fails, `SELECT` on its views is revoked. Through the CLI: apply, a second apply `+0 ~0 -0` with no statement, dry-run prints and runs nothing, dropping the semantics drops the view, `fluid validate` refuses a derived metric. All 19 industry-model tower-management contracts and the 4 forge templates that bind a table compile with no refusal. | What Databricks accepts and answers (the 2026-11-12 live row); Genie and AI/BI dashboards themselves; a view left in a schema the contract stops using (its table moved, its last expose removed) is not found, and OpenTofu's destroy of that schema then fails until it is dropped by hand, as for the column masks' functions. | diff --git a/fluid_build/build_runners/_databricks_job_run.py b/fluid_build/build_runners/_databricks_job_run.py index 49616b99..60852cd4 100644 --- a/fluid_build/build_runners/_databricks_job_run.py +++ b/fluid_build/build_runners/_databricks_job_run.py @@ -281,9 +281,23 @@ def run_job_build( env=os.environ if env is None else env, sleep=sleep, clock=clock, + dq_checks=_dq_checks(contract, build), ) +def _dq_checks(contract: Mapping[str, Any], build: Mapping[str, Any]) -> Tuple[Any, ...]: + """The contract rules the job runs as dbt tests (``iac/providers/databricks_dbt_tests.py``), + which the report names one by one; none when they cannot be read, since the report is a + courtesy and the apply already refused a rule it could not run.""" + try: + from fluid_build.iac.providers.databricks_dbt_tests import expected_tests + + return tuple(expected_tests(contract, build)) + except Exception: # noqa: BLE001 — never fail a build over its report + LOG.debug("databricks_job_run_dq_checks_unread", exc_info=True) + return () + + def start_and_wait( client: Any, job: JobRef, @@ -295,8 +309,12 @@ def start_and_wait( say: Optional[Say] = None, sleep: Callable[[float], None] = time.sleep, clock: Callable[[], float] = time.monotonic, + dq_checks: Sequence[Any] = (), ) -> int: - """Run ``job`` now through ``client`` and wait for the run to end. Returns 0 or 1.""" + """Run ``job`` now through ``client`` and wait for the run to end. Returns 0 or 1. + + ``dq_checks`` are the contract rules its dbt task runs as tests: the report says what + dbt found for each (:func:`fluid_build.iac.providers.databricks_dbt_tests.report`).""" say = say or _say api = client.api_client label = f"Databricks job '{name}' ({job.job_id})" @@ -343,7 +361,9 @@ def start_and_wait( if ended is None: return 1 # The run has ended: an interrupt from here on stops the report, never a run. - return _report_end(api, ended.run, run_id, build_id, ended.elapsed, ended.url, say) + return _report_end( + api, ended.run, run_id, build_id, ended.elapsed, ended.url, say, dq_checks=dq_checks + ) @dataclass(frozen=True) @@ -506,15 +526,23 @@ def _report_end( elapsed: float, url: str, say: Say, + *, + dq_checks: Sequence[Any] = (), ) -> int: + from fluid_build.iac.providers.databricks_jobs import DBT_TASK_KEY + code, message = _outcome(run) say(f" {_clock(elapsed)} TERMINATED {code}") LOG.debug("databricks_job_run_ended run_id=%s result=%s", run_id, code) if code == "SUCCESS": for task_key, task in _latest_attempts(run): - summary = _dbt_summary(api, task["run_id"]) + output, _unread = _task_output(api, task["run_id"]) + summary = _dbt_summary(output) if summary: say(f" task '{task_key}': {summary}") + if task_key == DBT_TASK_KEY: + for line in _rule_lines(output, dq_checks): + say(line) say(f"✅ Build '{build_id}' - Databricks run {run_id} succeeded in {_duration(elapsed)}") return 0 console_error( @@ -522,8 +550,16 @@ def _report_end( + (f": {message}" if message else "") ) for task_key, task in _latest_attempts(run): - if _outcome(task)[0] != "SUCCESS": - _report_task(api, task_key, task["run_id"], say) + failed = _outcome(task)[0] != "SUCCESS" + if not failed and task_key != DBT_TASK_KEY: + continue + output, unread = _task_output(api, task["run_id"]) + rules = _rule_lines(output, dq_checks) if task_key == DBT_TASK_KEY else [] + if failed: + _report_task(task_key, output, unread, say, rules) + else: # dbt passed and a later task failed: its tests still ran + for line in rules: + say(line) if url: say(f" run page: {url}") return 1 @@ -558,26 +594,47 @@ def _queued_behind(api: Any, job_id: int, run_id: int) -> str: ) -def _dbt_summary(api: Any, task_run: Any) -> str: - """dbt's own last ``Done. PASS=...`` line of a task's output, or ``""``: what a run that - succeeded built, at the cost of one read that may fail without failing the build.""" +def _task_output(api: Any, task_run: Any) -> Tuple[Mapping[str, Any], Optional[BaseException]]: + """A task's ``runs/get-output``, read once for every line about it: ``({}, error)`` when + it cannot be read.""" try: - output = api.do("GET", f"{_JOBS}/runs/get-output", query={"run_id": task_run}) or {} - except Exception: # noqa: BLE001 — the run succeeded; its summary is a courtesy - return "" + return api.do("GET", f"{_JOBS}/runs/get-output", query={"run_id": task_run}) or {}, None + except Exception as exc: # noqa: BLE001 — what is said about a task never fails the build + return {}, exc + + +def _dbt_summary(output: Mapping[str, Any]) -> str: + """dbt's own last ``Done. PASS=...`` line of a task's output, or ``""``: what a run that + succeeded built.""" done = [line for line in clean(output.get("logs") or "").splitlines() if "Done. PASS=" in line] return " ".join(_DBT_TIMESTAMP_RE.sub("", done[-1]).split()) if done else "" -def _report_task(api: Any, task_key: str, task_run: Any, say: Say) -> None: - try: - output = api.do("GET", f"{_JOBS}/runs/get-output", query={"run_id": task_run}) or {} - except Exception as exc: # noqa: BLE001 - say(f" task '{task_key}': its output could not be read ({_failure(exc)})") +def _rule_lines(output: Mapping[str, Any], dq_checks: Sequence[Any]) -> List[str]: + """What dbt found for each contract rule the dbt task ran as a test.""" + if not dq_checks: + return [] + from fluid_build.iac.providers.databricks_dbt_tests import report + + logs = clean(output.get("logs") or "") + return report(logs, dq_checks, truncated=bool(output.get("logs_truncated"))) + + +def _report_task( + task_key: str, + output: Mapping[str, Any], + unread: Optional[BaseException], + say: Say, + rules: Sequence[str] = (), +) -> None: + if unread is not None: + say(f" task '{task_key}': its output could not be read ({_failure(unread)})") return # One line: Databricks quotes the failed shell command with its newlines escaped. error = " ".join(clean(output.get("error") or "").replace("\\n", " ").split()) say(f" task '{task_key}'" + (f": {error}" if error else " failed")) + for line in rules: + say(line) logs = clean(output.get("logs") or output.get("error_trace") or "") lines = [line.rstrip() for line in logs.splitlines() if line.strip()] if not lines: diff --git a/fluid_build/iac/providers/databricks.py b/fluid_build/iac/providers/databricks.py index c4fc0b67..b2cf333b 100644 --- a/fluid_build/iac/providers/databricks.py +++ b/fluid_build/iac/providers/databricks.py @@ -2807,16 +2807,18 @@ def _job_actions(jobs: Sequence[DbtJob]) -> List[Dict[str, Any]]: """Plan actions for the dbt jobs and the project files they upload.""" actions: List[Dict[str, Any]] = [] for job in jobs: - actions.append( - { - "op": "databricks.job.ensure", - "resource_type": "databricks_job", - "resource_id": job.name, - "build": job.build_id, - "source": job.source, - "trigger": job.trigger, - } - ) + action: Dict[str, Any] = { + "op": "databricks.job.ensure", + "resource_type": "databricks_job", + "resource_id": job.name, + "build": job.build_id, + "source": job.source, + "trigger": job.trigger, + } + if job.tests: + # The contract's DQ rules, each a dbt test the job runs (databricks_dbt_tests.py). + action["dq_tests"] = list(job.tests) + actions.append(action) actions.extend( { "op": "databricks.workspace_file.ensure", diff --git a/fluid_build/iac/providers/databricks_dbt_tests.py b/fluid_build/iac/providers/databricks_dbt_tests.py new file mode 100644 index 00000000..bf9fa84a --- /dev/null +++ b/fluid_build/iac/providers/databricks_dbt_tests.py @@ -0,0 +1,779 @@ +# Copyright 2025 Agentics Transformation Limited +# SPDX-License-Identifier: Apache-2.0 + +"""The dbt tests a Databricks dbt job runs for its contract's data-quality rules. + +Each check of the rules on the exposes a build writes (:func:`fluid_build.util.dq_checks. +read_rules`) becomes one singular dbt test, ``/fluid_dq//.sql``, in the +copy of the dbt project the job uploads (``databricks_jobs.py``). The user's own files upload +byte for byte: dbt describes a model in one properties entry only (a second entry fails the +parse), and that entry is the user's, so the tests are files of their own rather than entries +in it. + +``dbt build`` runs each test right after the model it tests, so the job's run page lists every +one by its check name (the rule id), passed, warned or failed. A test of severity ``error`` that +fails skips the models below it and fails the task, the run and ``fluid apply``. + +What runs: + +* ``not_null``, ``unique``, ``accepted_values`` and ``reference``: dbt's own built-in test SQL, + called as a ``unique:`` test in YAML calls it (``adapter.dispatch('test_unique', 'dbt')``). A + reference is dbt's ``relationships`` against the source table the project declares for the + consumed expose (the table the model reads), found by the expose's id. +* ``regex``, ``bounds``, ``freshness``, ``schema`` and every share: forge-cli's own SQL, after + dbt_utils' ``recency`` and ``not_null_proportion`` and dbt_expectations' + ``expect_column_values_to_match_regex`` and ``expect_table_columns_to_match_set`` (both + Apache-2.0). No dbt package is installed: a serverless job's ``dbt deps`` may not reach + hub.getdbt.com (Databricks Free Edition limits outbound traffic to trusted domains), and + nothing is merged into the user's ``packages.yml``. + +Each file sets ``config(severity=..., enabled=true)``, which wins over ``dbt_project.yml``, and +the tag ``fluid_dq``. A name or value from the contract or the project reaches the file only as +an escaped Jinja string (:func:`_jinja_str`) or a validated, backtick-quoted identifier, and an +SQL literal only through ``databricks_access.string_literal``. + +After a run, :func:`report` reads dbt's own result line for each test (``N of M PASS ``) +from the task's output for ``fluid apply`` to print. +""" + +from __future__ import annotations + +import json +import re +from dataclasses import dataclass, field +from pathlib import PurePosixPath +from typing import Any, Dict, Iterator, List, Mapping, Optional, Sequence, Tuple + +from ...util.binding_paths import ENV_PLACEHOLDER_RE +from ...util.contract import builds_writing_expose +from ...util.dq_checks import DqCheck, describe, has_rules, read_rules +from ..base import UnsupportedBindingError + +#: The folder, below the project's first test path, that holds the generated tests. +TEST_DIR = "fluid_dq" +#: The dbt tag every generated test carries (``dbt build --select tag:fluid_dq``). +TAG = "fluid_dq" + +_SQL_OPERATORS = {">=": ">=", ">": ">", "<=": "<=", "<": "<", "==": "=", "!=": "!="} +#: The file names dbt itself reads at the top of a project, never a properties file. +_PROJECT_YAML = frozenset( + {"dbt_project.yml", "packages.yml", "dependencies.yml", "package-lock.yml", "selectors.yml"} +) +#: dbt's own line for a finished test: ``12 of 20 PASS name``, ``FAIL 3 name``, ``WARN 1``, +#: ``ERROR name``, ``SKIP test name`` (``format_fancy_output_line``, dbt-core). +_RESULT_RE = re.compile( + r"\b\d+ of \d+ (PASS|FAIL|WARN|ERROR|SKIP)(?: (\d+))?(?: (?:test|relation))? ([0-9A-Za-z_]+)" +) +_ANSI_RE = re.compile(r"\x1b\[[0-?]*[ -/]*[@-~]") + + +@dataclass(frozen=True) +class DqTest: + """One generated test: the check it runs and the file that runs it.""" + + check: DqCheck + model: str + path: str + + @property + def name(self) -> str: + return self.check.name + + +@dataclass +class CompiledTests: + """The test files for one build's job, its tests, and every refusal.""" + + files: List[Tuple[str, bytes]] = field(default_factory=list) + tests: List[DqTest] = field(default_factory=list) + refusals: List[UnsupportedBindingError] = field(default_factory=list) + + +def _refusal(kind: str, message: str, fix: str) -> UnsupportedBindingError: + # A name from the project's own YAML (a path, a source) can hold an unpaired surrogate, + # which a YAML escape loads and no terminal prints: the refusal spells it as that escape. + message, fix = ( + text.encode("utf-8", "backslashreplace").decode("utf-8") for text in (message, fix) + ) + return UnsupportedBindingError(f"databricks-job-dq-{kind}", message, (fix,)) + + +def written_tables(build: Mapping[str, Any], builds: Sequence[Any], tables: Sequence[Any]): + """The Databricks tables (``databricks.py::_Table``) ``build`` writes, in contract order.""" + return [ + table + for table in tables + if any(b is build for b in builds_writing_expose(table.exposure.get("exposeId"), builds)) + ] + + +def compile_tests( + contract: Mapping[str, Any], + build: Mapping[str, Any], + where: str, + tables: Sequence[Any], + files: Optional[Sequence[Tuple[str, bytes]]], + *, + git: bool = False, +) -> CompiledTests: + """The test files for the rules on ``tables`` (the Databricks tables ``build`` writes). + + ``files`` is the project as it is uploaded, ``(path, bytes)``; ``None`` when no file can + be read (``fluid validate`` without a contract location), when only the rules themselves + are checked. ``git``: the job runs a Git checkout, which no file can be added to. + """ + out = CompiledTests() + exposures = [table.exposure for table in tables] + if not has_rules(exposures): + return out + rules = read_rules(contract, exposures) + monitored = _monitored(exposures) + for refused in rules.refusals: + if refused.rule_type == "anomaly_detection" and refused.needs_history: + if refused.expose_id in monitored: + continue # Databricks data quality monitoring runs it (databricks_quality.py) + out.refusals.append( + _refusal( + "rule", + f"{where} expose {refused.expose_id!r}: {refused.message}, and the expose " + "turns Databricks data quality monitoring off (contract.dq.monitoring." + "enabled: false), which would learn its bounds.", + "Turn the monitoring back on, check the rule with fluid test, or remove it.", + ) + ) + continue + out.refusals.append( + _refusal( + "rule", f"{where} expose {refused.expose_id!r}: {refused.message}.", refused.fix + ) + ) + if git: + out.refusals.append( + _refusal( + "git-source", + f"{where} runs its dbt project from Git, and the contract rules on " + f"{_exposes(tables)} run as dbt tests in the job: forge-cli adds them to the " + "project it uploads, and a Git checkout runs as it was committed.", + "Point repository at the local dbt project (fluid apply uploads it).", + ) + ) + if git or files is None or out.refusals: + return out + project = _Project(files) + checked = {check.expose_id for check in rules.checks} + single = len(tables) == 1 + models = { + _expose_key(table): _model(build, table, project, where, out.refusals, single=single) + for table in tables + if _expose_key(table) in checked + } + if any(model == _DEFERRED for model in models.values()): + # ``fluid validate`` reads the contract before ``{{ env.* }}`` is substituted: the + # model is named only then, and the apply, which substitutes first, writes the tests. + return out + folder = _test_folder(project, where, out.refusals) + if folder is None or out.refusals: + return out + for check in rules.checks: + _add_test(out, check, str(models[check.expose_id]), folder, project, where) + if out.refusals: + out.files, out.tests = [], [] + return out + + +#: What :func:`_model` answers for a model name that is still an ``{{ env.* }}`` placeholder. +_DEFERRED = "\0deferred" + + +def _expose_key(table: Any) -> str: + """The key :func:`read_rules` gives a table's checks (``DqCheck.expose_id``).""" + return str(table.exposure.get("exposeId") or "") + + +class _Project: + """The uploaded dbt project as the tests read it: its paths, and its declared sources.""" + + def __init__(self, files: Sequence[Tuple[str, bytes]]) -> None: + self.files = files + self.present = {rel for rel, _ in files} + paths = _project_paths(files) + self.name = paths["name"][0] + self.model_paths, self.test_paths = paths["model-paths"], paths["test-paths"] + self._sources: Optional[Dict[str, List[Tuple[str, str]]]] = None + + def singular_tests(self) -> Dict[str, str]: + """``{test name: path}`` of the project's own singular tests: dbt names a singular + test after its file, and a test name once per project.""" + return { + PurePosixPath(rel).stem: rel + for rel in sorted(self.present) + if rel.endswith(".sql") and any(rel.startswith(f"{path}/") for path in self.test_paths) + } + + @property + def sources(self) -> Dict[str, List[Tuple[str, str]]]: + if self._sources is None: + self._sources = _declared_sources(self.files) + return self._sources + + +def _monitored(exposures: Sequence[Mapping[str, Any]]) -> set: + """The exposeIds whose tables Databricks data quality monitoring watches: it runs their + ``anomaly_detection`` rules (learning the bounds), which a dbt test cannot.""" + from .databricks_quality import monitoring_request + + return {str(e.get("exposeId") or "") for e in exposures if monitoring_request(e)} + + +def _test_folder( + project: _Project, where: str, refusals: List[UnsupportedBindingError] +) -> Optional[PurePosixPath]: + """``/fluid_dq``, where the generated tests go.""" + if not project.test_paths: + refusals.append( + _refusal( + "test-path", + f"{where} dbt_project.yml sets test-paths to an empty list, so dbt reads no " + "test file and the contract rules would not run.", + "Leave test-paths out (dbt reads tests/), or name a directory.", + ) + ) + return None + first = project.test_paths[0] + try: + first.encode("utf-8") + except UnicodeEncodeError: # an unpaired surrogate (a YAML escape loads one) + refusals.append( + _refusal( + "test-path", + f"{where} dbt_project.yml's first test path {first!r} holds an unpaired " + "surrogate, which no file path can hold, and the contract rules' tests are " + "written under it.", + "Write the test path in plain text.", + ) + ) + return None + return PurePosixPath(first) / TEST_DIR + + +def _add_test( + out: CompiledTests, + check: DqCheck, + model: str, + folder: PurePosixPath, + project: _Project, + where: str, +) -> None: + """The file for ``check`` into ``out``, or its refusal.""" + source = None + if check.reference is not None: + source = _source_for(check, project.sources, where, out.refusals) + if source is None: + return + try: + text = render(check, model, source) + except ValueError: + out.refusals.append( + _refusal( + "column", + f"{where} rule {check.rule_id!r} on expose {check.expose_id!r} reads a " + "column whose name is not a plain identifier (letters, digits and _), and " + "the tests name columns only that way.", + "Rename the column in the contract and the model.", + ) + ) + return + path = str(folder / model / f"{check.name}.sql") + taken = project.singular_tests().get(check.name) + if taken is not None and taken != path: + out.refusals.append( + _refusal( + "name", + f"{where} the test of rule {check.rule_id!r} is named {check.name!r}, and the " + f"dbt project's own test {taken} has that name: dbt names a test once per " + "project, so the whole job would fail to parse.", + "Rename the rule id, or the project's test.", + ) + ) + return + if path in project.present: + out.refusals.append( + _refusal( + "path", + f"{where} the dbt project already holds {path}, where forge-cli writes the " + f"test for rule {check.rule_id!r}: {folder}/ is the generated tests' folder.", + f"Move the file out of {folder}/.", + ) + ) + return + try: + data = text.encode("utf-8") + except UnicodeEncodeError: # an unpaired surrogate from the project's own YAML + out.refusals.append( + _refusal( + "text", + f"{where} the test of rule {check.rule_id!r} names a value from the dbt project " + "(a source or table name) holding an unpaired surrogate, which no file can hold.", + "Write the name in plain text in the project's properties YAML.", + ) + ) + return + out.files.append((path, data)) + out.tests.append(DqTest(check=check, model=model, path=path)) + + +def expected_tests(contract: Mapping[str, Any], build: Mapping[str, Any]) -> List[DqCheck]: + """The checks a build's job tests, from the contract alone: what :func:`report` looks for + in a run's output. The rules on each Databricks expose the build writes.""" + builds = [b for b in contract.get("builds") or [] if isinstance(b, Mapping)] + exposures = [] + for exposure in contract.get("exposes") or []: + if not isinstance(exposure, Mapping): + continue + binding = exposure.get("binding") + platform = binding.get("platform") if isinstance(binding, Mapping) else None + if str(platform or "").strip().lower() != "databricks": + continue + if any(b is build for b in builds_writing_expose(exposure.get("exposeId"), builds)): + exposures.append(exposure) + return read_rules(contract, exposures).checks + + +# ── project ────────────────────────────────────────────────────────────────── + + +def _project_paths(files: Sequence[Tuple[str, bytes]]) -> Dict[str, List[str]]: + """``name``, ``model-paths`` and ``test-paths`` of the uploaded ``dbt_project.yml``, with + dbt's defaults; the upload has already read it, so it parses.""" + import yaml + + config: Any = {} + for rel, data in files: + if rel == "dbt_project.yml": + try: + config = yaml.safe_load(data.decode("utf-8")) or {} + except (UnicodeDecodeError, RecursionError, yaml.YAMLError): + config = {} + config = config if isinstance(config, Mapping) else {} + out: Dict[str, List[str]] = {"name": [str(config.get("name") or "")]} + for key, default in (("model-paths", ["models"]), ("test-paths", ["tests"])): + raw = config.get(key, default) + entries = [raw] if isinstance(raw, str) else list(raw or []) + out[key] = [str(PurePosixPath(str(e))) for e in entries if str(e).strip()] + return out + + +def _model( + build: Mapping[str, Any], + table: Any, + project: _Project, + where: str, + refusals: List[UnsupportedBindingError], + *, + single: bool = True, +) -> Optional[str]: + """The dbt model that writes ``table``: the build's ``properties.model`` when it writes one + output (as ``build_selectors`` selects it), else the model named after the table. Refused + when the project does not hold it, or when the build's selection provably leaves it out + (its tests would never run); a selection by tag, config or state cannot be told here, + and the run's report says which tests did not run.""" + props = build.get("properties") if isinstance(build.get("properties"), Mapping) else {} + outputs = build.get("outputs") if isinstance(build.get("outputs"), list) else [] + named = props.get("model") + # ``properties.model`` names the model of the one table a build writes; a build that + # writes several (no ``outputs`` writes every expose) writes each through its own model. + model = str(named) if named and len(outputs) <= 1 and single else str(table.name) + if ENV_PLACEHOLDER_RE.search(model): + return _DEFERRED + files = sorted( + rel + for rel in project.present + if PurePosixPath(rel).stem == model + and PurePosixPath(rel).suffix in (".sql", ".py") + and any(rel.startswith(f"{path}/") for path in project.model_paths) + ) + if not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", model) or not files: + refusals.append( + _refusal( + "model", + f"{where} writes expose {table.expose_id!r}, whose contract rules run as dbt " + f"tests on the model that builds it: model {model!r}, which the dbt project's " + f"model-paths ({', '.join(project.model_paths) or 'none'}) do not hold.", + "Name the model in builds[].properties.model (one output per build), or name it " + "after the expose's binding.location.table.", + ) + ) + return None + from ...build_runners.dbt.runner import build_selectors + from .databricks_retention import DbtModel, selects_model + + path = PurePosixPath(files[0]) + below = [p for p in project.model_paths if files[0].startswith(f"{p}/")][0] + folders = path.parent.relative_to(below).parts + node = DbtModel(name=model, materialized="", fqn=(project.name, *folders, model), path=files[0]) + selectors = build_selectors(dict(build)) + unresolved = any(ENV_PLACEHOLDER_RE.search(str(item)) for item in selectors) + if not unresolved and selects_model(selectors, node) is False: + refusals.append( + _refusal( + "not-selected", + f"{where} writes expose {table.expose_id!r} with model {model!r}, but the " + f"build's selection ({' '.join(selectors)!r}) leaves the model out, so dbt " + "would never run the tests of its contract rules.", + f"Select the model: properties.select: [{model!r}] (or properties.model).", + ) + ) + return None + return model + + +def _declared_sources(files: Sequence[Tuple[str, bytes]]) -> Dict[str, List[Tuple[str, str]]]: + """``{source table name: [(source name, table name)]}`` from every properties file the + upload carries.""" + found: Dict[str, List[Tuple[str, str]]] = {} + for doc in _properties_docs(files): + for source_name, table_name in _source_tables(doc): + found.setdefault(table_name.lower(), []).append((source_name, table_name)) + return found + + +def _properties_docs(files: Sequence[Tuple[str, bytes]]) -> Iterator[Mapping[str, Any]]: + """Each YAML document of the upload but dbt's top-level project files. One that does not + parse is passed over: dbt reports it itself.""" + import yaml + + for rel, data in files: + if PurePosixPath(rel).suffix not in (".yml", ".yaml") or rel in _PROJECT_YAML: + continue + try: + doc = yaml.safe_load(data.decode("utf-8")) + except (UnicodeDecodeError, RecursionError, yaml.YAMLError): + continue + if isinstance(doc, Mapping): + yield doc + + +def _source_tables(doc: Mapping[str, Any]) -> Iterator[Tuple[str, str]]: + """``(source name, table name)`` for each table a properties document declares.""" + sources = doc.get("sources") + for source in sources if isinstance(sources, list) else []: + if not isinstance(source, Mapping) or not isinstance(source.get("name"), str): + continue + tables = source.get("tables") + for table in tables if isinstance(tables, list) else []: + if isinstance(table, Mapping) and isinstance(table.get("name"), str): + yield source["name"], table["name"] + + +def _source_for( + check: DqCheck, + sources: Mapping[str, List[Tuple[str, str]]], + where: str, + refusals: List[UnsupportedBindingError], +) -> Optional[Tuple[str, str]]: + reference = check.reference + assert reference is not None + if reference.expose_id is None: + refusals.append( + _refusal( + "reference", + f"{where} rule {check.rule_id!r} on expose {check.expose_id!r} checks that every " + f"{check.column} exists in {reference.product_id!r}.{reference.column!r}, and " + f"consumes[] does not list {reference.product_id!r}: a dbt test reads the other " + "product from the dbt source the project declares for the expose it consumes.", + f"List {reference.product_id!r} in consumes[] with the exposeId of its table, and " + "declare that table as a dbt source the model reads.", + ) + ) + return None + candidates = sorted(set(sources.get(reference.expose_id.lower(), []))) + if len(candidates) == 1: + return candidates[0] + listed = ", ".join(f"{s}.{t}" for s, t in candidates) + refusals.append( + _refusal( + "reference", + f"{where} rule {check.rule_id!r} on expose {check.expose_id!r} checks that every " + f"{check.column} exists in {reference.product_id!r}.{reference.column!r}, read from the " + f"dbt source table named {reference.expose_id!r} (the expose this product consumes " + "it by), and the project declares " + + (f"{len(candidates)} ({listed})." if candidates else "none."), + f"Declare exactly one dbt source table named {reference.expose_id!r} for " + f"{reference.product_id!r}, as the model reads it.", + ) + ) + return None + + +def _exposes(tables: Sequence[Any]) -> str: + return ", ".join(repr(t.expose_id) for t in tables) + + +# ── rendering ──────────────────────────────────────────────────────────────── + + +def _jinja_str(text: str) -> str: + """``text`` as a Jinja string literal holding no ``{``, ``}``, ``%`` or ``#``: JSON, whose + escapes Jinja's string literals decode, with those four escaped too, so no value can open + or close a Jinja block or comment. Other characters stay as they are: ``ensure_ascii`` + would write an emoji as a surrogate pair, which Jinja decodes into two lone surrogates + that dbt cannot encode (the whole project's parse fails on one in a rule id).""" + encoded = json.dumps(text, ensure_ascii=False) + for char, escape in (("{", "\\u007b"), ("}", "\\u007d"), ("%", "\\u0025"), ("#", "\\u0023")): + encoded = encoded.replace(char, escape) + return encoded + + +def _text(sql: str) -> str: + """Literal SQL text, emitted through a Jinja expression so the template never reads it.""" + return "{{ " + _jinja_str(sql) + " }}" + + +def _number(value: float) -> str: + return str(int(value)) if float(value).is_integer() else repr(float(value)) + + +def _databricks_type(fluid_type: str) -> str: + """The type Databricks reports for a column the contract declares as ``fluid_type``, + lower case, no spaces: the type table the table and the model share + (``engines/dbt/_types.py``) in the table emitter's spelling (``decimal(12)`` is + ``decimal(12,0)``), ``decimal`` at Databricks' default ``decimal(10,0)``.""" + from ...engines.dbt._types import sql_type + from .databricks import _canonical_type + + resolved = _canonical_type(sql_type(fluid_type, "databricks").lower().replace(" ", "")) + return "decimal(10,0)" if resolved == "decimal" else resolved + + +def render(check: DqCheck, model: str, source: Optional[Tuple[str, str]] = None) -> str: + """The singular test file for ``check`` on ``model``. ``source`` is the ``(source, + table)`` a reference reads. Raises ``ValueError`` for a column that is no plain + identifier.""" + from .databricks_access import quote + + config = ( + f"{{{{ config(severity={_jinja_str(check.severity)}, enabled=true, " + f"tags=[{_jinja_str(TAG)}], meta={{'fluid_rule_id': {_jinja_str(check.rule_id)}, " + f"'fluid_expose_id': {_jinja_str(check.expose_id)}}}) }}}}" + ) + header = ( + f"-- Contract rule {check.name} ({check.rule_type}): written by fluid apply from the " + "contract. Change the contract, not this file.\n" + config + "\n" + ) + ref = f"ref({_jinja_str(model)})" + if check.kind == "schema": + return header + _schema_sql(check, ref) + column = quote(str(check.column)) + parent = None + if check.kind == "reference": + assert source is not None and check.reference is not None + parent = ( + f"source({_jinja_str(source[0])}, {_jinja_str(source[1])})", + quote(check.reference.column), + ) + if check.share is not None: + return header + _share_sql(check, ref, column, parent) + return header + _row_sql(check, ref, column, parent) + + +def _builtin(test: str, *args: str) -> str: + return f"{{{{ adapter.dispatch('{test}', 'dbt')({', '.join(args)}) }}}}\n" + + +def _literals(check: DqCheck) -> List[str]: + from .databricks_access import string_literal + + return [string_literal(value) for value in check.values] + + +def _predicate(check: DqCheck, column: str) -> str: + """The SQL text of the row condition an allowed value meets (regex, bounds, values).""" + from .databricks_access import string_literal + + if check.kind == "regex": + return f"cast({column} as string) rlike {_text(string_literal(str(check.pattern)))}" + if check.kind == "bounds": + return " and ".join( + f"{column} {_SQL_OPERATORS[op]} {_number(value)}" for op, value in check.bounds + ) + if check.kind == "accepted_values": + return f"{column} in ({', '.join(_text(lit) for lit in _literals(check))})" + raise AssertionError(check.kind) + + +def _row_sql(check: DqCheck, ref: str, column: str, parent: Optional[Tuple[str, str]]) -> str: + """Every row: a failing row is one the check rejects.""" + col = _jinja_str(column) + if check.kind == "not_null": + return _builtin("test_not_null", ref, col) + if check.kind == "unique": + return _builtin("test_unique", ref, col) + if check.kind == "accepted_values": + values = ", ".join(_jinja_str(lit) for lit in _literals(check)) + return _builtin("test_accepted_values", ref, col, f"[{values}]", "false") + if check.kind == "reference": + assert parent is not None + return _builtin("test_relationships", ref, col, parent[0], _jinja_str(parent[1])) + if check.kind in ("regex", "bounds"): + return ( + f"select {column} as failing_value\n" + f"from {{{{ {ref} }}}}\n" + f"where {column} is not null and not ({_predicate(check, column)})\n" + ) + if check.kind == "freshness": + older = "<=" if check.strict else "<" + return ( + "select newest_value\n" + f"from (select max({column}) as newest_value from {{{{ {ref} }}}}) as newest\n" + "where newest_value is null\n" + f" or newest_value {older} timestampadd(SECOND, -{int(check.max_age_seconds or 0)}, " + "current_timestamp())\n" + ) + raise AssertionError(check.kind) + + +def _share_sql(check: DqCheck, ref: str, column: str, parent: Optional[Tuple[str, str]]) -> str: + """A share: one failing row, holding the share measured, when it misses the threshold. + An empty table (no share to measure) passes, as every per-row test does.""" + operator, threshold = check.share or (">=", 1.0) + source = f"{{{{ {ref} }}}}" + if check.kind == "not_null": + passing, total = f"count({column})", "count(*)" + elif check.kind == "unique": + passing, total = f"count(distinct {column})", f"count({column})" + elif check.kind in ("accepted_values", "regex", "bounds"): + passing, total = f"count_if({_predicate(check, column)})", f"count({column})" + elif check.kind == "reference": + assert parent is not None + passing, total = "count(parent.reference_value)", f"count(child.{column})" + source = ( + f"{{{{ {ref} }}}} as child\n" + f" left join (select distinct {parent[1]} as reference_value " + f"from {{{{ {parent[0]} }}}}) as parent\n" + f" on child.{column} = parent.reference_value" + ) + else: + raise AssertionError(check.kind) + limit = _number(threshold) + return ( + f"select passing_share, {limit} as required_share\n" + f"from (\n select try_divide({passing}, {total}) as passing_share\n from {source}\n" + ") as measured\n" + f"where passing_share is not null and not (passing_share {_SQL_OPERATORS[operator]} " + f"{limit})\n" + ) + + +def _schema_sql(check: DqCheck, ref: str) -> str: + """One failing row per contract column the table lacks or types otherwise, read from the + built table (``adapter.get_columns_in_relation``, as dbt_expectations reads it). + + ``ref`` is called outside the ``execute`` guard: dbt records a test's parents while it + parses (``execute`` false then), and selects a test by them, so a ``ref`` only inside the + guard leaves the test without a parent and ``dbt build --select +model+`` never runs it + (seen live: 11 tests found, 10 run).""" + from .databricks_access import quote + + for name, _kind in check.columns: + quote(name) # a plain identifier, or ValueError + expected = ", ".join( + f"{_jinja_str(name.lower())}: {_jinja_str(_databricks_type(kind))}" + for name, kind in check.columns + ) + return ( + f"{{%- set relation = {ref} -%}}\n" + f"{{%- set expected = {{{expected}}} -%}}\n" + "{%- set problems = [] -%}\n" + "{%- if execute -%}\n" + " {%- set actual = {} -%}\n" + " {%- for column in adapter.get_columns_in_relation(relation) -%}\n" + " {%- do actual.update({column.name | lower: column.dtype | lower" + " | replace(' ', '')}) -%}\n" + " {%- endfor -%}\n" + " {%- for name, kind in expected.items() -%}\n" + " {%- if name not in actual -%}\n" + " {%- do problems.append([name, 'missing', kind, '']) -%}\n" + " {%- elif actual[name] != kind -%}\n" + " {%- do problems.append([name, 'type', kind, actual[name]]) -%}\n" + " {%- endif -%}\n" + " {%- endfor -%}\n" + "{%- endif -%}\n" + "{%- for problem in problems %}\n" + "select '{{ problem[0] }}' as column_name, '{{ problem[1] }}' as problem," + ' \'{{ problem[2] | replace("\\\\", "\\\\\\\\") | replace("\'", "\\\\\'") }}\'' + " as expected_type," + ' \'{{ problem[3] | replace("\\\\", "\\\\\\\\") | replace("\'", "\\\\\'") }}\'' + " as actual_type\n" + "{%- if not loop.last %}\nunion all{% endif %}\n" + "{%- else %}\n" + "select cast(null as string) as column_name, cast(null as string) as problem," + " cast(null as string) as expected_type, cast(null as string) as actual_type\n" + "where 1 = 0\n" + "{%- endfor %}\n" + ) + + +# ── results ────────────────────────────────────────────────────────────────── + + +def results(logs: str) -> Dict[str, Tuple[str, Optional[int]]]: + """``{test name: (status, failures)}`` from dbt's output; the last line wins (a retried + task prints its tests again).""" + found: Dict[str, Tuple[str, Optional[int]]] = {} + for match in _RESULT_RE.finditer(_ANSI_RE.sub("", logs or "")): + status, count, name = match.group(1), match.group(2), match.group(3) + found[name] = (status, int(count) if count else None) + return found + + +def _outcome(status: Optional[str], failures: Optional[int], truncated: bool) -> str: + """What dbt said of one test, in words.""" + if status is None: + return "not in the task's output" + (" (Databricks kept only its end)" if truncated else "") + if failures is not None: + return f"{status} {failures} {'row' if failures == 1 else 'rows'}" + return status + + +def report(logs: str, checks: Sequence[DqCheck], *, truncated: bool = False) -> List[str]: + """Lines for ``fluid apply``: a summary, then one line per contract rule's test with what + dbt printed for it. Empty when the build tests no rule.""" + if not checks: + return [] + found = results(logs) + if not any(check.name in found for check in checks): + kept = " (Databricks kept only its end)" if truncated else "" + return [ + f" contract rules as dbt tests: none of the {len(checks)} reported in the task's " + f"output{kept}; dbt stopped before running them, or the run page has the rest" + ] + lines: List[str] = [] + tally: Dict[str, int] = {} + marks = {"PASS": "✓", "WARN": "!", "FAIL": "✗", "ERROR": "✗", "SKIP": "-"} + for check in checks: + status, failures = found.get(check.name, (None, None)) + key = status or "MISSING" + tally[key] = tally.get(key, 0) + 1 + outcome = _outcome(status, failures, truncated) + lines.append( + f" {marks.get(key, '?')} {check.name}: {outcome} - {describe(check)}, " + f"severity {check.severity}" + ) + order = ("PASS", "WARN", "FAIL", "ERROR", "SKIP", "MISSING") + words = {"MISSING": "not reported"} + summary = ", ".join(f"{tally[k]} {words.get(k, k.lower())}" for k in order if tally.get(k)) + return [f" contract rules as dbt tests: {summary}"] + lines + + +__all__ = [ + "CompiledTests", + "DqTest", + "TAG", + "TEST_DIR", + "compile_tests", + "expected_tests", + "render", + "report", + "results", + "written_tables", +] diff --git a/fluid_build/iac/providers/databricks_jobs.py b/fluid_build/iac/providers/databricks_jobs.py index 243c56da..0d323232 100644 --- a/fluid_build/iac/providers/databricks_jobs.py +++ b/fluid_build/iac/providers/databricks_jobs.py @@ -31,6 +31,11 @@ * ``execution.retries``, ``notifications``, ``runtime.timeout`` and ``runtime.serviceAccount``: the task's retries and timeout, the job's email notifications and its ``run_as``. +* ``exposes[].contract.dq.rules`` of each expose the build writes: one dbt test + per check, uploaded beside the project under ``/fluid_dq/`` + (``databricks_dbt_tests.py``), which ``dbt build`` runs after the model; a + rule no test can run is refused, and so is a Git source with rules, since no + file can be added to a checkout. The catalog and schema dbt writes to are those of the first Databricks expose the build writes (``builds[].outputs``). Everything the job would accept with @@ -88,6 +93,9 @@ PRODUCT_TAG = "fluid_product" BUILD_TAG = "fluid_build" +#: The key of the job's dbt task: what a run's report finds the dbt output under. +DBT_TASK_KEY = "dbt" + #: The data source that names the deploying identity's home folder. CURRENT_USER = "databricks_current_user" _HOME = tofu_ref(f"data.{CURRENT_USER}.me.home") @@ -155,6 +163,8 @@ class DbtJob: files: Tuple[str, ...] #: Where the files go, below the applying identity's home (WORKSPACE source). root: str = "" + #: The dbt tests its contract's DQ rules run as, by name (``databricks_dbt_tests.py``). + tests: Tuple[str, ...] = () @dataclass @@ -745,6 +755,8 @@ class _Source: git: Optional[Dict[str, Any]] = None directory: Optional[str] = None files: Tuple[Tuple[str, bytes], ...] = () + #: The dbt tests of the contract's DQ rules among ``files``, by name (``_with_tests``). + tests: Tuple[str, ...] = () def job_product(contract: Mapping[str, Any]) -> str: @@ -801,12 +813,54 @@ def compile_dbt_jobs( settings = _job_settings(item, compiled) table = _target_table(item, builds, tables, compiled.refusals) source = _project_source(item, compiled.refusals, validating=validating) + if source is not None: + source = _with_tests(contract, item, builds, tables, source, compiled.refusals) if len(compiled.refusals) == before and table is not None and source is not None: script = rebuild_script(build) if rebuild_script is not None else None _emit_job(compiled, item, table, settings, source, script) return compiled +def _with_tests( + contract: Mapping[str, Any], + item: _Build, + builds: Sequence[Mapping[str, Any]], + tables: Sequence[Any], + source: _Source, + refusals: List[UnsupportedBindingError], +) -> _Source: + """``source`` with the dbt tests the contract's DQ rules run as (``databricks_dbt_tests.py``). + A Git checkout takes none, so its rules are refused.""" + from .databricks_dbt_tests import compile_tests, written_tables + + git = source.git is not None + compiled = compile_tests( + contract, + item.build, + item.where, + written_tables(item.build, builds, tables), + # No file at all: ``fluid validate`` without a contract location (``_project_source``). + None if git or not source.files else source.files, + git=git, + ) + refusals.extend(compiled.refusals) + if not compiled.files: + return source + files = tuple(sorted((*source.files, *compiled.files))) + if len(files) > MAX_PROJECT_FILES: + refusals.append( + _refusal( + "project-size", + f"{item.where} the dbt project and the {len(compiled.files)} tests of its " + f"contract rules upload {len(files)} files; the workspace upload takes " + f"{MAX_PROJECT_FILES} at most, since each is a resource in the module.", + "Use a GIT source for a project this size.", + ) + ) + tests = tuple(test.name for test in compiled.tests) + return _Source(git=source.git, directory=source.directory, files=files, tests=tests) + + def _job_settings(item: _Build, compiled: CompiledJobs) -> _Settings: build, where, refusals = item.build, item.where, compiled.refusals engine = str(build.get("engine") or "").strip().lower() @@ -957,10 +1011,8 @@ def _emit_job( if depends: job["depends_on"] = depends compiled.data.setdefault(CURRENT_USER, {})["me"] = {} - job["task"] = [ - {"task_key": "dbt", "environment_key": "dbt", "dbt_task": dbt_task, **settings.task}, - *([_keys_task(item)] if keys_file else []), - ] + dbt = {"task_key": DBT_TASK_KEY, "environment_key": "dbt", "dbt_task": dbt_task} + job["task"] = [{**dbt, **settings.task}, *([_keys_task(item)] if keys_file else [])] compiled.resources.append(("databricks_job", item.key, f"{item.where} job", job)) compiled.jobs.append( DbtJob( @@ -971,16 +1023,21 @@ def _emit_job( trigger=settings.trigger, files=tuple(rel for rel, _ in uploads), root=item.root, + tests=source.tests, ) ) def _keys_task(item: _Build) -> Dict[str, Any]: - """The ``fluid_keys`` task: once dbt has succeeded, the uploaded script on the same - warehouse.""" + """The ``fluid_keys`` task: once dbt has ended, the uploaded script on the same warehouse. + ``run_if: ALL_DONE``, not the default ALL_SUCCESS: a contract rule's test of severity + error fails dbt after its model's CREATE OR REPLACE dropped the table's keys, and the + run that finds bad data must not also leave the table keyless (the script passes over a + key it cannot add, such as one on a table dbt did not rebuild).""" return { "task_key": REBUILD_TASK, - "depends_on": [{"task_key": "dbt"}], + "depends_on": [{"task_key": DBT_TASK_KEY}], + "run_if": "ALL_DONE", "timeout_seconds": REBUILD_TIMEOUT_SECONDS, "sql_task": { "warehouse_id": item.warehouse, @@ -1044,6 +1101,7 @@ def folder(rel_dir: str) -> str: __all__ = [ + "DBT_TASK_KEY", "CompiledJobs", "DbtJob", "compile_dbt_jobs", diff --git a/fluid_build/util/dq_checks.py b/fluid_build/util/dq_checks.py new file mode 100644 index 00000000..f9960b9d --- /dev/null +++ b/fluid_build/util/dq_checks.py @@ -0,0 +1,702 @@ +# Copyright 2025 Agentics Transformation Limited +# SPDX-License-Identifier: Apache-2.0 + +"""A contract's data-quality rules, read as checks an engine can run. + +``exposes[].contract.dq.rules[]`` (``$defs.dqRule``, closed in FLUID 0.7.x: ``id``, ``type``, +``selector``, ``threshold``, ``operator``, ``window``, ``severity``, ``description``, ``tags``, +``labels``) says what must hold. :func:`read_rules` reads each rule once, into +:class:`DqCheck` values that say how, for whichever engine runs them (the Databricks dbt job +first: ``iac/providers/databricks_dbt_tests.py``), and into a :class:`DqRefusal` for each rule +no check can run as written. A rule is never skipped in silence. + +How each rule type reads: + +* ``completeness``: ``not_null`` on each selected column. ``selector`` may list several, + comma-separated; each gets its own check. +* ``uniqueness``: ``unique`` on the one selected column. A list is refused: it could mean each + column or their combination, and ``fluid test`` reads one column too. +* ``valid_values``: a valid value of each selected column is what ``fluid test`` takes it + to be (``providers/quality_engine.py::value_domain``): the values the rule itself lists + (``" valid values: A, B."`` in its description), or else every declaration the + column makes, which must all hold: its first ``enum`` (``"A | B"`` or ``"A,B"``), ``regex`` + and ``range`` (``">= 0"``, ``">= 0 and <= 100"`` or ``"between 0 and 100"``) + validationRule, and its ``labels.references: .``, the foreign-key + convention the Unity Catalog keys read too; ``consumes[]`` says which of that product's + exposes holds the column. Each is a check. With none, the rule is refused. +* ``accuracy``: every value of each selected column meets `` ``, the bound + ``fluid test`` checks (``providers/quality_engine.py::_check_accuracy``). +* ``freshness``: the newest value of the one selected date or timestamp column is no older than + ``window`` (ISO 8601). ``operator`` ``<=`` (the default) allows exactly the window's age; + ``<`` does not. +* ``schema``: each contract column, or each selected one, exists with its declared type. + Columns the contract does not declare are left alone (additive evolution). +* ``anomaly_detection`` and ``drift_detection`` compare a table with its own history, which a + check of the table as it stands cannot do: refused. + +``threshold`` is the share of rows that pass, from 0 to 1: the non-null share for +completeness and the distinct share of non-null values for uniqueness, as ``fluid test`` +computes them, and for valid_values the share of non-null values that are allowed (``fluid +test`` checks every value of a valid_values rule, whatever its threshold). ``1.0`` with ``>=`` +or ``==`` (or no threshold at all) means every row, a per-row check; anything else is a +:attr:`DqCheck.share`. Accuracy's threshold is its bound, not a share; freshness and schema +rules take none. + +``severity``: ``critical`` and ``error`` fail the run (``error``); ``warn`` and ``info`` are +reported and do not (``warn``), the two levels a dbt test has. +""" + +from __future__ import annotations + +import math +import re +from dataclasses import dataclass, field, replace +from typing import Any, Dict, List, Mapping, Optional, Sequence, Set, Tuple + +#: ``dqRule.severity`` -> the level its checks run at. +SEVERITIES: Dict[str, str] = {"critical": "error", "error": "error", "warn": "warn", "info": "warn"} + +#: The comparisons a threshold or a bound may use (``$defs.dqRule.operator``). +OPERATORS = (">=", ">", "<=", "<", "==", "!=") + +#: What a check asserts; an engine renders each kind or refuses the check. +KINDS = ( + "not_null", + "unique", + "accepted_values", + "regex", + "bounds", + "reference", + "freshness", + "schema", +) + +#: The base types a freshness check can measure. +_TEMPORAL = ("date", "datetime", "smalldatetime", "timestamp") + +_NAME_RE = re.compile(r"[^0-9a-z_]+") +#: A control character, or a UTF-16 surrogate with no partner (YAML's "\ud83d" escape +#: loads one), which no UTF-8 file can hold: refused wherever a rule's text reaches a check. +_CONTROL_RE = re.compile(r"[\x00-\x1f\x7f-\x9f\ud800-\udfff]") + + +@dataclass(frozen=True) +class Reference: + """Where a value must exist: ``column`` of a table of ``product_id``. ``expose_id`` is the + expose ``consumes[]`` lists that product by, or ``None`` when it does not list it.""" + + product_id: str + expose_id: Optional[str] + column: str + + +@dataclass(frozen=True) +class DqCheck: + """One check of one rule, on one column or on the table.""" + + #: Unique among the checks :func:`read_rules` returned together: the rule id (lower case, + #: ``[0-9a-z_]``), then ``__`` for one of several columns, then ``__`` for + #: one of several declarations on a column. + name: str + rule_id: str + rule_type: str + expose_id: str + #: One of :data:`KINDS`. + kind: str + #: ``error`` or ``warn`` (:data:`SEVERITIES`). + severity: str + column: Optional[str] = None + #: ``(operator, threshold)``: the share of rows that must pass. ``None``: every row. + share: Optional[Tuple[str, float]] = None + #: ``accepted_values``: the allowed values, as text. + values: Tuple[str, ...] = () + #: ``regex``: the pattern every value matches. + pattern: Optional[str] = None + #: ``bounds``: ``(operator, number)`` pairs every value meets. + bounds: Tuple[Tuple[str, float], ...] = () + #: ``reference``: where every value must exist. + reference: Optional[Reference] = None + #: ``freshness``: the oldest the newest value may be, in seconds. + max_age_seconds: Optional[int] = None + #: ``freshness``: the newest value must be younger than the window, not at it. + strict: bool = False + #: ``schema``: ``(column, FLUID type)`` for each column that must exist. + columns: Tuple[Tuple[str, str], ...] = () + #: The rule as written, for an engine with more levels than two (Lakeflow fails an + #: update on ``critical`` and drops the row on ``error``) or a report with one row per + #: rule: its ``severity`` (lower case), ``threshold`` and ``operator``. + rule_severity: str = "" + rule_threshold: Optional[float] = None + rule_operator: Optional[str] = None + + +@dataclass(frozen=True) +class DqRefusal: + """A rule that cannot run as written: what is wrong, and what to change.""" + + rule_id: str + expose_id: str + message: str + fix: str + #: The rule's ``type``: an engine that runs some types another way (Databricks data quality + #: monitoring learns ``anomaly_detection`` bounds itself) can tell which refusals are its own. + rule_type: str = "" + #: The one refusal such an engine takes over: the rule is judged against the table's + #: history. A rule it cannot read either (a bad severity, an id holding a control + #: character) is refused with ``needs_history`` false, as any other. + needs_history: bool = False + + +@dataclass +class DqRules: + """The checks for some exposes' rules, and every refusal.""" + + checks: List[DqCheck] = field(default_factory=list) + refusals: List[DqRefusal] = field(default_factory=list) + + +def read_rules(contract: Mapping[str, Any], exposes: Sequence[Mapping[str, Any]]) -> DqRules: + """The checks for every ``contract.dq.rules[]`` entry of ``exposes``, named uniquely + across them, and a refusal for each rule none can run.""" + out = DqRules() + consumes = [c for c in contract.get("consumes") or [] if isinstance(c, Mapping)] + product = str(contract.get("id") or "") + taken: Set[str] = set() + for exposure in exposes: + expose_id, columns, rules = _parts(exposure) + for rule in rules: + reader = _Rule(rule, expose_id, columns, consumes, product) + checks = reader.read() + out.refusals.extend(reader.refusals) + out.checks.extend(_unique(check, taken) for check in checks) + return out + + +def has_rules(exposes: Sequence[Mapping[str, Any]]) -> bool: + """Whether any of ``exposes`` declares a ``contract.dq.rules[]`` entry.""" + return any(_parts(exposure)[2] for exposure in exposes if isinstance(exposure, Mapping)) + + +def _parts( + exposure: Mapping[str, Any] +) -> Tuple[str, List[Mapping[str, Any]], List[Mapping[str, Any]]]: + """An expose's id, its named schema columns and its DQ rules.""" + spec = exposure.get("contract") + spec = spec if isinstance(spec, Mapping) else {} + dq = spec.get("dq") + dq = dq if isinstance(dq, Mapping) else {} + columns = [ + c + for c in spec.get("schema") or [] + if isinstance(c, Mapping) and isinstance(c.get("name"), str) and c["name"] + ] + rules = dq.get("rules") + rules = [r for r in rules if isinstance(r, Mapping)] if isinstance(rules, list) else [] + return str(exposure.get("exposeId") or ""), columns, rules + + +def _base_name(rule_id: str) -> str: + return _NAME_RE.sub("_", rule_id.lower()).strip("_") or "rule" + + +def _unique(check: DqCheck, taken: Set[str]) -> DqCheck: + """``check``, renamed when another check already has its name: ``__``, then a + count. Rule ids are unique per expose at most, and nothing promises even that.""" + name = check.name + if name in taken: + name = f"{check.name}__{_base_name(check.expose_id)}" + count = 2 + while name in taken: + name = f"{check.name}_{count}" + count += 1 + taken.add(name) + if name == check.name: + return check + return replace(check, name=name) + + +def _finite(value: Any) -> Optional[float]: + """``value`` as a finite float; ``None`` for a bool, a non-number, an infinity, NaN, or + an integer too large for a float (YAML loads any number of digits).""" + if isinstance(value, bool) or not isinstance(value, (int, float)): + return None + try: + number = float(value) + except OverflowError: + return None + return number if math.isfinite(number) else None + + +def _empty(bounds: Sequence[Tuple[str, float]]) -> bool: + """Whether no number meets every bound (``between 100 and 0``, ``> 5 and < 1``).""" + lows = [(v, op == ">") for op, v in bounds if op in (">", ">=")] + highs = [(v, op == "<") for op, v in bounds if op in ("<", "<=")] + if not lows or not highs: + return False + low = max(lows, key=lambda b: (b[0], b[1])) + high = min(highs, key=lambda b: (b[0], not b[1])) + return low[0] > high[0] or (low[0] == high[0] and (low[1] or high[1])) + + +def _number(value: float) -> str: + return str(int(value)) if float(value).is_integer() else repr(float(value)) + + +def _declared(column: Mapping[str, Any], kind: str) -> Optional[str]: + """The constraint of the column's first ``kind`` validationRule, as ``fluid test`` picks + it (``providers/quality_engine.py::_column_constraint``).""" + rules = column.get("validationRules") + for entry in rules if isinstance(rules, list) else []: + if not isinstance(entry, Mapping): + continue + if str(entry.get("type") or "").strip().lower() != kind: + continue + constraint = entry.get("constraint") + if isinstance(constraint, str) and constraint.strip(): + return constraint.strip() + return None + + +def describe(check: DqCheck) -> str: + """What ``check`` asserts, in a few words, for a report.""" + column = f" {check.column}" if check.column else "" + if check.share is not None: + return f"{check.rule_type}{column} (share {check.share[0]} {_number(check.share[1])})" + return f"{check.rule_type}{column}" + + +class _Rule: + """One rule being read; its refusals collect in :attr:`refusals`.""" + + def __init__( + self, + rule: Mapping[str, Any], + expose_id: str, + columns: Sequence[Mapping[str, Any]], + consumes: Sequence[Mapping[str, Any]], + product: str, + ) -> None: + self.rule = rule + self.rule_id = str(rule.get("id") or "") + self.type = str(rule.get("type") or "").strip().lower() + self.expose_id = expose_id + self.columns = {str(c["name"]): c for c in columns} + self.consumes = consumes + self.product = product + self.refusals: List[DqRefusal] = [] + + # ── reading ─────────────────────────────────────────────────────────────── + + def read(self) -> List[DqCheck]: + if _CONTROL_RE.search(self.rule_id) or _CONTROL_RE.search(self.expose_id): + self.refuse( + "has an id (or its expose's) holding a control character or an unpaired " + "surrogate", + "Write the id in plain text.", + ) + return [] + severity = SEVERITIES.get(str(self.rule.get("severity") or "").strip().lower()) + if severity is None: + self.refuse( + f"has severity {self.rule.get('severity')!r}", + "Set severity to critical, error, warn or info.", + ) + return [] + reader = { + "completeness": self._completeness, + "uniqueness": self._uniqueness, + "valid_values": self._valid_values, + "accuracy": self._accuracy, + "freshness": self._freshness, + "schema": self._schema, + }.get(self.type) + if reader is None: + if self.type in ("anomaly_detection", "drift_detection"): + self.refuse( + "has no check of the table as it stands here: an anomaly or a drift is " + "judged against the table's history", + "Check it with fluid test against the deployed table (it reads an " + "anomaly_detection threshold as a row count), or remove it from this expose.", + needs_history=True, + ) + else: + self.refuse( + "is not a type a check can run", + "Use completeness, uniqueness, valid_values, accuracy, freshness or schema.", + ) + return [] + found = reader() + if self.refusals: + return [] + return self._named(found, severity) + + def _named(self, found: List[Dict[str, Any]], severity: str) -> List[DqCheck]: + base = _base_name(self.rule_id) + columns = {item.get("column") for item in found} + per_column: Dict[Any, int] = {} + for item in found: + per_column[item.get("column")] = per_column.get(item.get("column"), 0) + 1 + checks = [] + for item in found: + parts = [base] + if len(columns) > 1 and item.get("column"): + parts.append(_base_name(item["column"])) + if per_column[item.get("column")] > 1: + parts.append(item["kind"]) + checks.append( + DqCheck( + name="__".join(parts), + rule_id=self.rule_id, + rule_type=self.type, + expose_id=self.expose_id, + severity=severity, + rule_severity=str(self.rule.get("severity") or "").strip().lower(), + rule_threshold=self._number_or_none(self.rule.get("threshold")), + rule_operator=self._operator_or_none(), + **item, + ) + ) + return checks + + def _completeness(self) -> List[Dict[str, Any]]: + columns = self._selected(required=True) + share = self._share() + return [{"kind": "not_null", "column": c, "share": share} for c in columns] + + def _uniqueness(self) -> List[Dict[str, Any]]: + columns = self._selected(required=True) + share = self._share() + if len(columns) > 1: + self.refuse( + f"selects {len(columns)} columns ({', '.join(columns)}); a uniqueness rule checks " + "one column, as fluid test reads it, and a list could mean each column or their " + "combination", + "Write one uniqueness rule per column.", + ) + return [] + return [{"kind": "unique", "column": c, "share": share} for c in columns] + + def _valid_values(self) -> List[Dict[str, Any]]: + columns = self._selected(required=True) + share = self._share() + found: List[Dict[str, Any]] = [] + for name in columns: + for item in self._allowed(name) or []: + found.append({**item, "column": name, "share": share}) + return found + + def _accuracy(self) -> List[Dict[str, Any]]: + columns = self._selected(required=True) + threshold = _finite(self.rule.get("threshold")) + operator = self._operator() + if threshold is None: + self.refuse( + "has no finite numeric threshold, the bound each value of an accuracy rule " + "meets", + "Set threshold to the bound and operator to the comparison (>= 0, say).", + ) + return [] + if operator is None: + return [] + return [ + {"kind": "bounds", "column": c, "bounds": ((operator, threshold),)} for c in columns + ] + + def _freshness(self) -> List[Dict[str, Any]]: + columns = self._selected(required=True) + if self.rule.get("threshold") is not None: + self.refuse( + "has a threshold, which a freshness rule does not take: its window is the limit", + "Remove threshold and put the age limit in window (PT2H, say).", + ) + operator = str(self.rule.get("operator") or "<=").strip() + if operator not in ("<=", "<"): + self.refuse( + f"has operator {operator!r}; a freshness rule bounds the newest value's age from " + "above", + "Use <= (no older than the window) or < (younger than it).", + ) + if len(columns) > 1: + self.refuse( + f"selects {len(columns)} columns; a freshness rule measures one date or " + "timestamp column", + "Select the one column the age is measured on.", + ) + window = self.rule.get("window") + from fluid_build.providers.quality_engine import parse_duration_seconds + + try: + seconds = parse_duration_seconds(str(window)) if window else None + except (OverflowError, ValueError): + seconds = None # more digits than a duration holds + if seconds is None: + self.refuse( + f"has window {window!r}, which is not a fixed ISO 8601 duration", + "Set window to a duration such as PT15M, PT2H or P1D (months and years have " + "no fixed length).", + ) + for name in columns: + declared = str(self.columns[name].get("type") or "").strip().lower() + if not declared.startswith(_TEMPORAL): + self.refuse( + f"measures column {name!r}, of type {declared or 'unknown'}, and the age of " + "a value needs a date or timestamp column", + "Select a DATE or TIMESTAMP column.", + ) + if self.refusals: + return [] + return [ + { + "kind": "freshness", + "column": columns[0], + "max_age_seconds": seconds, + "strict": operator == "<", + } + ] + + def _schema(self) -> List[Dict[str, Any]]: + if self.rule.get("threshold") is not None: + self.refuse( + "has a threshold, which a schema rule does not take", + "Remove threshold from the rule.", + ) + return [] + columns = self._selected(required=False) + if self.refusals: + return [] + names = columns if columns else list(self.columns) + if not names: + self.refuse( + "checks the expose's schema, and exposes[].contract.schema declares no column", + "Declare the expose's columns.", + ) + return [] + return [ + { + "kind": "schema", + "columns": tuple((n, str(self.columns[n].get("type") or "")) for n in names), + } + ] + + # ── parts ───────────────────────────────────────────────────────────────── + + def _selected(self, *, required: bool) -> List[str]: + """The contract columns ``selector`` names, in its order; ``[]`` for ``*`` or none + (a refusal when ``required``).""" + raw = self.rule.get("selector") + text = raw.strip() if isinstance(raw, str) else "" + if not text or text == "*": + if required: + self.refuse( + "selects no column, and it checks columns", + "Set selector to the column, or to several, comma-separated.", + ) + return [] + names: List[str] = [] + for part in text.split(","): + name = self._column(part.strip()) + if name is None: + self.refuse( + f"selects {part.strip()!r}, which exposes[].contract.schema does not " + "declare", + "Name a column of the expose's schema, or declare the column there.", + ) + elif name not in names: + names.append(name) + return names + + def _column(self, name: str) -> Optional[str]: + """The schema's spelling of column ``name``: exact, else the one that differs only + in case (Databricks and most warehouses fold column names).""" + if name in self.columns: + return name + folded = [c for c in self.columns if c.lower() == name.lower()] + return folded[0] if len(folded) == 1 else None + + @staticmethod + def _number_or_none(value: Any) -> Optional[float]: + return _finite(value) + + def _operator_or_none(self) -> Optional[str]: + raw = self.rule.get("operator") + return str(raw).strip() if raw is not None else None + + def _operator(self) -> Optional[str]: + operator = str(self.rule.get("operator") or ">=").strip() + if operator not in OPERATORS: + self.refuse( + f"has operator {operator!r}", + f"Use one of {', '.join(OPERATORS)}.", + ) + return None + return operator + + def _share(self) -> Optional[Tuple[str, float]]: + threshold = self.rule.get("threshold") + if threshold is None: + return None + value = _finite(threshold) + if value is None: + self.refuse( + f"has threshold {threshold!r}, which is not a finite number", + "Set threshold to the share of rows that must pass, from 0 to 1.", + ) + return None + operator = self._operator() + if not 0.0 <= value <= 1.0: + self.refuse( + f"has threshold {threshold}, and a threshold is the share of rows that pass, " + "from 0 to 1 (1.0: every row)", + "Write the share as a fraction: 0.99, not 99.", + ) + return None + if operator is None or (value == 1.0 and operator in (">=", "==")): + return None + return (operator, value) + + def _allowed(self, name: str) -> Optional[List[Dict[str, Any]]]: + """What a valid value of column ``name`` is, read as ``fluid test`` reads it + (``providers/quality_engine.py::value_domain``), each part a check; ``None`` after a + refusal, which there is when neither the rule nor the column declares one. A + ``custom`` validationRule declares no allowed values.""" + from fluid_build.providers.quality_engine import extract_valid_values, value_domain + + rule = {**self.rule, "selector": name} + column = self.columns[name] + refused = len(self.refusals) + listed = extract_valid_values(rule) + if listed: # the values the rule lists win, as value_domain has it + found = [self._values(name, listed)] + else: + domain = value_domain(rule, [column]) + found = [ + self._values(name, domain.values) if domain.values else None, + self._regex(name, domain.pattern) if domain.pattern else None, + self._range(name, _declared(column, "range"), domain.bounds), + self._labelled_reference(name, column), + ] + if len(self.refusals) > refused: + return None + declared = [item for item in found if item is not None] + if not declared: + self.refuse( + f"reads column {name!r}, which declares no allowed values, and the rule lists " + "none", + f"Declare them on column {name!r}: validationRules [{{type: enum, constraint: " + "'A | B'}] (or regex, or range), or labels.references: . for " + "a value that must exist in another product's column.", + ) + return None + return declared + + def _values(self, name: str, values: Sequence[Any]) -> Optional[Dict[str, Any]]: + texts = tuple(str(value) for value in values) + if any(_CONTROL_RE.search(value) for value in texts): + self.refuse( + f"reads column {name!r}, whose allowed values hold a control character or an " + "unpaired surrogate", + "Remove it from the value.", + ) + return None + return {"kind": "accepted_values", "values": texts} + + def _regex(self, name: str, pattern: str) -> Optional[Dict[str, Any]]: + if _CONTROL_RE.search(pattern): + self.refuse( + f"reads column {name!r}, whose regex holds a control character or an unpaired " + "surrogate", + "Write it with an escape (\\t, \\n) instead.", + ) + return None + return {"kind": "regex", "pattern": pattern} + + def _range( + self, name: str, constraint: Optional[str], bounds: Sequence[Tuple[str, float]] + ) -> Optional[Dict[str, Any]]: + """``bounds`` (``quality_engine.parse_range`` of ``constraint``), or the refusal of a + range ``fluid test`` cannot read, which it passes over, or that no value can meet.""" + if constraint is None: + return None + if not bounds: + self.refuse( + f"reads column {name!r}, whose range {constraint!r} is not '>= 0', " + "'>= 0 and <= 100' or 'between 0 and 100'", + "Write the range in one of those forms.", + ) + return None + if _empty(bounds): + self.refuse( + f"reads column {name!r}, whose range {constraint!r} no value can meet", + "Write the lower bound first: 'between 0 and 100', '>= 0 and <= 100'.", + ) + return None + return {"kind": "bounds", "bounds": tuple(bounds)} + + def _labelled_reference(self, name: str, column: Mapping[str, Any]) -> Optional[Dict[str, Any]]: + labels = column.get("labels") + target = labels.get("references") if isinstance(labels, Mapping) else None + reference = self._reference(name, target) if target is not None else None + return None if reference is None else {"kind": "reference", "reference": reference} + + def _reference(self, name: str, target: Any) -> Optional[Reference]: + text = target.strip() if isinstance(target, str) else "" + product_id, _, column = text.rpartition(".") + if not product_id or not column: + self.refuse( + f"reads column {name!r}, whose labels.references {target!r} is not " + ".", + "Write it as the referenced product's id, a dot, and its column.", + ) + return None + if product_id == self.product: + self.refuse( + f"reads column {name!r}, which references {text!r} in this same product", + "Reference another product's column; a key within the product is not checked " + "here.", + ) + return None + exposes = sorted( + { + str(c.get("exposeId")) + for c in self.consumes + if c.get("productId") == product_id and c.get("exposeId") + } + ) + if len(exposes) > 1: + self.refuse( + f"reads column {name!r}, which references {text!r}, and consumes[] lists " + f"{product_id} by {len(exposes)} exposes ({', '.join(exposes)}), so which table " + "holds the column is ambiguous", + f"List {product_id} in consumes[] once, with the exposeId of the table that " + f"holds {column}.", + ) + return None + # Not consumed (a source-aligned product consumes nothing): an engine that finds the + # product's table another way can still check it; one that cannot refuses. + return Reference(product_id, exposes[0] if exposes else None, column) + + def refuse(self, problem: str, fix: str, *, needs_history: bool = False) -> None: + self.refusals.append( + DqRefusal( + rule_id=self.rule_id, + expose_id=self.expose_id, + message=f"rule {self.rule_id!r} ({self.type or 'no type'}) {problem}", + fix=fix, + rule_type=self.type, + needs_history=needs_history, + ) + ) + + +__all__ = [ + "DqCheck", + "DqRefusal", + "DqRules", + "KINDS", + "OPERATORS", + "Reference", + "SEVERITIES", + "describe", + "has_rules", + "read_rules", +] diff --git a/tests/build_runners/test_databricks_job_run.py b/tests/build_runners/test_databricks_job_run.py index 3acde3bf..08648d10 100644 --- a/tests/build_runners/test_databricks_job_run.py +++ b/tests/build_runners/test_databricks_job_run.py @@ -785,3 +785,111 @@ def sleep(self, seconds: float) -> None: assert signal.getsignal(signal.SIGHUP) is signal.SIG_IGN finally: signal.signal(signal.SIGHUP, before) + + +# ── the contract's DQ rules, as dbt tests of the run ─────────────────────── + +DQ_BUILD: Dict[str, Any] = { + "id": "transform", + "engine": "dbt", + "outputs": ["orders"], + "execution": {"runtime": {"platform": "databricks"}}, +} +DQ_CONTRACT: Dict[str, Any] = { + "id": "sales.orders", + "exposes": [ + { + "exposeId": "orders", + "binding": {"platform": "databricks"}, + "contract": { + "schema": [{"name": "id", "type": "bigint"}, {"name": "status", "type": "string"}], + "dq": { + "rules": [ + { + "id": "id_unique", + "type": "uniqueness", + "selector": "id", + "severity": "critical", + }, + { + "id": "status_set", + "type": "completeness", + "selector": "status", + "severity": "warn", + }, + ] + }, + }, + } + ], + "builds": [DQ_BUILD], +} +DQ_LOGS = ( + "\x1b[0m12:00:02 1 of 3 \x1b[32mPASS\x1b[0m id_unique ............ [\x1b[32mPASS\x1b[0m in 1.02s]\n" + "\x1b[0m12:00:03 2 of 3 \x1b[33mWARN 4\x1b[0m status_set .......... [\x1b[33mWARN 4\x1b[0m in 0.50s]\n" + "\x1b[0m12:00:04 Done. PASS=2 WARN=1 ERROR=0 SKIP=0 NO-OP=0 TOTAL=3\n" +) + + +def _go_dq(api: _Api, capsys: pytest.CaptureFixture) -> tuple: + time_ = _Time() + rc = run_job_build( + DQ_CONTRACT, + DQ_BUILD, + JOBS, + mode="amend-and-build", + client_factory=lambda: _Client(api), + env={}, + sleep=time_.sleep, + clock=time_.clock, + ) + captured = capsys.readouterr() + return rc, captured.out + captured.err + + +def test_a_success_reports_each_contract_rule_dbt_tested(capsys) -> None: + api = _Api( + [_run("TERMINATED", code="SUCCESS", tasks=[_task("SUCCESS")])], + outputs={78: {"logs": DQ_LOGS}}, + ) + rc, out = _go_dq(api, capsys) + assert rc == 0, out + assert "contract rules as dbt tests: 1 pass, 1 warn" in out + assert "✓ id_unique: PASS - uniqueness id, severity error" in out + assert "! status_set: WARN 4 rows - completeness status, severity warn" in out + # One read of the task's output serves the summary and the rules. + assert [c[1] for c in api.calls].count("runs/get-output") == 1 + + +def test_a_failed_test_is_named_before_the_end_of_the_output(capsys) -> None: + logs = DQ_LOGS.replace("PASS\x1b[0m id_unique", "FAIL 2\x1b[0m id_unique") + api = _Api( + [_run("TERMINATED", code="RUN_EXECUTION_ERROR", tasks=[_task("RUN_EXECUTION_ERROR")])], + outputs={78: {"error": "dbt command failed", "logs": logs}}, + ) + rc, out = _go_dq(api, capsys) + assert rc == 1 + flat = " ".join(out.split()) + assert "task 'dbt': dbt command failed" in flat + assert "✗ id_unique: FAIL 2 rows - uniqueness id, severity error" in flat + assert flat.index("id_unique: FAIL 2 rows") < flat.index("dbt output:") + + +def test_rules_are_reported_when_a_later_task_fails(capsys) -> None: + keys = {**_task("RUN_EXECUTION_ERROR", run_id=79), "task_key": "fluid_keys"} + api = _Api( + [_run("TERMINATED", code="RUN_EXECUTION_ERROR", tasks=[_task("SUCCESS"), keys])], + outputs={78: {"logs": DQ_LOGS}, 79: {"error": "key failed"}}, + ) + rc, out = _go_dq(api, capsys) + assert rc == 1 + assert "✓ id_unique: PASS" in out and "task 'fluid_keys': key failed" in out + + +def test_a_contract_without_rules_reports_none(capsys) -> None: + api = _Api( + [_run("TERMINATED", code="SUCCESS", tasks=[_task("SUCCESS")])], + outputs={78: {"logs": DQ_LOGS}}, + ) + rc, out = _go(api, capsys) + assert rc == 0 and "contract rules" not in out diff --git a/tests/iac/test_iac_databricks_dbt_tests.py b/tests/iac/test_iac_databricks_dbt_tests.py new file mode 100644 index 00000000..3b0fdee7 --- /dev/null +++ b/tests/iac/test_iac_databricks_dbt_tests.py @@ -0,0 +1,772 @@ +# Copyright 2025 Agentics Transformation Limited +# SPDX-License-Identifier: Apache-2.0 + +"""Contract DQ rules as dbt tests in a Databricks dbt job (``databricks_dbt_tests.py``). + +Stage 1 unit: no ``tofu``, no network. Each generated file is rendered with Jinja in a stand-in +for dbt's context (``config``, ``ref``, ``source``, ``adapter``), so what dbt would run is what +is asserted. The live proof (Databricks Free Edition) is recorded in HONESTLY_TESTED.md. +""" + +from __future__ import annotations + +import base64 +import copy +import shutil +import subprocess +import sys +from pathlib import Path +from typing import Any, Dict, List, Optional + +import jinja2 +import pytest + +from fluid_build.iac.providers import databricks as dbx +from fluid_build.iac.providers import databricks_dbt_tests as dq +from fluid_build.iac.providers.databricks_access import string_literal +from fluid_build.iac.source import contract_source +from fluid_build.util.dq_checks import DqCheck, Reference + +pytestmark = [pytest.mark.unit, pytest.mark.databricks] + +WAREHOUSE = "wh-0123456789abcdef" +GIT = "https://github.com/o/r?tag=v1.2.0" + +RULES: List[Dict[str, Any]] = [ + {"id": "orders_id_unique", "type": "uniqueness", "selector": "id", "severity": "critical"}, + { + "id": "orders_complete", + "type": "completeness", + "selector": "id,status", + "threshold": 1.0, + "operator": ">=", + "severity": "error", + }, + { + "id": "orders_status_valid", + "type": "valid_values", + "selector": "status", + "severity": "error", + }, + { + "id": "orders_country_valid", + "type": "valid_values", + "selector": "country", + "severity": "warn", + }, + { + "id": "orders_amount_positive", + "type": "valid_values", + "selector": "amount", + "severity": "error", + }, + { + "id": "orders_customer_ref", + "type": "valid_values", + "selector": "customer_id", + "severity": "warn", + }, + { + "id": "orders_fresh", + "type": "freshness", + "selector": "loaded_at", + "window": "PT2H", + "operator": "<=", + "severity": "warn", + }, + {"id": "orders_schema", "type": "schema", "selector": "*", "severity": "error"}, + { + "id": "orders_status_mostly", + "type": "completeness", + "selector": "status", + "threshold": 0.95, + "operator": ">=", + "severity": "error", + }, +] +SCHEMA: List[Dict[str, Any]] = [ + {"name": "id", "type": "bigint"}, + { + "name": "status", + "type": "string", + "validationRules": [{"type": "enum", "constraint": "NEW | SHIPPED"}], + }, + { + "name": "country", + "type": "string", + "validationRules": [{"type": "regex", "constraint": "^[A-Z]{2}$"}], + }, + { + "name": "amount", + "type": "decimal(12,2)", + "validationRules": [{"type": "range", "constraint": ">= 0"}], + }, + {"name": "customer_id", "type": "string", "labels": {"references": "crm.customers_v1.id"}}, + {"name": "loaded_at", "type": "timestamp"}, +] +SOURCES = ( + "version: 2\nsources:\n - name: crm\n tables:\n - name: customers\n" + " - name: other\n tables:\n - name: payments\n" +) +USER_YAML = "# the user's own model YAML, comments and all\nversion: 2\nmodels:\n - name: orders\n" + + +def _contract(repository: str = "./dbt", rules: Optional[List[Dict[str, Any]]] = None): + return { + "fluidVersion": "0.7.6", + "kind": "DataProduct", + "id": "sales.orders", + "name": "Orders", + "consumes": [{"productId": "crm.customers_v1", "exposeId": "customers"}], + "exposes": [ + { + "exposeId": "orders", + "kind": "table", + "binding": { + "platform": "databricks", + "format": "delta_table", + "location": {"database": "main", "schema": "sales", "table": "orders"}, + }, + "contract": { + "schema": copy.deepcopy(SCHEMA), + "dq": {"rules": copy.deepcopy(RULES if rules is None else rules)}, + }, + } + ], + "builds": [ + { + "id": "transform", + "engine": "dbt", + "repository": repository, + "outputs": ["orders"], + "properties": {"model": "orders"}, + "execution": {"runtime": {"platform": "databricks"}}, + } + ], + } + + +def _project(tmp_path: Path, files: Optional[Dict[str, str]] = None, config: str = "") -> Path: + root = tmp_path / "dbt" + root.mkdir(exist_ok=True) + (root / "dbt_project.yml").write_text(f"name: p\n{config}", encoding="utf-8") + default = { + "models/orders.sql": "select 1 as id\n", + "models/orders.yml": USER_YAML, + "models/sources.yml": SOURCES, + } + for rel, text in (default if files is None else files).items(): + path = root / rel + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text, encoding="utf-8") + return root + + +def _compile(contract, tmp_path: Optional[Path], *, validating: bool = False): + source = tmp_path / "contract.fluid.yaml" if tmp_path is not None else None + with contract_source(source): + return dbx._compile(contract, {dbx.WAREHOUSE_ENV: WAREHOUSE}, validating=validating) + + +def _kinds(compiled) -> List[str]: + return [refusal.kind for refusal in compiled.refusals] + + +def _uploaded(compiled) -> Dict[str, str]: + """``{path below the upload folder: text}`` of every workspace file.""" + out = {} + for body in compiled.resources.get("databricks_workspace_file", {}).values(): + rel = str(body["path"]).split("/files/", 1)[1] + out[rel] = base64.b64decode(body["content_base64"]).decode("utf-8") + return out + + +# ── a dbt-shaped Jinja context ─────────────────────────────────────────────── + + +class _Column: + def __init__(self, name: str, dtype: str) -> None: + self.name, self.dtype = name, dtype + + +class _Adapter: + def __init__(self, columns: List[_Column]) -> None: + self.columns = columns + self.calls: List[tuple] = [] + + def dispatch(self, name: str, namespace: str): + def call(*args: Any) -> str: + self.calls.append((name, namespace, args)) + return f"<{name}>" + + return call + + def get_columns_in_relation(self, relation: Any) -> List[_Column]: + return self.columns + + +def _render(text: str, columns: Optional[List[_Column]] = None, *, execute: bool = True): + """Render a generated file as dbt would: ``(sql, config kwargs, adapter calls)``. + ``execute=False`` is dbt's parse, which records the ``ref`` calls as the test's parents.""" + configs: Dict[str, Any] = {} + adapter = _Adapter(columns or []) + env = jinja2.Environment(extensions=["jinja2.ext.do"], undefined=jinja2.StrictUndefined) + refs: List[str] = [] + + def ref(name: str) -> str: + refs.append(name) + return f"`main`.`sales`.`{name}`" + + sql = env.from_string(text).render( + config=lambda **kwargs: configs.update(kwargs) or "", + ref=ref, + source=lambda source, table: f"`main`.`{source}`.`{table}`", + adapter=adapter, + execute=execute, + ) + _render.refs = refs # type: ignore[attr-defined] + return sql, configs, adapter.calls + + +def _check(kind: str, **fields: Any) -> DqCheck: + base = {"name": "r", "rule_id": "r", "rule_type": "valid_values", "expose_id": "orders"} + return DqCheck(**{**base, "kind": kind, "severity": "error", "column": "c", **fields}) + + +# ── the job ────────────────────────────────────────────────────────────────── + + +class TestTheJobUploadsATestPerCheck: + def test_each_check_is_a_test_file_beside_the_users_files(self, tmp_path): + _project(tmp_path) + compiled = _compile(_contract(), tmp_path) + assert not compiled.refusals, _kinds(compiled) + uploaded = _uploaded(compiled) + tests = sorted(p for p in uploaded if p.startswith("tests/fluid_dq/orders/")) + assert tests == sorted( + f"tests/fluid_dq/orders/{name}.sql" + for name in [ + "orders_id_unique", + "orders_complete__id", + "orders_complete__status", + "orders_status_valid", + "orders_country_valid", + "orders_amount_positive", + "orders_customer_ref", + "orders_fresh", + "orders_schema", + "orders_status_mostly", + ] + ) + # The user's own files upload byte for byte: no entry is merged into them. + assert uploaded["models/orders.yml"] == USER_YAML + (job,) = compiled.jobs + assert len(job.tests) == 10 + (task,) = compiled.resources["databricks_job"][ + next(iter(compiled.resources["databricks_job"])) + ]["task"] + assert task["dbt_task"]["commands"] == ["dbt deps", "dbt build --select +orders+"] + + def test_the_plan_lists_the_tests_of_the_job(self, tmp_path, monkeypatch): + monkeypatch.setenv(dbx.WAREHOUSE_ENV, WAREHOUSE) + _project(tmp_path) + with contract_source(tmp_path / "contract.fluid.yaml"): + actions = dbx.planned_objects(_contract()) + (job,) = [a for a in actions if a["op"] == "databricks.job.ensure"] + assert "orders_id_unique" in job["dq_tests"] + + def test_a_contract_without_rules_uploads_no_test(self, tmp_path): + _project(tmp_path) + compiled = _compile(_contract(rules=[]), tmp_path) + assert not compiled.refusals + assert not [p for p in _uploaded(compiled) if "fluid_dq" in p] + assert compiled.jobs[0].tests == () + + def test_the_first_test_path_holds_the_tests(self, tmp_path): + _project(tmp_path, config="test-paths: ['checks', 'more']\n") + compiled = _compile(_contract(), tmp_path) + assert any(p.startswith("checks/fluid_dq/orders/") for p in _uploaded(compiled)) + + +class TestRefusals: + def _refused(self, tmp_path, contract, kind: str) -> str: + compiled = _compile(contract, tmp_path) + assert f"databricks-job-dq-{kind}" in _kinds(compiled), _kinds(compiled) + assert "databricks_job" not in compiled.resources # never half a job + return " ".join(str(r) + " " + " ".join(r.remediation) for r in compiled.refusals) + + def test_a_git_source_with_rules_is_refused(self, tmp_path): + text = self._refused(tmp_path, _contract(GIT), "git-source") + assert "local dbt project" in text + + def test_a_git_source_without_rules_still_runs(self, tmp_path): + compiled = _compile(_contract(GIT, rules=[]), tmp_path) + assert not compiled.refusals + assert "databricks_job" in compiled.resources + + def test_a_rule_no_test_can_run_is_refused(self, tmp_path): + _project(tmp_path) + rules = [{"id": "drift", "type": "drift_detection", "selector": "*", "severity": "warn"}] + assert "the table's history" in self._refused(tmp_path, _contract(rules=rules), "rule") + + def test_an_anomaly_rule_is_left_to_data_quality_monitoring(self, tmp_path): + # databricks_quality.py (#864) turns monitoring on for an expose with such a rule, + # and the monitor learns its bounds: no dbt test, no refusal. + _project(tmp_path) + anomaly = {"id": "volume", "type": "anomaly_detection", "selector": "*", "severity": "warn"} + compiled = _compile(_contract(rules=[*RULES, anomaly]), tmp_path) + assert not compiled.refusals, _kinds(compiled) + assert "volume" not in compiled.jobs[0].tests + + @pytest.mark.parametrize("field, value", [("severity", "bogus"), ("id", "vol\nume")]) + def test_a_monitored_anomaly_rule_written_wrong_is_still_refused(self, tmp_path, field, value): + # Monitoring takes over only what needs the table's history. + _project(tmp_path) + anomaly = {"id": "volume", "type": "anomaly_detection", "selector": "*", "severity": "warn"} + anomaly[field] = value + text = self._refused(tmp_path, _contract(rules=[*RULES, anomaly]), "rule") + assert "monitoring back on" not in text + + def test_an_anomaly_rule_with_monitoring_turned_off_is_refused(self, tmp_path): + _project(tmp_path) + anomaly = {"id": "volume", "type": "anomaly_detection", "selector": "*", "severity": "warn"} + contract = _contract(rules=[anomaly]) + contract["exposes"][0]["contract"]["dq"]["monitoring"] = {"enabled": False} + assert "monitoring back on" in self._refused(tmp_path, contract, "rule") + + def test_a_model_the_project_does_not_hold_is_refused(self, tmp_path): + _project(tmp_path, files={"models/other.sql": "select 1\n", "models/sources.yml": SOURCES}) + assert "'orders'" in self._refused(tmp_path, _contract(), "model") + + @pytest.mark.parametrize( + "sources", + [ + "version: 2\nsources: []\n", + SOURCES + " - name: crm_copy\n tables:\n - name: customers\n", + ], + ) + def test_a_reference_needs_exactly_one_source_table_for_the_consumed_expose( + self, tmp_path, sources + ): + _project(tmp_path, files={"models/orders.sql": "select 1\n", "models/sources.yml": sources}) + text = self._refused(tmp_path, _contract(), "reference") + assert "named 'customers'" in text + + def test_a_reference_to_a_product_not_consumed_is_refused(self, tmp_path): + _project(tmp_path) + contract = _contract() + contract["consumes"] = [] + text = self._refused(tmp_path, contract, "reference") + assert "consumes[] does not list 'crm.customers_v1'" in text + + def test_a_user_file_where_a_test_goes_is_refused(self, tmp_path): + _project(tmp_path) + clash = tmp_path / "dbt/tests/fluid_dq/orders/orders_schema.sql" + clash.parent.mkdir(parents=True) + clash.write_text("select 1\n", encoding="utf-8") + self._refused(tmp_path, _contract(), "path") + + def test_a_test_name_the_projects_own_tests_use_is_refused(self, tmp_path): + # dbt names singular tests project-wide: a clash fails the whole job's parse + # ("dbt found two data_tests with the name"), so the plan refuses it first. + _project( + tmp_path, + files={ + "models/orders.sql": "select 1\n", + "models/sources.yml": SOURCES, + "tests/mine/orders_schema.sql": "select 1 where false\n", + }, + ) + assert "tests/mine/orders_schema.sql" in self._refused(tmp_path, _contract(), "name") + + def test_a_selection_that_leaves_the_model_out_is_refused(self, tmp_path): + _project(tmp_path) + contract = _contract() + contract["builds"][0]["properties"] = {"select": ["other_model"]} + assert "leaves the model out" in self._refused(tmp_path, contract, "not-selected") + + def test_a_selection_it_cannot_evaluate_is_left_to_the_runs_report(self, tmp_path): + _project(tmp_path) + contract = _contract() + contract["builds"][0]["properties"] = {"select": ["tag:nightly"]} + compiled = _compile(contract, tmp_path) + assert not compiled.refusals, _kinds(compiled) + assert compiled.jobs[0].tests + + def test_a_sibling_output_without_rules_needs_no_model(self, tmp_path): + # Only the exposes with rules are tested, so only their models are looked for. + _project(tmp_path) + contract = _contract() + sibling = copy.deepcopy(contract["exposes"][0]) + sibling.update(exposeId="extra") + sibling["binding"]["location"]["table"] = "extra" + sibling["contract"]["dq"] = {"rules": []} + contract["exposes"].append(sibling) + contract["builds"][0]["outputs"] = ["orders", "extra"] + compiled = _compile(contract, tmp_path) + assert not compiled.refusals, _kinds(compiled) + assert all( + p.startswith("tests/fluid_dq/orders/") for p in _uploaded(compiled) if "fluid_dq" in p + ) + + def test_each_table_of_a_build_without_outputs_is_tested_through_its_own_model(self, tmp_path): + files = { + "models/orders.sql": "select 1\n", + "models/extra.sql": "select 1\n", + "models/sources.yml": SOURCES, + } + _project(tmp_path, files=files) + contract = _contract(rules=[]) + extra = copy.deepcopy(contract["exposes"][0]) + extra.update(exposeId="extra") + extra["binding"]["location"]["table"] = "extra" + extra["contract"]["dq"] = { + "rules": [ + {"id": "extra_id", "type": "uniqueness", "selector": "id", "severity": "error"} + ] + } + contract["exposes"].append(extra) + del contract["builds"][0]["outputs"] # writes every expose + compiled = _compile(contract, tmp_path) + assert not compiled.refusals, _kinds(compiled) + assert "tests/fluid_dq/extra/extra_id.sql" in _uploaded(compiled) + + def test_a_user_test_of_that_name_under_fluid_dq_is_refused_too(self, tmp_path): + _project( + tmp_path, + files={ + "models/orders.sql": "select 1\n", + "models/sources.yml": SOURCES, + "tests/fluid_dq/legacy/orders_schema.sql": "select 1 where false\n", + }, + ) + self._refused(tmp_path, _contract(), "name") + + def test_a_project_yaml_too_deep_to_read_is_passed_over(self, tmp_path): + files = { + "models/orders.sql": "select 1\n", + "models/sources.yml": SOURCES, + "models/deep.yml": "x: " + "[" * 5000 + "]" * 5000 + "\n", + } + _project(tmp_path, files=files) + compiled = _compile(_contract(), tmp_path) + assert not compiled.refusals, _kinds(compiled) + + def test_validate_cannot_judge_a_selection_named_by_an_env_placeholder(self, tmp_path): + _project(tmp_path) + contract = _contract() + contract["builds"][0]["properties"] = {"model": "orders", "select": ["{{ env.SEL }}"]} + compiled = _compile(contract, tmp_path, validating=True) + assert "databricks-job-dq-not-selected" not in _kinds(compiled), _kinds(compiled) + + @pytest.mark.parametrize("where", ["rule id", "enum", "regex", "source name"]) + def test_an_unpaired_surrogate_is_refused_not_raised(self, tmp_path, where): + # YAML's "\ud83d" escape loads a lone surrogate, which no UTF-8 file can hold. + lone = "x\ud83d" + # The project's YAML carries the escape itself, which yaml.safe_load decodes. + escaped = 'name: "crm\\ud83d"' + sources = SOURCES.replace("name: crm", escaped) if where == "source name" else SOURCES + _project(tmp_path, files={"models/orders.sql": "select 1\n", "models/sources.yml": sources}) + contract = _contract() + rules = contract["exposes"][0]["contract"]["dq"]["rules"] + schema = contract["exposes"][0]["contract"]["schema"] + if where == "rule id": + rules[0]["id"] = lone + elif where == "enum": + schema[1]["validationRules"][0]["constraint"] = f"NEW | {lone}" + elif where == "regex": + schema[2]["validationRules"][0]["constraint"] = f"^{lone}$" + compiled = _compile(contract, tmp_path) + assert compiled.refusals, "a lone surrogate must be refused" + + def test_validate_defers_a_model_named_by_an_env_placeholder(self, tmp_path): + # fluid validate reads the contract before {{ env.* }} is substituted. + _project(tmp_path) + contract = _contract() + contract["exposes"][0]["binding"]["location"]["table"] = "{{ env.ORDERS_TABLE }}" + del contract["builds"][0]["properties"] + compiled = _compile(contract, tmp_path, validating=True) + assert "databricks-job-dq-model" not in _kinds(compiled), _kinds(compiled) + + def test_a_test_path_no_file_can_hold_is_refused_not_raised(self, tmp_path): + # YAML's escape loads a lone surrogate: no test file could be written under it. + _project(tmp_path, config='test-paths: ["t\\ud83d"]\n') + text = self._refused(tmp_path, _contract(), "test-path") + assert "plain text" in text + text.encode("utf-8") # the refusal prints: it names the path escaped + + def test_a_refusal_names_a_lone_surrogate_from_the_project_escaped(self, tmp_path): + # The model-paths reach the refusal's text, which fluid plan prints. + _project(tmp_path, config='model-paths: ["m\\ud83d"]\n') + text = self._refused(tmp_path, _contract(), "model") + assert "m\\ud83d" in text + text.encode("utf-8") + + def test_an_empty_test_paths_is_refused(self, tmp_path): + _project(tmp_path, config="test-paths: []\n") + self._refused(tmp_path, _contract(), "test-path") + + def test_a_referenced_column_that_is_no_plain_identifier_is_refused(self, tmp_path): + # The table's own columns are identifiers already (the table emitter refuses others); + # a reference names a column of another product's table. + _project(tmp_path) + contract = _contract() + column = contract["exposes"][0]["contract"]["schema"][4] + column["labels"]["references"] = "crm.customers_v1.odd name" + self._refused(tmp_path, contract, "column") + + def test_fluid_validate_without_a_contract_file_still_reads_the_rules(self): + rules = [{"id": "a", "type": "drift_detection", "selector": "*", "severity": "warn"}] + compiled = _compile(_contract(rules=rules), None, validating=True) + assert "databricks-job-dq-rule" in _kinds(compiled) + + +# ── what dbt runs ──────────────────────────────────────────────────────────── + + +PARSED = [ + ("not_null", {}), + ("unique", {}), + ("accepted_values", {"values": ("A",)}), + ("regex", {"pattern": "^A$"}), + ("bounds", {"bounds": ((">=", 0.0),)}), + ("freshness", {"max_age_seconds": 60}), + ("schema", {"column": None, "columns": (("id", "bigint"),)}), + ("not_null", {"share": (">=", 0.5)}), +] + + +class TestRendering: + @pytest.mark.parametrize("kind, fields", PARSED) + def test_every_test_refs_its_model_while_dbt_parses(self, kind, fields): + # dbt selects a test by the parents its parse recorded (``execute`` false): + # a ``ref`` only behind ``if execute`` left the schema test unselected, live. + _render(dq.render(_check(kind, **fields), "orders"), execute=False) + assert _render.refs == ["orders"] # type: ignore[attr-defined] + + def test_the_config_sets_severity_enabled_tag_and_the_rule(self): + _, configs, _ = _render(dq.render(_check("not_null", severity="warn"), "orders")) + assert configs == { + "severity": "warn", + "enabled": True, + "tags": ["fluid_dq"], + "meta": {"fluid_rule_id": "r", "fluid_expose_id": "orders"}, + } + + @pytest.mark.parametrize( + "kind, test", [("not_null", "test_not_null"), ("unique", "test_unique")] + ) + def test_not_null_and_unique_run_dbts_own_tests(self, kind, test): + _, _, calls = _render(dq.render(_check(kind), "orders")) + assert calls == [(test, "dbt", ("`main`.`sales`.`orders`", "`c`"))] + + def test_accepted_values_passes_databricks_literals_unquoted_by_dbt(self): + hostile = "O'Re}}il{%ly\\ #}" + _, _, calls = _render(dq.render(_check("accepted_values", values=("A", hostile)), "orders")) + ((name, _, args),) = calls + assert name == "test_accepted_values" + assert args[2] == [string_literal("A"), string_literal(hostile)] + assert args[3] is False # dbt adds no quotes of its own + + def test_a_reference_is_dbts_relationships_against_the_source_table(self): + check = _check("reference", reference=Reference("crm.customers_v1", "customers", "id")) + _, _, calls = _render(dq.render(check, "orders", ("crm", "customers"))) + assert calls == [ + ( + "test_relationships", + "dbt", + ("`main`.`sales`.`orders`", "`c`", "`main`.`crm`.`customers`", "`id`"), + ) + ] + + def test_a_regex_selects_the_non_null_values_that_do_not_match(self): + sql, _, _ = _render(dq.render(_check("regex", pattern="^\\d{2}'}}$"), "orders")) + literal = string_literal("^\\d{2}'}}$") + assert f"where `c` is not null and not (cast(`c` as string) rlike {literal})" in sql, sql + + def test_bounds_select_the_values_out_of_range(self): + check = _check("bounds", bounds=((">=", 0.0), ("<", 1.5))) + sql, _, _ = _render(dq.render(check, "orders")) + assert "where `c` is not null and not (`c` >= 0 and `c` < 1.5)" in sql + + @pytest.mark.parametrize("strict, older", [(False, "<"), (True, "<=")]) + def test_freshness_fails_on_no_row_or_an_old_newest_value(self, strict, older): + check = _check("freshness", max_age_seconds=7200, strict=strict) + sql, _, _ = _render(dq.render(check, "orders")) + assert "max(`c`) as newest_value from `main`.`sales`.`orders`" in sql + assert "where newest_value is null" in sql + assert f"newest_value {older} timestampadd(SECOND, -7200, current_timestamp())" in sql + + def test_a_share_fails_with_the_measured_share_when_it_misses(self): + check = _check("not_null", share=(">=", 0.95)) + sql, _, _ = _render(dq.render(check, "orders")) + assert "try_divide(count(`c`), count(*)) as passing_share" in sql + assert "where passing_share is not null and not (passing_share >= 0.95)" in sql + + def test_a_reference_share_counts_the_values_found_in_the_parent(self): + check = _check( + "reference", + share=(">", 0.5), + reference=Reference("crm.customers_v1", "customers", "id"), + ) + sql, _, _ = _render(dq.render(check, "orders", ("crm", "customers"))) + assert "try_divide(count(parent.reference_value), count(child.`c`))" in sql + assert "from `main`.`crm`.`customers`) as parent" in sql + assert "not (passing_share > 0.5)" in sql + + def test_schema_reports_each_missing_or_retyped_column(self): + check = _check( + "schema", + column=None, + columns=( + ("id", "bigint"), + ("amount", "decimal(12,2)"), + ("gone", "string"), + ("n", "number"), + ), + ) + text = dq.render(check, "orders") + columns = [ + _Column("ID", "BIGINT"), + _Column("amount", "decimal(10, 2)"), + _Column("n", "decimal(10,0)"), + ] + sql, _, _ = _render(text, columns) + rows = [line for line in sql.splitlines() if line.startswith("select ")] + assert rows == [ + "select 'amount' as column_name, 'type' as problem, " + + "'decimal(12,2)' as expected_type, 'decimal(10,2)' as actual_type", + "select 'gone' as column_name, 'missing' as problem, " + + "'string' as expected_type, '' as actual_type", + ] + assert "union all" in sql + + def test_a_conforming_schema_selects_no_row(self): + check = _check("schema", column=None, columns=(("id", "bigint"),)) + sql, _, _ = _render(dq.render(check, "orders"), [_Column("id", "bigint")]) + assert "where 1 = 0" in sql + + def test_a_decimal_without_scale_expects_the_scale_databricks_reports(self): + check = _check("schema", column=None, columns=(("amount", "decimal(12)"),)) + sql, _, _ = _render(dq.render(check, "orders"), [_Column("amount", "decimal(12,0)")]) + assert "where 1 = 0" in sql, sql + + def test_a_character_outside_the_bmp_reaches_dbt_whole(self): + # ensure_ascii wrote it as a surrogate pair, which dbt could not encode. + text = dq.render(_check("accepted_values", rule_id="r😀", values=("NEW😀",)), "orders") + _, configs, calls = _render(text) + assert configs["meta"]["fluid_rule_id"] == "r😀" + assert calls[0][2][2] == ["'NEW😀'"] + text.encode("utf-8") + + def test_a_type_the_warehouse_reports_cannot_break_out_of_its_literal(self): + check = _check("schema", column=None, columns=(("id", "bigint"),)) + sql, _, _ = _render(dq.render(check, "orders"), [_Column("id", "x' or '1'='1\\")]) + # Spaces are folded away first (``decimal(10, 2)`` is ``decimal(10,2)``). + assert "'x\\'or\\'1\\'=\\'1\\\\' as actual_type" in sql + + +# ── the run's results ─────────────────────────────────────────────────────── + +LOG = """\ +12:00:01 1 of 4 START test orders_id_unique .................................. [RUN] +12:00:02 1 of 4 PASS orders_id_unique ........................................ [PASS in 1.02s] +12:00:02 2 of 4 FAIL 3 orders_complete__id ................................... [FAIL 3 in 0.88s] +12:00:03 3 of 4 WARN 1 orders_fresh .......................................... [WARN 1 in 0.51s] +12:00:03 4 of 4 SKIP test orders_schema ...................................... [SKIP] +12:00:04 Done. PASS=1 WARN=1 ERROR=1 SKIP=1 NO-OP=0 TOTAL=4 +""" + + +class TestReport: + def test_results_read_dbts_line_for_each_test(self): + assert dq.results(LOG) == { + "orders_id_unique": ("PASS", None), + "orders_complete__id": ("FAIL", 3), + "orders_fresh": ("WARN", 1), + "orders_schema": ("SKIP", None), + } + + def test_the_report_lists_every_rule_and_what_dbt_said(self): + checks = [ + _check("unique", name="orders_id_unique", rule_type="uniqueness", column="id"), + _check("not_null", name="orders_complete__id", rule_type="completeness", column="id"), + _check("freshness", name="orders_fresh", rule_type="freshness", severity="warn"), + _check("schema", name="orders_schema", rule_type="schema", column=None), + _check("not_null", name="orders_gone", rule_type="completeness"), + ] + lines = dq.report(LOG, checks, truncated=True) + assert lines[0] == ( + " contract rules as dbt tests: 1 pass, 1 warn, 1 fail, 1 skip, 1 not reported" + ) + assert "✗ orders_complete__id: FAIL 3 rows - completeness id, severity error" in lines[2] + assert lines[-1].endswith( + "orders_gone: not in the task's output (Databricks kept only its end) - " + "completeness c, severity error" + ) + + def test_no_rules_no_report(self): + assert dq.report(LOG, []) == [] + + def test_a_run_that_reached_no_test_says_so_once(self): + checks = [_check("not_null", name="a"), _check("unique", name="b")] + (line,) = dq.report("12:00:01 Encountered an error:\nCompilation Error", checks) + assert "none of the 2 reported" in line + + def test_expected_tests_come_from_the_contract_alone(self): + contract = _contract() + names = [c.name for c in dq.expected_tests(contract, contract["builds"][0])] + assert names[0] == "orders_id_unique" and len(names) == 10 + + +# ── real dbt: the generated tests parse, and the build's selection picks each one ─────── + + +def _find_dbt() -> Optional[str]: + candidate = Path(sys.executable).parent / "dbt" + return str(candidate) if candidate.exists() else shutil.which("dbt") + + +@pytest.mark.integration +def test_dbt_parses_the_upload_and_selects_every_rules_test(tmp_path): + """dbt (any adapter) parses the uploaded project and ``dbt ls --select +orders+`` lists + each generated test: a test whose ``ref`` dbt cannot see at parse time would be missing, + as the schema test was on a live run before it was fixed.""" + dbt = _find_dbt() + if dbt is None: + pytest.skip("dbt CLI not available") + pytest.importorskip("dbt.adapters.duckdb") + _project(tmp_path, config="profile: p\nconfig-version: 2\nversion: '1.0'\n") + compiled = _compile(_contract(), tmp_path) + assert not compiled.refusals, _kinds(compiled) + upload = tmp_path / "upload" + for rel, text in _uploaded(compiled).items(): + target = upload / rel + target.parent.mkdir(parents=True, exist_ok=True) + target.write_text(text, encoding="utf-8") + (upload / "profiles.yml").write_text( + "p:\n target: dev\n outputs:\n dev:\n type: duckdb\n path: ':memory:'\n", + encoding="utf-8", + ) + listed = subprocess.run( + [dbt, "ls", "--resource-type", "test", "--select", "+orders+", "--output", "name"] + + ["--project-dir", str(upload), "--profiles-dir", str(upload), "--quiet"], + capture_output=True, + text=True, + check=False, + cwd=upload, + timeout=300, + ) + assert listed.returncode == 0, listed.stdout + listed.stderr + assert set(listed.stdout.split()) == set(compiled.jobs[0].tests) + + +def test_the_keys_task_runs_once_dbt_has_ended_whatever_its_outcome(): + # A failing error-severity rule fails dbt after CREATE OR REPLACE dropped the table's + # keys; the fluid_keys task (main's #860) must still declare them again. + from fluid_build.iac.providers import databricks_jobs as jobs + + item = jobs._Build({"id": "transform"}, "transform", "builds[0]", "sales.orders", WAREHOUSE) + task = jobs._keys_task(item) + assert task["depends_on"] == [{"task_key": jobs.DBT_TASK_KEY}] + assert task["run_if"] == "ALL_DONE" diff --git a/tests/util/test_dq_checks.py b/tests/util/test_dq_checks.py new file mode 100644 index 00000000..6cbadb63 --- /dev/null +++ b/tests/util/test_dq_checks.py @@ -0,0 +1,367 @@ +# Copyright 2025 Agentics Transformation Limited +# SPDX-License-Identifier: Apache-2.0 + +"""``util/dq_checks.py``: a contract's DQ rules read once, as checks or refusals. + +The module docstring is the spec. Shapes follow the industry packs' contracts (FLUID 0.7.6): +pipe-separated enums, comma-separated completeness selectors, ``labels.references`` keys. +""" + +from __future__ import annotations + +from typing import Any, Dict, List, Optional + +import pytest + +from fluid_build.providers.quality_engine import value_domain +from fluid_build.util.dq_checks import ( + Reference, + read_rules, +) + +pytestmark = [pytest.mark.unit] + +COLUMNS: List[Dict[str, Any]] = [ + {"name": "tenancy_id", "type": "STRING", "required": True}, + { + "name": "site_id", + "type": "STRING", + "labels": {"references": "bronze.asset.site_master_v1.site_id"}, + }, + { + "name": "tenant_operator", + "type": "STRING", + "validationRules": [{"type": "enum", "constraint": "ZAIN | OOREDOO | ISP"}], + }, + { + "name": "country_code", + "type": "STRING", + "validationRules": [{"type": "regex", "constraint": "^[A-Z]{2}$"}], + }, + { + "name": "grid_kwh", + "type": "DECIMAL(12,3)", + "validationRules": [{"type": "range", "constraint": ">= 0"}], + }, + {"name": "antenna_count", "type": "INTEGER"}, + {"name": "_ingested_at", "type": "TIMESTAMP"}, + {"name": "snapshot_date", "type": "DATE"}, +] +CONSUMES = [{"productId": "bronze.asset.site_master_v1", "exposeId": "site_master"}] + + +def _contract(*rules: Dict[str, Any], consumes: Optional[list] = None) -> Dict[str, Any]: + return { + "id": "silver.commercial.active_tenancy_v1", + "consumes": CONSUMES if consumes is None else consumes, + "exposes": [ + { + "exposeId": "active_tenancy", + "contract": {"schema": COLUMNS, "dq": {"rules": list(rules)}}, + } + ], + } + + +def _read(*rules: Dict[str, Any], **kwargs: Any): + contract = _contract(*rules, **kwargs) + return read_rules(contract, contract["exposes"]) + + +def _rule(type_: str, selector: str = "tenancy_id", **fields: Any) -> Dict[str, Any]: + return {"id": f"r_{type_}", "type": type_, "selector": selector, "severity": "error", **fields} + + +def _with_column(column: Dict[str, Any], **fields: Any): + """The checks of one valid_values rule on a product whose only column is ``column``.""" + rule = _rule("valid_values", column["name"], **fields) + contract = { + "id": "p", + "exposes": [{"exposeId": "t", "contract": {"schema": [column], "dq": {"rules": [rule]}}}], + } + return read_rules(contract, contract["exposes"]) + + +def _one(*rules: Dict[str, Any], **kwargs: Any): + read = _read(*rules, **kwargs) + assert not read.refusals, [r.message for r in read.refusals] + (check,) = read.checks + return check + + +def _refused(*rules: Dict[str, Any], **kwargs: Any) -> str: + read = _read(*rules, **kwargs) + assert not read.checks + assert read.refusals + return " ".join(f"{r.message} {r.fix}" for r in read.refusals) + + +class TestTypes: + def test_uniqueness_is_unique_on_its_column(self): + check = _one(_rule("uniqueness", threshold=1.0, operator=">=")) + assert (check.kind, check.column, check.share) == ("unique", "tenancy_id", None) + + def test_completeness_with_several_columns_checks_each(self): + read = _read(_rule("completeness", "site_id, tenant_operator,antenna_count")) + assert [(c.kind, c.column) for c in read.checks] == [ + ("not_null", "site_id"), + ("not_null", "tenant_operator"), + ("not_null", "antenna_count"), + ] + assert [c.name for c in read.checks] == [ + "r_completeness__site_id", + "r_completeness__tenant_operator", + "r_completeness__antenna_count", + ] + + def test_uniqueness_on_several_columns_is_refused(self): + assert "one uniqueness rule per column" in _refused( + _rule("uniqueness", "tenancy_id,site_id") + ) + + def test_valid_values_reads_the_columns_enum_split_on_pipes(self): + check = _one(_rule("valid_values", "tenant_operator")) + assert (check.kind, check.values) == ("accepted_values", ("ZAIN", "OOREDOO", "ISP")) + + def test_valid_values_reads_the_columns_regex(self): + check = _one(_rule("valid_values", "country_code")) + assert (check.kind, check.pattern) == ("regex", "^[A-Z]{2}$") + + def test_valid_values_reads_the_columns_range(self): + check = _one(_rule("valid_values", "grid_kwh")) + assert (check.kind, check.bounds) == ("bounds", ((">=", 0.0),)) + + def test_valid_values_reads_a_reference_to_a_consumed_product(self): + check = _one(_rule("valid_values", "site_id", severity="warn")) + assert check.kind == "reference" + assert check.reference == Reference("bronze.asset.site_master_v1", "site_master", "site_id") + assert check.severity == "warn" + + def test_the_values_the_rule_lists_win_over_the_columns_as_fluid_test_reads_them(self): + # providers/quality_engine.py::value_domain: the rule's own list, else the column + # (grid_kwh declares a range, which then does not apply). + rule = _rule("valid_values", "grid_kwh", description="grid_kwh valid values: 1, 2.") + assert value_domain(rule, COLUMNS) == value_domain(rule) + check = _one(rule) + assert (check.kind, check.values) == ("accepted_values", ("1", "2")) + + def test_an_enum_splits_on_pipes_and_commas_alike(self): + column = {"name": "code", "type": "STRING"} + column["validationRules"] = [{"type": "enum", "constraint": "A | B, C"}] + (check,) = _with_column(column).checks + assert check.values == ("A", "B", "C") == value_domain(_rule("x", "code"), [column]).values + + def test_only_the_first_declaration_of_a_kind_counts_as_in_fluid_test(self): + column = {"name": "code", "type": "STRING"} + column["validationRules"] = [ + {"type": "regex", "constraint": " "}, # no pattern: passed over + {"type": "regex", "constraint": "^[A-Z]+$"}, + {"type": "regex", "constraint": "^[0-9]+$"}, + ] + (check,) = _with_column(column).checks + assert (check.kind, check.pattern) == ("regex", "^[A-Z]+$") + + def test_valid_values_falls_back_to_values_the_description_lists(self): + check = _one( + _rule("valid_values", "antenna_count", description="antenna_count valid values: 1, 2.") + ) + assert check.values == ("1", "2") + + def test_valid_values_without_any_allowed_value_is_refused_with_the_fix(self): + text = _refused(_rule("valid_values", "antenna_count")) + assert "declares no allowed values" in text + assert "validationRules" in text and "labels.references" in text + + def test_a_column_with_two_declarations_gets_a_check_for_each(self): + columns = [ + { + "name": "code", + "type": "STRING", + "validationRules": [ + {"type": "enum", "constraint": "AA,BB"}, + {"type": "regex", "constraint": "^[A-Z]+$"}, + ], + } + ] + contract = { + "id": "p", + "exposes": [ + { + "exposeId": "t", + "contract": { + "schema": columns, + "dq": {"rules": [_rule("valid_values", "code")]}, + }, + } + ], + } + read = read_rules(contract, contract["exposes"]) + assert [(c.name, c.kind) for c in read.checks] == [ + ("r_valid_values__accepted_values", "accepted_values"), + ("r_valid_values__regex", "regex"), + ] + + def test_accuracy_is_a_bound_on_every_value(self): + check = _one(_rule("accuracy", "antenna_count", threshold=0, operator=">=")) + assert (check.kind, check.bounds, check.share) == ("bounds", ((">=", 0.0),), None) + + def test_accuracy_without_a_threshold_is_refused(self): + assert "no finite numeric threshold" in _refused(_rule("accuracy", "antenna_count")) + + def test_freshness_is_the_newest_values_age_in_seconds(self): + check = _one( + _rule("freshness", "_ingested_at", window="PT2H", operator="<=", severity="warn") + ) + assert (check.kind, check.max_age_seconds, check.strict) == ("freshness", 7200, False) + assert _one(_rule("freshness", "snapshot_date", window="P35D", operator="<")).strict + + @pytest.mark.parametrize( + "fields, needle", + [ + ({"window": "P1M"}, "not a fixed ISO 8601 duration"), + ({"window": "PT1H", "operator": ">="}, "bounds the newest value's age from above"), + ({"window": "PT1H", "threshold": 1.0}, "does not take"), + ], + ) + def test_a_freshness_rule_that_cannot_run_is_refused(self, fields, needle): + assert needle in _refused(_rule("freshness", "_ingested_at", **fields)) + + def test_freshness_on_a_non_temporal_column_is_refused(self): + assert "needs a date or timestamp column" in _refused( + _rule("freshness", "antenna_count", window="PT1H") + ) + + def test_schema_checks_every_contract_column_with_its_declared_type(self): + check = _one({"id": "s", "type": "schema", "selector": "*", "severity": "error"}) + assert check.kind == "schema" + assert check.columns[0] == ("tenancy_id", "STRING") + assert len(check.columns) == len(COLUMNS) + + @pytest.mark.parametrize("kind", ["anomaly_detection", "drift_detection"]) + def test_history_rules_are_refused(self, kind): + assert "the table's history" in _refused(_rule(kind, "*")) + assert all(refusal.needs_history for refusal in _read(_rule(kind, "*")).refusals) + + def test_a_history_rule_written_wrong_is_refused_for_that(self): + # An engine that keeps the history (data quality monitoring) still needs a rule it + # can read: this refusal is not one it takes over. + (refusal,) = _read(_rule("anomaly_detection", "*", severity="bogus")).refusals + assert "severity" in refusal.message and not refusal.needs_history + + +class TestTheRuleAsWritten: + def test_each_check_keeps_the_rules_severity_threshold_and_operator(self): + check = _one(_rule("completeness", "site_id", severity="critical", threshold=1.0)) + assert (check.severity, check.rule_severity) == ("error", "critical") + assert (check.rule_threshold, check.rule_operator) == (1.0, None) + check = _one(_rule("completeness", "site_id", threshold=0.95, operator=">")) + assert (check.rule_threshold, check.rule_operator, check.share) == (0.95, ">", (">", 0.95)) + + def test_a_freshness_check_has_no_threshold(self): + check = _one(_rule("freshness", "_ingested_at", window="PT1H", severity="info")) + assert (check.rule_severity, check.rule_threshold) == ("info", None) + + +class TestShares: + def test_every_row_when_the_threshold_is_one(self): + assert _one(_rule("completeness", "site_id", threshold=1.0, operator="==")).share is None + + def test_a_lower_threshold_is_a_share(self): + check = _one(_rule("completeness", "site_id", threshold=0.95)) + assert check.share == (">=", 0.95) + + def test_a_threshold_above_one_reads_as_a_percentage_and_is_refused(self): + assert "0.99, not 99" in _refused(_rule("completeness", "site_id", threshold=99)) + + +class TestRefusals: + @pytest.mark.parametrize("threshold", [10**400, float("inf"), float("nan")]) + def test_a_threshold_no_float_holds_is_refused_not_raised(self, threshold): + assert "finite" in _refused(_rule("accuracy", "antenna_count", threshold=threshold)) + assert "finite" in _refused(_rule("completeness", "site_id", threshold=threshold)) + + def test_a_window_too_long_to_count_is_refused_not_raised(self): + window = "PT" + "9" * 30 + "S" + assert "fixed ISO 8601" in _refused(_rule("freshness", "_ingested_at", window=window)) + + @pytest.mark.parametrize("constraint", ["between 100 and 0", "> 5 and < 1", "> 3 and <= 3"]) + def test_a_range_no_value_meets_is_refused(self, constraint): + columns = [ + { + "name": "x", + "type": "INTEGER", + "validationRules": [{"type": "range", "constraint": constraint}], + } + ] + contract = { + "id": "p", + "exposes": [ + { + "exposeId": "t", + "contract": {"schema": columns, "dq": {"rules": [_rule("valid_values", "x")]}}, + } + ], + } + read = read_rules(contract, contract["exposes"]) + assert "no value can meet" in read.refusals[0].message + + def test_an_unknown_column_is_refused(self): + assert "does not declare" in _refused(_rule("completeness", "nope")) + + def test_a_column_differing_only_in_case_is_the_schemas_column(self): + assert _one(_rule("completeness", "SITE_ID")).column == "site_id" + + def test_a_reference_to_a_product_not_consumed_has_no_expose(self): + # A source-aligned product consumes nothing: the engine decides whether it can find + # the other product's table without an expose (the dbt job cannot, and refuses). + check = _one(_rule("valid_values", "site_id"), consumes=[]) + assert check.reference == Reference("bronze.asset.site_master_v1", None, "site_id") + + def test_a_reference_consumed_by_two_exposes_is_refused(self): + consumes = [ + {"productId": "bronze.asset.site_master_v1", "exposeId": "a"}, + {"productId": "bronze.asset.site_master_v1", "exposeId": "b"}, + ] + assert "ambiguous" in _refused(_rule("valid_values", "site_id"), consumes=consumes) + + def test_an_unknown_severity_is_refused(self): + assert "severity" in _refused(_rule("completeness", "site_id", severity="fatal")) + + @pytest.mark.parametrize("severity, level", [("critical", "error"), ("info", "warn")]) + def test_severities_map_to_the_two_dbt_levels(self, severity, level): + assert _one(_rule("completeness", "site_id", severity=severity)).severity == level + + +class TestNames: + def test_names_are_unique_across_exposes(self): + rule = _rule("completeness", "site_id") + contract = _contract(rule) + second = {**contract["exposes"][0], "exposeId": "other"} + read = read_rules(contract, [contract["exposes"][0], second]) + assert [c.name for c in read.checks] == ["r_completeness", "r_completeness__other"] + + def test_a_rule_id_folds_to_lower_case_letters_digits_and_underscores(self): + check = _one({**_rule("completeness", "site_id"), "id": "Site-ID check!"}) + assert check.name == "site_id_check" + + +@pytest.mark.parametrize( + "text, bounds", + [ + (">= 0", ((">=", 0.0),)), + (">= 0 and < 1.5", ((">=", 0.0), ("<", 1.5))), + ("between -1 and 10", ((">=", -1.0), ("<=", 10.0))), + ("0 to 10", None), + (">= 0 or <= 2", None), + (">= " + "9" * 400, None), + ], +) +def test_a_range_reads_as_fluid_test_reads_it_or_is_refused(text, bounds): + column = {"name": "n", "type": "DOUBLE"} + column["validationRules"] = [{"type": "range", "constraint": text}] + read = _with_column(column) + if bounds is None: # fluid test passes such a range over; here it is refused, with the fix + assert not read.checks and "is not '>= 0'" in read.refusals[0].message + else: + assert [check.bounds for check in read.checks] == [bounds] + assert value_domain(_rule("x", "n"), [column]).bounds == bounds