Compute table_mapping in plan evaluator

Author: treysp

sqlmesh/core/plan/evaluator.py

@@ -344,6 +344,7 @@ def visit_virtual_layer_update_stage(
                     stage.demoted_environment_naming_info,
                     deployability_index=stage.deployability_index,
                     on_complete=lambda s: self.console.update_promotion_progress(s, False),
+                    snapshots=stage.all_snapshots,
                 )
 
             completed = True
@@ -385,12 +386,19 @@ 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,
+            table_mapping=to_view_mapping(
+                snapshots.values(),
+                environment_naming_info,
+                default_catalog=self.default_catalog,
+                dialect=self.snapshot_evaluator.adapter.dialect,
+            ),
             deployability_index=deployability_index,
             on_complete=on_complete,
         )

sqlmesh/core/snapshot/evaluator.py

@@ -278,7 +278,7 @@ def demote(
         self,
         target_snapshots: t.Iterable[Snapshot],
         environment_naming_info: EnvironmentNamingInfo,
-        snapshots: t.Optional[t.Dict[SnapshotId, Snapshot]] = None,
+        table_mapping: t.Optional[t.Dict[str, str]] = None,
         deployability_index: t.Optional[DeployabilityIndex] = None,
         on_complete: t.Optional[t.Callable[[SnapshotInfoLike], None]] = None,
     ) -> None:
@@ -293,7 +293,11 @@ def demote(
             concurrent_apply_to_snapshots(
                 target_snapshots,
                 lambda s: self._demote_snapshot(
-                    s, environment_naming_info, deployability_index, on_complete
+                    s,
+                    environment_naming_info,
+                    deployability_index=deployability_index,
+                    on_complete=on_complete,
+                    table_mapping=table_mapping,
                 ),
                 self.ddl_concurrent_tasks,
             )
@@ -1022,6 +1026,7 @@ def _demote_snapshot(
         environment_naming_info: EnvironmentNamingInfo,
         deployability_index: t.Optional[DeployabilityIndex],
         on_complete: t.Optional[t.Callable[[SnapshotInfoLike], None]],
+        table_mapping: t.Optional[t.Dict[str, str]] = None,
     ) -> None:
         if not snapshot.is_model:
             return
@@ -1040,6 +1045,7 @@ def _demote_snapshot(
                 snapshot.model.render_session_properties(
                     engine_adapter=adapter,
                     deployability_index=deployability_index,
+                    table_mapping=table_mapping,
                     runtime_stage=RuntimeStage.DEMOTING,
                 )
             ),