feat(databricks): Unity Catalog metric views from exposes[].semantics

.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
 

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 `<catalog>.<schema>.<table>_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`

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)

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

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,
+  ``<catalog>.<schema>.<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]] = {}