diff --git a/.github/workflows/iac-tests.yml b/.github/workflows/iac-tests.yml index 66fb0694..81215c4a 100644 --- a/.github/workflows/iac-tests.yml +++ b/.github/workflows/iac-tests.yml @@ -214,12 +214,14 @@ jobs: tests/iac/test_iac_databricks_fake_e2e.py tests/iac/test_iac_databricks_jobs_fake_e2e.py tests/iac/test_iac_databricks_genie_fake_e2e.py + tests/iac/test_iac_databricks_metrics_fake_e2e.py pytest-args: "-p no:randomly" junit: databricks-fake.xml require: | Databricks fake tier=tests/iac/test_iac_databricks_fake_e2e.py Databricks jobs fake tier=tests/iac/test_iac_databricks_jobs_fake_e2e.py Databricks Genie fake tier=tests/iac/test_iac_databricks_genie_fake_e2e.py + Databricks metric views fake tier=tests/iac/test_iac_databricks_metrics_fake_e2e.py steps: - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 diff --git a/.secrets.baseline b/.secrets.baseline index 52b748b2..1c59454d 100644 --- a/.secrets.baseline +++ b/.secrets.baseline @@ -139,7 +139,7 @@ "filename": ".github/workflows/iac-tests.yml", "hashed_secret": "a94a8fe5ccb19ba61c4c0873d391e987982fbbd3", "is_verified": false, - "line_number": 348 + "line_number": 350 } ], ".github/workflows/integration.yml": [ diff --git a/CHANGELOG.md b/CHANGELOG.md index ff37eab8..5434468e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -72,6 +72,30 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 workspace over the quota stops all compute for the rest of the day, so on Free Edition opt out the products you do not need watched. +- **Databricks: a Unity Catalog metric view from each expose's `semantics` block.** `fluid apply` + creates `.._semantic_model` beside the table, in the product's own + schema, as metric view YAML 1.1, so Genie, AI/BI dashboards and SQL + (`SELECT country_code, MEASURE(site_count) ... GROUP BY ALL`) read the contract's semantic model + natively. Entities and dimensions become dimensions (an entity as its key column, a time + dimension at its `timeGranularity`); a measure is its aggregate; a simple metric is that + aggregate with its filter as `FILTER (WHERE ...)`; a ratio is `MEASURE(numerator) / + NULLIF(MEASURE(denominator), 0)`; a `nonAdditiveDimension` is a window measure + (`semiadditive: last` for `max`, `first` for `min`). Descriptions become comments, a column's + `businessName` the dimension's `display_name` and its `labels.synonyms` its synonyms; the + model's `ai-synonyms` and `ai-instructions` labels go into the view's comment. Reading the view + is exactly the principals that may read the table: `SELECT` granted to them, revoked from + anyone else, revokes first and before the apply. The provider has no metric view resource + (terraform-provider-databricks#4827), so the view is created on the SQL warehouse after + `tofu apply`, like the column masks: `CREATE VIEW` when it is missing, `ALTER VIEW ... AS` + (which keeps its grants) when its definition changed, nothing when it did not, and + `DROP VIEW`, before the apply, once the contract no longer declares it (in the schemas its + tables are in, so another environment of the product keeps its own). `--mode dry-run` prints + every statement. What metric view YAML cannot express is refused at `fluid validate`, not + approximated: derived metrics, a ratio with a filter, a non-additive dimension that is no time + dimension, `hour` or `minute` on a `DATE`, colliding names, a name that is no column, and + calls that would lend the view owner's rights to its readers. Verified live on Databricks Free + Edition with the semantics block of an industry-model gold product (`HONESTLY_TESTED.md`). + - **`fluid test` checks a Databricks product live, and can keep its results in Unity Catalog.** `fluid test` used to answer "Live validation against Unity Catalog is not available" for `platform: databricks`. It now reads the table's columns from the Unity Catalog tables API @@ -123,6 +147,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 path exits 1 where it used to write a template instead. A `fluid ai setup` choice and a single keyring key still come first, and `GEMINI_API_KEY` with `GOOGLE_API_KEY` is one provider. Pick one with `--llm-provider` or `FLUID_LLM_PROVIDER`. +- **`fluid apply` on a Databricks contract with `exposes[].semantics` needs the `[databricks]` + extra** (databricks-sdk), which creates the metric view, and refuses before `tofu apply` when it + is missing, as it does for column masks. A semantics block the metric view cannot hold (a + derived metric, say) now fails `fluid validate` for a Databricks binding. +- `fluid apply` prints what the Databricks access hooks say (statements, server errors) as data, + not Rich markup: a `[word]` in a description no longer vanishes from the dry-run, and a `[/word]` + no longer ends the run with an unexpected error. +- The MCP query compiler and the Databricks metric view render a measure's aggregate through one + helper (`forge_datamodel/semantics_builder.py::aggregate_call`); the compiler's SQL and errors + are unchanged. - **The native quality engine runs the rules a contract already declares in full.** A `valid_values` rule that names no values takes them from its column: its `enum` validationRule (`"A | B"` or `"A,B"`), its `regex` validationRule, its `range` diff --git a/HONESTLY_TESTED.md b/HONESTLY_TESTED.md index a60121bf..8836001a 100644 --- a/HONESTLY_TESTED.md +++ b/HONESTLY_TESTED.md @@ -91,6 +91,8 @@ Markers declared in `pyproject.toml`. | Stage 1 real tofu, stand-in workspace: dbt jobs | `tests/iac/test_iac_databricks_jobs_fake_e2e.py` | ✅ 6. A local dbt build applies as one serverless job plus the workspace files dbt reads, each in a folder created before it (no upload meets a missing parent) (never `profiles.yml`, `target/` or a hidden file), and a second plan is empty. A changed model updates its file in place. A GIT build runs its tag on a continuous trigger and uploads nothing. Turning a job build local removes the job, its files and their folders without `--allow-data-loss`, and says so; a folder holding a file the apply did not put there is not deleted. The table a dbt build writes keeps what dbt's CREATE OR REPLACE made of it, and the apply reports the column drift instead of failing on a type change. | Whether Databricks accepts and runs the job: serverless environment version 2, the `dbt-databricks>=1.8.0,<2.0.0` range, the commands, a WORKSPACE `project_directory` under `/Workspace`, `run_as` permissions, a Git credential for a private repository, `dbt deps` with a local package (not uploaded). The manual live row proves the first four. Whether the jobs API echoes a server default the stand-in does not (the fake's builder found `performance_target` and an all-false `notification_settings` would each plan an update), which would make a second plan non-empty (live, it was empty). What dbt's CREATE OR REPLACE does to a table with column mapping (the module sets none on a dbt-written table). Not bound by a plan: an apply uploads the dbt project as it is when it runs (`planDigest` and `bundleDigest` cover the contract, not the project's files, as for a local dbt run), and a bundle does not carry the project. Known gap, by design: a job is not adopted. Its import id is a job id, which neither the contract nor `tofu` can look up by name, so an apply that lost its state creates a second job. `fluid verify` flags the duplicate (a stubbed client proves it). Workspace files need no import: the provider's create overwrites. | | Stage 1 unit: Databricks retention (`lifecycle.expire`) | `tests/iac/test_iac_databricks_retention.py`, `tests/cli/test_verify_databricks_retention.py` | ✅ 195 + 18. ISO 8601 durations (years and months at their shortest, P1Y = 365 and P1M = 28 days; a fraction, a sign, a lower-case letter, `PT`, `P1DT`, out-of-order and over-long numbers refused), Delta `interval` properties (rounded up, months refused) and the history as the larger of the two properties with Delta's defaults. A dbt model's `materialized` resolved from `config()` in the SQL, the model's YAML `config:` and `dbt_project.yml` (the `+` prefix or none, root, project, folder and model-name levels, the most specific wins, SQL over YAML over project), the default `view`, the model found by alias or file name, commented-out and `raw` configs ignored, a `config()` inside a Jinja block, `var()`, `**`, a custom schema or database, a Python model, a versioned or twin model, a linked file and a Git repository refused as undeterminable. The build's `--select` must include the resolved model (name, `fqn`, `path:`, `*`, unions and intersections; a graph operator on another node, a tag, a config or run state is refused as not evaluable). Accepted: a dbt `table` model and a DuckDB acquisition with P30D, P365D, P6Y; refused: P7D, P29D, P4W, P1M, a history set longer than the retention, `incremental`, `snapshot`, `view`, another acquisition engine, no writer, a mixed set of writers. `fluid validate` and the emitter agree on kind and message. `fluid verify`'s retention dimension against a stubbed client and, over HTTP, the stand-in workspace (pass, fail with the `ALTER TABLE`, the runtime property over the declared one, skip, predictive optimization off as a warning). | Live behaviour: no real Databricks endpoint was called. Whether Unity Catalog's GetTable serves a table's `delta.logRetentionDuration` and `delta.deletedFileRetentionDuration` in `properties` or `delta_runtime_properties_kvpairs` (read from databricks-sdk-go v0.190.0 `model.go`, not observed), whether `effective_predictive_optimization_flag` is filled for a managed table, that predictive optimization runs `VACUUM` within the retention (a Databricks guarantee, not tested), and that dbt-databricks `table` replaces the table each run. Delta history is only half the promise: the acquisition acceptance rests on the landing being an `INSERT OVERWRITE` of the whole table in the one mode it runs (`full_refresh`, read from the landing's own `LOAD_MODES`; another source mode is refused as `incremental`, none as `unmapped`), which the bronze landing row covers against the stand-in, not live. A GetTable answer with no properties is a verify warning (Delta's defaults assumed). Embedded-logic DuckDB builds, `dlt`, `airbyte` and the other acquisition engines are refused. Selection is matched on names and paths only, not on dbt's graph: `+other+` is refused even if `other` is upstream of the model. | | Stage 1: Databricks Unity Catalog metadata (comments, product tags, column tags; live: the 2026-11-04 row) | `tests/iac/test_iac_databricks_metadata.py` (the emitter), `tests/iac/test_iac_databricks_metadata_fake_e2e.py` (real `fluid` + `tofu` + databricks/databricks 1.140.0 against `tests/iac/_fake_databricks.py`), `tests/cli/test_verify_databricks.py` (the comment and column-tag cases) | ✅ 43 + 5 + 7. The table's `comment` is the expose's description, else the product's (its `purpose` when structured), each described column's `comment` its schema description, stripped; none on a table a dbt build writes (dbt's CREATE OR REPLACE owns them; `fluid validate` warns). The product's identity and owner team are tags on the table only (`fluid_product_id`, `fluid_product_name`, `fluid_domain`, `fluid_layer`, `fluid_product_type`, `fluid_product_version`, `fluid_version`, `fluid_owner_team`), an unstated field makes none, and they count with the governance labels toward 50; `lifecycle.state: deprecated` (the expose's, else the product's) sets `system.certification_status = deprecated`, else the label `certification: certified` (the expose's, else the product's) sets `certified` and is no plain tag (a product-level label certifies every expose's table, as intended; an expose-level one only its own); another label value is `databricks-certification-invalid`; a plan notice says the identity needs ASSIGN on that governed tag. Against the stand-in: certified, then deprecated in place, then removed. A classified column (`sensitivity`) carries `fluid_sensitivity` as a `columns` tag named `catalog.schema.table.column`, imported best-effort by the provider's id. Refused through `validate` and `emit` alike: a backslash in any comment (`databricks-comment-unsafe`; the provider's `parseComment` escapes only quotes) and a `${` (Spark variable substitution), a label meeting a product key (`databricks-tag-collision`), a value over 256 characters (`databricks-tag-value-too-long`), over 1,000 column tags (`databricks-tags-too-many`); a `${...}` in a tag value is rendered as a literal. Against the stand-in: apply writes the comments (an apostrophe included, through the provider's escaping), the product tags and the column tag, never an `OWNER TO`; a second plan is empty; a changed description is an in-place `COMMENT ON TABLE`, never a replace; removing a classification destroys only its column tag; a table made before this is adopted and gains its comments in place (`+5 ~1` for the product tags and the table update). `fluid verify` passes after the apply and fails a comment or column tag changed by hand; against a stubbed client it reads each column's tag list to its last page, ignores a tag or comment someone else added, and checks no comment on a dbt-written table. | **Not run live.** What real Unity Catalog stores and returns for a table or column comment (whitespace, newlines, length limits, which the docs do not state), whether it accepts a tag value with `, . : / = -` (the docs restrict only keys), whether `entity-tag-assignments/columns/{catalog.schema.table.column}` resolves as the SDK documents, and whether a dbt CREATE OR REPLACE keeps column tags (undocumented; `fluid verify` is the gate), and whether the Entity Tag Assignments API sets the system tag `system.certification_status` (its docs show the UI and `SET TAG`; the SDK documents governed tags through the API, needing ASSIGN; the stand-in checks no permission). A contract that adds a column while comments change for the first time hits the provider's mixed-change refusal (its fix-it says to apply in two steps). | +| Stage 1: Databricks 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. | +| Manual live, Free Edition (founder-approved), 2026-11-12, metric views | the real `fluid` CLI, no test file: this branch, databricks-sdk 0.166.0, a scratch catalog `fluid_iactest_metrics` dropped `CASCADE` after | ✅ The semantics block of industry-model `dp019_site_360` verbatim (13 dimensions, 3 non-additive measures, 4 filtered metrics) on a table shaped like it: dry-run printed the `CREATE VIEW` and grants; apply `+77`, then the view created and `SELECT` granted to the 3 reader principals; `SELECT country_code, MEASURE(site_count), MEASURE(avg_tenancy_ratio), ... GROUP BY ALL` returned each market at its latest snapshot (8 rows seeded, 4 sites x 2 days); by date, each day's own values; a second apply `+0 ~0 -0` ran no statement. `information_schema.tables` lists it as `METRIC_VIEW` with its comment, `.columns` every dimension and measure with its comment, `.table_privileges` the 3 `SELECT` grants. Probes on the same warehouse (Databricks SQL 2026.45): `CREATE OR REPLACE VIEW` drops a view's grants and `ALTER VIEW ... AS` keeps them and its properties; the stored `view_definition` leaves out `comment`, `display_name` and `synonyms`; a dimension named like a column shadows it in a measure's filter and a window measure named like a column breaks the window measures that read it (`METRIC_VIEW_WINDOW_MEASURE_REFERENCES_WINDOW_MEASURE`), and `source.` reads the column; `semiadditive: last` is per output group; `FILTER (WHERE ...)` works on a window measure, `COUNT(DISTINCT ...)` and `PERCENTILE_CONT ... WITHIN GROUP`; `TBLPROPERTIES` combine with `WITH METRICS`; a metric view over a row-filtered table applies the filter to the reader; a table at the view's name fails `EXPECT_VIEW_NOT_TABLE`. | A query as a reader persona (the run had the deploy user's token only; the persona's effective `SELECT` was read from the permissions API); Genie and dashboards; a contract with column masks. | | Stage 1: Databricks data quality monitoring (live: the 2026-11-12 row) | `tests/iac/test_iac_databricks_quality.py` (what asks for it, the monitor per schema, the derived import, the notice, the engine), `tests/iac/test_iac_databricks_quality_fake_e2e.py` (real `fluid` + `tofu` + databricks/databricks 1.140.0 against `tests/iac/_fake_databricks.py`), `tests/iac/test_iac_tofu_validate.py::test_a_databricks_data_quality_monitor_passes_tofu_validate` | ✅ 37 + 6 + 1. A freshness or completeness SLO (`qos.freshnessSLO`, `qos.completenessTarget`, an `observability.defaultSLIs` freshness or completeness block, a `contract.dq.rules` entry of type `freshness`, `completeness` or `anomaly_detection`) puts one `databricks_data_quality_monitor` on the owned schema, watching it by `schema_id`, created after the schema's tables, the product's other tables there in `excluded_table_full_names`; `contract.dq.monitoring.enabled: false` removes only the monitor; a `dq.monitoring.window` other than one day is a `fluid validate` warning. Against the stand-in: a second plan is empty; a monitor turned on outside forge-cli is adopted by `schema,` taken from state after the schema's import (no create); the last exclusion leaving is a PATCH whose `update_mask` names the whole config (the stand-in honours the mask), and the plan converges; a monitor Databricks refuses fails the apply after the tables, its suggestions saying what it needs and naming the opt-out (`apply_failure_hints`, after the three generic ones); opting out deletes it, not gated; `tofu destroy` deletes it before the schema. | What Databricks scans and reports (the live row); the stand-in serves schema monitors only. | | Manual live, Free Edition (founder-approved) | the real `fluid` CLI, no test file: run 2026-11-01 from main dfd6f145, databricks-sdk 0.158.0, every `fluid_iactest_*` object swept after (1 schema, 1 table, 1 folder; 0 left) | ✅ validate, `generate iac --validate`, plan, dry-run and apply `+9`; a second plan is EMPTY against the real Jobs API; Databricks accepts the Quartz schedule, serverless environment 2 with `dbt-databricks>=1.8.0,<2.0.0` (dbt 1.12.6, adapter 1.13.0), the WORKSPACE `project_directory` and the commands as emitted; one `run-now` ends SUCCESS and dbt builds the table; `fluid verify` with the real SDK (columns, types, one job); `fluid diff`'s state check (its live check has no Databricks inspector); a model change `~1`; the build made local removes job, files and folders `-5` without `--allow-data-loss` (the empty `.fluid//` folders above `files/` stay: only `files/` and below are managed). It found the USE CATALOG check reading a group-conveyed privilege as none. After the fix, `fluid verify` on a fresh one-table product (`+4`) passes every dimension, catalog access included (1 schema and 1 table swept, 0 left). | A GIT source, profile or OAuth auth, a run-as service principal, table adoption and a larger project, live; the region residency check (the metastore precondition and `fluid verify`'s residency dimension), which was added after this run and not run live; dbt's CREATE OR REPLACE drops a contract's NOT NULL, so verify flags a `required` column a dbt model does not constrain itself. | | Manual live, Free Edition (founder-approved), masks and row filters | the real `fluid` CLI, no test file: run 2026-11-03 at the tip of #785, a real workspace with an AWS us-west-2 metastore; every object swept after (0 left) | ✅ 10 steps passed. Functions are created and read back (owner is the user's email; `routine_definition` is the body without `RETURN`, backticks kept; `full_data_type` is STRING or BOOLEAN; a zero-parameter function's FunctionInfo has no `input_params`). The row filter and masks are bound and enforced on the restricted user's own SELECT: only the consented rows, `msisdn` as `XXXXX4567`, a 3-character value hashed, the denied column NULL. `SET ROW FILTER ... ON ()` is accepted and a `SET` replaces a live filter. Tags are present, `fluid verify` passes every dimension, a second apply is `+0 ~0 -0`, the residency precondition fails the plan and names both regions, `generate iac` refuses, and the sweep left 0. | Still open, as this run showed: superseded functions were left behind (orphans; the follow-up that drops them after a successful apply is not run live); `tofu destroy` of the module fails at a schema that still holds functions, so a teardown must drop them first; and the residency failure text was forge-cli's 800-character tail of the tofu JSON, which cut off the sentence naming the live region (the one-line explanation is not run live). A metastore in another cloud or region, a service principal as the run identity, and a group-conveyed grant on the live workspace are still not run. | diff --git a/fluid_build/cli/_apply_opentofu_engine.py b/fluid_build/cli/_apply_opentofu_engine.py index d0c7310f..e9266f9e 100644 --- a/fluid_build/cli/_apply_opentofu_engine.py +++ b/fluid_build/cli/_apply_opentofu_engine.py @@ -406,6 +406,8 @@ def _enforce_access( run = getattr(plugin, hook, None) if not callable(run): return + # What the hooks print quotes the contract and the server (statements, descriptions, errors): + # data, not markup, so a "[word]" in it neither vanishes nor raises. try: if dry_run: run(contract, say=cprint_literal, dry_run=True) diff --git a/fluid_build/forge_datamodel/semantics_builder.py b/fluid_build/forge_datamodel/semantics_builder.py index 177d5c25..94b3199f 100644 --- a/fluid_build/forge_datamodel/semantics_builder.py +++ b/fluid_build/forge_datamodel/semantics_builder.py @@ -313,6 +313,76 @@ def normalized_time_type_params(grain: Optional[str]) -> Optional[Dict[str, str] return {"timeGranularity": canonical} +#: ``measures[].agg`` (the schema's enum) → the SQL aggregate it compiles to. Shared by the +#: consumers that render a measure as SQL (the MCP query compiler and the Databricks metric +#: view), so one contract aggregates the same way on every path. ``count_distinct`` renders +#: ``COUNT(DISTINCT …)`` and ``percentile`` the ordered-set ``PERCENTILE_CONT`` (see +#: :func:`aggregate_call`). +AGGREGATE_FUNCTIONS: Dict[str, str] = { + "sum": "SUM", + "avg": "AVG", + "count": "COUNT", + "count_distinct": "COUNT", + "min": "MIN", + "max": "MAX", + "median": "MEDIAN", + "percentile": "PERCENTILE_CONT", +} + +#: ``aggParams.percentile`` when an ``agg: percentile`` measure gives none: 0.5, the median. +#: The dbt MetricFlow bridge keeps its own copy, pinned equal by tests, so every consumer +#: answers the same number for one contract. +DEFAULT_PERCENTILE = 0.5 + + +def percentile_params(measure_name: str, measure: Mapping[str, Any]) -> Tuple[float, bool]: + """``(percentile, use_discrete)`` from ``measure.aggParams``; :class:`ValueError` if bad. + + The percentile is interpolated into SQL as a literal (an aggregate's argument must be a + constant), so this check is the safety boundary: a real number in [0, 1], never a bool. + """ + params = measure.get("aggParams") or {} + if not isinstance(params, Mapping): + raise ValueError(f"Measure {measure_name!r} has non-object aggParams") + raw = params.get("percentile", DEFAULT_PERCENTILE) + if isinstance(raw, bool) or not isinstance(raw, (int, float)) or not 0 <= float(raw) <= 1: + raise ValueError( + f"Measure {measure_name!r} has invalid aggParams.percentile " + f"{raw!r}; expected a number in [0, 1]" + ) + return float(raw), bool(params.get("useDiscretePercentile")) + + +def aggregate_call(measure_name: str, measure: Mapping[str, Any], expr: str) -> str: + """``AGG(expr)`` for one ``measures[]`` entry; :class:`ValueError` on an unknown ``agg``. + + ``expr`` is the caller's: already validated, and rewritten where its target needs it. + ``count_distinct`` is ``COUNT(DISTINCT expr)``; ``percentile`` is the ANSI ordered-set + ``PERCENTILE_CONT(p) WITHIN GROUP (ORDER BY expr)`` (``p`` from ``aggParams.percentile``, + default :data:`DEFAULT_PERCENTILE`; ``useDiscretePercentile`` selects ``PERCENTILE_DISC``). + A target without that aggregate refuses ``percentile`` before calling this. + """ + agg = measure.get("agg") + if not isinstance(agg, str) or agg not in AGGREGATE_FUNCTIONS: + raise ValueError( + f"Measure {measure_name!r} has unsupported agg " + f"{agg!r}; supported: {sorted(AGGREGATE_FUNCTIONS)}" + ) + if agg == "count_distinct": + return f"COUNT(DISTINCT {expr})" + if agg == "percentile": + percentile, use_discrete = percentile_params(measure_name, measure) + fn = "PERCENTILE_DISC" if use_discrete else "PERCENTILE_CONT" + return f"{fn}({percentile:g}) WITHIN GROUP (ORDER BY {expr})" + return f"{AGGREGATE_FUNCTIONS[agg]}({expr})" + + +def parens_balanced(expr: str) -> bool: + """Every ``(`` closes in order and none closes early: a contract expression is spliced + into a larger statement, and an unbalanced one would escape its place in it.""" + return _parens_balanced(expr) + + def default_agg_time_dimension(dimensions: List[Mapping[str, Any]]) -> Optional[str]: """The model-level default aggregation time dimension: the first declared time dimension. Populating it (instead of leaving the diff --git a/fluid_build/iac/providers/databricks.py b/fluid_build/iac/providers/databricks.py index 7da9aa14..b091c846 100644 --- a/fluid_build/iac/providers/databricks.py +++ b/fluid_build/iac/providers/databricks.py @@ -87,6 +87,12 @@ A predicate that is not one boolean expression over the table's columns is ``databricks-row-filter-invalid``; ``access: allow``, a principal with no SQL test and a text mask on a non-string column are refused too. ``fluid verify`` checks the bindings. +* An expose's ``semantics`` block is a Unity Catalog metric view beside its table, + ``..
_semantic_model``, created on the SQL warehouse after + ``tofu apply`` (the provider has no metric view resource, terraform-provider-databricks#4827), + readable by exactly the principals that read the table, and dropped before the apply once the + contract no longer declares it, in the schemas its tables are in. What metric view YAML cannot + express is refused (``databricks_metrics.py``). * An expose that states a freshness or completeness SLO (``qos.freshnessSLO``, ``qos.completenessTarget``, ``observability.defaultSLIs``, a ``contract.dq.rules`` entry of type ``freshness``, ``completeness`` or ``anomaly_detection``) turns on @@ -190,6 +196,7 @@ ) from ..provider_match import canonical_cloud, is_cloud from ..versions import required_providers +from . import databricks_metrics as _metric_views from .databricks_access import TableEnforcement, compile_enforcement from .databricks_auth import tofu_env_overlay from .databricks_enforce import ( @@ -373,6 +380,8 @@ class _Compiled: warnings: List[str] = field(default_factory=list) #: ``(entity type, full name, tag key, tag value)`` per tag assignment emitted. tags: List[Tuple[str, str, str, str]] = field(default_factory=list) + #: The metric views of the exposes that declare ``semantics`` (``databricks_metrics.py``). + metric_views: _metric_views.MetricViews = field(default_factory=_metric_views.MetricViews) #: The data quality monitor emitted per schema (``databricks_quality.py``). monitors: List[SchemaMonitor] = field(default_factory=list) @@ -1408,6 +1417,9 @@ def _compile( _emit_grants(compiled, table, cid, privileges.get(table.full_name, {})) _emit_tags(compiled, table, cid, contract) _check_own_keys(compiled, contract) + compiled.metric_views = _metric_views.compile_metric_views( + contract, compiled.tables, privileges, refusals, compiled.warnings + ) _emit_quality_monitors(compiled, cid) _merge_jobs( compiled, @@ -2419,11 +2431,14 @@ def standalone_module_refusal( ) @staticmethod - def _enforcement_targets(contract: Mapping[str, Any]) -> List[Target]: + def _hook_state( + contract: Mapping[str, Any], + ) -> Tuple[List[Target], _metric_views.MetricViews]: compiled = _compile(contract, os.environ) if compiled.refusals: raise compiled.refusals[0] - return [Target(table.full_name, table.enforcement) for table in compiled.tables] + targets = [Target(table.full_name, table.enforcement) for table in compiled.tables] + return targets, compiled.metric_views def pre_apply( self, @@ -2432,20 +2447,25 @@ def pre_apply( dry_run: bool = False, client_factory: Optional[Callable[[], Any]] = None, ) -> None: - """Before ``tofu apply``: bind the contract's column masks and row filter on the - tables that already exist (``databricks_enforce.pre_apply``), so a grant the apply - adds lands on a protected table. With ``dry_run`` it prints every statement either - hook would run, and runs none. Raises :class:`UnsupportedBindingError` to stop the + """Before ``tofu apply``: check the Genie space's client (``databricks_genie.preflight``); + settle the metric views (``databricks_metrics.pre_apply``: refuse a name an object this + module did not create holds, revoke readers the contract no longer names, drop this + product's views it no longer declares); then bind the contract's column masks and row + filter on the tables that already exist (``databricks_enforce.pre_apply``), so a grant + the apply adds lands on a protected table. With ``dry_run`` it prints every statement + the hooks would run, and runs none. Raises :class:`UnsupportedBindingError` to stop the run before the apply.""" - targets = self._enforcement_targets(contract) + targets, views = self._hook_state(contract) say = say or print if dry_run: preview(targets, say, client_factory) _declare_keys(contract, say, client_factory, dry_run=True) + _metric_views.preview(views, say, client_factory) else: from .databricks_genie import preflight preflight(_genie(contract, _compile(contract, os.environ), os.environ), client_factory) + _metric_views.pre_apply(views, say, client_factory) pre_apply(targets, say, client_factory) def post_apply( @@ -2457,18 +2477,23 @@ def post_apply( ) -> None: """After ``tofu apply``: bind on the tables it created, drop what the contract no longer restricts, and revoke ``SELECT`` from the restricted principals of any table - it could not bind (``databricks_enforce.post_apply``). ``after_failed_apply`` is the - bind-only mode for an apply that failed part way: it drops nothing and skips a table - that was never created. Then, after an apply that succeeded, the contract's Unity - Catalog keys (:func:`_declare_keys`).""" - post_apply( - self._enforcement_targets(contract), - say or print, - client_factory, - after_failed_apply=after_failed_apply, - ) + it could not bind (``databricks_enforce.post_apply``). When a binding fails, ``SELECT`` on + the metric views over restricted tables is revoked before the error is raised. After an + apply that succeeded, the contract's Unity Catalog keys (:func:`_declare_keys`), then its + metric views created, updated and granted (``databricks_metrics.post_apply``). + ``after_failed_apply`` is the bind-only mode for an apply that failed part way: it drops + nothing, skips a table that was never created and leaves the keys and the metric views + to the next apply.""" + targets, views = self._hook_state(contract) + say = say or print + try: + post_apply(targets, say, client_factory, after_failed_apply=after_failed_apply) + except UnsupportedBindingError: + _metric_views.lock_restricted(views, say, client_factory) + raise if not after_failed_apply: - _declare_keys(contract, say or print, client_factory) + _declare_keys(contract, say, client_factory) + _metric_views.post_apply(views, say, client_factory) def post_build( self, @@ -2843,6 +2868,7 @@ def planned_objects(contract: Mapping[str, Any]) -> List[Dict[str, Any]]: "tag_value": tag_value, } ) + actions.extend(compiled.metric_views.plan_actions()) actions.extend(_monitor_actions(compiled.monitors)) actions.extend(_job_actions(compiled.jobs)) unique: Dict[Tuple[str, str], Dict[str, Any]] = {} diff --git a/fluid_build/iac/providers/databricks_metrics.py b/fluid_build/iac/providers/databricks_metrics.py new file mode 100644 index 00000000..5df2d2d8 --- /dev/null +++ b/fluid_build/iac/providers/databricks_metrics.py @@ -0,0 +1,1540 @@ +# Copyright 2025 Agentics Transformation Limited +# SPDX-License-Identifier: Apache-2.0 + +"""Unity Catalog metric views from ``exposes[].semantics``: compile, create, grant, drop. + +Each Databricks expose that declares a ``semantics`` block gets one metric view, so Genie, +AI/BI dashboards and SQL read the contract's semantic model natively +(``SELECT country_code, MEASURE(site_count) FROM ... GROUP BY ALL``). The view is +``..
_semantic_model``, beside its table in the product's own schema, so +nothing lands outside what the product owns. It is named after its table, not after +``semantics.name``: the schema calls that a human-readable name, and forge's own emitters set it to +the table's name, which would be the table's own name in Unity Catalog. + +The definition is metric view YAML version 1.1 (Databricks, "Metric view YAML syntax +reference"). The mapping follows the Apache Ossie Databricks converter +(``apache/ossie-converters`` ``databricks/``, Apache-2.0) where FLUID and OSI meet, and refuses +what the YAML cannot hold rather than approximating it: + +* ``source`` is the expose's table. ``dimensions`` are the view's dimensions, and so are the + ``entities`` (the primary key, foreign keys): an entity over a bare column is that column under + the column's name (``customer`` over ``customer_id`` is ``customer_id``), and one whose key is no + column of the table is left out with a warning, since a metric view holds no join. A time + dimension's ``timeGranularity`` is applied + the way MetricFlow presents a dimension at its grain: a ``DATE`` column at ``day`` is used as it + is, a coarser grain on a ``DATE`` is ``TRUNC(col, 'MONTH')`` (still a ``DATE``), anything else + ``DATE_TRUNC('MONTH', col)``; ``hour`` or ``minute`` on a ``DATE`` is refused. +* A measure is its aggregate over its ``expr`` (the shared rendering of + ``forge_datamodel/semantics_builder.py::aggregate_call``, the MCP query compiler's too). A + ``simple`` metric is its measure's aggregate under the metric's name with its ``filter`` as + ``FILTER (WHERE ...)``; a ``ratio`` metric is ``MEASURE(numerator) / NULLIF(MEASURE(denominator), + 0)``, NULL rather than a division-by-zero error. ``derived`` metrics, a ratio that carries a + filter, and an unknown metric type are refused. +* ``nonAdditiveDimension`` is a window measure ordered by that dimension: ``range: current`` + and ``semiadditive: last`` for ``windowChoice: max`` (``first`` for ``min``, MetricFlow's + default). One difference is kept, not hidden: Databricks takes the last value of the dimension + per output group (a country whose sites stopped reporting earlier shows its own last day; + measured live), where MetricFlow without ``window_groupings`` takes one latest value for the + whole query. A dimension that is not a declared time dimension is refused. +* Every reference to a column of the table, in a dimension, a measure or a filter, is written + ``source.`` (``
.`` too); a word in a unit or type position + (``EXTRACT(YEAR FROM d)``, ``INTERVAL 1 DAY``, ``x::DATE``) is left as it is. A bare name in a metric view expression reads a field or a measure of the + view before a column (measured live: a dimension shadowing ``site_status`` made a filter on it + read the dimension, and a window measure named like a column broke every window measure that + read the column, ``METRIC_VIEW_WINDOW_MEASURE_REFERENCES_WINDOW_MEASURE``). A name that is no + column of the table but names a field or measure of the model is refused for the same reason. +* Expressions pass the same allowlist as the governed query path + (``providers/_sql_safety.validate_sql_expression_allowlist``: no statement keyword, no comment, + no ``;`` or ``$``), balanced parentheses, no aggregate where none belongs, no ``MEASURE()``, and + no call that would lend the view owner's rights to its readers (``secret()``, + ``http_request()``, ``ai_*()``, ``read_*()``, reflection, the querying user's identity). +* Text: the view's ``comment`` is ``semantics.description`` (else the expose's), then the model's + ``ai-synonyms`` label as ``Synonyms: ...`` and its ``ai-instructions`` label as + ``Instructions: ...``, since the YAML has no model-level synonym field. A dimension's + ``comment`` is its description, else its column's; its ``display_name`` is the column's + ``businessName`` and its ``synonyms`` the column's ``labels.synonyms`` (``;``-separated), at most + 10 of at most 255 characters each. Text holding ``$$`` (which would end the YAML body) or ``${`` + (which Spark may substitute) is refused. Not carried: ``aggTimeDimension`` and + ``defaultAggTimeDimension`` (MetricFlow's ``metric_time``; a metric view groups by a named time + dimension), ``createMetric`` (every measure is queryable), a metric's ``owner``, the model's + ``tags`` and its ``ai-example-questions`` label (a Genie space's sample questions). + +The databricks/databricks provider has no metric view resource (terraform-provider-databricks +#4827), so the view is created on the SQL warehouse through the SQL Statement Execution API, +like the column masks (``databricks_enforce.py``), around ``tofu apply``: + +* :func:`pre_apply` refuses, before anything is applied, a view whose name is taken by an object + this module did not create: another kind of table, a view without this product's + ``fluid.product_id`` property, or one the deploy identity does not own. It then revokes reading + each existing view from whoever the contract no longer names, so a ``tofu apply`` that fails + part way leaves no removed reader able to read. +* :func:`post_apply` runs after the masks and row filters are bound. It creates a missing view + with ``CREATE VIEW`` (never ``CREATE OR REPLACE``: replacing a view drops its grants, measured + live, and makes whoever replaces it its owner) and updates a changed one with ``ALTER VIEW ... + AS``, which keeps its grants. The view carries ``fluid.product_id``, ``fluid.expose_id`` and + ``fluid.definition_sha256`` as properties; one whose digest and stored structure match is + left alone, so a second apply runs nothing. Reading the view is exactly the principals that + may read its table (``SELECT`` or ``ALL_PRIVILEGES`` there, from ``accessPolicy.grants`` and + ``policy.authz.readers``): ``SELECT`` granted to them, ``SELECT`` and ``ALL PRIVILEGES`` revoked + from anyone else, revokes first and each on its own. Readers also need ``USE SCHEMA`` (the + module grants it) and ``USE CATALOG`` (named by the plan notice). +* A view of this product that the contract no longer declares, in a schema its tables are in + (its table renamed, its semantics or its expose removed), is dropped before ``tofu apply``: + dropping a view only narrows who reads. It is found by name (``*_semantic_model``, + ``table-summaries``) and dropped only with this product's ``fluid.product_id`` and the deploy + identity as owner. Only those schemas are looked in, so another environment of the product (an + overlay that changes the schema, in the same catalog, under the same identity) keeps its views. + The cost is the one the column masks' functions share: 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 (never ``force_destroy``) then fails while the schema holds it; drop it by hand + (``DROP VIEW ``) and re-run. +* A metric view reads its table with its owner's rights, and Unity Catalog applies the table's + row filter and column masks to whoever queries the view (a row filter measured live). When + binding them fails, :func:`lock_restricted` revokes ``SELECT`` on the views over restricted + tables from everyone but the owner before the run fails, so a view cannot serve rows the + fail-safe closed on the table. + +``fluid apply --mode dry-run`` prints every statement (:func:`preview`); a module written by +``fluid generate iac`` holds no metric view (it is no OpenTofu resource), and ``fluid apply`` +creates it. +""" + +from __future__ import annotations + +import copy +import hashlib +import os +import re +from dataclasses import dataclass, field +from typing import Any, Callable, Dict, Iterator, List, Mapping, Optional, Sequence, Set, Tuple +from urllib.parse import quote + +import yaml + +from ...forge_datamodel import semantics_builder as _semantics +from ...forge_datamodel.time_grains import normalize_time_grain +from ...providers._sql_safety import validate_ident, validate_sql_expression_allowlist +from ..base import UnsupportedBindingError +from . import databricks_enforce as _enforce +from .databricks_access import quote_principal, quote_table, revoke_all_statement, string_literal +from .databricks_enforce import ( + DATABRICKS_EXTRA_INSTALL, + WAREHOUSE_ENV, + Session, + SqlRunner, + StatementFailed, + _list_named, + _list_pages, + _privileges, + owner_matches, + read_table, + scrub, +) + +#: The metric view YAML version this module writes: the one with ``display_name``, ``synonyms`` +#: and window measures, measured on Databricks SQL 2026.45 (Free Edition). +YAML_VERSION = 1.1 +#: The ``table_type`` Unity Catalog reports for a metric view. +METRIC_VIEW = "METRIC_VIEW" +#: The view name when ``semantics.name`` is not given: ``
`` + this. +DEFAULT_SUFFIX = "_semantic_model" +#: Properties every view this module creates carries: whose it is, and what it was created from. +PRODUCT_PROPERTY = "fluid.product_id" +EXPOSE_PROPERTY = "fluid.expose_id" +DIGEST_PROPERTY = "fluid.definition_sha256" +#: Metric view YAML 1.1 limits: at most 10 synonyms, and 255 characters in a synonym or a +#: ``display_name``. +MAX_SYNONYMS = 10 +MAX_TEXT = 255 +#: How a metric view expression names its source table. +SOURCE = "source" +#: Table privileges that read: who holds one on the table may read its metric view. +_READ = frozenset({"SELECT", "ALL_PRIVILEGES"}) +#: Text keys Unity Catalog keeps as metadata and leaves out of the stored ``view_definition`` +#: (measured live): the structural comparison ignores them, the digest does not. +_METADATA_KEYS = ("comment", "display_name", "synonyms") +#: Grains a ``DATE`` column does not have. +_SUB_DAY = frozenset({"hour", "minute"}) + +#: A quoted identifier, a string literal (either quote), or an identifier that is not the tail +#: of a number or of a longer word. Backslashes cannot occur: the allowlist refuses them. +_TOKEN_RE = re.compile( + r"`(?:``|[^`])*`|'(?:''|[^'])*'|\"(?:\"\"|[^\"])*\"|(?)``. +_AFTER_AS_RE = re.compile(r"(?i)\bAS$") +#: What a word may follow and be a unit or a type, not a column: ``::``, ``EXTRACT(``, an +#: ``INTERVAL`` amount, the unit argument of the date-arithmetic functions. +_UNIT_CONTEXT_RE = re.compile( + r"(?is)(?:::|\bextract\s*\(|\binterval\s+[-+]?(?:\d+(?:\.\d+)?|'(?:''|[^'])*'))\s*$" +) +#: The date-arithmetic functions whose first argument is a unit in their 3-argument form +#: (``DATEADD(MONTH, 1, d)``) and a date in their 2-argument one (``date_add(d, 1)``): there a +#: word is a unit only when it is one of these. +_UNIT_ARGUMENT_RE = re.compile( + r"(?is)\b(?:dateadd|datediff|date_add|date_diff|timestampadd|timestampdiff)\s*\(\s*$" +) +_TIME_UNITS = frozenset( + { + "year", + "years", + "quarter", + "quarters", + "month", + "months", + "week", + "weeks", + "day", + "days", + "dayofyear", + "dayofweek", + "dow", + "doy", + "hour", + "hours", + "minute", + "minutes", + "second", + "seconds", + "millisecond", + "milliseconds", + "microsecond", + "microseconds", + } +) + +Say = Callable[[str], None] +ClientFactory = Callable[[], Any] +Step = Tuple[str, str] + + +def _refusal(kind: str, message: str, *remediation: str) -> UnsupportedBindingError: + return UnsupportedBindingError(f"databricks-metric-view-{kind}", message, tuple(remediation)) + + +class MetricViewError(UnsupportedBindingError): + """A metric view could not be created, updated, granted or dropped on the warehouse.""" + + def __init__(self, message: str, remediation: Sequence[str] = ()): + super().__init__("databricks-metric-view-failed", message, tuple(remediation)) + + +class _Dumper(yaml.SafeDumper): + """Plain YAML for a reader: no ``&id`` anchors for a repeated value, multi-line text as a + ``|`` block.""" + + def ignore_aliases(self, data: Any) -> bool: + return True + + +def _represent_str(dumper: yaml.SafeDumper, data: str) -> yaml.ScalarNode: + style = "|" if "\n" in data else None + return dumper.represent_scalar("tag:yaml.org,2002:str", data, style=style) + + +_Dumper.add_representer(str, _represent_str) + + +def _dump(document: Mapping[str, Any]) -> str: + return yaml.dump( + dict(document), Dumper=_Dumper, sort_keys=False, allow_unicode=True, width=1_000_000 + ) + + +def _plain_ident(raw: Any) -> Optional[str]: + """``raw`` stripped when it is a plain identifier, else ``None``.""" + if not isinstance(raw, str): + return None + try: + return validate_ident(raw.strip()) + except ValueError: + return None + + +# ── the compiled views ───────────────────────────────────────────────── + + +@dataclass(frozen=True) +class MetricView: + """One metric view: where it goes, what it says and who reads it.""" + + full_name: str + #: The table it reads. + source: str + expose_id: str + product_id: str + #: The metric view YAML. + definition: str + #: The principals ``SELECT`` on the view goes to, as the table's grants name them. + readers: Tuple[str, ...] + #: The table it reads has column masks or a row filter. + restricted: bool + where: str + + @property + def digest(self) -> str: + return hashlib.sha256(self.definition.encode("utf-8")).hexdigest() + + def _properties(self) -> str: + pairs = ( + (PRODUCT_PROPERTY, self.product_id), + (EXPOSE_PROPERTY, self.expose_id), + (DIGEST_PROPERTY, self.digest), + ) + return ", ".join(f"{string_literal(k)} = {string_literal(v)}" for k, v in pairs) + + @property + def create(self) -> str: + return ( + f"CREATE VIEW {quote_table(self.full_name)} WITH METRICS LANGUAGE YAML " + f"TBLPROPERTIES ({self._properties()}) AS $${self.definition}$$" + ) + + @property + def update(self) -> Tuple[str, str]: + """``ALTER VIEW ... AS`` (it keeps the view's grants), then the new digest.""" + name = quote_table(self.full_name) + return ( + f"ALTER VIEW {name} AS $${self.definition}$$", + f"ALTER VIEW {name} SET TBLPROPERTIES ({self._properties()})", + ) + + +@dataclass +class MetricViews: + """Every metric view a contract declares, and where a view it no longer declares may be.""" + + product_id: str = "" + views: List[MetricView] = field(default_factory=list) + #: ``catalog.schema`` of each schema the product's tables are in. + schemas: Tuple[str, ...] = () + + def plan_actions(self) -> List[Dict[str, Any]]: + """The views as the native planner's actions (``databricks.planned_objects``).""" + return [ + { + # No provider resource: fluid apply creates it on the SQL warehouse, so the + # op names no kind ``iac/shadow.py`` compares with the OpenTofu module (as a + # grant names none); "view" in it would read as a resource the module lacks. + "op": "databricks.semantic_model.ensure", + "resource_type": "databricks_metric_view", + "resource_id": view.full_name, + "source": view.source, + "readers": list(view.readers), + } + for view in self.views + ] + + +@dataclass(frozen=True) +class _Field: + name: str + #: The expression as the contract wrote it. + raw: str + is_time: bool + + +@dataclass(frozen=True) +class _Measure: + """A base measure, for the simple metrics over it.""" + + entry: Mapping[str, Any] + #: Its ``expr``, columns qualified. + expr: str + window: Optional[List[Dict[str, str]]] + + +@dataclass(frozen=True) +class _Ratio: + name: str + numerator: str + denominator: str + what: str + description: Any + + +class _ViewCompiler: + """One expose's ``semantics`` block → metric view YAML, or refusals.""" + + def __init__( + self, table: Any, semantics: Mapping[str, Any], refusals: List[Any], warnings: List[str] + ): + self.table = table + self.semantics = semantics + self.where = f"{table.where} semantics" + self.refusals = refusals + self.warnings = warnings + self.failed = False + #: The table's columns by the lower-case name Unity Catalog stores. + self.columns: Dict[str, Mapping[str, Any]] = {c["name"]: c for c in table.columns} + self.schema_columns: Dict[str, Mapping[str, Any]] = {} + for column in (table.exposure.get("contract") or {}).get("schema") or []: + if isinstance(column, Mapping): + self.schema_columns[str(column.get("name") or "").strip().lower()] = column + #: Every field or measure name the view will hold, lower-cased: a bare one in an + #: expression would read it. An entity over a bare column is that column, no name of its own. + self.declared: Set[str] = { + entry["name"].strip().lower() + for kind in ("entities", "dimensions", "measures", "metrics") + for entry in semantics.get(kind) or [] + if isinstance(entry, Mapping) + and isinstance(entry.get("name"), str) + and not (kind == "entities" and _plain_ident(entry.get("expr") or entry["name"])) + } + #: Names taken in the view (fields and measures share one namespace, ignoring case). + self.taken: Dict[str, str] = {} + self.dimension_docs: List[Dict[str, Any]] = [] + self.fields: Dict[str, _Field] = {} + self.measure_docs: List[Dict[str, Any]] = [] + self.base: Dict[str, _Measure] = {} + self.ratios: List[_Ratio] = [] + + # ── checks ── + + def refuse(self, kind: str, message: str, *remediation: str) -> None: + self.failed = True + self.refusals.append(_refusal(kind, message, *remediation)) + + def name(self, raw: Any, what: str) -> Optional[str]: + text = _plain_ident(raw) + if text is None: + self.refuse( + "invalid", + f"{what} is {raw!r}, which is not a plain identifier (letters, digits and " + "underscores, not starting with a digit).", + "Rename it: metric view dimensions and measures are queried by this name.", + ) + return text + + def claim(self, name: str, what: str) -> bool: + lower = name.lower() + if lower == SOURCE: + self.refuse( + "invalid", + f"{what} is named {name!r}, the name a metric view expression gives its source " + "table: every source. in the view would read it instead.", + "Rename it.", + ) + return False + if lower in self.taken: + self.refuse( + "collision", + f"{what} is named {name!r}, as is {self.taken[lower]}: a metric view's dimensions " + "and measures share one namespace, ignoring case.", + "Rename one of them.", + ) + return False + self.taken[lower] = what + return True + + def text(self, raw: Any, what: str, *, limit: Optional[int] = None) -> Optional[str]: + if raw is None or isinstance(raw, (Mapping, list)): + return None + text = str(raw).strip() + if not text: + return None + if "$$" in text or "${" in text: + self.refuse( + "invalid", + f"{what} holds {'$$' if '$$' in text else '${'}: the metric view's YAML is sent " + "between $$ quotes, which $$ would end, and Spark may substitute ${...} with a " + "variable, setting or environment value.", + "Write it without '$$' and '${' (for example '$ {').", + ) + return None + if limit is not None and len(text) > limit: + self.refuse( + "invalid", + f"{what} is {len(text)} characters; a metric view holds at most {limit}.", + f"Shorten it to {limit} characters.", + ) + return None + return text + + def expression(self, raw: Any, what: str) -> Optional[str]: + """``raw`` with every column of the table written ``source.``; ``None`` if refused.""" + text = self._allowed(raw, what) + if text is None: + return None + scan = _blank(text, keep="`") # string literals out, quoted names kept + refused_call = _refused_call(_blank(text, keep='`"')) + if refused_call is not None: + self.refuse( + "invalid", + f"{what} calls {refused_call}: a metric view evaluates its expressions with its " + "owner's rights for everyone who queries it, so it may not read secrets, reach a " + "connection, a model or a file, reflect, ask who is querying, name a function " + "from a string, or call a function by a quoted or qualified name (another " + "schema's function runs with the owner's rights too).", + "Compute the value in the table's build and reference the column.", + ) + return None + nested = _semantics.first_aggregate_call(scan) + if _MEASURE_CALL_RE.search(scan) or nested is not None: + called = "MEASURE()" if _MEASURE_CALL_RE.search(scan) else f"{str(nested).upper()}(...)" + self.refuse( + "invalid", + f"{what} is {raw!r}, which calls {called}: a contract expression reads the " + "table's columns, and the metric view applies the measure's agg itself (a " + "dimension or filter cannot hold an aggregate).", + "Split it: 'expr: SUM(amount)' + 'agg: sum' becomes 'expr: amount' + 'agg: sum'; " + "a ratio of two measures is a ratio metric.", + ) + return None + qualified, bare = _qualify(text, self.columns, self.table.name) + shadowed = sorted({n for n in bare if n.lower() in self.declared}) + if shadowed: + self.refuse( + "invalid", + f"{what} reads {', '.join(repr(n) for n in shadowed)}, which is not a column of " + f"{self.table.full_name} but names a dimension or measure of this model: in a " + "metric view the name would read that dimension or measure, not a column.", + "Declare the column in contract.schema, or rename the dimension or measure.", + ) + return None + return qualified + + def _allowed(self, raw: Any, what: str) -> Optional[str]: + """``raw`` if the allowlist takes it and its parentheses balance.""" + try: + text = validate_sql_expression_allowlist(raw) if isinstance(raw, str) else "" + except ValueError: + text = "" + if not text: + self.refuse( + "invalid", + f"{what} is {raw!r}, which is not an expression this module writes into a metric " + "view: only names, literals, operators and parentheses, with no statement " + "keyword (SELECT, DROP, ...), no comment, and no ';' or '$'.", + "Rewrite it as a plain SQL expression over the table's columns.", + ) + return None + code = _blank(text) # a '(' in '...' or `...` opens nothing + if any(quote in code for quote in "'\"`"): + self.refuse( + "invalid", + f"{what} is {raw!r}, which opens a quote or a quoted name it does not close.", + "Close every quote and backtick it opens.", + ) + return None + if not _semantics.parens_balanced(code): + self.refuse( + "invalid", + f"{what} is {raw!r}, whose parentheses do not balance.", + "Close every parenthesis it opens.", + ) + return None + return text + + def no_column(self, raw: str, name: str, what: str) -> bool: + """Refuse a bare name that is no column of the table: the view would read nothing.""" + column = _plain_ident(raw) + if column is None or column.lower() in self.columns: + return False + self.refuse( + "invalid", + f"{what} ({name}) reads {raw!r}, which is no column of {self.table.full_name}.", + "Name a column of contract.schema, or write the expression over its columns.", + ) + return True + + # ── dimensions ── + + def compile_fields(self) -> None: + for kind in ("entities", "dimensions"): + for position, entry in enumerate(self.semantics.get(kind) or []): + if isinstance(entry, Mapping): + self._add_field(kind, f"{self.where}.{kind}[{position}]", entry) + + def _add_field(self, kind: str, what: str, entry: Mapping[str, Any]) -> None: + name = self.name(entry.get("name"), f"{what}.name") + if name is None: + return + raw = str(entry.get("expr") or name).strip() + column = _plain_ident(raw) + if kind == "entities" and column is not None: + if column.lower() not in self.columns: + self.warnings.append( + f"{what} ({name}) is the key {raw!r}, which is no column of " + f"{self.table.full_name}: an entity is a join key, and the metric view, which " + "holds no join, leaves it out." + ) + return + name = column # an entity is its key column, under the column's own name + elif self.no_column(raw, name, what): + return + is_time = kind == "dimensions" and str(entry.get("type") or "") == "time" + prior = self.fields.get(name.lower()) + if prior is not None and not is_time and raw == prior.raw: + return # an entity and a dimension over one column: one field + if not self.claim(name, f"{what} ({name})"): + return + expr = self.expression(raw, f"{what}.expr") + if expr is not None and is_time: + expr = self._time_expr(entry, raw, expr, what) + if expr is None: + return + document: Dict[str, Any] = {"name": name, "expr": expr} + self._describe(document, entry, raw, what) + self.dimension_docs.append(document) + self.fields[name.lower()] = _Field(name=name, raw=raw, is_time=is_time) + + def _time_expr(self, entry: Mapping[str, Any], raw: str, expr: str, what: str) -> Optional[str]: + params = entry.get("typeParams") + declared = params.get("timeGranularity") if isinstance(params, Mapping) else None + if declared is None: + return expr + grain = normalize_time_grain(str(declared)) + if grain is None: + self.refuse( + "unsupported", + f"{what}.typeParams.timeGranularity is {declared!r}, which is no grain a metric " + "view can truncate to.", + "Use day, week, month, quarter, year, hour or minute.", + ) + return None + column = self.columns.get(raw.strip().lower()) + if column is None or str(column.get("type")) != "date": + return f"DATE_TRUNC('{grain.upper()}', {expr})" + if grain in _SUB_DAY: + self.refuse( + "unsupported", + f"{what} is the DATE column {raw!r} with timeGranularity {grain!r}, which a " + "date does not have.", + "Use a day or coarser grain, or a TIMESTAMP column.", + ) + return None + return expr if grain == "day" else f"TRUNC({expr}, '{grain.upper()}')" + + def _describe( + self, document: Dict[str, Any], entry: Mapping[str, Any], raw: str, what: str + ) -> None: + """The field's comment, and the display name and synonyms of the column it is.""" + bare = raw.strip().lower() + column = self.schema_columns.get(bare) if bare in self.columns else None + comment = self.text(entry.get("description"), f"{what}.description") + if comment is None and column is not None: + comment = self.text(column.get("description"), f"{self.table.where} column {raw!r}") + if comment is not None: + document["comment"] = comment + if column is None: + return + display = self.text( + column.get("businessName"), + f"{self.table.where} column {raw!r} businessName", + limit=MAX_TEXT, + ) + if display is not None: + document["display_name"] = display + synonyms = self._synonyms(column, raw) + if synonyms: + document["synonyms"] = synonyms + + def _synonyms(self, column: Mapping[str, Any], raw: str) -> List[str]: + labels = column.get("labels") + value = labels.get("synonyms") if isinstance(labels, Mapping) else None + if not isinstance(value, str): + return [] + what = f"{self.table.where} column {raw!r} labels.synonyms" + out: List[str] = [] + for part in value.split(";"): + text = self.text(part, what, limit=MAX_TEXT) + if text is not None and text.lower() not in {s.lower() for s in out}: + out.append(text) + if len(out) > MAX_SYNONYMS: + self.refuse( + "invalid", + f"{what} lists {len(out)} synonyms; a metric view field holds at most " + f"{MAX_SYNONYMS}.", + f"Keep the {MAX_SYNONYMS} that matter most.", + ) + return [] + return out + + # ── measures ── + + def compile_measures(self) -> None: + for position, entry in enumerate(self.semantics.get("measures") or []): + if isinstance(entry, Mapping): + self._add_measure(f"{self.where}.measures[{position}]", entry) + for position, entry in enumerate(self.semantics.get("metrics") or []): + if isinstance(entry, Mapping): + self._add_metric(f"{self.where}.metrics[{position}]", entry) + self._place_ratios() + + def _add_measure(self, what: str, entry: Mapping[str, Any]) -> None: + name = self.name(entry.get("name"), f"{what}.name") + if name is None or not self.claim(name, f"{what} ({name})"): + return + raw = str(entry.get("expr") or name).strip() + if self.no_column(raw, name, what): + return + expr = self.expression(raw, f"{what}.expr") + window = self._window(entry, what) + if expr is None or window is False: + return + try: + aggregate = _semantics.aggregate_call(name, entry, expr) + except ValueError as exc: + self.refuse("unsupported", f"{what}: {exc}.", "Use an agg the contract schema lists.") + return + self.base[name.lower()] = _Measure(entry=entry, expr=expr, window=window or None) + self.measure_docs.append(self._measure(name, aggregate, entry.get("description"), what)) + if window: + self.measure_docs[-1]["window"] = copy.deepcopy(window) + + def _add_metric(self, what: str, entry: Mapping[str, Any]) -> None: + name = self.name(entry.get("name"), f"{what}.name") + if name is None: + return + kind = str(entry.get("type") or "") + if kind == "simple": + self._simple(name, entry, what) + elif kind == "ratio": + self._ratio(name, entry, what) + elif kind == "derived": + self.refuse( + "unsupported", + f"{what} ({name}) is a derived metric: its expr names other metrics, which a " + "metric view reads only through MEASURE(), and forge-cli does not rewrite the " + "expression, so it is refused rather than approximated.", + "Express it as a ratio of two metrics, or remove it for Databricks.", + ) + else: + self.refuse( + "unsupported", + f"{what} ({name}) has type {kind!r}; a metric view takes simple and ratio metrics.", + "Use type simple or ratio.", + ) + + def _measure(self, name: str, expr: str, description: Any, what: str) -> Dict[str, Any]: + document: Dict[str, Any] = {"name": name, "expr": expr} + comment = self.text(description, f"{what}.description") + if comment is not None: + document["comment"] = comment + return document + + def _window(self, entry: Mapping[str, Any], what: str) -> Any: + """The window of a non-additive measure; ``None`` for none, ``False`` when refused.""" + nad = entry.get("nonAdditiveDimension") + if nad is None: + return None + name = nad.get("name") if isinstance(nad, Mapping) else None + target = self.fields.get(name.strip().lower()) if isinstance(name, str) else None + if target is None or not target.is_time: + self.refuse( + "unsupported", + f"{what}.nonAdditiveDimension names {name!r}, which is not a time dimension of " + "this model: a metric view orders the window of a non-additive measure by one of " + "its own dimensions.", + "Name a dimension of type time declared in semantics.dimensions.", + ) + return False + choice = nad.get("windowChoice", "min") + if choice not in ("min", "max"): + self.refuse( + "unsupported", + f"{what}.nonAdditiveDimension.windowChoice is {choice!r}; it is min or max.", + "Use min (the first value) or max (the last).", + ) + return False + semiadditive = "last" if choice == "max" else "first" + return [{"order": target.name, "range": "current", "semiadditive": semiadditive}] + + def _simple(self, name: str, entry: Mapping[str, Any], what: str) -> None: + ref = entry.get("measure") + base = self.base.get(ref.strip().lower()) if isinstance(ref, str) else None + if base is None: + self.refuse( + "unsupported", + f"{what} ({name}) is a simple metric over measure {ref!r}, which this model does " + "not declare (or which was refused).", + "Name a measure declared in semantics.measures.", + ) + return + aggregate = _semantics.aggregate_call(str(ref).strip(), base.entry, base.expr) + condition = entry.get("filter") + if condition is None and name.lower() == str(ref).strip().lower(): + return # the metric is its measure, which the view already holds + if condition is not None: + qualified = self.expression(condition, f"{what}.filter") + if qualified is None: + return + aggregate = f"{aggregate} FILTER (WHERE {qualified})" + if not self.claim(name, f"{what} ({name})"): + return + self.measure_docs.append(self._measure(name, aggregate, entry.get("description"), what)) + if base.window: + self.measure_docs[-1]["window"] = copy.deepcopy(base.window) + + def _ratio(self, name: str, entry: Mapping[str, Any], what: str) -> None: + if entry.get("filter") is not None: + self.refuse( + "unsupported", + f"{what} ({name}) is a ratio metric with a filter, which a metric view cannot " + "apply to MEASURE(numerator) / MEASURE(denominator).", + "Put the filter on its numerator and denominator metrics instead.", + ) + return + inputs = [m for m in entry.get("inputMetrics") or [] if isinstance(m, str)] + numerator = self.name( + entry.get("numerator") or (inputs[0] if inputs else None), f"{what}.numerator" + ) + denominator = self.name( + entry.get("denominator") or (inputs[1] if len(inputs) > 1 else None), + f"{what}.denominator", + ) + if numerator and denominator and self.claim(name, f"{what} ({name})"): + self.ratios.append(_Ratio(name, numerator, denominator, what, entry.get("description"))) + + def _place_ratios(self) -> None: + """Ratios after the measures they read: a metric view resolves ``MEASURE()`` against + the measures defined before it.""" + placed = {document["name"].lower() for document in self.measure_docs} + pending = list(self.ratios) + while True: + ready = [r for r in pending if {r.numerator.lower(), r.denominator.lower()} <= placed] + if not ready: + break + for ratio in ready: + expr = f"MEASURE({ratio.numerator}) / NULLIF(MEASURE({ratio.denominator}), 0)" + self.measure_docs.append( + self._measure(ratio.name, expr, ratio.description, ratio.what) + ) + placed.add(ratio.name.lower()) + pending.remove(ratio) + for ratio in pending: + missing = [m for m in (ratio.numerator, ratio.denominator) if m.lower() not in placed] + self.refuse( + "unsupported", + f"{ratio.what} ({ratio.name}) divides {ratio.numerator!r} by " + f"{ratio.denominator!r}, but {', '.join(repr(m) for m in missing)} is no measure " + "or metric this view holds (undeclared, refused, or a ratio that reads this one).", + "Name metrics or measures this model declares.", + ) + + # ── the view ── + + def comment(self) -> Optional[str]: + description = self.text(self.semantics.get("description"), f"{self.where}.description") + if description is None: + description = self.text( + self.table.exposure.get("description"), f"{self.table.where} description" + ) + parts = [description] if description else [] + labels = self.semantics.get("labels") + labels = labels if isinstance(labels, Mapping) else {} + synonyms = self.text(labels.get("ai-synonyms"), f"{self.where}.labels.ai-synonyms") + if synonyms: + parts.append(f"Synonyms: {synonyms.rstrip('.')}.") + instructions = self.text( + labels.get("ai-instructions"), f"{self.where}.labels.ai-instructions" + ) + if instructions: + parts.append(f"Instructions: {instructions}") + return "\n\n".join(parts) or None + + def property_values(self, product_id: str) -> None: + """Refuse a property value the CREATE statement could not hold as itself.""" + for value, what in ( + (product_id, "the contract's id"), + (str(self.table.expose_id), f"{self.table.where} exposeId"), + ): + if not value or _CONTROL_RE.search(value) or "${" in value or "$$" in value: + self.refuse( + "invalid", + f"{what} is {value!r}; the metric view records it as a property, which takes " + "no control character, '${' or '$$', and no empty value.", + "Rename it.", + ) + + +def _blank(text: str, keep: str = "") -> str: + """``text`` with each quoted span replaced by a space unless its quote is in ``keep``: + what SQL reads as code, the spans found in one pass as Databricks lexes them.""" + return _SPAN_RE.sub(lambda m: m.group(0) if m.group(0)[0] in keep else " ", text) + + +def _refused_call(text: str) -> Optional[str]: + """The first call in ``text`` (string literals blanked, quoted names kept) that may not run + with the owner's rights, as written, or ``None``.""" + for match in _CALL_RE.finditer(text): + call = match.group(1) + name = call.lower() + if "`" in call or '"' in call or "." in call: + return f"{call}() (a quoted or qualified function name)" + if name in _FORBIDDEN_FUNCTIONS or name.startswith(_FORBIDDEN_PREFIXES): + return f"{call}()" + return None + + +def _qualify(text: str, columns: Mapping[str, Any], table: str = "") -> Tuple[str, List[str]]: + """``text`` with each name of a column in ``columns`` written ``source.``, and the + other names it reads. + + A name is a column reference unless it is in a string literal, follows a ``.`` (a field of + a struct, or a name already qualified), is called (``name(``), prefixes a typed literal + (``DATE'2026-01-01'``), names a type (``CAST(x AS DATE)``, ``x::DATE``) or a unit + (``EXTRACT(YEAR FROM d)``, ``INTERVAL 1 DAY``, ``DATEADD(MONTH, 1, d)``). ``
.``, + the table qualifying its own column, is ``source.``. + """ + out: List[str] = [] + bare: List[str] = [] + last = 0 + for match in _TOKEN_RE.finditer(text): + token = match.group(0) + if token[0] in "'\"": + continue + before = text[: match.start()].rstrip() + after = text[match.end() :].lstrip() + typed = after.startswith(("'", '"')) or _AFTER_AS_RE.search(before) # DATE'..', AS DATE + unit = _UNIT_CONTEXT_RE.search(before) or ( + _UNIT_ARGUMENT_RE.search(before) and token.lower() in _TIME_UNITS + ) + if before.endswith(".") or after.startswith("(") or typed or unit: + continue + name = token[1:-1].replace("``", "`") if token[0] == "`" else token + if name.lower() in columns: + out += [text[last : match.start()], f"{SOURCE}.{token}"] + last = match.end() + elif table and name.lower() == table.lower() and after.startswith("."): + out += [text[last : match.start()], SOURCE] + last = match.end() + else: + bare.append(name) + out.append(text[last:]) + return "".join(out), bare + + +def compile_metric_views( + contract: Mapping[str, Any], + tables: Sequence[Any], + privileges: Mapping[str, Mapping[str, Set[str]]], + refusals: List[UnsupportedBindingError], + warnings: List[str], +) -> MetricViews: + """The metric view of each table whose expose declares ``semantics``; refusals appended. + + ``tables`` are the module's resolved tables (``databricks._Table``: ``where``, + ``expose_id``, ``catalog``, ``schema``, ``name``, ``full_name``, ``columns``, ``exposure``, + ``enforcement``); ``privileges`` is ``{table full name: {principal: privileges}}``. + """ + product_id = str(contract.get("id") or contract.get("name") or "").strip() + plan = MetricViews( + product_id=product_id, + schemas=tuple(sorted({f"{t.catalog}.{t.schema}" for t in tables})), + ) + names = {t.full_name.lower(): t.where for t in tables} + for table in tables: + if not isinstance(table.exposure.get("semantics"), Mapping): + continue + view = _compile_view(table, product_id, privileges, refusals, warnings) + if view is None: + continue + clash = names.get(view.full_name.lower()) + if clash is not None: + refusals.append( + _refusal( + "collision", + f"{table.where} semantics would create the metric view {view.full_name}, a " + f"name {clash} already takes.", + "Rename one of the two tables: the view is named after its table.", + ) + ) + continue + names[view.full_name.lower()] = view.where + plan.views.append(view) + return plan + + +def _compile_view( + table: Any, + product_id: str, + privileges: Mapping[str, Mapping[str, Set[str]]], + refusals: List[UnsupportedBindingError], + warnings: List[str], +) -> Optional[MetricView]: + compiler = _ViewCompiler(table, table.exposure["semantics"], refusals, warnings) + view_name = f"{table.name}{DEFAULT_SUFFIX}" + compiler.compile_fields() + compiler.compile_measures() + comment = compiler.comment() + compiler.property_values(product_id) + if compiler.failed: + return None + if not compiler.dimension_docs and not compiler.measure_docs: + warnings.append( + f"{table.where} semantics declares no entity, dimension, measure or metric, so no " + "metric view is created for it." + ) + return None + document: Dict[str, Any] = {"version": YAML_VERSION} + if comment: + document["comment"] = comment + document["source"] = table.full_name + if compiler.dimension_docs: + document["dimensions"] = compiler.dimension_docs + if compiler.measure_docs: + document["measures"] = compiler.measure_docs + definition = _dump(document) + if "$$" in definition or "${" in definition: # every text is checked; this is the backstop + refusals.append( + _refusal( + "invalid", + f"{table.where} semantics renders a metric view definition holding '$$' or '${{'.", + "Remove '$$' and '${' from the semantics block and the column texts it reads.", + ) + ) + return None + readers = sorted( + principal + for principal, granted in privileges.get(table.full_name, {}).items() + if set(granted) & _READ + ) + return MetricView( + full_name=f"{table.catalog}.{table.schema}.{view_name}", + source=table.full_name, + expose_id=str(table.expose_id), + product_id=product_id, + definition=definition, + readers=tuple(readers), + restricted=table.enforcement is not None, + where=compiler.where, + ) + + +# ── live state ───────────────────────────────────────────────────────── + + +def _structure(document: Any) -> Any: + """A metric view document without the text Unity Catalog keeps outside it.""" + if not isinstance(document, Mapping): + return document + out = {k: v for k, v in copy.deepcopy(dict(document)).items() if k != "comment"} + for key in ("dimensions", "fields", "measures"): + items = out.get(key) + if isinstance(items, list): + out[key] = [ + ( + {k: v for k, v in item.items() if k not in _METADATA_KEYS} + if isinstance(item, Mapping) + else item + ) + for item in items + ] + return out + + +@dataclass(frozen=True) +class _Live: + """What holds a view's name: ``absent``, ``ours`` (``current`` when it matches) or + ``foreign``, with why.""" + + state: str + info: Optional[Mapping[str, Any]] = None + current: bool = False + why: str = "" + + +def _stored(info: Mapping[str, Any]) -> Any: + try: + return yaml.safe_load(str(info.get("view_definition") or "")) + except yaml.YAMLError: + return None + + +#: How a ``table_type`` reads in a message. +_KINDS = { + "MANAGED": "a managed table", + "EXTERNAL": "an external table", + "VIEW": "a view", + "MATERIALIZED_VIEW": "a materialized view", + "STREAMING_TABLE": "a streaming table", +} + + +def _classify(session: Session, view: MetricView) -> _Live: + info = read_table(session.client, view.full_name) + if info is None: + return _Live("absent") + kind = str(info.get("table_type") or "").upper() + found = (info.get("properties") or {}).get(PRODUCT_PROPERTY) + owner = str(info.get("owner") or "") + if kind != METRIC_VIEW: + why = f"it is {_KINDS.get(kind, 'a ' + (kind.lower() or 'table'))}, not a metric view" + elif found != view.product_id: + why = ( + f"it is the metric view of product {found!r}" + if found + else "it is a metric view this module did not create (no fluid.product_id)" + ) + elif not owner_matches(owner, {session.identity}): + why = ( + f"it is owned by {owner or 'an unknown owner'}, not the deploy identity " + f"{session.identity}" + ) + else: + current = (info.get("properties") or {}).get(DIGEST_PROPERTY) == view.digest and _stored( + info + ) == _structure(yaml.safe_load(view.definition)) + return _Live("ours", info, current=current) + return _Live("foreign", info, why=why) + + +def _session( + client_factory: Optional[ClientFactory], say: Say, *, required: bool +) -> Optional[Session]: + """A client, the warehouse runner and the run's identity; ``None`` (or raised, when + ``required``) without one.""" + try: + # The enforcement module's, looked up per call: the one client a test patches. + client = (client_factory or _enforce.default_client)() + except UnsupportedBindingError: + if required: + raise + return None + except ImportError: + if required: + raise MetricViewError( + "databricks-sdk is not installed, and the contract's metric views are created " + "through it.", + (f"Install the extra: {DATABRICKS_EXTRA_INSTALL}",), + ) from None + return None # nothing declared now, and no SDK an earlier view could have come from + except Exception as exc: # noqa: BLE001 — a bad credential must not become a traceback + if required: + raise MetricViewError( + f"could not create the Databricks client for the metric views: " + f"{type(exc).__name__}: {scrub(exc)}" + ) from exc + say( + " note: metric views an earlier apply created were not checked for removal (no " + "Databricks client); any left in place are still readable by their grantees." + ) + return None + warehouse = str(os.environ.get(WAREHOUSE_ENV) or "").strip() + if not warehouse: + if not required: + return None + raise MetricViewError( + f"{WAREHOUSE_ENV} is not set, and the metric views are created on that warehouse.", + (f"Set {WAREHOUSE_ENV} to the id of a SQL warehouse in the workspace.",), + ) + return Session(client, SqlRunner(client, warehouse)) + + +def _permissions(session: Session, full_name: str) -> Dict[str, Set[str]]: + """``{principal: privileges}`` granted directly on a table or view.""" + path = f"/api/2.1/unity-catalog/permissions/table/{quote(full_name, safe='.')}" + return { + str(entry["principal"]): set(_privileges(entry)) + for entry in _list_pages(session.client, path) + if entry.get("principal") + } + + +def _exempt(session: Session, info: Optional[Mapping[str, Any]]) -> Set[str]: + """Who a grant change never touches: the view's owner and this run's identity.""" + return {p.lower() for p in (str((info or {}).get("owner") or ""), session.identity) if p} + + +def _grant_steps( + session: Session, view: MetricView, live: Optional[Mapping[str, Any]] +) -> Tuple[List[Step], List[Step]]: + """``(revokes, grants)`` that make reading the view exactly its readers: ``SELECT`` granted to + each reader, ``SELECT`` and ``ALL PRIVILEGES`` revoked from everyone else but the view's owner + and this run's identity.""" + name = quote_table(view.full_name) + exempt = _exempt(session, live) + held = _permissions(session, view.full_name) if live is not None else {} + wanted = {p.lower(): p for p in view.readers} + revokes: List[Step] = [] + for principal, privileges in sorted(held.items()): + if principal.lower() in wanted or principal.lower() in exempt: + continue + if "SELECT" in privileges: + revokes.append( + ( + f"REVOKE SELECT ON TABLE {name} FROM {quote_principal(principal)}", + f"revoke SELECT from {principal}", + ) + ) + if "ALL_PRIVILEGES" in privileges: + revokes.append( + ( + revoke_all_statement(view.full_name, principal), + f"revoke ALL PRIVILEGES from {principal}", + ) + ) + holding = {p.lower() for p, privileges in held.items() if "SELECT" in privileges} + grants = [ + (f"GRANT SELECT ON TABLE {name} TO {quote_principal(p)}", f"grant SELECT to {p}") + for lower, p in sorted(wanted.items()) + if lower not in holding and lower not in exempt + ] + return revokes, grants + + +def _definition_steps(view: MetricView, live: _Live) -> List[Step]: + if live.state == "absent": + return [(view.create, f"create metric view {view.full_name}")] + if live.current: + return [] + return [ + (view.update[0], f"update metric view {view.full_name}"), + (view.update[1], "record its definition digest"), + ] + + +def _say_statement(say: Say, statement: str, indent: str = " ") -> None: + for line in statement.splitlines(): + say(f"{indent}{line}") + + +# ── the hooks ────────────────────────────────────────────────────────── + + +def preview(plan: MetricViews, say: Say, client_factory: Optional[ClientFactory] = None) -> None: + """Print what an apply would run for the metric views, in order; run none.""" + if not plan.views and not plan.schemas: + return + session = _session(client_factory, say, required=False) + if plan.views: + say("\n Unity Catalog metric views (run on the SQL warehouse after tofu apply):") + for view in plan.views: + _preview_view(session, view, say) + if session is not None: + _drop_stale(session, plan, say, dry_run=True) + if plan.views: + say(" dry-run: none of these statements ran.") + + +def _preview_view(session: Optional[Session], view: MetricView, say: Say) -> None: + try: + live = _classify(session, view) if session is not None else _Live("absent") + except Exception as exc: # noqa: BLE001 — a dry run reports, it does not stop + say(f" {view.full_name}: could not be read ({type(exc).__name__}: {scrub(exc)})") + live = _Live("absent") + if live.state == "foreign": + say(f" {view.full_name}: REFUSED before the apply: {live.why}.") + return + say(f" {view.full_name}" + ("" if session else " (live state not read)")) + if live.state == "absent": + say(f" after apply: create the metric view over {view.source}") + elif live.current: + say(" unchanged") + else: + say(" after apply: update the definition (ALTER VIEW keeps its grants)") + for statement, _what in _definition_steps(view, live): + _say_statement(say, statement) + try: + assert session is not None + revokes, grants = _grant_steps(session, view, live.info if live.state == "ours" else None) + except Exception: # noqa: BLE001 — unread: say what the contract asks + revokes, grants = [], [("", f"grant SELECT to {p}") for p in view.readers] + for _statement, what in revokes: + say(f" before apply: {what}") + for _statement, what in grants: + say(f" after apply: {what}") + + +def pre_apply(plan: MetricViews, say: Say, client_factory: Optional[ClientFactory] = None) -> None: + """Before ``tofu apply``: refuse a view whose name an object this module did not create holds + (creating ours would fail, and replacing it would take it over), then revoke reading each + existing view from whoever the contract no longer names, so no later failure leaves it open, + and drop this product's views the contract no longer declares (:func:`_drop_stale`).""" + if not plan.views and not plan.schemas: + return + session = _session(client_factory, say, required=bool(plan.views)) + if session is None: + return + conflicts, narrowing = [], [] + for view in plan.views: + try: + live = _classify(session, view) + if live.state == "ours": + narrowing.append((view, _grant_steps(session, view, live.info)[0])) + except Exception as exc: # noqa: BLE001 — every failure here is a typed refusal + raise MetricViewError( + f"could not read {view.full_name} from Unity Catalog: {type(exc).__name__}: " + f"{scrub(exc)}", + ("Check DATABRICKS_HOST and the credential, then re-run fluid apply.",), + ) from exc + if live.state == "foreign": + conflicts.append(f"{view.full_name} ({view.where}): {live.why}") + if conflicts: + raise MetricViewError( + "the metric view name is taken, so nothing was applied: " + "; ".join(conflicts) + ".", + ( + "Rename the table, or drop or rename the existing object in Databricks. A view " + "of this product that another identity owns can be handed back with ALTER VIEW " + " OWNER TO .", + ), + ) + for view, revokes in narrowing: + if revokes: + say(f"\n {view.full_name}: metric view readers, before the apply") + for statement, what in revokes: + try: + session.runner.run(statement) + except StatementFailed as exc: + raise MetricViewError( + f"{view.full_name}: {what} failed before the apply, so nothing was applied: " + f"{exc}", + ("Fix the cause (this identity must own the view), then re-run fluid apply.",), + ) from exc + say(f" done: {what}") + _drop_stale(session, plan, say) + + +def post_apply(plan: MetricViews, say: Say, client_factory: Optional[ClientFactory] = None) -> None: + """Create or update every view and make its grants the contract's; raise + :class:`MetricViewError` naming what failed.""" + if not plan.views: + return + session = _session(client_factory, say, required=True) + assert session is not None + failed = [problem for view in plan.views for problem in _converge(session, view, say)] + if failed: + raise MetricViewError( + "the apply succeeded, but these metric views are not as the contract says: " + + " | ".join(failed), + ( + "Fix the cause (this identity needs CREATE TABLE and USE SCHEMA on the schema, " + "SELECT on the table and CAN USE on the warehouse), then re-run fluid apply.", + ), + ) + + +def _converge(session: Session, view: MetricView, say: Say) -> List[str]: + """Make one view the contract's; what failed. Access narrows first, each revoke on its own, + then the definition, then each grant: no failure leaves a removed reader able to read.""" + try: + live = _classify(session, view) + if live.state == "foreign": + return [f"{view.full_name}: {live.why}"] + revokes, grants = _grant_steps(session, view, live.info if live.state == "ours" else None) + definition = _definition_steps(view, live) + except Exception as exc: # noqa: BLE001 + return [f"{view.full_name}: it could not be read ({type(exc).__name__}: {scrub(exc)})"] + if revokes or definition or grants: + say(f"\n {view.full_name}: metric view") + failed = [f for step in revokes if (f := _run_step(session, view, step, say))] + for step in definition: + problem = _run_step(session, view, step, say) + if problem: + return [*failed, problem] # no grant on a view that is not what the contract says + return failed + [f for step in grants if (f := _run_step(session, view, step, say))] + + +def _run_step(session: Session, view: MetricView, step: Step, say: Say) -> Optional[str]: + statement, what = step + try: + session.runner.run(statement) + except StatementFailed as exc: + say(f" FAILED: {what}: {exc}") + return f"{view.full_name}: {what} failed: {exc}" + say(f" done: {what}") + return None + + +def _schema_views(session: Session, where: str, say: Say) -> List[Mapping[str, Any]]: + """The metric views of one schema named as this module names one (``*_semantic_model``, + ``GET /api/2.1/unity-catalog/table-summaries``); none, with a note, when unlisted.""" + catalog, schema = where.split(".", 1) + query = { + "catalog_name": catalog, + "schema_name_pattern": schema, + "table_name_pattern": f"%{DEFAULT_SUFFIX}", + } + try: + rows = _list_named( + session.client, "/api/2.1/unity-catalog/table-summaries", query, "tables" + ) + except Exception as exc: # noqa: BLE001 — unknown contents: drop nothing + if not _enforce.is_not_found(exc): # a schema not created yet holds no view + say( + f" note: metric views in {where} were not checked for removal: they could not " + f"be listed ({type(exc).__name__}: {scrub(exc)})." + ) + return [] + # The pattern is a SQL LIKE, so `_` matches any character: keep this schema's rows only. + return [ + row + for row in rows + if str(row.get("table_type") or "").upper() == METRIC_VIEW + and str(row.get("full_name") or "").lower().rsplit(".", 1)[0] == where.lower() + ] + + +def _stale_views(session: Session, plan: MetricViews, say: Say) -> Iterator[Mapping[str, Any]]: + """This product's metric views in the schemas its tables are in that the contract no longer + declares. Only those schemas: another environment of the product (an overlay that changes the + schema, the same catalog, the same deploy identity) keeps its views.""" + expected = {view.full_name.lower() for view in plan.views} + for where in plan.schemas: + for row in _schema_views(session, where, say): + full = str(row.get("full_name") or "") + if not full or full.lower() in expected: + continue + try: + info = read_table(session.client, full) or {} + except Exception as exc: # noqa: BLE001 — unread: leave it + say( + f" note: metric view {full} was not checked for removal ({type(exc).__name__})" + ) + continue + ours = (info.get("properties") or {}).get(PRODUCT_PROPERTY) == plan.product_id + if ours and str(info.get("table_type") or "").upper() == METRIC_VIEW: + yield {**info, "full_name": full} + + +def _drop_stale(session: Session, plan: MetricViews, say: Say, *, dry_run: bool = False) -> None: + """Drop this product's metric views in its schemas that the contract no longer declares.""" + if not plan.product_id: + return + for info in _stale_views(session, plan, say): + full, owner = str(info["full_name"]), str(info.get("owner") or "") + try: + mine = owner_matches(owner, {session.identity}) + except Exception: # noqa: BLE001 — unknown identity: leave it + mine = False + if not mine: + say( + f" left metric view {full}: owned by {owner or 'unknown'}, not the deploy identity" + ) + elif dry_run: + say(f" before apply: drop metric view {full}: the contract no longer declares it") + else: + try: + statement = f"DROP VIEW IF EXISTS {quote_table(full)}" + except ValueError: + say(f" left metric view {full}: its name is no plain identifier; drop it by hand") + continue + _run_noted( + session, + statement, + say, + done=f" dropped metric view {full}: the contract no longer declares it", + failed=f" note: metric view {full} was not dropped", + ) + + +def _run_noted(session: Session, statement: str, say: Say, *, done: str, failed: str) -> None: + try: + session.runner.run(statement) + except StatementFailed as exc: + say(f"{failed}: {exc}") + else: + say(done) + + +def lock_restricted( + plan: MetricViews, say: Say, client_factory: Optional[ClientFactory] = None +) -> None: + """After column masks or a row filter failed to bind: revoke ``SELECT`` (and ``ALL + PRIVILEGES``) on each existing metric view over a restricted table from every grantee but + its owner and this run's identity. Best effort; the run fails with the binding's error.""" + views = [view for view in plan.views if view.restricted] + if not views: + return + try: + session = _session(client_factory, say, required=True) + except UnsupportedBindingError as exc: + say(f" fail-safe: the metric views over restricted tables were not closed: {exc}") + return + assert session is not None + for view in views: + _close(session, view, say) + + +def _close(session: Session, view: MetricView, say: Say) -> None: + """Revoke every read on one view but its owner's and this run's.""" + try: + info = read_table(session.client, view.full_name) + if info is None or str(info.get("table_type") or "").upper() != METRIC_VIEW: + return + exempt = _exempt(session, info) + held = _permissions(session, view.full_name) + except Exception as exc: # noqa: BLE001 + say( + f" fail-safe: NOT CLOSED: the grants on metric view {view.full_name} could not " + f"be read ({type(exc).__name__}: {scrub(exc)}); revoke SELECT on it by hand." + ) + return + name = quote_table(view.full_name) + for principal, privileges in sorted(held.items()): + if principal.lower() in exempt: + continue + statements = [] + try: + if "SELECT" in privileges: + statements.append( + f"REVOKE SELECT ON TABLE {name} FROM {quote_principal(principal)}" + ) + if "ALL_PRIVILEGES" in privileges: + statements.append(revoke_all_statement(view.full_name, principal)) + except ValueError: + say( + f" fail-safe: NOT CLOSED for {principal!r} on {view.full_name}: revoke it by hand" + ) + continue + for statement in statements: + _run_noted( + session, + statement, + say, + done=f" fail-safe: revoked: {statement}", + failed=f" fail-safe: NOT REVOKED: {statement} failed", + ) + + +__all__ = [ + "DEFAULT_SUFFIX", + "DIGEST_PROPERTY", + "EXPOSE_PROPERTY", + "METRIC_VIEW", + "MetricView", + "MetricViewError", + "MetricViews", + "PRODUCT_PROPERTY", + "compile_metric_views", + "lock_restricted", + "post_apply", + "pre_apply", + "preview", +] diff --git a/fluid_build/output_ports/mcp/query_compiler.py b/fluid_build/output_ports/mcp/query_compiler.py index ab8a9d71..618c857b 100644 --- a/fluid_build/output_ports/mcp/query_compiler.py +++ b/fluid_build/output_ports/mcp/query_compiler.py @@ -56,6 +56,7 @@ from dataclasses import dataclass, field from typing import Any, Dict, FrozenSet, Iterable, List, Mapping, Optional, Set, Tuple +from fluid_build.forge_datamodel import semantics_builder as _semantics_builder from fluid_build.output_ports.mcp.auth import MTLS_IDENTITY_ATTRS from fluid_build.providers._sql_safety import ( validate_ident, @@ -740,24 +741,14 @@ def _declared_grain(dimension: Mapping[str, Any]) -> Optional[str]: return normalize_time_grain(type_params.get("timeGranularity")) -_AGG_FUNCTIONS = { - "sum": "SUM", - "avg": "AVG", - "count": "COUNT", - "count_distinct": "COUNT", # rendered with DISTINCT below - "min": "MIN", - "max": "MAX", - "median": "MEDIAN", - "percentile": "PERCENTILE_CONT", -} - - -# The contract-schema default when a percentile measure carries no -# ``aggParams.percentile``. MUST stay identical to the dbt MetricFlow -# bridge's default (``engines/dbt/semantic_models.py``) so the governed -# query path and the exported dbt project answer the same number for the +# ``measures[].agg`` → SQL aggregate, and the default percentile, shared with the +# Databricks metric view (``forge_datamodel/semantics_builder.py``) so every path that +# renders a measure as SQL aggregates it the same way. The default MUST stay identical +# to the dbt MetricFlow bridge's (``engines/dbt/semantic_models.py``, pinned by tests) so +# the governed query path and the exported dbt project answer the same number for the # same contract. 0.5 == median. -DEFAULT_PERCENTILE = 0.5 +_AGG_FUNCTIONS = _semantics_builder.AGGREGATE_FUNCTIONS +DEFAULT_PERCENTILE = _semantics_builder.DEFAULT_PERCENTILE def _resolve_percentile_params(measure_name: str, measure: Mapping[str, Any]) -> Tuple[float, bool]: @@ -767,18 +758,13 @@ def _resolve_percentile_params(measure_name: str, measure: Mapping[str, Any]) -> dbt-semantic-interfaces' ``agg_params``. The percentile value is validated to a real number in [0, 1] — it is interpolated into SQL as a literal (aggregate arguments must be constants), so validation - here is the safety boundary. + here is the safety boundary + (:func:`~fluid_build.forge_datamodel.semantics_builder.percentile_params`). """ - params = measure.get("aggParams") or {} - if not isinstance(params, Mapping): - raise QueryValidationError(f"Measure {measure_name!r} has non-object aggParams") - raw = params.get("percentile", DEFAULT_PERCENTILE) - if isinstance(raw, bool) or not isinstance(raw, (int, float)) or not 0 <= float(raw) <= 1: - raise QueryValidationError( - f"Measure {measure_name!r} has invalid aggParams.percentile " - f"{raw!r}; expected a number in [0, 1]" - ) - return float(raw), bool(params.get("useDiscretePercentile")) + try: + return _semantics_builder.percentile_params(measure_name, measure) + except ValueError as exc: + raise QueryValidationError(str(exc)) from None def _render_measure_expression( @@ -818,10 +804,7 @@ def _render_measure_expression( if not isinstance(expr_raw, str): raise QueryValidationError(f"Measure {measure_name!r} has non-string expr") validate_sql_expression_allowlist(expr_raw) - sql_func = _AGG_FUNCTIONS[agg_raw] projection_alias = validate_ident(alias or measure_name) - if agg_raw == "count_distinct": - return f"COUNT(DISTINCT {expr_raw}) AS {projection_alias}" if agg_raw == "percentile": if dialect in ("bigquery", "athena"): raise QueryValidationError( @@ -830,10 +813,9 @@ def _render_measure_expression( "Use 'median'-free approximations engine-side or query a " "different measure." ) - percentile, use_discrete = _resolve_percentile_params(measure_name, measure) - fn = "PERCENTILE_DISC" if use_discrete else "PERCENTILE_CONT" - return f"{fn}({percentile:g}) WITHIN GROUP (ORDER BY {expr_raw}) " f"AS {projection_alias}" - return f"{sql_func}({expr_raw}) AS {projection_alias}" + # Validated here so the error is a QueryValidationError, as it always was. + _resolve_percentile_params(measure_name, measure) + return f"{_semantics_builder.aggregate_call(measure_name, measure, expr_raw)} AS {projection_alias}" def compile_semantic_query( diff --git a/tests/forge_datamodel/test_semantics_builder.py b/tests/forge_datamodel/test_semantics_builder.py index 82c4b228..31c76db2 100644 --- a/tests/forge_datamodel/test_semantics_builder.py +++ b/tests/forge_datamodel/test_semantics_builder.py @@ -463,3 +463,47 @@ def test_distinct_names_produce_no_collision_warning(): ) assert errors == [] assert warnings == [] + + +# ── one aggregate rendering for every SQL consumer of a measure ───────────── + + +def test_the_mcp_query_compiler_and_the_shared_helper_render_the_same_aggregate(): + from fluid_build.forge_datamodel import semantics_builder as sb + from fluid_build.output_ports.mcp import query_compiler as qc + + assert qc.DEFAULT_PERCENTILE == sb.DEFAULT_PERCENTILE == 0.5 + assert qc._AGG_FUNCTIONS is sb.AGGREGATE_FUNCTIONS + measure = { + "agg": "percentile", + "aggParams": {"percentile": 0.25, "useDiscretePercentile": True}, + } + assert sb.aggregate_call("m", measure, "x") == "PERCENTILE_DISC(0.25) WITHIN GROUP (ORDER BY x)" + assert qc._render_measure_expression("m", {**measure, "expr": "x"}) == ( + "PERCENTILE_DISC(0.25) WITHIN GROUP (ORDER BY x) AS m" + ) + assert sb.aggregate_call("m", {"agg": "count_distinct"}, "id") == "COUNT(DISTINCT id)" + assert sb.aggregate_call("m", {"agg": "avg"}, "x") == "AVG(x)" + assert sb.aggregate_call("m", {"agg": "percentile"}, "x") == ( + "PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY x)" + ) + + +@pytest.mark.parametrize("bad", [True, 2, -0.1, "0.5", float("nan")]) +def test_a_bad_percentile_is_refused_by_both(bad): + from fluid_build.forge_datamodel import semantics_builder as sb + from fluid_build.output_ports.mcp import query_compiler as qc + + measure = {"agg": "percentile", "aggParams": {"percentile": bad}, "expr": "x"} + with pytest.raises(ValueError, match="invalid aggParams.percentile"): + sb.aggregate_call("m", measure, "x") + with pytest.raises(qc.QueryValidationError, match="invalid aggParams.percentile"): + qc._render_measure_expression("m", measure) + + +def test_an_agg_outside_the_schema_is_refused(): + from fluid_build.forge_datamodel import semantics_builder as sb + + with pytest.raises(ValueError, match="unsupported agg 'mode'"): + sb.aggregate_call("m", {"agg": "mode"}, "x") + assert sb.parens_balanced("(a) and (b)") and not sb.parens_balanced("a) or (b") diff --git a/tests/iac/_fake_databricks.py b/tests/iac/_fake_databricks.py index 64973154..cf97cc37 100644 --- a/tests/iac/_fake_databricks.py +++ b/tests/iac/_fake_databricks.py @@ -440,6 +440,146 @@ def _coldef_param(txt: str) -> Dict[str, str]: return {"name": m.group(1), "type": m.group(2).strip().lower()} +#: What a metric view keeps outside its stored ``view_definition`` (measured on Databricks SQL +#: 2026.45: the YAML read back has no ``comment``, ``display_name`` or ``synonyms``). +_METRIC_VIEW_METADATA = ("comment", "display_name", "synonyms") +_PROPERTY_RE = re.compile(r"'((?:[^'\\]|\\.)*)'\s*=\s*'((?:[^'\\]|\\.)*)'") + + +def _metric_view_properties(text: str) -> Dict[str, str]: + def unescape(v: str) -> str: + return re.sub(r"\\(.)", r"\1", v) + + return {unescape(k): unescape(v) for k, v in _PROPERTY_RE.findall(text or "")} + + +def _metric_view_doc(body: str) -> Tuple[Optional[Dict[str, Any]], str]: + """The parsed definition, or ``(None, error)`` as Databricks words a bad one.""" + import yaml + + try: + doc = yaml.safe_load(body) + except yaml.YAMLError as exc: + return None, f"[METRIC_VIEW_INVALID_VIEW_DEFINITION] invalid YAML: {exc}" + if not isinstance(doc, dict) or doc.get("version") != 1.1: + return None, "[METRIC_VIEW_INVALID_VIEW_DEFINITION] version must be 1.1" + if not doc.get("dimensions") and not doc.get("measures"): + return None, "[METRIC_VIEW_INVALID_VIEW_DEFINITION] no dimensions or measures" + return doc, "" + + +def _metric_view_record(fake: "FakeDatabricks", doc: Dict[str, Any]) -> Dict[str, Any]: + import yaml + + stored = {k: v for k, v in doc.items() if k != "comment"} + columns = [] + for kind in ("dimensions", "measures"): + items = [] + for item in doc.get(kind) or []: + items.append({k: v for k, v in item.items() if k not in _METRIC_VIEW_METADATA}) + columns.append( + { + "name": item["name"], + "type": "string" if kind == "dimensions" else "double", + "nullable": True, + "comment": item.get("comment"), + } + ) + if kind in stored: + stored[kind] = items + return { + "table_type": "METRIC_VIEW", + "columns": columns, + "partitions": [], + "cluster_keys": [], + "comment": doc.get("comment"), + "view_definition": yaml.safe_dump(stored, sort_keys=False), + "created_at": int(time.time() * 1000), + } + + +def _metric_view_statement(fake: "FakeDatabricks", s: str) -> Optional[Tuple[bool, str]]: + """CREATE / ALTER / DROP of a metric view, and GRANT on a table: ``None`` for any other.""" + m = re.match( + rf"CREATE\s+(OR\s+REPLACE\s+)?VIEW\s+{_FULL}\s+WITH\s+METRICS\s+LANGUAGE\s+YAML\s+" + r"(?:TBLPROPERTIES\s*\((.*?)\)\s+)?AS\s+\$\$(.*)\$\$$", + s, + re.S | re.I, + ) + if m: + replace, cat, sch, name, props, body = m.groups() + full = f"{cat}.{sch}.{name}" + if f"{cat}.{sch}" not in fake.schemas: + return False, f"[SCHEMA_NOT_FOUND] {cat}.{sch}" + old = fake.tables.get(full) + if old is not None and old.get("table_type") != "METRIC_VIEW": + return False, f"[EXPECT_VIEW_NOT_TABLE.NO_ALTERNATIVE] {full}" + if old is not None and not replace: + return False, f"[TABLE_OR_VIEW_ALREADY_EXISTS] {full}" + doc, err = _metric_view_doc(body) + if doc is None: + return False, err + source = str(doc.get("source") or "") + if fake.tables.get(source, {}).get("table_type") not in ("MANAGED", "EXTERNAL"): + return False, f"[TABLE_OR_VIEW_NOT_FOUND] {source}" + record = _metric_view_record(fake, doc) + record["properties"] = _metric_view_properties(props or "") + record["owner"] = _USER # whoever creates (or replaces) a view owns it + fake.tables[full] = record + # Replacing a view drops its grants (measured live). + fake.grants.pop(("table", full), None) + return True, "" + + m = re.match(rf"ALTER\s+VIEW\s+{_FULL}\s+AS\s+\$\$(.*)\$\$$", s, re.S | re.I) + if m: + cat, sch, name, body = m.groups() + full = f"{cat}.{sch}.{name}" + old = fake.tables.get(full) + if old is None or old.get("table_type") != "METRIC_VIEW": + return False, f"[TABLE_OR_VIEW_NOT_FOUND] {full}" + doc, err = _metric_view_doc(body) + if doc is None: + return False, err + record = _metric_view_record(fake, doc) + record["properties"] = old["properties"] # ALTER VIEW ... AS keeps them, and the grants + record["owner"] = old.get("owner", _USER) + fake.tables[full] = record + return True, "" + + m = re.match(rf"ALTER\s+VIEW\s+{_FULL}\s+SET\s+TBLPROPERTIES\s*\((.*)\)$", s, re.S | re.I) + if m: + cat, sch, name, props = m.groups() + full = f"{cat}.{sch}.{name}" + if full not in fake.tables: + return False, f"[TABLE_OR_VIEW_NOT_FOUND] {full}" + fake.tables[full]["properties"].update(_metric_view_properties(props)) + return True, "" + + m = re.match(rf"DROP\s+VIEW\s+(IF\s+EXISTS\s+)?{_FULL}\s*$", s, re.I) + if m: + ie, cat, sch, name = m.groups() + full = f"{cat}.{sch}.{name}" + if full not in fake.tables: + return (True, "") if ie else (False, f"[TABLE_OR_VIEW_NOT_FOUND] {full}") + if fake.tables[full].get("table_type") != "METRIC_VIEW": + return False, f"[WRONG_COMMAND_FOR_OBJECT_TYPE] {full} is a table" + del fake.tables[full] + fake.grants.pop(("table", full), None) + return True, "" + + m = re.match(rf"GRANT\s+(\w+)\s+ON\s+TABLE\s+{_FULL}\s+TO\s+`((?:[^`]|``)+)`\s*$", s, re.I) + if m: + priv, cat, sch, name, who = m.groups() + full = f"{cat}.{sch}.{name}" + if full not in fake.tables: + return False, f"[TABLE_OR_VIEW_NOT_FOUND] {full}" + held = fake.grants.setdefault(("table", full), {}) + who = who.replace("``", "`") + held[who] = sorted(set(held.get(who, [])) | {priv.upper()}) + return True, "" + return None + + def _execute(fake: "FakeDatabricks", sql: str) -> Tuple[bool, str]: """Run one statement against ``fake``; return (ok, error message).""" s = sql.strip().rstrip(";").strip() @@ -564,6 +704,10 @@ def _execute(fake: "FakeDatabricks", sql: str) -> Tuple[bool, str]: t["comment"] = _unescape(text) return True, "" + handled = _metric_view_statement(fake, s) + if handled is not None: + return handled + m = re.match(rf"ALTER\s+TABLE\s+{_FULL}\s+(.*)$", s, re.S | re.I) if not m: return False, f"fake_databricks: unsupported statement: {s[:200]}" @@ -1324,6 +1468,7 @@ def _register(self) -> None: a("GET", r"/api/2\.1/unity-catalog/functions/([^/]+)", self._get_function) a("GET", r"/api/2\.1/unity-catalog/functions", self._list_functions) a("GET", r"/api/2\.1/unity-catalog/tables", self._list_tables) + a("GET", r"/api/2\.1/unity-catalog/table-summaries", self._table_summaries) a("GET", r"/api/2\.1/unity-catalog/tables/([^/]+)", self._get_table) a("DELETE", r"/api/2\.1/unity-catalog/tables/([^/]+)", self._delete_table) a("GET", r"/api/2\.1/unity-catalog/permissions/([a-z_]+)/([^/]+)", self._get_perms) @@ -1654,7 +1799,14 @@ def _table_row(self, full: str) -> Dict[str, Any]: if view.get("comment"): row["comment"] = view["comment"] else: - row["table_type"] = "MANAGED" + # A table, or a metric view made by ``CREATE VIEW ... WITH METRICS``, whose record + # keeps its type, its stored YAML and its comment. + table = self.tables[full] + row["table_type"] = table.get("table_type") or "MANAGED" + if row["table_type"] == "METRIC_VIEW": + for key in ("view_definition", "comment"): + if table.get(key): + row[key] = table[key] return row def add_metric_view(self, full_name: str, source: str, comment: str = "") -> None: @@ -1894,6 +2046,30 @@ def add_genie_space( self.genie_acl[space_id] = {} return space_id + def _table_summaries(self, m: Any, q: Any, b: Any) -> Tuple[int, Any]: + """ListTableSummaries: the tables of a catalog; the name patterns are SQL LIKEs.""" + cat = (q.get("catalog_name") or [""])[0] + + def like(key: str) -> "re.Pattern[str]": + text = (q.get(key) or ["%"])[0] + body = "".join(".*" if c == "%" else "." if c == "_" else re.escape(c) for c in text) + return re.compile(f"^{body}$") + + schemas, tables = like("schema_name_pattern"), like("table_name_pattern") + names = sorted( + n + for n in self.tables + if n.split(".")[0] == cat + and schemas.match(n.split(".")[1]) + and tables.match(n.split(".")[2]) + ) + return 200, self._page( + names, + q, + "tables", + lambda n: {"full_name": n, "table_type": self.tables[n].get("table_type")}, + ) + def _page( self, names: List[str], q: Any, key: str, row: Callable[[str], Any] ) -> Dict[str, Any]: @@ -1958,6 +2134,7 @@ def _table_doc(self, full: str) -> Dict[str, Any]: "comment": t.get("comment"), "storage_location": f"s3://fake-root/{full}", "created_at": t["created_at"], + **({"view_definition": t["view_definition"]} if t.get("view_definition") else {}), } def _delete_table(self, m: Any, q: Any, b: Any) -> Tuple[int, Any]: diff --git a/tests/iac/test_iac_databricks_genie.py b/tests/iac/test_iac_databricks_genie.py index 35392738..95f8fe11 100644 --- a/tests/iac/test_iac_databricks_genie.py +++ b/tests/iac/test_iac_databricks_genie.py @@ -129,6 +129,15 @@ def site_contract(**block: Any) -> Dict[str, Any]: "name": "site_360_semantic_model", "defaultAggTimeDimension": "snapshot_date", "entities": [{"name": "site_id", "type": "primary", "expr": "site_id"}], + # A non-additive measure is ordered by a time dimension of its own model + # (MetricFlow's rule, and the metric view's window). + "dimensions": [ + { + "name": "snapshot_date", + "type": "time", + "typeParams": {"timeGranularity": "day"}, + } + ], "measures": [ { "name": "site_count", @@ -1114,7 +1123,11 @@ def missing(): with pytest.raises(genie.GenieError) as exc: PLUGIN.pre_apply(site_contract(), say=print, client_factory=missing) assert "databricks-sdk is not installed" in str(exc.value) - PLUGIN.pre_apply(without_block(site_contract()), say=print, client_factory=missing) + # Without the block, and with no metric view to make (its semantics would declare one), + # nothing needs the client before tofu apply. + plain = without_block(site_contract()) + del plain["exposes"][0]["semantics"] + PLUGIN.pre_apply(plain, say=print, client_factory=missing) class Plugin: def pre_apply(self, contract, say=None, dry_run=False): diff --git a/tests/iac/test_iac_databricks_metrics.py b/tests/iac/test_iac_databricks_metrics.py new file mode 100644 index 00000000..1f43a2cd --- /dev/null +++ b/tests/iac/test_iac_databricks_metrics.py @@ -0,0 +1,932 @@ +# Copyright 2025 Agentics Transformation Limited +# SPDX-License-Identifier: Apache-2.0 + +"""Unity Catalog metric views from ``exposes[].semantics`` (``iac/providers/databricks_metrics.py``). + +Pinned offline: the YAML a site-360-shaped semantics block compiles to (an entity, categorical +and time dimensions, non-additive measures, filtered simple metrics, a ratio), the +``source.`` qualification the live warehouse needs, the time grains, and every refusal. +The hooks run against the in-process stand-in (``_fake_databricks.py``): a first apply creates +the view and grants it to the table's readers, a second runs nothing, a changed definition is an +``ALTER VIEW`` that keeps the grants, ``SELECT`` is exactly the readers, a view the contract no +longer declares is dropped, a view this module did not create is refused before the apply, and +the views over a restricted table are closed when its masks or row filter cannot be bound. + +Not proven here: what real Databricks accepts and answers (the Free Edition run recorded in the +PR, HONESTLY_TESTED.md). +""" + +from __future__ import annotations + +import copy +from dataclasses import replace +from typing import Any, Callable, Dict, Iterator, List + +import pytest +import yaml + +from fluid_build.iac import IAC_PLUGINS +from fluid_build.iac.base import UnsupportedBindingError +from fluid_build.iac.providers import databricks as dbx +from fluid_build.iac.providers import databricks_metrics as mv + +from . import _databricks_governed as governed +from ._databricks_cli import make_contract +from ._fake_databricks import DEPLOY_USER, FakeDatabricks, _execute + +pytestmark = [pytest.mark.unit, pytest.mark.databricks] + +PLUGIN = IAC_PLUGINS["databricks"] +ENV = {"DATABRICKS_WAREHOUSE_ID": "wh-1"} +TABLE = "main.sales.site_360" +VIEW = "main.sales.site_360_semantic_model" +WINDOW = {"order": "snapshot_date", "range": "current", "semiadditive": "last"} + +COLUMNS = [ + ("site_id", "string", True), + ("country_code", "string", False), + ("site_status", "string", False), + ("colocation_ready_flag", "boolean", False), + ("tenancy_count", "int", False), + ("availability_90d_pct", "double", False), + ("snapshot_date", "date", False), + ("captured_at", "timestamp", False), +] + +SEMANTICS: Dict[str, Any] = { + "name": "site_360_semantic_model", + "description": "Business semantics of Site 360.", + "defaultAggTimeDimension": "snapshot_date", + "entities": [ + {"name": "site_id", "type": "primary", "expr": "site_id", "description": "Primary key."} + ], + "dimensions": [ + { + "name": "country_code", + "type": "categorical", + "expr": "country_code", + "description": "ISO market", + }, + {"name": "site_status", "type": "categorical", "expr": "site_status"}, + {"name": "colocation_ready_flag", "type": "categorical", "expr": "colocation_ready_flag"}, + { + "name": "snapshot_date", + "type": "time", + "expr": "snapshot_date", + "typeParams": {"timeGranularity": "day"}, + }, + ], + "measures": [ + { + "name": "site_count", + "description": "Count of distinct sites.", + "agg": "count_distinct", + "expr": "site_id", + "createMetric": False, + "aggTimeDimension": "snapshot_date", + "nonAdditiveDimension": {"name": "snapshot_date", "windowChoice": "max"}, + }, + { + "name": "tenancy_count", + "agg": "avg", + "expr": "tenancy_count", + "nonAdditiveDimension": {"name": "snapshot_date", "windowChoice": "max"}, + }, + ], + "metrics": [ + { + "name": "avg_tenancy_ratio", + "type": "simple", + "measure": "tenancy_count", + "filter": "site_status = 'ON_AIR'", + "description": "Average tenancies per on-air site.", + "owner": "tm-data-platform", + }, + { + "name": "colocation_ready_site_count", + "type": "simple", + "measure": "site_count", + "filter": "colocation_ready_flag = true AND site_status = 'ON_AIR'", + }, + { + "name": "ready_share", + "type": "ratio", + "numerator": "colocation_ready_site_count", + "denominator": "site_count", + }, + ], + "tags": ["semantic-model"], + "labels": { + "ai-synonyms": "Site 360; Tower Passport", + "ai-instructions": "One row per site.", + "ai-example-questions": "Which sites are ready?", + }, +} + +EXPECTED: Dict[str, Any] = { + "version": 1.1, + "comment": ( + "Business semantics of Site 360.\n\nSynonyms: Site 360; Tower Passport.\n\n" + "Instructions: One row per site." + ), + "source": TABLE, + "dimensions": [ + { + "name": "site_id", + "expr": "source.site_id", + "comment": "Primary key.", + "display_name": "Site ID", + "synonyms": ["site code", "tower id"], + }, + { + "name": "country_code", + "expr": "source.country_code", + "comment": "ISO market", + "display_name": "Country Code", + "synonyms": ["market"], + }, + {"name": "site_status", "expr": "source.site_status"}, + {"name": "colocation_ready_flag", "expr": "source.colocation_ready_flag"}, + {"name": "snapshot_date", "expr": "source.snapshot_date"}, + ], + "measures": [ + { + "name": "site_count", + "expr": "COUNT(DISTINCT source.site_id)", + "comment": "Count of distinct sites.", + "window": [WINDOW], + }, + {"name": "tenancy_count", "expr": "AVG(source.tenancy_count)", "window": [WINDOW]}, + { + "name": "avg_tenancy_ratio", + "expr": "AVG(source.tenancy_count) FILTER (WHERE source.site_status = 'ON_AIR')", + "comment": "Average tenancies per on-air site.", + "window": [WINDOW], + }, + { + "name": "colocation_ready_site_count", + "expr": ( + "COUNT(DISTINCT source.site_id) FILTER (WHERE source.colocation_ready_flag = true " + "AND source.site_status = 'ON_AIR')" + ), + "window": [WINDOW], + }, + { + "name": "ready_share", + "expr": "MEASURE(colocation_ready_site_count) / NULLIF(MEASURE(site_count), 0)", + }, + ], +} + + +def site_contract(semantics: Any = None, **kw: Any) -> Dict[str, Any]: + contract = make_contract(columns=COLUMNS, readers=kw.pop("readers", ("group:analysts",)), **kw) + expose = contract["exposes"][0] + expose["binding"]["location"]["table"] = "site_360" + for column in expose["contract"]["schema"]: + if column["name"] == "site_id": + column.update( + description="Tower site the row consolidates", + businessName="Site ID", + labels={"synonyms": "site code; tower id; Site Code"}, + ) + elif column["name"] == "country_code": + column.update(businessName="Country Code", labels={"synonyms": "market"}) + expose["semantics"] = copy.deepcopy(SEMANTICS) if semantics is None else semantics + return contract + + +def compile_views(contract: Dict[str, Any]) -> dbx._Compiled: + return dbx._compile(contract, ENV) + + +def refusals_of(contract: Dict[str, Any]) -> List[UnsupportedBindingError]: + return [r for r in compile_views(contract).refusals if r.kind.startswith("databricks-metric")] + + +def only_view(contract: Dict[str, Any]) -> mv.MetricView: + compiled = compile_views(contract) + assert compiled.refusals == [] + (view,) = compiled.metric_views.views + return view + + +# ── compile ───────────────────────────────────────────────────────────── + + +def test_site_360_semantics_compile_to_the_metric_view_yaml(): + view = only_view(site_contract()) + assert view.full_name == VIEW + assert view.source == TABLE + assert view.readers == ("analysts",) + assert view.restricted is False + assert yaml.safe_load(view.definition) == EXPECTED + # Plain YAML for a reader: no anchors for the windows the metrics share with their measure. + assert "&id" not in view.definition and "*id" not in view.definition + + +def test_statements_create_with_properties_and_alter_keeps_the_view(): + view = only_view(site_contract()) + name = "`main`.`sales`.`site_360_semantic_model`" + assert view.create.startswith( + f"CREATE VIEW {name} WITH METRICS LANGUAGE YAML TBLPROPERTIES ('fluid.product_id' = " + f"'dbx.fake.orders', 'fluid.expose_id' = 'orders', 'fluid.definition_sha256' = " + f"'{view.digest}') AS $$version: 1.1\n" + ) + assert view.create.endswith("\n$$") + assert "OR REPLACE" not in view.create + alter, mark = view.update + assert alter == f"ALTER VIEW {name} AS $${view.definition}$$" + assert mark.startswith(f"ALTER VIEW {name} SET TBLPROPERTIES ('fluid.product_id' = ") + + +def test_the_view_is_named_after_its_table_whatever_the_model_is_called(): + # Forge's own emitters name the model after its table; the schema calls the name + # human-readable. Neither may become the view's name. + for name in ("site_360", "Site 360 model", None, 42): + semantics = copy.deepcopy(SEMANTICS) + semantics["name"] = name + assert only_view(site_contract(semantics)).full_name == VIEW + + +def test_a_table_named_like_another_tables_view_is_refused(): + contract = site_contract() + clash = copy.deepcopy(contract["exposes"][0]) + clash["exposeId"] = "clash" + clash["binding"]["location"]["table"] = "site_360_semantic_model" + del clash["semantics"] + contract["exposes"].append(clash) + assert any("already takes" in str(r) for r in refusals_of(contract)) + + +def test_an_entity_is_its_key_column_and_one_on_no_column_is_left_out(): + semantics = copy.deepcopy(SEMANTICS) + semantics["entities"] = [ + {"name": "site", "type": "primary", "expr": "site_id"}, + {"name": "tower", "type": "foreign"}, # no column "tower": a join key the view cannot hold + ] + semantics["dimensions"].insert( + 0, {"name": "site", "type": "categorical", "expr": "site_status"} + ) + compiled = compile_views(site_contract(semantics)) + assert compiled.refusals == [] + fields = [ + d["name"] for d in yaml.safe_load(compiled.metric_views.views[0].definition)["dimensions"] + ] + assert fields[:2] == ["site_id", "site"] + assert "tower" not in fields + assert any("entities[1] (tower) is the key 'tower'" in w for w in compiled.warnings) + + +def test_a_table_qualifying_its_own_column_is_the_source(): + semantics = copy.deepcopy(SEMANTICS) + semantics["measures"][1]["expr"] = "site_360.tenancy_count" + measures = yaml.safe_load(only_view(site_contract(semantics)).definition)["measures"] + assert measures[1]["expr"] == "AVG(source.tenancy_count)" + + +@pytest.mark.parametrize( + ("expr", "expected"), + [ + ("EXTRACT(YEAR FROM snapshot_date)", "EXTRACT(YEAR FROM source.snapshot_date)"), + ("snapshot_date + INTERVAL 1 DAY", "source.snapshot_date + INTERVAL 1 DAY"), + ("captured_at::date", "source.captured_at::date"), + ("DATEADD(MONTH, 1, snapshot_date)", "DATEADD(MONTH, 1, source.snapshot_date)"), + ("CAST(year AS STRING)", "CAST(source.year AS STRING)"), + ], +) +def test_a_unit_or_type_keyword_named_like_a_column_stays_a_keyword(expr, expected): + contract = site_contract() + for name in ("year", "day", "date", "month"): + contract["exposes"][0]["contract"]["schema"].append({"name": name, "type": "int"}) + semantics = contract["exposes"][0]["semantics"] + semantics["dimensions"].append({"name": "derived", "type": "categorical", "expr": expr}) + semantics["dimensions"].append( + {"name": "year_of", "type": "categorical", "expr": "EXTRACT(YEAR FROM snapshot_date)"} + ) + fields = { + d["name"]: d["expr"] for d in yaml.safe_load(only_view(contract).definition)["dimensions"] + } + assert fields["derived"] == expected + + +def test_a_dimension_named_like_a_keyword_it_uses_compiles(): + semantics = copy.deepcopy(SEMANTICS) + semantics["dimensions"].append( + {"name": "year", "type": "categorical", "expr": "EXTRACT(YEAR FROM snapshot_date)"} + ) + fields = { + d["name"]: d["expr"] + for d in yaml.safe_load(only_view(site_contract(semantics)).definition)["dimensions"] + } + assert fields["year"] == "EXTRACT(YEAR FROM source.snapshot_date)" + + +def test_two_fields_differing_only_in_a_literals_case_collide(): + semantics = copy.deepcopy(SEMANTICS) + semantics["dimensions"] += [ + {"name": "flag", "type": "categorical", "expr": "site_status = 'A'"}, + {"name": "flag", "type": "categorical", "expr": "site_status = 'a'"}, + ] + assert any("share one namespace" in str(r) for r in refusals_of(site_contract(semantics))) + + +def test_the_plan_and_validate_see_the_view(monkeypatch): + monkeypatch.setenv(dbx.WAREHOUSE_ENV, "wh-1") + actions = dbx.planned_objects(site_contract()) + (action,) = [a for a in actions if a["resource_type"] == "databricks_metric_view"] + assert action["resource_id"] == VIEW + assert action["source"] == TABLE + assert action["readers"] == ["analysts"] + semantics = copy.deepcopy(SEMANTICS) + semantics["metrics"].append({"name": "net", "type": "derived", "expr": "a - b"}) + errors, _warnings = dbx.validate_databricks_binding(site_contract(semantics), ENV) + assert any("derived metric" in e for e in errors) + + +def test_qualify_writes_columns_as_source_and_leaves_the_rest(): + columns = {"a": {}, "b": {}, "d": {}} + text, bare = mv._qualify("a = 'a' AND b.c > 1 AND upper(a) <> `d` OR e IS NULL", columns) + assert ( + text == "source.a = 'a' AND source.b.c > 1 AND upper(source.a) <> source.`d` OR e IS NULL" + ) + assert bare == ["AND", "AND", "OR", "e", "IS", "NULL"] + assert mv._qualify("COUNT(*)", columns) == ("COUNT(*)", []) + assert mv._qualify("x1e5 + 1e5", {"e5": {}})[0] == "x1e5 + 1e5" + typed = "CAST(a AS date) > DATE'2026-01-01' AND date IS NOT NULL" + assert mv._qualify(typed, {"a": {}, "date": {}})[0] == ( + "CAST(source.a AS date) > DATE'2026-01-01' AND source.date IS NOT NULL" + ) + + +@pytest.mark.parametrize( + ("column_type", "grain", "expected"), + [ + ("date", "day", "source.snapshot_date"), + ("date", "month", "TRUNC(source.snapshot_date, 'MONTH')"), + ("date", "weekly", "TRUNC(source.snapshot_date, 'WEEK')"), + ("timestamp", "day", "DATE_TRUNC('DAY', source.snapshot_date)"), + ("timestamp", "hour", "DATE_TRUNC('HOUR', source.snapshot_date)"), + ("date", None, "source.snapshot_date"), + ], +) +def test_time_dimension_is_at_its_grain(column_type, grain, expected): + contract = site_contract() + for column in contract["exposes"][0]["contract"]["schema"]: + if column["name"] == "snapshot_date": + column["type"] = column_type + dimension = contract["exposes"][0]["semantics"]["dimensions"][3] + if grain is None: + del dimension["typeParams"] + else: + dimension["typeParams"]["timeGranularity"] = grain + fields = yaml.safe_load(only_view(contract).definition)["dimensions"] + assert fields[-1] == {"name": "snapshot_date", "expr": expected} + + +def test_percentile_and_median_measures_take_a_filter(): + semantics = copy.deepcopy(SEMANTICS) + semantics["measures"].append( + { + "name": "p90_availability", + "agg": "percentile", + "expr": "availability_90d_pct", + "aggParams": {"percentile": 0.9}, + } + ) + semantics["measures"].append( + {"name": "median_availability", "agg": "median", "expr": "availability_90d_pct"} + ) + semantics["metrics"].append( + { + "name": "p90_on_air", + "type": "simple", + "measure": "p90_availability", + "filter": "site_status = 'ON_AIR'", + } + ) + measures = { + m["name"]: m + for m in yaml.safe_load(only_view(site_contract(semantics)).definition)["measures"] + } + assert measures["p90_availability"]["expr"] == ( + "PERCENTILE_CONT(0.9) WITHIN GROUP (ORDER BY source.availability_90d_pct)" + ) + assert measures["median_availability"]["expr"] == "MEDIAN(source.availability_90d_pct)" + assert measures["p90_on_air"]["expr"] == ( + "PERCENTILE_CONT(0.9) WITHIN GROUP (ORDER BY source.availability_90d_pct) " + "FILTER (WHERE source.site_status = 'ON_AIR')" + ) + + +def test_a_metric_that_is_its_measure_is_one_measure_and_ratios_follow_their_inputs(): + semantics = copy.deepcopy(SEMANTICS) + semantics["metrics"] = [ + {"name": "share_of_share", "type": "ratio", "inputMetrics": ["ready_share", "site_count"]}, + {"name": "site_count", "type": "simple", "measure": "site_count"}, + { + "name": "ready_share", + "type": "ratio", + "numerator": "tenancy_count", + "denominator": "site_count", + }, + ] + names = [ + m["name"] + for m in yaml.safe_load(only_view(site_contract(semantics)).definition)["measures"] + ] + assert names == ["site_count", "tenancy_count", "ready_share", "share_of_share"] + + +def test_comment_falls_back_to_the_expose_description_and_may_be_absent(): + semantics = copy.deepcopy(SEMANTICS) + del semantics["description"], semantics["labels"] + contract = site_contract(semantics) + contract["exposes"][0]["description"] = "One row per site." + assert yaml.safe_load(only_view(contract).definition)["comment"] == "One row per site." + del contract["exposes"][0]["description"] + assert "comment" not in yaml.safe_load(only_view(contract).definition) + + +def test_readers_are_who_may_select_the_table_and_writers_are_not(): + contract = site_contract(readers=("group:analysts",)) + contract["exposes"][0]["policy"]["authz"]["writers"] = ["group:loaders"] + contract["accessPolicy"] = { + "grants": [ + {"principal": "group:admins", "permissions": ["admin"]}, + {"principal": "group:bi", "permissions": ["read"]}, + {"principal": "group:etl", "permissions": ["write"]}, + ] + } + assert only_view(contract).readers == ("admins", "analysts", "bi") + + +def test_semantics_with_nothing_to_compile_warns_and_makes_no_view(): + compiled = compile_views(site_contract({"name": "empty", "description": "Nothing yet."})) + assert compiled.refusals == [] and compiled.metric_views.views == [] + assert any("no metric view is created" in w for w in compiled.warnings) + + +def _with(path: str, value: Any) -> Callable[[Dict[str, Any]], None]: + """Set (or, with ``...``, delete) one key of the semantics block: ``measures.0.agg``.""" + + def change(semantics: Dict[str, Any]) -> None: + *parents, last = path.split(".") + node: Any = semantics + for part in parents: + node = node[int(part)] if part.isdigit() else node[part] + key: Any = int(last) if last.isdigit() else last + if value is ...: + del node[key] + else: + node[key] = value + + return change + + +def _append(kind: str, entry: Dict[str, Any]) -> Callable[[Dict[str, Any]], None]: + return lambda semantics: semantics[kind].append(entry) + + +REFUSALS = [ + (_append("metrics", {"name": "net", "type": "derived", "expr": "a - b"}), "derived metric"), + (_with("metrics.2.filter", "site_status = 'X'"), "ratio metric with a filter"), + (_with("metrics.2.denominator", "nope"), "'nope' is no measure or metric"), + (_with("metrics.0.measure", "missing"), "over measure 'missing'"), + (_with("metrics.0.type", "cumulative"), "has type 'cumulative'"), + (_with("measures.0.nonAdditiveDimension.name", "site_status"), "not a time dimension"), + (_with("measures.0.nonAdditiveDimension.windowChoice", "last"), "windowChoice is 'last'"), + (_with("dimensions.3.typeParams.timeGranularity", "hour"), "which a date does not have"), + (_with("dimensions.3.typeParams.timeGranularity", "fortnight"), "no grain"), + (_with("measures.1.name", "Country_Code"), "share one namespace"), + (_with("measures.1.name", "tenancy count"), "not a plain identifier"), + (_with("dimensions.0.name", "source"), "the name a metric view expression gives"), + (_with("measures.0.expr", "site_id; DROP TABLE x"), "not an expression"), + (_with("measures.0.expr", "(site_id"), "do not balance"), + (_with("measures.0.expr", "SUM(site_id)"), "calls SUM(...)"), + (_with("dimensions.0.expr", "MAX(country_code)"), "calls MAX(...)"), + (_with("measures.0.expr", "MEASURE(site_count)"), "calls MEASURE()"), + (_with("metrics.0.filter", "tenancy_count > (SELECT 1)"), "not an expression"), + (_with("measures.1.expr", "avg_tenancy_ratio + 1"), "not a column of"), + (_with("measures.1.expr", "length(secret('scope', 'key'))"), "calls secret()"), + (_with("dimensions.0.expr", "ai_query('endpoint', country_code)"), "calls ai_query()"), + ( + _with("metrics.0.filter", "is_account_group_member('admins')"), + "calls is_account_group_member()", + ), + (_with("measures.1.expr", "length(`secret`('s', 'k'))"), "a quoted or qualified function"), + (_with("measures.1.expr", "length(system.builtin.secret('s', 'k'))"), "quoted or qualified"), + (_with("dimensions.0.expr", "builtin . ai_query('ep', country_code)"), "quoted or qualified"), + (_with("dimensions.0.expr", "main.tools.lookup(country_code)"), "quoted or qualified"), + (_with("metrics.0.filter", "IDENTIFIER('secret')('a', 'b') = 'x'"), "calls IDENTIFIER()"), + (_with("metrics.0.filter", "upper(site_status) = 'AI_QUERY('"), None), # a literal: fine + # A quote inside a backtick-quoted name opens no literal: the call between them is seen. + ( + _with( + "metrics.0.filter", + 'exists(array(1), `a\'` -> true) AND length(secret("s", "k")) > 0 ' + "AND exists(array(1), `'b` -> true)", + ), + "calls secret()", + ), + ( + _with( + "metrics.0.filter", + "exists(array(1), `a'` -> true)) + 0 * (exists(array(1), `'b` -> true)", + ), + "parentheses do not balance", + ), + (_with("metrics.0.filter", "site_status = 'ON_AIR"), "does not close"), + (_with("metrics.0.filter", "`site_status = 'x'"), "does not close"), + (_with("metrics.0.filter", "`we(ird` = 'x' OR site_status = 'y'"), None), + (_append("measures", {"name": "ghost", "agg": "sum"}), "'ghost', which is no column of"), + (_with("dimensions.0.expr", "region_code"), "'region_code', which is no column of"), + (_with("measures.1.expr", "length(\"secret\"('s', 'k'))"), "quoted or qualified"), + (_with("measures.1.expr", "length(`x`secret('s', 'k'))"), "calls secret()"), + (_with("measures.0.agg", "stddev"), "unsupported agg 'stddev'"), + (_with("measures.0.agg", "percentile"), None), # default 0.5: compiles + ( + _append( + "measures", + { + "name": "p", + "agg": "percentile", + "aggParams": {"percentile": 2}, + "expr": "tenancy_count", + }, + ), + "invalid aggParams.percentile", + ), + (_with("description", "Price in $$"), "holds $$"), + (_with("labels", {"ai-instructions": "Use ${env:SECRET}"}), "holds ${"), +] + + +@pytest.mark.parametrize(("change", "fragment"), REFUSALS) +def test_what_the_yaml_cannot_hold_is_refused(change, fragment): + semantics = copy.deepcopy(SEMANTICS) + change(semantics) + found = refusals_of(site_contract(semantics)) + if fragment is None: + assert found == [] + return + assert found, f"expected a refusal mentioning {fragment!r}" + assert any(fragment in str(r) for r in found), [str(r) for r in found] + assert all(r.remediation for r in found) + + +def test_column_text_limits_are_refused_not_truncated(): + contract = site_contract() + site_id = contract["exposes"][0]["contract"]["schema"][0] + site_id["labels"] = {"synonyms": "; ".join(f"s{i}" for i in range(11))} + assert any("at most 10" in str(r) for r in refusals_of(contract)) + site_id["labels"] = {} + site_id["businessName"] = "x" * 256 + assert any("at most 255" in str(r) for r in refusals_of(contract)) + + +def test_a_property_value_spark_would_substitute_is_refused(): + contract = site_contract() + contract["id"] = "dbx.${env:X}" + assert any("records it as a property" in str(r) for r in refusals_of(contract)) + + +# ── the hooks, against the stand-in ───────────────────────────────────── + + +@pytest.fixture +def fake(monkeypatch) -> Iterator[FakeDatabricks]: + server = FakeDatabricks().start() + monkeypatch.setenv(dbx.WAREHOUSE_ENV, server.warehouse_id) + monkeypatch.setenv("DATABRICKS_TOKEN", "dapi-secret-token-value") # pragma: allowlist secret + server.create_table(TABLE, COLUMNS) + try: + yield server + finally: + server.stop() + + +class Run: + """The plugin's hooks against ``fake``, with what they print kept.""" + + def __init__(self, fake: FakeDatabricks): + self.fake = fake + self.said: List[str] = [] + + def pre(self, doc: Dict[str, Any], **kw: Any) -> None: + PLUGIN.pre_apply(doc, say=self.said.append, client_factory=self.fake.client, **kw) + + def post(self, doc: Dict[str, Any], **kw: Any) -> None: + PLUGIN.post_apply(doc, say=self.said.append, client_factory=self.fake.client, **kw) + + def statements(self, since: int = 0) -> List[str]: + return [ + s for s in self.fake.sql_log[since:] if "VIEW" in s or "GRANT" in s or "REVOKE" in s + ] + + @property + def text(self) -> str: + return "\n".join(self.said) + + +def test_first_apply_creates_and_grants_and_a_second_runs_nothing(fake): + run, doc = Run(fake), site_contract() + run.pre(doc) + assert run.statements() == [] # nothing before the apply + run.post(doc) + created = fake.tables[VIEW] + assert created["table_type"] == "METRIC_VIEW" + assert created["properties"]["fluid.product_id"] == "dbx.fake.orders" + assert created["properties"]["fluid.expose_id"] == "orders" + assert created["properties"]["fluid.definition_sha256"] == only_view(doc).digest + assert created["comment"].startswith("Business semantics of Site 360.") + assert fake.grants[("table", VIEW)] == {"analysts": ["SELECT"]} + assert [s.split(" WITH ")[0].split(" TO ")[0] for s in run.statements()] == [ + "CREATE VIEW `main`.`sales`.`site_360_semantic_model`", + "GRANT SELECT ON TABLE `main`.`sales`.`site_360_semantic_model`", + ] + assert "done: create metric view main.sales.site_360_semantic_model" in run.text + before = len(fake.sql_log) + run.pre(doc) + run.post(doc) + assert run.statements(before) == [] + + +def test_a_changed_definition_is_altered_and_keeps_its_grants(fake): + run, doc = Run(fake), site_contract() + run.post(doc) + fake.grant("table", VIEW, "analysts", "MANAGE") + doc["exposes"][0]["semantics"]["measures"][0]["description"] = "Distinct sites." + before = len(fake.sql_log) + run.post(doc) + assert [s[:12] for s in run.statements(before)] == ["ALTER VIEW `", "ALTER VIEW `"] + assert fake.grants[("table", VIEW)] == {"analysts": ["MANAGE", "SELECT"]} + assert fake.tables[VIEW]["properties"]["fluid.definition_sha256"] == only_view(doc).digest + # Changed by hand in Databricks: the stored structure no longer matches, so it is restored. + fake.tables[VIEW]["view_definition"] = "version: 1.1\nsource: main.sales.site_360\n" + before = len(fake.sql_log) + run.post(doc) + assert [s[:12] for s in run.statements(before)] == ["ALTER VIEW `", "ALTER VIEW `"] + + +def test_select_on_the_view_is_exactly_the_tables_readers(fake): + run, doc = Run(fake), site_contract(readers=("group:analysts", "group:bi")) + run.post(doc) + fake.grant("table", VIEW, "intruder", "SELECT") + fake.grant("table", VIEW, DEPLOY_USER, "SELECT") + before = len(fake.sql_log) + run.post(site_contract(readers=("group:bi",))) + assert sorted(run.statements(before)) == [ + "REVOKE SELECT ON TABLE `main`.`sales`.`site_360_semantic_model` FROM `analysts`", + "REVOKE SELECT ON TABLE `main`.`sales`.`site_360_semantic_model` FROM `intruder`", + ] + assert fake.grants[("table", VIEW)]["bi"] == ["SELECT"] + assert fake.grants[("table", VIEW)][DEPLOY_USER] == ["SELECT"] # this run's own: left + + +def _view_statement(full: str, product: str) -> str: + return ( + f"CREATE VIEW {full} WITH METRICS LANGUAGE YAML TBLPROPERTIES ('fluid.product_id' = " + f"'{product}') AS $$version: 1.1\nsource: {TABLE}\nmeasures:\n- name: n\n expr: COUNT(1)\n$$" + ) + + +def test_a_view_the_contract_no_longer_declares_is_dropped_before_the_apply(fake): + run, doc = Run(fake), site_contract() + run.post(doc) + fake.create_table("main.archive.kept", [("x", "string", True)]) + ours_moved = "main.archive.site_360_old_semantic_model" # its table moved away + ours_renamed = "main.sales.site_360_v0_semantic_model" # its table was renamed + theirs = "main.sales.someone_elses_semantic_model" + for full, product in ( + (ours_moved, "dbx.fake.orders"), + (ours_renamed, "dbx.fake.orders"), + (theirs, "other.product"), + ): + ok, err = _execute(fake, _view_statement(full, product)) + assert ok, err + before = len(fake.sql_log) + run.pre(doc) + dropped = [s for s in run.statements(before) if s.startswith("DROP VIEW")] + assert dropped == ["DROP VIEW IF EXISTS `main`.`sales`.`site_360_v0_semantic_model`"] + # Only the product's own schemas are looked in: one it no longer uses is not (a known limit). + assert ours_moved in fake.tables + assert theirs in fake.tables and VIEW in fake.tables + del doc["exposes"][0]["semantics"] + run.pre(doc) + assert VIEW not in fake.tables + assert f"dropped metric view {VIEW}" in run.text + + +def test_a_stale_view_whose_name_sql_cannot_hold_is_left_and_said(fake): + run, doc = Run(fake), site_contract() + run.post(doc) + ok, err = _execute(fake, _view_statement("main.sales.old_semantic_model", "dbx.fake.orders")) + assert ok, err + fake.tables["main.sales.odd-name_semantic_model"] = fake.tables.pop( + "main.sales.old_semantic_model" + ) + run.pre(site_contract()) + assert "main.sales.odd-name_semantic_model" in fake.tables + assert "its name is no plain identifier" in run.text + + +def test_dry_run_prints_the_statements_and_runs_none(fake): + run, doc = Run(fake), site_contract() + run.pre(doc, dry_run=True) + assert run.statements() == [] + assert "after apply: create the metric view over main.sales.site_360" in run.text + assert "CREATE VIEW `main`.`sales`.`site_360_semantic_model` WITH METRICS" in run.text + assert "after apply: grant SELECT to analysts" in run.text + assert "dry-run: none of these statements ran." in run.text + + +@pytest.mark.parametrize( + ("seed", "why"), + [ + ( + lambda fake: fake.create_table(VIEW, [("x", "string", True)]), + "it is a managed table, not a metric view", + ), + ( + lambda fake: _execute( + fake, + f"CREATE VIEW {VIEW} WITH METRICS LANGUAGE YAML AS $$version: 1.1\nsource: " + f"{TABLE}\nmeasures:\n- name: n\n expr: COUNT(1)\n$$", + ), + "did not create (no fluid.product_id)", + ), + ( + lambda fake: _execute( + fake, + f"CREATE VIEW {VIEW} WITH METRICS LANGUAGE YAML TBLPROPERTIES ('fluid.product_id' " + f"= 'other.product') AS $$version: 1.1\nsource: {TABLE}\nmeasures:\n- name: n\n " + "expr: COUNT(1)\n$$", + ), + "the metric view of product 'other.product'", + ), + ], +) +def test_a_name_taken_by_an_object_this_module_did_not_create_is_refused_before_the_apply( + fake, seed, why +): + seed(fake) + before = len(fake.sql_log) + with pytest.raises(UnsupportedBindingError, match="the metric view name is taken") as caught: + Run(fake).pre(site_contract()) + assert why in str(caught.value) + assert caught.value.kind == "databricks-metric-view-failed" + assert [s for s in fake.sql_log[before:] if "VIEW" in s] == [] + + +def test_a_view_another_identity_owns_is_refused(fake): + Run(fake).post(site_contract()) + fake.tables[VIEW]["owner"] = "someone@example.test" + with pytest.raises(UnsupportedBindingError, match="owned by someone@example.test"): + Run(fake).pre(site_contract()) + + +def test_a_failed_statement_is_a_typed_error_naming_the_view(fake): + fake.fail_on("CREATE VIEW", "[PERMISSION_DENIED] no CREATE TABLE on schema") + run = Run(fake) + with pytest.raises(UnsupportedBindingError, match="create metric view .* failed") as caught: + run.post(site_contract()) + assert caught.value.kind == "databricks-metric-view-failed" + assert "PERMISSION_DENIED" in str(caught.value) + assert ("table", VIEW) not in fake.grants + + +def test_no_sdk_is_refused_only_when_a_view_is_declared(fake): + def no_sdk() -> Any: + raise ImportError("no databricks-sdk") + + with pytest.raises(UnsupportedBindingError, match="databricks-sdk is not installed"): + PLUGIN.pre_apply(site_contract(), say=lambda _l: None, client_factory=no_sdk) + plain = site_contract() + del plain["exposes"][0]["semantics"] + PLUGIN.post_apply(plain, say=lambda _l: None, client_factory=no_sdk) # nothing to do: silent + + +def test_views_over_a_restricted_table_are_closed_when_its_binding_fails(fake): + profile = "main.cp360.profile" + view = "main.cp360.profile_semantic_model" + fake.create_table(profile, [(n, t, True) for n, t in governed.COLUMNS]) + doc = governed.contract() + doc["exposes"][0]["semantics"] = { + "dimensions": [{"name": "region", "type": "categorical"}], + "measures": [{"name": "profiles", "agg": "count", "expr": "1"}], + } + run = Run(fake) + run.pre(doc) + run.post(doc) + assert set(fake.grants[("table", view)]) == {governed.ANALYSTS_GROUP, governed.STEWARDS_GROUP} + # A rebuild dropped the row filter, and binding it again fails: the table is locked, and + # so is its metric view (it reads the table with its owner's rights). + fake.tables[profile].pop("row_filter", None) + fake.fail_on("SET ROW FILTER", "[INTERNAL_ERROR] injected") + with pytest.raises(UnsupportedBindingError): + run.post(doc) + assert all("SELECT" not in privs for privs in fake.grants[("table", view)].values()) + assert ( + "fail-safe: revoked: REVOKE SELECT ON TABLE `main`.`cp360`.`profile_semantic_model`" + in run.text + ) + + +def test_lock_restricted_leaves_the_views_of_unrestricted_tables(fake): + run, doc = Run(fake), site_contract() + run.post(doc) + plan = dbx._compile(doc, {"DATABRICKS_WAREHOUSE_ID": fake.warehouse_id}).metric_views + mv.lock_restricted(plan, run.said.append, fake.client) + assert fake.grants[("table", VIEW)] == {"analysts": ["SELECT"]} + restricted = mv.MetricViews(plan.product_id, [replace(plan.views[0], restricted=True)]) + mv.lock_restricted(restricted, run.said.append, fake.client) + assert fake.grants[("table", VIEW)] == {"analysts": []} + + +def test_a_failed_grant_or_update_still_revokes_who_the_contract_no_longer_names(fake): + run = Run(fake) + run.post(site_contract(readers=("group:analysts", "group:bi"))) + fake.fail_on("GRANT SELECT ON TABLE `main`.`sales`.`site_360_semantic_model` TO `ghost`") + with pytest.raises(UnsupportedBindingError, match="grant SELECT to ghost failed"): + run.post(site_contract(readers=("group:bi", "group:ghost"))) + assert "SELECT" not in fake.grants[("table", VIEW)].get("analysts", []) + changed = site_contract(readers=("group:ghost",)) + changed["exposes"][0]["semantics"]["description"] = "Changed." + fake.fail_on("ALTER VIEW") + with pytest.raises(UnsupportedBindingError, match="update metric view"): + run.post(changed) + assert "SELECT" not in fake.grants[("table", VIEW)].get("bi", []) + + +def test_pre_apply_narrows_who_reads_before_the_apply(fake): + run = Run(fake) + run.post(site_contract(readers=("group:analysts", "group:bi"))) + before = len(fake.sql_log) + run.pre(site_contract(readers=("group:bi",))) + assert run.statements(before) == [ + "REVOKE SELECT ON TABLE `main`.`sales`.`site_360_semantic_model` FROM `analysts`" + ] + assert "metric view readers, before the apply" in run.text + + +def test_all_privileges_held_by_a_non_reader_is_revoked(fake): + run, doc = Run(fake), site_contract() + run.post(doc) + fake.grant("table", VIEW, "intruder", "ALL_PRIVILEGES") + before = len(fake.sql_log) + run.post(doc) + assert run.statements(before) == [ + "REVOKE ALL PRIVILEGES ON TABLE `main`.`sales`.`site_360_semantic_model` FROM `intruder`" + ] + assert fake.grants[("table", VIEW)]["intruder"] == [] + + +def test_the_shadow_report_does_not_count_the_view_as_a_missing_opentofu_resource(monkeypatch): + from fluid_build.iac.shadow import native_logical_resources + + monkeypatch.setenv(dbx.WAREHOUSE_ENV, "wh-1") + kinds = {r.kind for r in native_logical_resources(dbx.planned_objects(site_contract()))} + assert "view" not in kinds + + +def test_the_engine_prints_what_the_hooks_say_as_data(monkeypatch, capsys): + from fluid_build.cli import _apply_opentofu_engine as engine + + said: List[str] = [] + + class Plugin: + def pre_apply(self, contract, say, dry_run=False): + say("comment: Share of rows [/total] and [bold]") + + monkeypatch.setattr(engine, "cprint_literal", said.append) + engine._enforce_access(Plugin(), "pre_apply", {}, dry_run=True) + assert said == ["comment: Share of rows [/total] and [bold]"] + + +def test_another_environment_of_the_product_keeps_its_view(fake): + """An overlay that changes only the schema (prod main.sales, staging main.sales_stg), in one + catalog under one deploy identity: neither environment's apply drops the other's view.""" + prod, staging = site_contract(), site_contract() + staging["exposes"][0]["binding"]["location"]["schema"] = "sales_stg" + fake.create_table("main.sales_stg.site_360", COLUMNS) + run = Run(fake) + for doc in (prod, staging, prod, staging): + run.pre(doc) + run.post(doc) + assert VIEW in fake.tables and "main.sales_stg.site_360_semantic_model" in fake.tables + assert not [s for s in fake.sql_log if s.startswith("DROP VIEW")] + + +def test_a_dates_own_column_is_qualified_in_a_two_argument_date_add(): + semantics = copy.deepcopy(SEMANTICS) + semantics["dimensions"] += [ + {"name": "next_day", "type": "categorical", "expr": "date_add(snapshot_date, 1)"}, + {"name": "age", "type": "categorical", "expr": "datediff(captured_at, snapshot_date)"}, + ] + fields = { + d["name"]: d["expr"] + for d in yaml.safe_load(only_view(site_contract(semantics)).definition)["dimensions"] + } + assert fields["next_day"] == "date_add(source.snapshot_date, 1)" + assert fields["age"] == "datediff(source.captured_at, source.snapshot_date)" diff --git a/tests/iac/test_iac_databricks_metrics_fake_e2e.py b/tests/iac/test_iac_databricks_metrics_fake_e2e.py new file mode 100644 index 00000000..427897a9 --- /dev/null +++ b/tests/iac/test_iac_databricks_metrics_fake_e2e.py @@ -0,0 +1,163 @@ +# Copyright 2025 Agentics Transformation Limited +# SPDX-License-Identifier: Apache-2.0 + +"""Unity Catalog metric views through the real ``fluid`` CLI, ``tofu`` and provider, against a stand-in. + +Pinned (Stage 1: real ``tofu`` and ``databricks/databricks`` 1.140.0, the stand-in workspace of +``_fake_databricks.py`` and its stand-in for the SDK client): ``fluid apply`` creates the table +with OpenTofu and then the metric view of its ``semantics`` on the warehouse, with ``SELECT`` for +the table's readers; a second apply plans nothing and runs no metric view statement; +``--mode dry-run`` prints the ``CREATE VIEW`` and runs none; a contract that drops the semantics +drops the view; ``fluid validate`` refuses what metric view YAML cannot express; the product's +Genie space (``extensions.databricksGenie``, converged after the metric views) lists the view the +same apply made among its tables. + +Not proven here: what real Databricks accepts and answers (the Free Edition run in the PR, +HONESTLY_TESTED.md). +""" + +from __future__ import annotations + +import copy +import json +from pathlib import Path +from typing import Any, Dict, Iterator, List + +import pytest + +from fluid_build.iac import runner +from fluid_build.iac.providers import databricks_enforce as enforce + +from ._databricks_cli import ( # noqa: F401 — plugin_cache is a fixture + apply_contract, + isolate_workspace, + make_contract, + plugin_cache, + run_fluid, + write_contract, +) +from ._fake_databricks import FakeDatabricks + +pytestmark = [ + pytest.mark.integration, + pytest.mark.provider, + pytest.mark.databricks, + pytest.mark.skipif(runner.tofu_path() is None, reason="tofu binary not installed"), +] + +VIEW = "main.sales.orders_semantic_model" +SEMANTICS: Dict[str, Any] = { + "name": "orders_metrics", + "description": "Orders, as the shop counts them.", + "entities": [{"name": "order_id", "type": "primary", "expr": "id"}], + "dimensions": [{"name": "region", "type": "categorical", "expr": "region"}], + "measures": [ + {"name": "revenue", "agg": "sum", "expr": "amount"}, + {"name": "orders", "agg": "count_distinct", "expr": "id"}, + ], + "metrics": [ + {"name": "eu_revenue", "type": "simple", "measure": "revenue", "filter": "region = 'eu'"}, + { + "name": "revenue_per_order", + "type": "ratio", + "numerator": "revenue", + "denominator": "orders", + }, + ], +} + + +@pytest.fixture +def fake() -> Iterator[FakeDatabricks]: + server = FakeDatabricks().start() + try: + yield server + finally: + server.stop() + + +@pytest.fixture +def workspace(monkeypatch, tmp_path: Path, fake: FakeDatabricks, plugin_cache: str) -> Path: + isolate_workspace( + monkeypatch, + tmp_path, + host=fake.endpoint, + warehouse_id=fake.warehouse_id, + plugin_cache=plugin_cache, + ) + # The stand-in's client in place of databricks-sdk's WorkspaceClient, so the hooks that run + # on the warehouse around tofu apply reach the same workspace without the [databricks] extra. + monkeypatch.setattr(enforce, "default_client", fake.client) + return tmp_path + + +def _contract(semantics: Any = SEMANTICS) -> Dict[str, Any]: + contract = make_contract() + if semantics is not None: + contract["exposes"][0]["semantics"] = copy.deepcopy(semantics) + return contract + + +def _view_statements(fake: FakeDatabricks, since: int = 0) -> List[str]: + return [s for s in fake.sql_log[since:] if " VIEW " in f" {s} " or s.startswith("GRANT")] + + +def test_apply_creates_the_metric_view_and_a_second_apply_changes_nothing(workspace, fake, capfd): + path = write_contract(workspace, _contract()) + rc, out = apply_contract(capfd, path) + assert rc == 0, out + view = fake.tables[VIEW] + assert view["table_type"] == "METRIC_VIEW" + assert view["properties"]["fluid.product_id"] == "dbx.fake.orders" + assert view["comment"] == "Orders, as the shop counts them." + assert fake.grants[("table", VIEW)] == {"analysts": ["SELECT"], "bi": ["SELECT"]} + assert "done: create metric view main.sales.orders_semantic_model" in out + assert "MEASURE(revenue) / NULLIF(MEASURE(orders), 0)" in view["view_definition"] + + before = len(fake.sql_log) + rc, out = apply_contract(capfd, path) + assert rc == 0, out + assert "tofu plan: +0 ~0 -0" in out + assert _view_statements(fake, before) == [] + + +def test_dry_run_prints_the_metric_view_and_runs_nothing(workspace, fake, capfd): + rc, out = apply_contract(capfd, write_contract(workspace, _contract()), "--mode", "dry-run") + assert rc == 0, out + assert "Unity Catalog metric views (run on the SQL warehouse after tofu apply)" in out + assert "CREATE VIEW `main`.`sales`.`orders_semantic_model` WITH METRICS LANGUAGE YAML" in out + assert "after apply: grant SELECT to analysts" in out + assert "dry-run: none of these statements ran." in out + assert VIEW not in fake.tables and _view_statements(fake) == [] + + +def test_dropping_the_semantics_drops_the_view(workspace, fake, capfd): + assert apply_contract(capfd, write_contract(workspace, _contract()))[0] == 0 + assert VIEW in fake.tables + rc, out = apply_contract(capfd, write_contract(workspace, _contract(None))) + assert rc == 0, out + assert VIEW not in fake.tables + assert "dropped metric view main.sales.orders_semantic_model" in out + + +def test_validate_refuses_what_the_yaml_cannot_express(workspace, fake, capfd): + semantics = copy.deepcopy(SEMANTICS) + semantics["metrics"].append({"name": "net", "type": "derived", "expr": "revenue - 1"}) + rc, out = run_fluid(capfd, "validate", write_contract(workspace, _contract(semantics))) + assert rc != 0, out + assert "derived metric" in out + assert fake.sql_log == [] + + +def test_the_products_genie_space_holds_the_metric_view_the_apply_made(workspace, fake, capfd): + # Genie reads the contract's semantic model through the metric view: the space is converged + # after the metric views, so the first apply already lists the view beside its table. + semantics = {**SEMANTICS, "labels": {"ai-example-questions": "What is revenue per order?"}} + contract = _contract(semantics) + contract["extensions"] = {"databricksGenie": {}} + rc, out = apply_contract(capfd, write_contract(workspace, contract)) + assert rc == 0, out + (space,) = fake.genie_spaces.values() + tables = json.loads(space["serialized_space"])["data_sources"]["tables"] + assert [t["identifier"] for t in tables] == ["main.sales.orders", VIEW] + assert "1 table(s), 1 metric view(s)" in out