test: pin warehouse definition binding as a shared fixture

pkg-py/src/commons/_definitions/__init__.py

@@ -12,7 +12,13 @@
 authority, not this code.
 """
 
-from ._compile import attach_compiled_definitions, mixed_grain
+from ._compile import (
+    DefinitionAbsentColumnError,
+    DefinitionMixedGrainError,
+    DefinitionUnmatchedTableError,
+    attach_compiled_definitions,
+    mixed_grain,
+)
 from ._registry import (
     ExportRecord,
     Registry,
@@ -26,6 +32,9 @@
 )
 
 __all__ = [
+    "DefinitionAbsentColumnError",
+    "DefinitionMixedGrainError",
+    "DefinitionUnmatchedTableError",
     "ExportRecord",
     "Registry",
     "applied_text",

pkg-py/src/commons/_definitions/_compile.py

@@ -18,7 +18,13 @@
 from ._export import DefinitionExport, Ir
 from ._registry import ExportRecord
 
-__all__ = ["attach_compiled_definitions", "mixed_grain"]
+__all__ = [
+    "DefinitionAbsentColumnError",
+    "DefinitionMixedGrainError",
+    "DefinitionUnmatchedTableError",
+    "attach_compiled_definitions",
+    "mixed_grain",
+]
 
 _TARGETS = {
     "duckdb": "SQL(duckdb)",
@@ -27,6 +33,18 @@
 }
 
 
+class DefinitionUnmatchedTableError(ValueError):
+    """An authored table declares definitions and matched no relation."""
+
+
+class DefinitionAbsentColumnError(ValueError):
+    """A definition names an authored column the matched relation lacks."""
+
+
+class DefinitionMixedGrainError(ValueError):
+    """A metric's dependency chain mixes row and aggregate grain."""
+
+
 def mixed_grain(definitions: dict[str, DefinitionExport]) -> dict[str, bool]:
     """Which definitions mix row and aggregate grain, directly or inherited.
 
@@ -126,59 +144,85 @@ def attach_compiled_definitions(
 
     Nothing is assigned until every table compiles, so a refusal leaves the
     dictionary untouched and the caller can attach it to another source.
+
+    The refusals are checked in the order the definition-binding contract
+    pins, so a dictionary with several faults reports the same one in either
+    package: metric grain on every table, then the dialect, then each table
+    in authored order, its match before its columns.
     """
-    exports: dict[str, dict[str, DefinitionExport]] = (
-        getattr(dictionary, "definition_exports", None) or {}
-    )
-    if bindings is not None:
-        _check_definitions_matched(exports, bindings)
-    compiled: dict[str, list[ExportRecord]] = {}
-    for table_name, entry in dictionary.tables.items():
-        # A catalog import re-keys the dictionary to the warehouse's labels
-        # and records what the author called each table, which is how its
-        # exports are still found afterwards.
-        authored = getattr(entry, "authored_name", None) or table_name
-        definitions = exports.get(authored) or {}
-        if not definitions:
-            compiled[table_name] = []
-            continue
-        if table_name not in exposed:
-            raise ValueError(
-                f"The data dictionary declares definitions on table "
-                f"{table_name!r}, which the data source does not expose. "
-                f"Exposed tables: {', '.join(sorted(exposed))}."
-            )
-        target = _TARGETS.get(dialect)
-        if target is None:
-            raise ValueError(
-                f"Definitions on table {table_name!r} cannot be compiled for a "
-                f"{dialect!r} data source. commons lowers definitions to "
-                f"{', '.join(sorted(_TARGETS))}."
-            )
+    exports: dict[str, dict[str, DefinitionExport]] = {
+        authored: definitions
+        for authored, definitions in (
+            getattr(dictionary, "definition_exports", None) or {}
+        ).items()
+        if definitions
+    }
+    for authored, definitions in exports.items():
+        _check_metric_grain(authored, definitions)
+    target = _TARGETS.get(dialect)
+    if exports and target is None:
+        raise ValueError(
+            f"Definitions on table {next(iter(exports))!r} cannot be compiled "
+            f"for a {dialect!r} data source. commons lowers definitions to "
+            f"{', '.join(sorted(_TARGETS))}."
+        )
+    compiled: dict[str, list[ExportRecord]] = {
+        table_name: [] for table_name in dictionary.tables
+    }
+    for authored, definitions in exports.items():
+        # A catalog import re-keys the dictionary to the warehouse's labels,
+        # so the bindings are how an authored table's exports find its entry.
+        table_name = _definition_table(authored, exposed, bindings)
         if bindings is not None:
             definitions = _bind_table(
                 definitions, bindings["columns"].get(authored) or {}, table_name
             )
+        assert target is not None
         compiled[table_name] = _compile_table(table_name, definitions, target)
     for table_name, entry in dictionary.tables.items():
         entry.compiled_definitions = compiled[table_name]
 
 
-def _check_definitions_matched(
-    exports: dict[str, dict[str, DefinitionExport]], bindings: dict[str, Any]
-) -> None:
-    """Refuse a definition whose table the catalog selection left out.
+def _definition_table(
+    authored: str, exposed: set[str], bindings: dict[str, Any] | None
+) -> str:
+    """The table an authored table's definitions compile against.
 
-    Compiling it is impossible and dropping it is worse than failing: the
-    agent would be told about a governed definition that quietly is not there.
+    A table the catalog selection left out is refused rather than skipped:
+    the agent would otherwise be told about a governed definition that
+    quietly is not there.
     """
-    for authored, definitions in exports.items():
-        if definitions and bindings["tables"].get(authored) is None:
+    if bindings is None:
+        if authored not in exposed:
             raise ValueError(
-                f"Authored table {authored!r} declares definitions, and does "
-                f"not match an exposed relation. Name it as the data source "
-                f"selects it, or drop it from the data dictionary."
+                f"The data dictionary declares definitions on table "
+                f"{authored!r}, which the data source does not expose. "
+                f"Exposed tables: {', '.join(sorted(exposed))}."
             )
+        return authored
+    label = bindings["tables"].get(authored)
+    if label is None:
+        raise DefinitionUnmatchedTableError(
+            f"Authored table {authored!r} declares definitions, and does "
+            f"not match an exposed relation. Name it as the data source "
+            f"selects it, or drop it from the data dictionary."
+        )
+    return label
+
+
+def _check_metric_grain(table: str, definitions: dict[str, DefinitionExport]) -> None:
+    grain = mixed_grain(definitions)
+    offenders = [
+        name
+        for name, definition in definitions.items()
+        if grain[name] and definition.kind == "metric"
+    ]
+    if offenders:
+        raise DefinitionMixedGrainError(
+            f"Metric definition {offenders[0]!r} on table {table!r} cannot be "
+            f"compiled into one SQL expression. Its dependency chain mixes row "
+            f"and aggregate grain and would need a subquery rewrite."
+        )
 
 
 def _bind_table(
@@ -231,53 +275,32 @@ def _bind_selection(
 ) -> dict[str, Any] | None:
     """Rewrite a `COLUMNS(...)` selection to the discovered spellings.
 
-    A selection the author wrote out by name is bound strictly: naming a
-    column the warehouse does not have is a mistake worth reporting. A
-    selection derived from a pattern or from the whole table is not, because
-    the author named no column at all. It is resolved over the authored
-    columns, so a dictionary that still documents a dropped column would
-    otherwise take every wildcard definition on the table down with it, and
-    the refusal would name a column that appears nowhere in the expression.
-    The other side of that intersection is already tolerated: a warehouse
-    column the dictionary does not describe is simply not selected.
+    A selection from a pattern or from the whole table is resolved over the
+    authored columns, so it binds as strictly as one written out by name: a
+    dictionary that documents a column the relation lacks is refused.
     """
     if selection is None:
         return None
-    named = selection.get("form") == "list"
-    bound = []
-    for column in selection["columns"]:
-        path = _bind_path(
-            column["path"], columns, definition, table, required=named
-        )
-        if path is not None:
-            bound.append({**column, "path": path})
-    if not bound:
-        raise ValueError(
-            f"Definition {definition!r} on table {table!r} selects columns, "
-            f"and the selected relation has none of the ones the data "
-            f"dictionary describes."
-        )
-    return {**selection, "columns": bound}
+    return {
+        **selection,
+        "columns": [
+            {**column, "path": _bind_path(column["path"], columns, definition, table)}
+            for column in selection["columns"]
+        ],
+    }
 
 
 def _bind_path(
     path: Any,
     columns: dict[str, str | None],
     definition: str,
     table: str,
-    required: bool = True,
 ) -> Any:
-    """The physical spelling of the column a path leads with.
-
-    Returns None for a column the relation does not have, when the caller
-    can drop it rather than refuse.
-    """
+    """The physical spelling of the column a path leads with."""
     segments = list(path) if isinstance(path, list) else [path]
     physical = columns.get(segments[0])
     if physical is None:
-        if not required:
-            return None
-        raise ValueError(
+        raise DefinitionAbsentColumnError(
             f"Definition {definition!r} on table {table!r} references "
             f"authored column {segments[0]!r}, which is absent from the "
             f"selected relation."
@@ -290,17 +313,6 @@ def _compile_table(
     table: str, definitions: dict[str, DefinitionExport], target: str
 ) -> list[ExportRecord]:
     grain = mixed_grain(definitions)
-    offenders = [
-        name
-        for name, definition in definitions.items()
-        if grain[name] and definition.kind == "metric"
-    ]
-    if offenders:
-        raise ValueError(
-            f"Metric definition {offenders[0]!r} on table {table!r} cannot be "
-            f"compiled into one SQL expression. Its dependency chain mixes row "
-            f"and aggregate grain and would need a subquery rewrite."
-        )
     markers = _reference_markers(definitions)
     # Only the source's own target is lowered, and ExportRecord carries that
     # one target. The R implementation emits every target up front and keeps

pkg-py/tests/_warehouse.py

@@ -10,6 +10,8 @@
 fake that answered both would hide a reader wired to the wrong warehouse.
 """
 
+from commons._catalog import MergedDictionary, Relation, merge_dictionary
+from commons._data_dictionary import DataDictionary
 from commons._data_source import TableId
 from tests._shared import load_shared_fixture
 
@@ -19,6 +21,7 @@
     "column_row",
     "databricks_column_row",
     "databricks_relation_row",
+    "merge_shared_case",
     "relation_row",
 ]
 
@@ -181,3 +184,26 @@ def list_tables(self) -> list[str]:
 
     def dialect(self) -> str:
         return "databricks"
+
+
+def merge_shared_case(case: dict) -> MergedDictionary:
+    """Merge a shared fixture case's dictionary into its listed relations."""
+    relations = {}
+    for entry in case["relations"]:
+        relations[entry["label"]] = Relation(
+            id=TableId(
+                catalog=entry.get("catalog"),
+                schema=entry.get("schema"),
+                table=entry["table"],
+            ),
+            kind=entry.get("kind"),
+            description=entry.get("description"),
+        )
+    columns = {entry["label"]: entry["columns"] for entry in case["relations"]}
+    dictionary = DataDictionary.model_validate(case["dictionary"])
+    return merge_dictionary(
+        dictionary,
+        relations,
+        describe_relation=lambda table_id: columns[table_id.label],
+        identifier_case=case["identifier_case"],
+    )

pkg-py/tests/test_catalog.py

@@ -18,7 +18,6 @@
     CatalogAmbiguousTableError,
     CatalogClaimedRelationError,
     Manifest,
-    MergedDictionary,
     Relation,
     Selector,
     check_exclude,
@@ -33,6 +32,7 @@
 from commons._data_dictionary import DataDictionary
 from commons._data_source import TableId
 from tests._shared import load_shared_fixture
+from tests._warehouse import merge_shared_case
 
 
 def relation(label: str, **kwargs) -> Relation:
@@ -489,28 +489,6 @@ def test_exclusion_globs_match_the_shared_contract():
         assert excluded(case["names"], case["patterns"]) == case["hidden"], case["name"]
 
 
-def run_merge_case(case: dict) -> MergedDictionary:
-    relations = {}
-    for entry in case["relations"]:
-        relations[entry["label"]] = Relation(
-            id=TableId(
-                catalog=entry.get("catalog"),
-                schema=entry.get("schema"),
-                table=entry["table"],
-            ),
-            kind=entry.get("kind"),
-            description=entry.get("description"),
-        )
-    columns = {entry["label"]: entry["columns"] for entry in case["relations"]}
-    dictionary = DataDictionary.model_validate(case["dictionary"])
-    return merge_dictionary(
-        dictionary,
-        relations,
-        describe_relation=lambda table_id: columns[table_id.label],
-        identifier_case=case["identifier_case"],
-    )
-
-
 _MERGE_ERRORS = {
     "ambiguous": CatalogAmbiguousTableError,
     "claimed": CatalogClaimedRelationError,
@@ -525,10 +503,10 @@ def test_the_dictionary_merge_matches_the_shared_contract():
         expect = case["expect"]
         if "error" in expect:
             with pytest.raises(_MERGE_ERRORS[expect["error"]]):
-                run_merge_case(case)
+                merge_shared_case(case)
             continue
 
-        merged = run_merge_case(case)
+        merged = merge_shared_case(case)
         assert list(merged.dictionary.tables) == list(expect["tables"]), case["name"]
         for label, want in expect["tables"].items():
             table = merged.dictionary.tables[label]