fix: preserve cross-dialect resolution in narrowed _resolve_table lookup

The narrowed single-snapshot lookup added in the previous commit did a raw
snapshots.get(table_name) dict lookup. table_name is normalized under the
referencing renderer's own dialect, while a snapshots dict key is each
model's fqn, normalized under that model's own dialect. These can disagree
in casing when models use different dialects (e.g. a case-uppercasing
dialect like snowflake referenced from a case-insensitive one like duckdb),
causing the lookup to silently miss an existing snapshot and leave the
table name unmapped, even though the old full-mapping + exp.replace_tables
path (which reconciles casing per-dialect during matching) would have
resolved it correctly.

_resolve_table now falls back to building the full mapping only when the
narrowed lookup misses and the name isn't in table_mapping either, so the
common same-dialect case stays O(1) while the rare cross-dialect miss still
gets exp.replace_tables' dialect-aware reconciliation.

Also adds tests for: the cross-dialect regression itself, table_mapping-only
resolution with no snapshots, the non-string exp.Expr branch (otherwise
unreachable from any real call site), expand-then-find-Table ordering in
_resolve_tables, a table reference appearing only inside a string literal,
and deployability_index handling through the narrowed path.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01DzEtt434q32KhtoGDtD424
Signed-off-by: Michael Day <mdaytn@gmail.com>

Author: mday-io

sqlmesh/core/renderer.py

@@ -338,10 +338,27 @@ def _resolve_table(
             # environment - building the full mapping made this call O(N) in the number of
             # snapshots in the environment for every table resolved.
             snapshot = snapshots.get(table_name) if snapshots else None
-            mapping = {
-                **self._to_table_mapping([snapshot] if snapshot else [], deployability_index),
-                **({table_name: table_mapping[table_name]} if table_name in table_mapping else {}),
-            }
+            if snapshot is None and snapshots and table_name not in table_mapping:
+                # table_name is normalized under this renderer's own dialect, but a snapshot's
+                # fqn (the snapshots dict key) is normalized under that model's own dialect -
+                # these can disagree in casing when models use different dialects (e.g. a
+                # case-uppercasing dialect referenced from a case-insensitive one). A direct
+                # dict lookup can miss in that case even though the table is present, so fall
+                # back to the full, dialect-reconciling mapping that exp.replace_tables itself
+                # performs. This only pays the O(N) cost on a miss, not on every resolution.
+                mapping = {
+                    **self._to_table_mapping(snapshots.values(), deployability_index),
+                    **table_mapping,
+                }
+            else:
+                mapping = {
+                    **self._to_table_mapping([snapshot] if snapshot else [], deployability_index),
+                    **(
+                        {table_name: table_mapping[table_name]}
+                        if table_name in table_mapping
+                        else {}
+                    ),
+                }
         else:
             mapping = {
                 **self._to_table_mapping((snapshots or {}).values(), deployability_index),

tests/core/test_model.py

@@ -9864,6 +9864,246 @@ def test_render_virtual_properties_skips_mapping_without_table_refs(
         assert len(call.args[0]) <= 1
 
 
+def test_resolve_table_cross_dialect_fqn_mismatch(make_snapshot: t.Callable):
+    """`_resolve_table`'s narrowed lookup keys `snapshots` by the caller's already-normalized
+    `table_name` string. That string is built with the *referencing* model's own dialect
+    (`self._dialect` in the `resolve_table` macro closure), while the entry in `snapshots` is
+    keyed by the *referenced* model's fqn, which is normalized using that model's own dialect.
+
+    When the two models use dialects with different identifier-casing rules (e.g. a
+    case-insensitive dialect like duckdb referencing a model whose fqn was computed under a
+    case-uppercasing dialect like snowflake), the raw string lookup can miss even though
+    `exp.replace_tables`'s own (dialect-aware) matching -- which is what ran before this
+    optimization, and which the narrowed lookup's own final `exp.replace_tables` call still
+    performs when the key IS found -- would have matched them.
+    """
+
+    @macro()
+    def resolve_named(evaluator, name):
+        return evaluator.resolve_table(name.name)
+
+    # parent is declared/rendered under snowflake, which uppercases unquoted identifiers, so its
+    # fqn (the key that will appear in `snapshots`) is uppercase-quoted.
+    parent = load_sql_based_model(
+        d.parse("MODEL (name parent); SELECT 1 AS c"), dialect="snowflake"
+    )
+    parent_snapshot = make_snapshot(parent)
+    parent_snapshot.categorize_as(SnapshotChangeCategory.BREAKING)
+    assert parent.fqn == '"PARENT"'
+
+    # child is declared/rendered under duckdb (case-insensitive), referencing `parent` in lowercase
+    child = load_sql_based_model(
+        d.parse(
+            """
+            MODEL (name child);
+            SELECT c FROM parent;
+            @resolve_named('parent')
+            """
+        ),
+        dialect="duckdb",
+    )
+
+    snapshots = {parent.fqn: parent_snapshot}
+    post_statements = child.render_post_statements(snapshots=snapshots)
+
+    assert len(post_statements) == 1
+    resolved_sql = post_statements[0].sql()
+    # BUG: if this assertion fails with the resolved name still literally "parent" (unmapped)
+    # instead of the physical table name, the narrowed single-snapshot lookup in `_resolve_table`
+    # failed to find `parent` in `snapshots` due to the cross-dialect casing mismatch between the
+    # lookup key and the dict key, even though the table legitimately exists in `snapshots`.
+    assert resolved_sql == f'"sqlmesh__default"."parent__{parent_snapshot.version}"', (
+        f"expected parent to resolve to its physical table name, but got {resolved_sql!r} -- "
+        "this indicates the narrowed snapshots.get(table_name) lookup in _resolve_table missed "
+        "a snapshot that the old full-mapping + exp.replace_tables path would have matched"
+    )
+
+
+def test_resolve_table_table_mapping_only_no_snapshots(make_snapshot: t.Callable):
+    """A `table_mapping` entry with no corresponding `snapshots` entry should still be honored
+    by the narrowed lookup in `_resolve_table` (mirrors the override case in
+    `test_resolve_table_large_environment`, but with `snapshots=None`/empty entirely, to make
+    sure the narrowed code path doesn't assume `snapshots` is non-empty before consulting
+    `table_mapping`)."""
+
+    @macro()
+    def resolve_named(evaluator, name):
+        return evaluator.resolve_table(name.name)
+
+    child = load_sql_based_model(
+        d.parse(
+            """
+            MODEL (name child);
+            SELECT 1 AS c;
+            @resolve_named('parent')
+            """
+        )
+    )
+
+    post_statements = child.render_post_statements(
+        snapshots=None, table_mapping={'"parent"': "explicit_physical_table"}
+    )
+    assert post_statements[0].sql() == '"explicit_physical_table"'
+
+
+def test_resolve_table_non_string_expr_path(make_snapshot: t.Callable):
+    """When `table_name` is an `exp.Expr` (not a `str`), `_resolve_table` falls back to building
+    the full snapshot mapping (the `else` branch of the new code). This exercises that branch --
+    which the `this_model`/`resolve_table` macro call sites never hit, since they always pass a
+    pre-normalized string -- directly at the renderer level, to make sure it's still reachable
+    and correct, and not dead code that silently bit-rots."""
+
+    from sqlmesh.core.renderer import ExpressionRenderer
+
+    parent = load_sql_based_model(d.parse("MODEL (name parent); SELECT 1 AS c"))
+    parent_snapshot = make_snapshot(parent)
+    parent_snapshot.categorize_as(SnapshotChangeCategory.BREAKING)
+
+    other = load_sql_based_model(d.parse("MODEL (name other); SELECT 1 AS c"))
+    other_snapshot = make_snapshot(other)
+    other_snapshot.categorize_as(SnapshotChangeCategory.BREAKING)
+
+    expr_renderer = ExpressionRenderer(
+        exp.select("*"),
+        dialect="",
+        macro_definitions=[],
+        path=Path("."),
+    )
+
+    table_expr = exp.to_table('"parent"')
+    resolved = expr_renderer._resolve_table(
+        table_expr,
+        snapshots={'"parent"': parent_snapshot, '"other"': other_snapshot},
+    )
+    assert (
+        resolved.sql(comments=False)
+        == f'"sqlmesh__default"."parent__{parent_snapshot.version}"'
+    )
+
+
+def test_resolve_tables_table_ref_only_in_string_literal_not_expanded(make_snapshot: t.Callable):
+    """Adversarial case for the `expression.find(exp.Table)` short-circuit in `_resolve_tables`:
+    an expression that references a table only inside a string literal (not a parsed `exp.Table`
+    node) has no `exp.Table` node for `find()` to see, so the mapping build is correctly skipped.
+    This documents/locks in that the short-circuit is safe because `exp.replace_tables` itself
+    only ever rewrites `exp.Table` nodes -- it would never have touched a string literal either,
+    mapping built or not -- so skipping the mapping cannot change behavior here."""
+
+    parent = load_sql_based_model(d.parse("MODEL (name parent); SELECT 1 AS c"))
+    parent_snapshot = make_snapshot(parent)
+    parent_snapshot.categorize_as(SnapshotChangeCategory.BREAKING)
+
+    model = load_sql_based_model(
+        d.parse(
+            """
+            MODEL (
+                name test_schema.string_ref_model,
+                virtual_properties (
+                    description = 'references parent as a plain string, not a table node'
+                ),
+            );
+            SELECT a FROM tbl;
+            """
+        )
+    )
+
+    snapshots = {'"parent"': parent_snapshot}
+    props = model.render_virtual_properties(snapshots=snapshots)
+    assert (
+        props["description"].this
+        == "references parent as a plain string, not a table node"
+    )
+
+
+def test_resolve_tables_expand_reveals_table_after_find_check(make_snapshot: t.Callable):
+    """Embedded-model expansion (`expand=`) runs as an `expression.transform` *before* the new
+    `expression.find(exp.Table)` short-circuit in `_resolve_tables`, so a table reference that
+    only exists *after* inlining an embedded model's query must still be seen by `find()` and
+    mapped. This locks in that ordering: `grandparent` is not a literal `exp.Table` node in
+    `child`'s original query -- it only appears once the embedded `mid` model is expanded -- and
+    must still resolve to its physical table name, not be silently skipped because it wasn't
+    present at the time `_resolve_tables` was first called."""
+
+    grandparent = load_sql_based_model(d.parse("MODEL (name grandparent); SELECT 1 AS c"))
+    grandparent_snapshot = make_snapshot(grandparent)
+    grandparent_snapshot.categorize_as(SnapshotChangeCategory.BREAKING)
+
+    mid = load_sql_based_model(
+        d.parse("MODEL (name mid, kind EMBEDDED); SELECT c FROM grandparent;")
+    )
+    mid_snapshot = make_snapshot(mid)
+    mid_snapshot.categorize_as(SnapshotChangeCategory.BREAKING)
+
+    child = load_sql_based_model(d.parse("MODEL (name child); SELECT c FROM mid;"))
+
+    snapshots = {'"grandparent"': grandparent_snapshot, '"mid"': mid_snapshot}
+    query = child.render_query(snapshots=snapshots)
+    assert query is not None
+    rendered_sql = query.sql()
+
+    # the physical table name for `grandparent` must appear -- if the find(exp.Table) check had
+    # run before expansion (or expansion didn't feed into it), `grandparent` would remain
+    # unmapped in the rendered output.
+    assert f"grandparent__{grandparent_snapshot.version}" in rendered_sql
+    assert "FROM grandparent" not in rendered_sql
+
+
+def test_resolve_table_deployability_index_consistency(make_snapshot: t.Callable):
+    """The narrowed `_resolve_table` single-snapshot mapping must respect `deployability_index`
+    identically to the full-mapping path: a non-deployable (dev-preview) snapshot should map to
+    its dev table, not its deployable/prod table.
+
+    A snapshot's dev table only differs from its prod table when `dev_version_` differs from
+    `version` (see `Snapshot._table_name`); that normally arises from a forward-only change
+    against a previous version. `SnapshotChangeCategory.FORWARD_ONLY` is deprecated/blocked by
+    `categorize_as`, so this sets `dev_version_` directly to force that condition deterministically
+    without relying on a deprecated code path.
+    """
+    from sqlmesh.core.snapshot import DeployabilityIndex
+
+    parent = load_sql_based_model(
+        d.parse("MODEL (name parent); SELECT 1 AS c"),
+        dialect="duckdb",
+    )
+    parent_snapshot = make_snapshot(parent)
+    parent_snapshot.categorize_as(SnapshotChangeCategory.BREAKING)
+    parent_snapshot.dev_version_ = "customdevversion123"
+    assert parent_snapshot.table_name(is_deployable=True) != parent_snapshot.table_name(
+        is_deployable=False
+    )
+
+    @macro()
+    def resolve_named(evaluator, name):
+        return evaluator.resolve_table(name.name)
+
+    child_sql = """
+        MODEL (name child);
+        SELECT 1 AS c;
+        @resolve_named('parent')
+        """
+
+    snapshots = {parent.fqn: parent_snapshot}
+
+    # separate model instances per render call so the statement-render cache (keyed independent
+    # of `deployability_index`) doesn't just return the first call's cached result.
+    deployable_result = load_sql_based_model(d.parse(child_sql)).render_post_statements(
+        snapshots=snapshots, deployability_index=DeployabilityIndex.all_deployable()
+    )[0].sql()
+    non_deployable_result = load_sql_based_model(d.parse(child_sql)).render_post_statements(
+        snapshots=snapshots,
+        deployability_index=DeployabilityIndex.all_deployable().with_non_deployable(
+            parent_snapshot
+        ),
+    )[0].sql()
+
+    # the narrowed single-snapshot mapping must still pick the right table for each index.
+    assert deployable_result != non_deployable_result
+    assert parent_snapshot.table_name(is_deployable=True) in deployable_result.replace('"', "")
+    assert parent_snapshot.table_name(is_deployable=False) in non_deployable_result.replace(
+        '"', ""
+    )
+
+
 def test_cluster_with_complex_expression():
     expressions = d.parse(
         """