perf: resolve promotion properties against a pre-normalized view mapping

Promotion passed snapshots keyed by SnapshotId into render_virtual_properties,
which expects them keyed by model name. Snapshot lookups, @model_kind_name and
embedded-model expansion therefore never worked in virtual properties during
promotion. _promote_snapshot now re-keys them by name once and uses that dict
for both virtual properties and on_virtual_update.

to_view_mapping now returns a TableMapping, which normalizes its keys once per
dialect. _resolve_table looks the table up in that index and passes only the
matching entry to exp.replace_tables, instead of re-normalizing every model in
the environment for each rendered property. The explicit mapping still takes
precedence over snapshots, and among equivalent keys the last one still wins.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Rwe8v9iP3iGKVygBfPEZaU

Author: claude

sqlmesh/core/renderer.py

@@ -45,6 +45,65 @@
 logger = logging.getLogger(__name__)
 
 
+class TableMapping(t.Dict[str, str]):
+    """A table name mapping that caches the dialect-normalized form of its keys.
+
+    `exp.replace_tables` normalizes every key of the mapping it's given, so resolving a single
+    table against a mapping of every model in an environment costs O(N). Resolving it against
+    this mapping costs a dictionary lookup, since each key is normalized once per dialect.
+    """
+
+    def __init__(self, *args: t.Any, **kwargs: t.Any):
+        super().__init__(*args, **kwargs)
+        self._normalized_keys: t.Dict[DialectType, t.Dict[str, str]] = {}
+
+    def normalized_keys(self, dialect: DialectType) -> t.Dict[str, str]:
+        """Returns a mapping from each normalized key to the last key that normalizes to it."""
+        normalized_keys = self._normalized_keys.get(dialect)
+        if normalized_keys is None:
+            normalized_keys = {exp.normalize_table_name(key, dialect=dialect): key for key in self}
+            self._normalized_keys[dialect] = normalized_keys
+        return normalized_keys
+
+    def __setitem__(self, key: str, value: str) -> None:
+        self._normalized_keys.clear()
+        super().__setitem__(key, value)
+
+    def __delitem__(self, key: str) -> None:
+        self._normalized_keys.clear()
+        super().__delitem__(key)
+
+    def __ior__(self, other: t.Any) -> TableMapping:  # type: ignore[override,misc]
+        self._normalized_keys.clear()
+        return super().__ior__(other)
+
+    def update(self, *args: t.Any, **kwargs: t.Any) -> None:
+        self._normalized_keys.clear()
+        super().update(*args, **kwargs)
+
+    def setdefault(self, key: str, default: str) -> str:  # type: ignore[override]
+        self._normalized_keys.clear()
+        return super().setdefault(key, default)
+
+    def pop(self, key: str, *args: t.Any) -> t.Any:
+        self._normalized_keys.clear()
+        return super().pop(key, *args)
+
+    def popitem(self) -> t.Tuple[str, str]:
+        self._normalized_keys.clear()
+        return super().popitem()
+
+    def clear(self) -> None:
+        self._normalized_keys.clear()
+        super().clear()
+
+
+def _normalize_keys(mapping: t.Dict[str, str], dialect: DialectType) -> t.Dict[str, str]:
+    if isinstance(mapping, TableMapping):
+        return mapping.normalized_keys(dialect)
+    return {exp.normalize_table_name(key, dialect=dialect): key for key in mapping}
+
+
 class BaseExpressionRenderer:
     def __init__(
         self,
@@ -330,35 +389,30 @@ def _resolve_table(
         table_mapping: t.Optional[t.Dict[str, str]] = None,
         deployability_index: t.Optional[DeployabilityIndex] = None,
     ) -> exp.Table:
-        table_mapping = table_mapping or {}
-        if isinstance(table_name, str):
+        table = t.cast(
+            exp.Table, exp.maybe_parse(table_name, into=exp.Table, dialect=self._dialect)
+        )
+
+        mapping: t.Dict[str, str] = {}
+        if table_mapping:
+            # An explicit mapping takes precedence over snapshots, so when one of its keys matches
+            # the table, that key alone decides the result. Among equivalent keys, the last wins.
+            key = _normalize_keys(table_mapping, self._dialect).get(
+                exp.normalize_table_name(table, dialect=self._dialect)
+            )
+            if key is not None:
+                mapping = {key: table_mapping[key]}
+
+        if not mapping and snapshots:
             # An exact FQN match avoids scanning unrelated snapshots.
-            snapshot = snapshots.get(table_name) if snapshots else None
-            if snapshot is None and table_name not in table_mapping:
-                # Keys normalized under different dialects may differ in casing or quoting.
-                # Fall back to the full mapping so exp.replace_tables can reconcile them.
-                mapping = {
-                    **self._to_table_mapping((snapshots or {}).values(), deployability_index),
-                    **table_mapping,
-                }
-            else:
-                mapping = {
-                    **self._to_table_mapping([snapshot] if snapshot else [], deployability_index),
-                    # Keep all explicit overrides to preserve precedence for equivalent keys.
-                    **table_mapping,
-                }
-        else:
-            mapping = {
-                **self._to_table_mapping((snapshots or {}).values(), deployability_index),
-                **table_mapping,
-            }
+            snapshot = snapshots.get(table_name) if isinstance(table_name, str) else None
+            # Keys normalized under different dialects may differ in casing or quoting.
+            # Fall back to the full mapping so exp.replace_tables can reconcile them.
+            mapping = self._to_table_mapping(
+                [snapshot] if snapshot else snapshots.values(), deployability_index
+            )
 
-        table = exp.replace_tables(
-            t.cast(exp.Table, exp.maybe_parse(table_name, into=exp.Table, dialect=self._dialect)),
-            mapping,
-            dialect=self._dialect,
-            copy=False,
-        )
+        table = exp.replace_tables(table, mapping, dialect=self._dialect, copy=False)
         # We quote the table here to mimic the behavior of _resolve_tables, otherwise we may end
         # up normalizing twice, because _to_table_mapping returns the mapped names unquoted.
         return (

sqlmesh/core/snapshot/definition.py

@@ -24,6 +24,7 @@
 from sqlmesh.core.model import Model, ModelKindMixin, ModelKindName, ViewKind, CustomKind
 from sqlmesh.core.model.definition import _Model
 from sqlmesh.core.node import IntervalUnit, NodeType
+from sqlmesh.core.renderer import TableMapping
 from sqlmesh.utils import sanitize_name, unique
 from sqlmesh.utils.dag import DAG
 from sqlmesh.utils.date import (
@@ -2007,14 +2008,17 @@ def to_view_mapping(
     environment_naming_info: EnvironmentNamingInfo,
     default_catalog: t.Optional[str] = None,
     dialect: t.Optional[str] = None,
-) -> t.Dict[str, str]:
-    return {
-        snapshot.name: snapshot.display_name(
-            environment_naming_info, default_catalog=default_catalog, dialect=dialect
+) -> TableMapping:
+    return TableMapping(
+        (
+            snapshot.name,
+            snapshot.display_name(
+                environment_naming_info, default_catalog=default_catalog, dialect=dialect
+            ),
         )
         for snapshot in snapshots
         if snapshot.is_model
-    }
+    )
 
 
 def has_paused_forward_only(

sqlmesh/core/snapshot/evaluator.py

@@ -1284,6 +1284,8 @@ def _promote_snapshot(
             table_mapping=table_mapping,
             runtime_stage=RuntimeStage.PROMOTING,
         )
+        # Renderers look snapshots up by model name, not by SnapshotId.
+        snapshots_by_name = {s.name: s for s in (snapshots or {}).values()}
 
         with (
             adapter.transaction(),
@@ -1294,14 +1296,16 @@ def _promote_snapshot(
                 view_name=view_name,
                 model=snapshot.model,
                 environment=environment_naming_info.name,
-                snapshots=snapshots,
+                snapshots=snapshots_by_name,
                 snapshot=snapshot,
                 **render_kwargs,
             )
 
-            snapshot_by_name = {s.name: s for s in (snapshots or {}).values()}
-            render_kwargs["snapshots"] = snapshot_by_name
-            adapter.execute(snapshot.model.render_on_virtual_update(**render_kwargs))
+            adapter.execute(
+                snapshot.model.render_on_virtual_update(
+                    snapshots=snapshots_by_name, **render_kwargs
+                )
+            )
 
         if on_complete is not None:
             on_complete(snapshot)

tests/core/test_model.py

@@ -15,6 +15,7 @@
 from sqlglot.schema import MappingSchema
 from sqlmesh.cli.project_init import init_example_project, ProjectTemplate
 from sqlmesh.core.environment import EnvironmentNamingInfo
+from sqlmesh.core.renderer import TableMapping
 from sqlmesh.core.model.kind import TimeColumn, ModelKindName, SeedKind
 
 from sqlmesh import CustomMaterialization, CustomKind
@@ -9818,9 +9819,10 @@ def resolve_named(evaluator, name):
     assert unmapped_result[0].sql() == '"does_not_exist"'
 
 
+@pytest.mark.parametrize("mapping_type", [dict, TableMapping])
 @pytest.mark.parametrize("include_exact_mapping", [False, True])
 def test_resolve_table_preserves_dialect_equivalent_table_mapping_override(
-    make_snapshot: t.Callable, include_exact_mapping: bool
+    make_snapshot: t.Callable, include_exact_mapping: bool, mapping_type: t.Callable
 ):
     """An explicit mapping should override a snapshot mapping when its key is dialect-equivalent
     to the resolved table name, even when the snapshot lookup is an exact match."""
@@ -9848,7 +9850,7 @@ def resolve_named(evaluator, name):
         table_mapping = {parent.fqn: "earlier_table", **table_mapping}
 
     post_statements = child.render_post_statements(
-        snapshots={parent.fqn: parent_snapshot}, table_mapping=table_mapping
+        snapshots={parent.fqn: parent_snapshot}, table_mapping=mapping_type(table_mapping)
     )
 
     assert post_statements[0].sql() == '"override_table"'
@@ -10221,6 +10223,87 @@ def items(self):
     assert ItemsCountingDict.items_call_count == 0
 
 
+def test_resolve_table_with_view_mapping_uses_single_entry(
+    make_snapshot: t.Callable, mocker: MockerFixture
+):
+    """During promotion `table_mapping` maps every model in the environment to its view. Resolving
+    one table against it must not normalize every key in that mapping on every call
+    (https://github.com/SQLMesh/sqlmesh/issues/6017)."""
+    from sqlmesh.core.snapshot.definition import to_view_mapping
+
+    @macro()
+    def resolve_named(evaluator, name):
+        return evaluator.resolve_table(name.name)
+
+    snapshots = {}
+    for i in range(50):
+        other = load_sql_based_model(d.parse(f"MODEL (name db.other_{i}); SELECT 1 AS c"))
+        other_snapshot = make_snapshot(other)
+        other_snapshot.categorize_as(SnapshotChangeCategory.BREAKING)
+        snapshots[other.fqn] = other_snapshot
+
+    children = [
+        load_sql_based_model(
+            d.parse(
+                f"""
+                MODEL (name db.child_{i});
+                SELECT 1 AS c;
+                @resolve_named('db.other_{i}')
+                """
+            )
+        )
+        for i in range(3)
+    ]
+    for child in children:
+        child_snapshot = make_snapshot(child)
+        child_snapshot.categorize_as(SnapshotChangeCategory.BREAKING)
+        snapshots[child.fqn] = child_snapshot
+
+    table_mapping = to_view_mapping(snapshots.values(), EnvironmentNamingInfo(name="dev"))
+    spy = mocker.spy(exp, "replace_tables")
+
+    for i, child in enumerate(children):
+        rendered = child.render_post_statements(snapshots=snapshots, table_mapping=table_mapping)
+        assert rendered[0].sql() == f'"db__dev"."other_{i}"'
+
+    # One call for `this_model` and one for the resolved table, per child.
+    assert spy.call_count == 6
+    for call in spy.call_args_list:
+        assert len(call.args[1]) == 1
+
+
+@pytest.mark.parametrize("dialect", ["duckdb", "snowflake"])
+def test_table_mapping_normalized_keys(dialect: str):
+    table_mapping = TableMapping({'"db"."a"': "view_a", "db.A": "view_a_upper"})
+
+    normalized = table_mapping.normalized_keys(dialect)
+    # Keys that normalize to the same name resolve to the last one, like exp.replace_tables.
+    if dialect == "snowflake":
+        assert normalized == {"db.a": '"db"."a"', "DB.A": "db.A"}
+    else:
+        assert normalized == {"db.a": "db.A"}
+    # Normalization happens once per dialect.
+    assert table_mapping.normalized_keys(dialect) is normalized
+
+    # Every mutation invalidates the cache.
+    table_mapping["db.b"] = "view_b"
+    assert "db.b" in table_mapping.normalized_keys("duckdb")
+    table_mapping.update({"db.c": "view_c"})
+    assert "db.c" in table_mapping.normalized_keys("duckdb")
+    table_mapping.setdefault("db.d", "view_d")
+    assert "db.d" in table_mapping.normalized_keys("duckdb")
+    table_mapping |= {"db.e": "view_e"}
+    assert "db.e" in table_mapping.normalized_keys("duckdb")
+    del table_mapping["db.b"]
+    assert "db.b" not in table_mapping.normalized_keys("duckdb")
+    table_mapping.pop("db.c")
+    assert "db.c" not in table_mapping.normalized_keys("duckdb")
+    table_mapping.popitem()
+    assert "db.e" not in table_mapping.normalized_keys("duckdb")
+    table_mapping.clear()
+    assert table_mapping.normalized_keys("duckdb") == {}
+
+
 def test_cluster_with_complex_expression():
     expressions = d.parse(
         """