diff --git a/sqlmesh/core/renderer.py b/sqlmesh/core/renderer.py index 9f403cbcb4..a2660a435e 100644 --- a/sqlmesh/core/renderer.py +++ b/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, @@ -325,20 +384,35 @@ def update_cache(self, expression: t.Optional[exp.Expr]) -> None: def _resolve_table( self, - table_name: str | exp.Expr, + table_name: str, snapshots: t.Optional[t.Dict[str, Snapshot]] = None, table_mapping: t.Optional[t.Dict[str, str]] = None, deployability_index: t.Optional[DeployabilityIndex] = None, ) -> exp.Table: - table = exp.replace_tables( - t.cast(exp.Table, exp.maybe_parse(table_name, into=exp.Table, dialect=self._dialect)), - { - **self._to_table_mapping((snapshots or {}).values(), deployability_index), - **(table_mapping or {}), - }, - dialect=self._dialect, - copy=False, + 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) + # 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(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 ( @@ -363,6 +437,11 @@ def _resolve_tables( expression = expression.copy() with self._normalize_and_quote(expression) as expression: + # An expression with no table (e.g. most session or virtual properties) has nothing + # to expand or replace, so skip building the O(N) expand set and mapping. + if not expression.find(exp.Table): + return expression + snapshots = snapshots or {} table_mapping = table_mapping or {} mapping = { diff --git a/sqlmesh/core/snapshot/definition.py b/sqlmesh/core/snapshot/definition.py index 0c9635a7c2..2d813a06b1 100644 --- a/sqlmesh/core/snapshot/definition.py +++ b/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( diff --git a/sqlmesh/core/snapshot/evaluator.py b/sqlmesh/core/snapshot/evaluator.py index ad935310f4..9d9ecf7672 100644 --- a/sqlmesh/core/snapshot/evaluator.py +++ b/sqlmesh/core/snapshot/evaluator.py @@ -312,6 +312,8 @@ def promote( self._get_virtual_data_objects(target_snapshots, environment_naming_info) deployability_index = deployability_index or DeployabilityIndex.all_deployable() + # Renderers look snapshots up by model name, not by SnapshotId. + snapshots_by_name = {s.name: s for s in (snapshots or {}).values()} with self.concurrent_context(): concurrent_apply_to_snapshots( target_snapshots, @@ -320,7 +322,7 @@ def promote( start=start, end=end, execution_time=execution_time, - snapshots=snapshots, + snapshots=snapshots_by_name, table_mapping=table_mapping, environment_naming_info=environment_naming_info, deployability_index=deployability_index, # type: ignore @@ -1260,7 +1262,7 @@ def _promote_snapshot( start: t.Optional[TimeLike] = None, end: t.Optional[TimeLike] = None, execution_time: t.Optional[TimeLike] = None, - snapshots: t.Optional[t.Dict[SnapshotId, Snapshot]] = None, + snapshots: t.Optional[t.Dict[str, Snapshot]] = None, table_mapping: t.Optional[t.Dict[str, str]] = None, ) -> None: if not snapshot.is_model: @@ -1299,9 +1301,9 @@ def _promote_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, **render_kwargs) + ) if on_complete is not None: on_complete(snapshot) diff --git a/tests/core/test_model.py b/tests/core/test_model.py index c9045c9506..c5d1a71bee 100644 --- a/tests/core/test_model.py +++ b/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 @@ -9748,6 +9749,357 @@ def resolve_parent(evaluator, name): assert post_statements[0].sql() == f'"main"."sqlmesh__schema"."schema__parent__{version}"' +def test_resolve_table_large_environment(make_snapshot: t.Callable, mocker: MockerFixture): + """`_resolve_table` should only build a mapping for the one table being resolved, not the + entire environment (https://github.com/SQLMesh/sqlmesh/issues/6017).""" + + @macro() + def resolve_named(evaluator, name): + return evaluator.resolve_table(name.name) + + target = load_sql_based_model(d.parse("MODEL (name target); SELECT 1 AS c")) + target_snapshot = make_snapshot(target) + target_snapshot.categorize_as(SnapshotChangeCategory.BREAKING) + + snapshots = {'"target"': target_snapshot} + for i in range(50): + other = load_sql_based_model(d.parse(f"MODEL (name other_{i}); SELECT 1 AS c")) + other_snapshot = make_snapshot(other) + other_snapshot.categorize_as(SnapshotChangeCategory.BREAKING) + snapshots[f'"other_{i}"'] = other_snapshot + + child = load_sql_based_model( + d.parse( + """ + MODEL (name child); + SELECT c FROM target; + @resolve_named('target') + """ + ) + ) + + spy = mocker.spy(exp, "replace_tables") + + post_statements = child.render_post_statements(snapshots=snapshots) + assert len(post_statements) == 1 + assert post_statements[0].sql() == f'"sqlmesh__default"."target__{target_snapshot.version}"' + + # every replace_tables call made while resolving the single `target` reference should only + # ever see that one mapping entry, not all 51 snapshots in the environment + for call in spy.call_args_list: + assert len(call.args[1]) <= 1 + + # a name absent from both snapshots and table_mapping resolves unchanged + unmapped = load_sql_based_model( + d.parse( + """ + MODEL (name unmapped_child); + SELECT 1 AS c; + @resolve_named('does_not_exist') + """ + ) + ) + unmapped_result = unmapped.render_post_statements(snapshots=snapshots) + 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, 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.""" + + @macro() + def resolve_named(evaluator, name): + return evaluator.resolve_table(name.name) + + 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) + + child = load_sql_based_model( + d.parse( + """ + MODEL (name child); + SELECT 1 AS c; + @resolve_named('parent') + """ + ) + ) + + table_mapping = {"parent": "override_table"} + if include_exact_mapping: + table_mapping = {parent.fqn: "earlier_table", **table_mapping} + + post_statements = child.render_post_statements( + snapshots={parent.fqn: parent_snapshot}, table_mapping=mapping_type(table_mapping) + ) + + assert post_statements[0].sql() == '"override_table"' + + +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. + """ + + # Use explicit per-model normalization settings so this regression is independent of + # mutable process-global SQLGlot dialect settings. + parent_dialect = "snowflake,normalization_strategy=uppercase" + child_dialect = "duckdb,normalization_strategy=case_insensitive" + + @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=parent_dialect + ) + 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=child_dialect, + ) + + 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_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_resolve_table_table_mapping_only_dialect_mismatch(make_snapshot: t.Callable): + """When `snapshots` is empty/None, `_resolve_table`'s narrowed lookup must still fall back to + the full, dialect-reconciling mapping on a miss - not just when `snapshots` is non-empty. + + `table_name` and a `table_mapping` key can be normalized under different dialects (e.g. a + unit-test `table_mapping` built from the project's dialect vs. a model's own dialect for the + macro-resolved name), so they can disagree in casing/quoting even though an entry for this + table exists. The old exp.replace_tables-based path reconciled this via its own + normalization; a raw `table_name in table_mapping` string-equality check does not. + """ + + @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('a.b') + """ + ) + ) + + # `table_mapping` key differs from the resolved name only in quoting - a raw dict lookup on + # `'"a"."b"'` would miss `'a.b'`, but exp.replace_tables' normalization matches them. + post_statements = child.render_post_statements(snapshots=None, table_mapping={"a.b": "c"}) + assert post_statements[0].sql(comments=False) == '"c"' + + +def test_resolve_tables_skips_expand_computation_without_table_refs( + make_snapshot: t.Callable, +): + """Rendering a table-less expression (e.g. `virtual_properties`) must skip building the + `expand` set and `model_mapping` entirely, not just the final mapping/replace_tables call - + both of those are themselves O(N) in the number of snapshots when any snapshot is embedded, + so doing them for an expression with no `exp.Table` node at all defeats the point of skipping + the mapping build.""" + + embedded = load_sql_based_model(d.parse("MODEL (name embedded, kind EMBEDDED); SELECT 1 AS c")) + embedded_snapshot = make_snapshot(embedded) + embedded_snapshot.categorize_as(SnapshotChangeCategory.BREAKING) + + class ItemsCountingDict(dict): + items_call_count = 0 + + def items(self): + ItemsCountingDict.items_call_count += 1 + return super().items() + + snapshots = ItemsCountingDict({embedded.fqn: embedded_snapshot}) + + model = load_sql_based_model( + d.parse( + """ + MODEL ( + name test_schema.test_model, + virtual_properties ( + labels = [('team', 'data')] + ), + ); + SELECT a FROM tbl; + """ + ) + ) + + assert model.render_virtual_properties(snapshots=snapshots) == { + "labels": exp.maybe_parse("[('team', 'data')]") + } + # `_resolve_tables` computing `expand` (which scans `snapshots.items()` for embedded + # snapshots) and `model_mapping` must not happen for a table-less expression, even though + # this environment has an embedded snapshot that would otherwise trigger both. + assert ItemsCountingDict.items_call_count == 0 + + +def test_table_mapping_normalized_keys(): + table_mapping = TableMapping({'"db"."a"': "view_a", "db.A": "view_a_upper"}) + + # Keys that normalize to the same name resolve to the last one, like exp.replace_tables. + duckdb_keys = table_mapping.normalized_keys("duckdb") + assert duckdb_keys == {"db.a": "db.A"} + # Normalization happens once per dialect, and each dialect gets its own normalization. + assert table_mapping.normalized_keys("duckdb") is duckdb_keys + assert table_mapping.normalized_keys("snowflake") == {"db.a": '"db"."a"', "DB.A": "db.A"} + + # 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( """ diff --git a/tests/core/test_snapshot_evaluator.py b/tests/core/test_snapshot_evaluator.py index 3950e0d53f..a9ca86496a 100644 --- a/tests/core/test_snapshot_evaluator.py +++ b/tests/core/test_snapshot_evaluator.py @@ -5656,3 +5656,113 @@ def test_grants_in_production_with_dev_only_vde( # Should still apply grants to physical table when target layer is ALL or PHYSICAL sync_grants_mock.assert_called_once() assert sync_grants_mock.call_args[0][1] == {"select": ["user1"], "insert": ["role1"]} + + +def test_promote_virtual_properties_see_snapshots_by_name(mocker: MockerFixture, make_snapshot): + """Promotion receives snapshots keyed by SnapshotId, but renderers expect them keyed by name. + Virtual properties must see the same name-keyed snapshots as `on_virtual_update` does.""" + + @macro() + def upstream_version(evaluator): + upstream = evaluator.snapshots.get('"test_schema"."upstream"') + return exp.Literal.string(upstream.version if upstream else "missing") + + @macro() + def local_or_missing(evaluator, name): + value = evaluator.locals.get(name.name) + if isinstance(value, exp.Expr): + value = value.sql(evaluator.dialect, comments=False) + return exp.Literal.string(value or "missing") + + adapter_mock = mocker.patch("sqlmesh.core.engine_adapter.EngineAdapter") + adapter_mock.dialect = "duckdb" + adapter_mock.with_settings.return_value = adapter_mock + evaluator = SnapshotEvaluator(adapter_mock) + + upstream = load_sql_based_model( + d.parse("MODEL (name test_schema.upstream, kind FULL); SELECT 1 AS a") + ) + upstream_snapshot = make_snapshot(upstream) + upstream_snapshot.categorize_as(SnapshotChangeCategory.BREAKING) + + model = load_sql_based_model( + d.parse( + """ + MODEL ( + name test_schema.test_model, + kind FULL, + virtual_properties ( + upstream_version = @upstream_version(), + kind_name = @local_or_missing('model_kind_name'), + this_view = @local_or_missing('this_model'), + ), + ); + SELECT a FROM test_schema.upstream + """ + ) + ) + snapshot = make_snapshot(model, nodes={upstream.fqn: upstream}) + snapshot.categorize_as(SnapshotChangeCategory.BREAKING) + + snapshots = {s.snapshot_id: s for s in (upstream_snapshot, snapshot)} + environment_naming_info = EnvironmentNamingInfo(name="test_env") + evaluator.promote( + [snapshot], + environment_naming_info, + snapshots=snapshots, + table_mapping=to_view_mapping(snapshots.values(), environment_naming_info), + ) + + view_properties = adapter_mock.create_view.call_args.kwargs["view_properties"] + assert view_properties["upstream_version"] == exp.Literal.string(upstream_snapshot.version) + assert view_properties["kind_name"] == exp.Literal.string("FULL") + # The environment view mapping still takes precedence over the physical table. + assert view_properties["this_view"] == exp.Literal.string( + '"test_schema__test_env"."test_model"' + ) + + +def test_promote_resolves_this_model_with_single_mapping_entry( + mocker: MockerFixture, make_snapshot +): + """Rendering a promoted view's properties must not re-normalize the whole environment's view + mapping for every view (https://github.com/SQLMesh/sqlmesh/issues/6017).""" + adapter_mock = mocker.patch("sqlmesh.core.engine_adapter.EngineAdapter") + adapter_mock.dialect = "duckdb" + adapter_mock.with_settings.return_value = adapter_mock + evaluator = SnapshotEvaluator(adapter_mock) + + snapshots = {} + for i in range(20): + model = load_sql_based_model( + d.parse( + f""" + MODEL ( + name test_schema.model_{i}, + kind FULL, + virtual_properties (description = 'model {i}'), + ); + SELECT 1 AS a + """ + ) + ) + snapshot = make_snapshot(model) + snapshot.categorize_as(SnapshotChangeCategory.BREAKING) + snapshots[snapshot.snapshot_id] = snapshot + + environment_naming_info = EnvironmentNamingInfo(name="test_env") + table_mapping = to_view_mapping(snapshots.values(), environment_naming_info) + spy = mocker.spy(exp, "replace_tables") + + evaluator.promote( + list(snapshots.values()), + environment_naming_info, + snapshots=snapshots, + table_mapping=table_mapping, + ) + + assert adapter_mock.create_view.call_count == 20 + # One call per view, to resolve `this_model`. + assert spy.call_count == 20 + for call in spy.call_args_list: + assert len(call.args[1]) == 1