From 1d1cfd06cdf531e5443dbe8498270ad7c6f15f01 Mon Sep 17 00:00:00 2001
From: Josh Taillon <joshua.taillon@posit.co>
Date: Fri, 4 Sep 2026 10:18:41 -0600
Subject: [PATCH 1/2] feat(py): bind definitions to the names the warehouse
 reported

A definition's expression names the columns the author wrote, and a
catalog import re-keys the dictionary to what the warehouse reported, so
the expression is rewritten before it is lowered: column paths in the IR
and in a COLUMNS(...) selection take the discovered spelling, references
to sibling definitions are left for composition, and an authored column
the relation does not have is refused rather than emitted.

An authored table with definitions that matched no selected relation is
refused too, since dropping it would leave the agent told about a
governed definition that is not there.
---
 pkg-py/src/commons/_catalog/_core.py        |   6 +-
 pkg-py/src/commons/_catalog/_import.py      |  71 -------
 pkg-py/src/commons/_data_source.py          |   5 +-
 pkg-py/src/commons/_definitions/_compile.py | 195 ++++++++++++++++----
 pkg-py/tests/test_catalog_import.py         |  24 ---
 pkg-py/tests/test_data_source_warehouse.py  |  20 --
 pkg-py/tests/test_definition_binding.py     | 185 +++++++++++++++++++
 7 files changed, 347 insertions(+), 159 deletions(-)
 create mode 100644 pkg-py/tests/test_definition_binding.py

diff --git a/pkg-py/src/commons/_catalog/_core.py b/pkg-py/src/commons/_catalog/_core.py
index f0ebe675..680cf119 100644
--- a/pkg-py/src/commons/_catalog/_core.py
+++ b/pkg-py/src/commons/_catalog/_core.py
@@ -402,11 +402,7 @@ def merge_dictionary(
     return MergedDictionary(
         dictionary=dictionary,
         relations=relations,
-        definition_bindings={
-            "tables": matches,
-            "columns": column_matches,
-            "strict": True,
-        },
+        definition_bindings={"tables": matches, "columns": column_matches},
     )
 
 
diff --git a/pkg-py/src/commons/_catalog/_import.py b/pkg-py/src/commons/_catalog/_import.py
index 696f6413..f0e77a71 100644
--- a/pkg-py/src/commons/_catalog/_import.py
+++ b/pkg-py/src/commons/_catalog/_import.py
@@ -17,11 +17,9 @@
 from . import _databricks, _snowflake
 from ._core import (
     Manifest,
-    MergedDictionary,
     Relation,
     Selector,
     check_exclude,
-    has_suffix,
     id_type,
     merge_dictionary,
     table_registry,
@@ -105,7 +103,6 @@ def access_check(table_id: TableId, label: str) -> None:
         access_check=access_check,
     )
     check_session(backend, session)
-    _check_definitions_bound(merged, registry.dropped, exclude, identifier_case)
 
     # The manifest starts with every relation unknown, including the ones
     # just probed. Carrying the construction-time answer forward would save a
@@ -125,74 +122,6 @@ def access_check(table_id: TableId, label: str) -> None:
     )
 
 
-def _check_definitions_bound(
-    merged: MergedDictionary,
-    dropped: list[Relation] | None = None,
-    exclude: list[str] | None = None,
-    identifier_case: str | None = None,
-) -> None:
-    """Refuse definitions the merge renamed out from under.
-
-    The merge re-keys an authored dictionary to the warehouse's own labels
-    and column spellings, but a definition's expression still names what the
-    author wrote. Lowering it as written would emit SQL against identifiers
-    the warehouse does not have, so it is refused until the compiler can bind
-    the two together.
-    """
-    bindings = merged.definition_bindings
-    exports = getattr(merged.dictionary, "definition_exports", None) or {}
-    if not bindings or not exports:
-        return
-    for authored_table, definitions in exports.items():
-        if not definitions:
-            continue
-        # An authored table that matched nothing is dropped by the merge, so
-        # its definitions would go with it and the agent would never be told.
-        if bindings["tables"].get(authored_table) is None:
-            if _was_excluded(authored_table, dropped, identifier_case):
-                raise ValueError(
-                    f"Authored table {authored_table!r} declares definitions, "
-                    f"and exclude dropped it from the catalog listing. Narrow "
-                    f"{exclude!r}, or drop the table from the data dictionary."
-                )
-            raise ValueError(
-                f"Authored table {authored_table!r} declares definitions, and "
-                f"does not match an exposed relation. Name it as the data "
-                f"source selects it, or drop it from the data dictionary."
-            )
-        # Every authored column is checked, not only the ones a definition
-        # reads: which columns an expression touches is in the compiler's
-        # parse tree, and the refusal is temporary either way. A column the
-        # warehouse never reported is caught here too, since a definition
-        # over it would lower to SQL naming nothing at all.
-        columns = bindings["columns"].get(authored_table) or {}
-        unbound = [name for name, discovered in columns.items() if discovered != name]
-        if unbound:
-            raise NotImplementedError(
-                f"Table {authored_table!r} declares definitions, and the "
-                f"warehouse does not have column {unbound[0]!r} under that "
-                f"name. Binding a definition to the discovered spelling is "
-                f"not available yet."
-            )
-
-
-def _was_excluded(
-    authored_table: str, dropped: list[Relation] | None, identifier_case: str | None
-) -> bool:
-    """Whether exclude is what removed the relation an authored name meant.
-
-    Compared against the relations exclude actually dropped rather than
-    against the patterns: a pattern is written in the warehouse's spelling
-    and an authored name need not be, so only the folded names line up. The
-    authored name may be qualified, so it is matched as a suffix, by the
-    same rule the merge uses to find the relation in the first place.
-    """
-    suffix = authored_table.split(".")
-    return any(
-        has_suffix(item, suffix, identifier_case) for item in dropped or []
-    )
-
-
 def _selectors(backend: Any, reader: Any, tables: Any) -> list[Selector]:
     """Read a `tables` selection, defaulting to the connection's namespace.
 
diff --git a/pkg-py/src/commons/_data_source.py b/pkg-py/src/commons/_data_source.py
index 4fe6cc40..4d6b2d21 100644
--- a/pkg-py/src/commons/_data_source.py
+++ b/pkg-py/src/commons/_data_source.py
@@ -54,7 +54,10 @@ def _with_compiled_definitions(source: DataSource) -> DataSource:
         from ._definitions import attach_compiled_definitions
 
         attach_compiled_definitions(
-            source.dictionary, source.dialect(), set(source.tables)
+            source.dictionary,
+            source.dialect(),
+            set(source.tables),
+            source.definition_bindings,
         )
     return source
 
diff --git a/pkg-py/src/commons/_definitions/_compile.py b/pkg-py/src/commons/_definitions/_compile.py
index 655683ce..4ea4cbbb 100644
--- a/pkg-py/src/commons/_definitions/_compile.py
+++ b/pkg-py/src/commons/_definitions/_compile.py
@@ -9,6 +9,8 @@
 
 from __future__ import annotations
 
+from collections.abc import Callable
+from dataclasses import replace
 from typing import Any
 
 from ._emit_duckdb import emit_duckdb
@@ -57,24 +59,49 @@ def _has_aggregate(ir: Ir) -> bool:
 
 
 def _children(ir: Ir) -> list[Ir]:
+    """Every node one level down, wherever the parent hung it.
+
+    A node's payload is a plain mapping, so a child can sit under a key, in a
+    list, or in a mapping inside a list: `CASE` holds its branches as a list
+    of condition-and-result pairs. Anything that stops at the first level it
+    does not recognize walks part of the tree and silently skips the rest.
+    """
     out: list[Ir] = []
-    for value in ir.attrs.values():
+
+    def collect(value: Any) -> None:
         if isinstance(value, Ir):
             out.append(value)
         elif isinstance(value, list):
             for item in value:
-                if isinstance(item, Ir):
-                    out.append(item)
-                elif isinstance(item, dict):
-                    # A `CASE` branch is a mapping of condition and result.
-                    out.extend(v for v in item.values() if isinstance(v, Ir))
+                collect(item)
         elif isinstance(value, dict):
-            out.extend(item for item in value.values() if isinstance(item, Ir))
+            for item in value.values():
+                collect(item)
+
+    collect(dict(ir.attrs))
     return out
 
 
+def _map_children(ir: Ir, transform: Callable[[Ir], Ir]) -> dict[str, Any]:
+    """A copy of a node's payload with `transform` applied to every child."""
+
+    def convert(value: Any) -> Any:
+        if isinstance(value, Ir):
+            return transform(value)
+        if isinstance(value, list):
+            return [convert(item) for item in value]
+        if isinstance(value, dict):
+            return {key: convert(item) for key, item in value.items()}
+        return value
+
+    return {key: convert(value) for key, value in ir.attrs.items()}
+
+
 def attach_compiled_definitions(
-    dictionary: Any, dialect: str, exposed: set[str]
+    dictionary: Any,
+    dialect: str,
+    exposed: set[str],
+    bindings: dict[str, Any] | None = None,
 ) -> None:
     """Compile every governed definition for `dialect`, onto the dictionary.
 
@@ -88,12 +115,18 @@ def attach_compiled_definitions(
     records have already reached the dictionary's retrieval chunks. Prose
     about an unexposed table is left alone, since only a definition emits SQL.
 
+    `bindings` is what a warehouse catalog import matched. With it, every
+    column a definition names is rewritten to the spelling the warehouse
+    reported, because the expression was written against the authored one.
+
     Nothing is assigned until every table compiles, so a refusal leaves the
     dictionary untouched and the caller can attach it to another source.
     """
     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
@@ -117,11 +150,109 @@ def attach_compiled_definitions(
                 f"{dialect!r} data source. commons lowers definitions to "
                 f"{', '.join(sorted(_TARGETS))}."
             )
+        if bindings is not None:
+            definitions = _bind_table(
+                definitions, bindings["columns"].get(authored) or {}, table_name
+            )
         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.
+
+    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.
+    """
+    for authored, definitions in exports.items():
+        if definitions and bindings["tables"].get(authored) is None:
+            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."
+            )
+
+
+def _bind_table(
+    definitions: dict[str, DefinitionExport],
+    columns: dict[str, str | None],
+    table: str,
+) -> dict[str, DefinitionExport]:
+    """Rewrite every definition's expression to the discovered spelling.
+
+    `DefinitionExport.columns`, the list of columns a definition reads, keeps
+    the authored spelling: it describes the dictionary the author wrote,
+    which is also what the R implementation leaves it as.
+    """
+    return {
+        name: replace(
+            definition,
+            ir=None
+            if definition.ir is None
+            else _bind_ir(definition.ir, columns, name, table),
+            selection=_bind_selection(definition.selection, columns, name, table),
+        )
+        for name, definition in definitions.items()
+    }
+
+
+def _bind_ir(ir: Ir, columns: dict[str, str | None], definition: str, table: str) -> Ir:
+    attrs = _map_children(
+        ir, lambda child: _bind_ir(child, columns, definition, table)
+    )
+    # A reference to a sibling definition is a name in the dictionary, not a
+    # column in the warehouse, and is inlined later rather than bound.
+    if ir.kind == "column" and ir.attrs.get("reference") != "definition":
+        attrs["path"] = _bind_path(attrs["path"], columns, definition, table)
+    # A COLUMNS(...) node carries the resolved selection alongside the copy
+    # the export record holds. The emitters read the record's, but leaving a
+    # stale one here would put two different answers in the same tree.
+    if "selection" in attrs:
+        attrs["selection"] = _bind_selection(
+            attrs["selection"], columns, definition, table
+        )
+    return Ir(kind=ir.kind, type=ir.type, shape=ir.shape, attrs=attrs)
+
+
+def _bind_selection(
+    selection: dict[str, Any] | None,
+    columns: dict[str, str | None],
+    definition: str,
+    table: str,
+) -> dict[str, Any] | None:
+    if selection is None:
+        return None
+    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
+) -> Any:
+    """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:
+        raise ValueError(
+            f"Definition {definition!r} on table {table!r} references "
+            f"authored column {segments[0]!r}, which is absent from the "
+            f"selected relation."
+        )
+    segments[0] = physical
+    return segments if isinstance(path, list) else segments[0]
+
+
 def _compile_table(
     table: str, definitions: dict[str, DefinitionExport], target: str
 ) -> list[ExportRecord]:
@@ -196,6 +327,7 @@ def _reference_markers(definitions: dict[str, DefinitionExport]) -> dict[str, st
         used.update(definition.columns)
         if definition.ir is not None:
             used.update(_ir_identifiers(definition.ir))
+        used.update(_selection_identifiers(definition.selection))
     markers: dict[str, str] = {}
     for index, name in enumerate(definitions, start=1):
         marker = f"__commons_definition_reference_{index:03d}__"
@@ -214,38 +346,25 @@ def _ir_identifiers(ir: Ir) -> set[str]:
     return out
 
 
+def _selection_identifiers(selection: dict[str, Any] | None) -> set[str]:
+    """The columns a `COLUMNS(...)` selection resolved to.
+
+    A selection reaches the emitted SQL as a plain mapping rather than as
+    column nodes in the tree, so `_ir_identifiers` never sees it. Binding can
+    put any physical spelling here, which is why it has to be reserved too.
+    """
+    if selection is None:
+        return set()
+    out: set[str] = set()
+    for column in selection["columns"]:
+        path = column["path"]
+        out.update(path if isinstance(path, list) else [path])
+    return out
+
+
 def _mark_references(ir: Ir, markers: dict[str, str]) -> Ir:
     """Rename each sibling-definition reference to its marker."""
-    attrs: dict[str, Any] = {}
-    for key, value in ir.attrs.items():
-        if isinstance(value, Ir):
-            attrs[key] = _mark_references(value, markers)
-        elif isinstance(value, list):
-            attrs[key] = [
-                _mark_references(item, markers)
-                if isinstance(item, Ir)
-                # A `CASE` branch is a mapping of condition and result.
-                else (
-                    {
-                        inner_key: _mark_references(inner, markers)
-                        if isinstance(inner, Ir)
-                        else inner
-                        for inner_key, inner in item.items()
-                    }
-                    if isinstance(item, dict)
-                    else item
-                )
-                for item in value
-            ]
-        elif isinstance(value, dict):
-            attrs[key] = {
-                inner_key: _mark_references(inner, markers)
-                if isinstance(inner, Ir)
-                else inner
-                for inner_key, inner in value.items()
-            }
-        else:
-            attrs[key] = value
+    attrs = _map_children(ir, lambda child: _mark_references(child, markers))
     if ir.kind == "column" and ir.attrs.get("reference") == "definition":
         path = list(attrs["path"])
         path[0] = markers[path[0]]
diff --git a/pkg-py/tests/test_catalog_import.py b/pkg-py/tests/test_catalog_import.py
index f44c1e95..fbcfa4ae 100644
--- a/pkg-py/tests/test_catalog_import.py
+++ b/pkg-py/tests/test_catalog_import.py
@@ -268,30 +268,6 @@ def test_an_exclude_that_matched_nothing_does_not_blame_exclude():
     assert "exclude dropped" not in str(refusal.value)
 
 
-@pytest.mark.parametrize(
-    "authored", ["sales", "PUBLIC.sales", "ANALYTICS.PUBLIC.sales"]
-)
-def test_definitions_on_an_excluded_table_name_exclude_however_qualified(authored):
-    # An authored name may be qualified, and a glob may not be, so the two
-    # are matched by the same suffix rule the merge uses.
-    dictionary = DataDictionary.model_validate(
-        {
-            "tables": [
-                {
-                    "name": authored,
-                    "columns": [{"name": "id", "type": "number(quantity)"}],
-                    "definitions": [{"name": "total", "expr": "sum(id)"}],
-                }
-            ]
-        }
-    )
-
-    with pytest.raises(ValueError, match="exclude dropped it"):
-        import_catalog(FakeWarehouse(), exclude=["SALES"], dictionary=dictionary)
-
-
-
-
 def test_excluding_a_name_the_warehouse_never_had_does_not_blame_exclude():
     # The relation was absent, not dropped, so exclude is not the reason
     # there is nothing left to expose.
diff --git a/pkg-py/tests/test_data_source_warehouse.py b/pkg-py/tests/test_data_source_warehouse.py
index 708bbf9e..65491559 100644
--- a/pkg-py/tests/test_data_source_warehouse.py
+++ b/pkg-py/tests/test_data_source_warehouse.py
@@ -94,26 +94,6 @@ def test_exclude_is_refused_for_named_frames():
         data_source(sales=pd.DataFrame({"id": [1]}), exclude=["tmp_*"])
 
 
-def test_definitions_the_warehouse_spells_differently_are_refused():
-    # Until the compiler can bind an authored name to the discovered one, a
-    # definition over a renamed column would lower to SQL naming a column
-    # that is not there, so construction fails instead.
-    dictionary = DataDictionary.model_validate(
-        {
-            "tables": [
-                {
-                    "name": "sales",
-                    "columns": [{"name": "id", "type": "number(quantity)"}],
-                    "definitions": [{"name": "total", "expr": "sum(id)"}],
-                }
-            ]
-        }
-    )
-
-    with pytest.raises(NotImplementedError, match="does not have column"):
-        warehouse_source(dictionary=dictionary)
-
-
 def sales_dictionary(**table):
     return DataDictionary.model_validate(
         {
diff --git a/pkg-py/tests/test_definition_binding.py b/pkg-py/tests/test_definition_binding.py
new file mode 100644
index 00000000..2b906a92
--- /dev/null
+++ b/pkg-py/tests/test_definition_binding.py
@@ -0,0 +1,185 @@
+"""Binding a definition to the names a warehouse actually uses.
+
+A definition's expression names the columns the author wrote. A catalog
+import re-keys the dictionary to what the warehouse reported, which on
+Snowflake means upper case, so the expression has to be rewritten before it
+is lowered or the SQL would name columns the warehouse does not have.
+"""
+
+import pytest
+
+from commons._catalog._import import import_catalog
+from commons._data_dictionary import DataDictionary
+from commons._definitions._compile import (
+    _bind_ir,
+    _reference_markers,
+    attach_compiled_definitions,
+)
+from commons._definitions._export import DefinitionExport, Ir
+from tests._warehouse import FakeWarehouse
+
+
+def dictionary(*definitions: dict, columns: list[dict] | None = None):
+    return DataDictionary.model_validate(
+        {
+            "tables": [
+                {
+                    "name": "sales",
+                    "columns": columns
+                    if columns is not None
+                    else [{"name": "id", "type": "number(quantity)"}],
+                    "definitions": list(definitions),
+                }
+            ]
+        }
+    )
+
+
+def imported_sql(*definitions: dict, columns: list[dict] | None = None):
+    """Every definition's SQL, as a warehouse source would compile it."""
+    imported = import_catalog(
+        FakeWarehouse(), dictionary=dictionary(*definitions, columns=columns)
+    )
+    attach_compiled_definitions(
+        imported.dictionary,
+        "snowflake",
+        set(imported.tables),
+        imported.definition_bindings,
+    )
+    assert imported.dictionary is not None
+    entry = imported.dictionary.tables["ANALYTICS.PUBLIC.SALES"]
+    return {record.name: record.sql for record in entry.compiled_definitions}
+
+
+def test_a_column_is_lowered_with_the_spelling_the_warehouse_reported():
+    compiled = imported_sql({"name": "total", "expr": "sum(id)"})
+
+    assert compiled["total"] == 'sum("ID")'
+
+
+def test_a_sibling_reference_is_bound_through_its_own_expression():
+    compiled = imported_sql(
+        {"name": "total", "expr": "sum(id)"},
+        {"name": "doubled", "expr": "total * 2"},
+    )
+
+    assert compiled["doubled"] == '(sum("ID")) * 2'
+
+
+def test_a_selection_is_bound_column_by_column():
+    compiled = imported_sql({"name": "any_set", "expr": "COLUMNS('^i') IS NOT NULL"})
+
+    assert '"ID"' in compiled["any_set"]
+    assert '"id"' not in compiled["any_set"]
+
+
+def test_a_column_inside_a_case_branch_is_bound_too():
+    # CASE keeps its branches as a list of condition-and-result pairs, so a
+    # walk that only descends into nodes it finds directly under a key walks
+    # past every column reference in them.
+    compiled = imported_sql(
+        {"name": "flagged", "expr": "CASE WHEN id > 0 THEN 1 ELSE 0 END"}
+    )
+
+    assert '"ID"' in compiled["flagged"]
+    assert '"id"' not in compiled["flagged"]
+
+
+def test_a_selection_is_bound_on_the_node_that_carries_it_too():
+    # A COLUMNS(...) node keeps its own copy of the resolved selection. The
+    # emitters read the export record's copy, but two copies that disagree
+    # are a trap for whoever reads the tree next.
+    node = Ir(
+        kind="selected",
+        type="any",
+        shape="row",
+        attrs={
+            "selection": {
+                "form": "regex",
+                "pattern": "^i",
+                "columns": [{"name": "id", "path": "id", "type": "number"}],
+            }
+        },
+    )
+
+    bound = _bind_ir(node, {"id": "ID"}, "any_set", "sales")
+
+    assert [column["path"] for column in bound.attrs["selection"]["columns"]] == ["ID"]
+
+
+def test_a_definition_over_a_column_the_relation_lacks_is_refused():
+    # An authored column the warehouse never reported survives the merge, so
+    # the definition type-checks against the dictionary and only binding can
+    # catch that there is no such column to read.
+    with pytest.raises(ValueError, match="absent from the selected relation"):
+        imported_sql(
+            {"name": "total", "expr": "sum(absent)"},
+            columns=[
+                {"name": "id", "type": "number(quantity)"},
+                {"name": "absent", "type": "number(quantity)"},
+            ],
+        )
+
+
+def test_an_authored_table_that_matched_nothing_takes_its_definitions_with_it():
+    imported = import_catalog(
+        FakeWarehouse(),
+        dictionary=DataDictionary.model_validate(
+            {
+                "tables": [
+                    {
+                        "name": "unmatched",
+                        "columns": [{"name": "id", "type": "number(quantity)"}],
+                        "definitions": [{"name": "total", "expr": "sum(id)"}],
+                    }
+                ]
+            }
+        ),
+    )
+
+    with pytest.raises(ValueError, match="does not match an exposed relation"):
+        attach_compiled_definitions(
+            imported.dictionary,
+            "snowflake",
+            set(imported.tables),
+            imported.definition_bindings,
+        )
+
+
+def test_without_bindings_a_definition_keeps_the_authored_spelling():
+    authored = dictionary({"name": "total", "expr": "sum(id)"})
+
+    attach_compiled_definitions(authored, "snowflake", {"sales"})
+
+    records = authored.tables["sales"].compiled_definitions
+    assert records[0].sql == 'sum("id")'
+
+
+def test_a_bound_selection_cannot_be_mistaken_for_a_reference_marker():
+    """A physical column named like a marker must not shadow a reference.
+
+    Binding chooses the physical spelling, so nothing stops a warehouse
+    column from being named exactly like the marker a sibling reference is
+    substituted through. The selection carries that name outside the tree's
+    column nodes, which is the path the collision set used to miss.
+    """
+    marker = "__commons_definition_reference_001__"
+    markers = _reference_markers(
+        {
+            "base": DefinitionExport(
+                name="base",
+                label=None,
+                description=None,
+                details=None,
+                todo=None,
+                expression="sum(id)",
+                kind="metric",
+                type="number",
+                columns=["id"],
+                definitions=[],
+                selection={"columns": [{"path": [marker]}]},
+            )
+        }
+    )
+
+    assert markers["base"] != marker

From 51ab0ac2f9f8876364d7e4aa974ff6e47bdfa3ce Mon Sep 17 00:00:00 2001
From: Josh Taillon <joshua.taillon@posit.co>
Date: Sun, 6 Sep 2026 16:56:11 -0600
Subject: [PATCH 2/2] fix(py): bind a derived column selection leniently, and
 correct what the docs promise

Review findings on this branch.

A `COLUMNS(...)` selection resolved from a pattern or from `*` was bound as
strictly as one the author wrote out by name, so a dictionary that still
documented a column the warehouse had dropped took every wildcard definition
on that table down with it, and the refusal named a column that appears
nowhere in the expression. A derived selection now drops what the relation
does not have, which is what already happened to the other side of the same
intersection: a warehouse column the dictionary does not describe was never
selected. A selection the author wrote out by name still refuses, since
naming a column that is not there is worth reporting, and a derived selection
left with nothing refuses too rather than lowering to an empty predicate.

`data_source()` still promised the refusal this commit's parent replaced, that
construction fails when the warehouse spells a column differently. That is now
the case binding exists to handle. It says what actually raises instead.

`attach_compiled_definitions()` documented the new `bindings` argument but not
the default: without it the authored spellings are lowered as written and
neither of the two refusals binding adds is reachable, which is what every
source with no catalog to match against does.

That `DefinitionExport.columns` keeps the authored spelling was explained only
in a private helper in another module, and justified by what the R package
does rather than by what the field is for. The reason is that it describes the
authored dictionary rather than the relation it matched, and it is now on the
field a reader lands on. `definition_bindings` says what its two keys hold.

Tests: the four selection cases above, and the public constructor compiling a
definition whose column the warehouse spells differently. Both existing
public-path definition tests used a warehouse whose spelling already matched,
so binding was an identity transform in them.

Pinning any of this as a shared fixture is kata 5hfw, which now also carries
the derived-vs-named split, since R does not make that distinction.
---
 pkg-py/src/commons/_catalog/_core.py         |  4 ++
 pkg-py/src/commons/_data_source.py           |  9 +--
 pkg-py/src/commons/_definitions/_compile.py  | 63 +++++++++++++++-----
 pkg-py/src/commons/_definitions/_registry.py |  3 +
 pkg-py/tests/test_catalog_import.py          |  4 +-
 pkg-py/tests/test_data_source_warehouse.py   | 18 ++++++
 pkg-py/tests/test_definition_binding.py      | 46 ++++++++++++++
 7 files changed, 126 insertions(+), 21 deletions(-)

diff --git a/pkg-py/src/commons/_catalog/_core.py b/pkg-py/src/commons/_catalog/_core.py
index 680cf119..25b7e804 100644
--- a/pkg-py/src/commons/_catalog/_core.py
+++ b/pkg-py/src/commons/_catalog/_core.py
@@ -87,6 +87,10 @@ class TableRegistry:
 class MergedDictionary:
     dictionary: Any
     relations: dict[str, Relation]
+    # What the merge matched, for the definition compiler: "tables" maps each
+    # authored table name to the relation label it matched, or None, and
+    # "columns" maps each authored column name to the spelling the warehouse
+    # reported. None when there was no catalog to match against.
     definition_bindings: dict[str, Any] | None
 
 
diff --git a/pkg-py/src/commons/_data_source.py b/pkg-py/src/commons/_data_source.py
index 4d6b2d21..46f4d749 100644
--- a/pkg-py/src/commons/_data_source.py
+++ b/pkg-py/src/commons/_data_source.py
@@ -390,10 +390,11 @@ def data_source(
     dialect during construction, so construction raises if the dialect has
     no emitter (DuckDB, Snowflake, and Databricks have one), if a definition
     sits on a table the source does not expose, or if a metric mixes row and
-    aggregate grain. On a warehouse it also raises if a table declaring
-    definitions matched no exposed relation, and, until the compiler can
-    bind an authored name to the discovered one, if the warehouse spells one
-    of that table's columns differently.
+    aggregate grain. On a warehouse the authored column spellings are bound
+    to the names the catalog reported before anything is lowered, so it
+    raises there only if a table declaring definitions matched no exposed
+    relation, or if a definition names an authored column the selected
+    relation does not have.
     """
     from ._data_dictionary import as_data_dictionary
 
diff --git a/pkg-py/src/commons/_definitions/_compile.py b/pkg-py/src/commons/_definitions/_compile.py
index 4ea4cbbb..4396fac1 100644
--- a/pkg-py/src/commons/_definitions/_compile.py
+++ b/pkg-py/src/commons/_definitions/_compile.py
@@ -117,7 +117,12 @@ def attach_compiled_definitions(
 
     `bindings` is what a warehouse catalog import matched. With it, every
     column a definition names is rewritten to the spelling the warehouse
-    reported, because the expression was written against the authored one.
+    reported, because the expression was written against the authored one,
+    and two further refusals apply: a table that declares definitions and
+    matched no relation, and a definition naming an authored column the
+    relation does not have. Without it, which is every source that has no
+    catalog to match against, the authored spellings are lowered as written
+    and neither refusal is reachable.
 
     Nothing is assigned until every table compiles, so a refusal leaves the
     dictionary untouched and the caller can attach it to another source.
@@ -184,8 +189,9 @@ def _bind_table(
     """Rewrite every definition's expression to the discovered spelling.
 
     `DefinitionExport.columns`, the list of columns a definition reads, keeps
-    the authored spelling: it describes the dictionary the author wrote,
-    which is also what the R implementation leaves it as.
+    the authored spelling. It describes the dictionary the author wrote,
+    rather than the relation it was matched to, so it is the one place the
+    authored names survive binding.
     """
     return {
         name: replace(
@@ -223,27 +229,54 @@ def _bind_selection(
     definition: str,
     table: str,
 ) -> 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.
+    """
     if selection is None:
         return None
-    return {
-        **selection,
-        "columns": [
-            {
-                **column,
-                "path": _bind_path(column["path"], columns, definition, table),
-            }
-            for column in selection["columns"]
-        ],
-    }
+    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}
 
 
 def _bind_path(
-    path: Any, columns: dict[str, str | None], definition: str, table: str
+    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."""
+    """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.
+    """
     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(
             f"Definition {definition!r} on table {table!r} references "
             f"authored column {segments[0]!r}, which is absent from the "
diff --git a/pkg-py/src/commons/_definitions/_registry.py b/pkg-py/src/commons/_definitions/_registry.py
index 1434803f..13ea600b 100644
--- a/pkg-py/src/commons/_definitions/_registry.py
+++ b/pkg-py/src/commons/_definitions/_registry.py
@@ -49,6 +49,9 @@ class ExportRecord:
     label: str | None
     description: str | None
     details: str | None
+    # The authored spellings, even on a warehouse source whose `sql` names
+    # the columns the catalog reported: this describes the dictionary the
+    # author wrote rather than the relation it was matched to.
     columns: list[str]
     definitions: list[str]
     sql: str
diff --git a/pkg-py/tests/test_catalog_import.py b/pkg-py/tests/test_catalog_import.py
index fbcfa4ae..60fc5e62 100644
--- a/pkg-py/tests/test_catalog_import.py
+++ b/pkg-py/tests/test_catalog_import.py
@@ -265,7 +265,7 @@ def test_an_exclude_that_matched_nothing_does_not_blame_exclude():
     with pytest.raises(ValueError, match="contains no objects") as refusal:
         import_catalog(backend, exclude=["NOTHING_*"])
 
-    assert "exclude dropped" not in str(refusal.value)
+    assert "dropped every relation" not in str(refusal.value)
 
 
 def test_excluding_a_name_the_warehouse_never_had_does_not_blame_exclude():
@@ -276,4 +276,4 @@ def test_excluding_a_name_the_warehouse_never_had_does_not_blame_exclude():
             FakeWarehouse(), "ANALYTICS.PUBLIC.MISSING", exclude=["MISSING"]
         )
 
-    assert "exclude dropped" not in str(refusal.value)
+    assert "dropped every relation" not in str(refusal.value)
diff --git a/pkg-py/tests/test_data_source_warehouse.py b/pkg-py/tests/test_data_source_warehouse.py
index 65491559..b78e0f8d 100644
--- a/pkg-py/tests/test_data_source_warehouse.py
+++ b/pkg-py/tests/test_data_source_warehouse.py
@@ -122,6 +122,24 @@ def test_a_definition_survives_the_rekeying_a_merge_does(monkeypatch):
     assert [record.name for record in entry.compiled_definitions] == ["total"]
 
 
+def test_a_definition_reaches_sql_with_the_warehouse_spelling(monkeypatch):
+    # The public constructor is the only path a user takes, and the two other
+    # definition tests here use a warehouse whose spelling already matches,
+    # so binding is an identity transform in them and proves nothing.
+    backend = FakeWarehouse()
+    monkeypatch.setattr("commons._data_source.EngineBackend", lambda engine: backend)
+    dictionary = sales_dictionary(definitions=[{"name": "total", "expr": "sum(id)"}])
+
+    source = data_source(sqlalchemy.create_engine("sqlite://"), dictionary=dictionary)
+
+    assert source.dictionary is not None
+    entry = source.dictionary.tables["ANALYTICS.PUBLIC.SALES"]
+    record = entry.compiled_definitions[0]
+    assert record.sql == 'sum("ID")'
+    # The record still reports the column the author wrote.
+    assert record.columns == ["id"]
+
+
 def test_a_definition_on_a_table_that_matched_nothing_is_refused(monkeypatch):
     backend = FakeWarehouse(columns=["id"])
     monkeypatch.setattr("commons._data_source.EngineBackend", lambda engine: backend)
diff --git a/pkg-py/tests/test_definition_binding.py b/pkg-py/tests/test_definition_binding.py
index 2b906a92..1dbccf2c 100644
--- a/pkg-py/tests/test_definition_binding.py
+++ b/pkg-py/tests/test_definition_binding.py
@@ -183,3 +183,49 @@ def test_a_bound_selection_cannot_be_mistaken_for_a_reference_marker():
     )
 
     assert markers["base"] != marker
+
+
+def test_a_derived_selection_drops_a_column_the_relation_does_not_have():
+    # A pattern names no column, so a dictionary that still documents one the
+    # warehouse dropped should not take every wildcard definition with it.
+    compiled = imported_sql(
+        {"name": "any_set", "expr": "COLUMNS('.*') IS NOT NULL"},
+        columns=[
+            {"name": "id", "type": "number(quantity)"},
+            {"name": "ghost", "type": "number(quantity)"},
+        ],
+    )
+
+    assert compiled["any_set"] == '"ID" IS NOT NULL'
+
+
+def test_a_star_selection_drops_a_column_the_relation_does_not_have():
+    compiled = imported_sql(
+        {"name": "any_set", "expr": "COLUMNS(*) IS NOT NULL"},
+        columns=[
+            {"name": "id", "type": "number(quantity)"},
+            {"name": "ghost", "type": "number(quantity)"},
+        ],
+    )
+
+    assert compiled["any_set"] == '"ID" IS NOT NULL'
+
+
+def test_a_named_selection_still_refuses_a_column_the_relation_lacks():
+    # The author wrote this column out, so its absence is worth reporting.
+    with pytest.raises(ValueError, match="authored column 'ghost'"):
+        imported_sql(
+            {"name": "any_set", "expr": "COLUMNS([id, ghost]) IS NOT NULL"},
+            columns=[
+                {"name": "id", "type": "number(quantity)"},
+                {"name": "ghost", "type": "number(quantity)"},
+            ],
+        )
+
+
+def test_a_derived_selection_left_with_nothing_is_refused():
+    with pytest.raises(ValueError, match="has none of the ones"):
+        imported_sql(
+            {"name": "any_set", "expr": "COLUMNS('.*') IS NOT NULL"},
+            columns=[{"name": "ghost", "type": "number(quantity)"}],
+        )