Do not pass all snapshots to _demote_snapshot

Author: treysp

sqlmesh/core/plan/evaluator.py

@@ -342,7 +342,6 @@ def visit_virtual_layer_update_stage(
                 self._demote_snapshots(
                     [stage.all_snapshots[s.snapshot_id] for s in stage.demoted_snapshots],
                     stage.demoted_environment_naming_info,
-                    snapshots=stage.all_snapshots,
                     deployability_index=stage.deployability_index,
                     on_complete=lambda s: self.console.update_promotion_progress(s, False),
                 )
@@ -386,14 +385,12 @@ def _demote_snapshots(
         self,
         target_snapshots: t.Iterable[Snapshot],
         environment_naming_info: EnvironmentNamingInfo,
-        snapshots: t.Dict[SnapshotId, Snapshot],
         deployability_index: t.Optional[DeployabilityIndex] = None,
         on_complete: t.Optional[t.Callable[[SnapshotInfoLike], None]] = None,
     ) -> None:
         self.snapshot_evaluator.demote(
             target_snapshots,
             environment_naming_info,
-            snapshots=snapshots,
             deployability_index=deployability_index,
             on_complete=on_complete,
         )

sqlmesh/core/snapshot/evaluator.py

@@ -293,7 +293,7 @@ def demote(
             concurrent_apply_to_snapshots(
                 target_snapshots,
                 lambda s: self._demote_snapshot(
-                    s, environment_naming_info, snapshots, deployability_index, on_complete
+                    s, environment_naming_info, deployability_index, on_complete
                 ),
                 self.ddl_concurrent_tasks,
             )
@@ -974,42 +974,44 @@ def _promote_snapshot(
         snapshots: t.Optional[t.Dict[SnapshotId, Snapshot]] = None,
         table_mapping: t.Optional[t.Dict[str, str]] = None,
     ) -> None:
-        if snapshot.is_model:
-            adapter = (
-                self.get_adapter(snapshot.model_gateway)
-                if environment_naming_info.gateway_managed
-                else self.adapter
-            )
-            table_name = snapshot.table_name(deployability_index.is_representative(snapshot))
-            view_name = snapshot.qualified_view_name.for_environment(
-                environment_naming_info, dialect=adapter.dialect
-            )
-            render_kwargs: t.Dict[str, t.Any] = dict(
-                start=start,
-                end=end,
-                execution_time=execution_time,
-                engine_adapter=adapter,
-                deployability_index=deployability_index,
-                table_mapping=table_mapping,
-                runtime_stage=RuntimeStage.PROMOTING,
-            )
+        if not snapshot.is_model:
+            return
 
-            with (
-                adapter.transaction(),
-                adapter.session(snapshot.model.render_session_properties(**render_kwargs)),
-            ):
-                _evaluation_strategy(snapshot, adapter).promote(
-                    table_name=table_name,
-                    view_name=view_name,
-                    model=snapshot.model,
-                    environment=environment_naming_info.name,
-                    snapshots=snapshots,
-                    **render_kwargs,
-                )
+        adapter = (
+            self.get_adapter(snapshot.model_gateway)
+            if environment_naming_info.gateway_managed
+            else self.adapter
+        )
+        table_name = snapshot.table_name(deployability_index.is_representative(snapshot))
+        view_name = snapshot.qualified_view_name.for_environment(
+            environment_naming_info, dialect=adapter.dialect
+        )
+        render_kwargs: t.Dict[str, t.Any] = dict(
+            start=start,
+            end=end,
+            execution_time=execution_time,
+            engine_adapter=adapter,
+            deployability_index=deployability_index,
+            table_mapping=table_mapping,
+            runtime_stage=RuntimeStage.PROMOTING,
+        )
+
+        with (
+            adapter.transaction(),
+            adapter.session(snapshot.model.render_session_properties(**render_kwargs)),
+        ):
+            _evaluation_strategy(snapshot, adapter).promote(
+                table_name=table_name,
+                view_name=view_name,
+                model=snapshot.model,
+                environment=environment_naming_info.name,
+                snapshots=snapshots,
+                **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))
+            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))
 
         if on_complete is not None:
             on_complete(snapshot)
@@ -1018,34 +1020,34 @@ def _demote_snapshot(
         self,
         snapshot: Snapshot,
         environment_naming_info: EnvironmentNamingInfo,
-        snapshots: t.Optional[t.Dict[SnapshotId, Snapshot]],
         deployability_index: t.Optional[DeployabilityIndex],
         on_complete: t.Optional[t.Callable[[SnapshotInfoLike], None]],
     ) -> None:
-        if snapshot.is_model:
-            adapter = (
-                self.get_adapter(snapshot.model_gateway)
-                if environment_naming_info.gateway_managed
-                else self.adapter
-            )
-            view_name = snapshot.qualified_view_name.for_environment(
-                environment_naming_info, dialect=adapter.dialect
-            )
-            with (
-                adapter.transaction(),
-                adapter.session(
-                    snapshot.model.render_session_properties(
-                        engine_adapter=adapter,
-                        snapshots=snapshots,
-                        deployability_index=deployability_index,
-                        runtime_stage=RuntimeStage.DEMOTING,
-                    )
-                ),
-            ):
-                _evaluation_strategy(snapshot, adapter).demote(view_name)
+        if not snapshot.is_model:
+            return
 
-            if on_complete is not None:
-                on_complete(snapshot)
+        adapter = (
+            self.get_adapter(snapshot.model_gateway)
+            if environment_naming_info.gateway_managed
+            else self.adapter
+        )
+        view_name = snapshot.qualified_view_name.for_environment(
+            environment_naming_info, dialect=adapter.dialect
+        )
+        with (
+            adapter.transaction(),
+            adapter.session(
+                snapshot.model.render_session_properties(
+                    engine_adapter=adapter,
+                    deployability_index=deployability_index,
+                    runtime_stage=RuntimeStage.DEMOTING,
+                )
+            ),
+        ):
+            _evaluation_strategy(snapshot, adapter).demote(view_name)
+
+        if on_complete is not None:
+            on_complete(snapshot)
 
     def _cleanup_snapshot(
         self,