Repository: TobikoData/sqlmesh
Pull request #2513 (2024-04-24 23:27:09)

fix: action bot properly output all changes
Prior to this change the bot displayed changed by using the output generated when categorizing changes which would give you the directly modified snapshot and then the indirectly modified models in a subtree. The problem is that this doesn't cover all possible changes and therefore removed snapshots (and likely other things) were not being represented.

This change uses the normal output that is created instead which captures everything. This was actually already written but not used since we switched to the above approach so it just needed updated. I tested all the new output that is generated against this to make sure it formats properly: https://jbt.github.io/markdown-editor/

The main regression here is that users no longer see the directly modified and then below it what was indirectly modified and instead they are listed in separate lists. Will consider improving this based on user feedback. 


Patch:
diff --git a/docs/integrations/github.md b/docs/integrations/github.md
index d3525b27..7af82c03 100644
--- a/docs/integrations/github.md
+++ b/docs/integrations/github.md
@@ -243,7 +243,6 @@ jobs:
       uses: actions/checkout@v4
       with:
         ref: main
-        path: wursthall
     ...
     - name: Run CI/CD Bot
       # Add if statement so we don't run-all on merged PR
@@ -392,8 +391,9 @@ Commands:
   update-pr-environment     Creates or updates the PR environments
 ```
 
-## Example Full Workflow
-This workflow involves configuring a SQLMesh connection to Databricks and configuring access to GCP to talk to Cloud Composer (Airflow)
+## Example Synchronized Full Workflow
+This workflow involves configuring a SQLMesh connection to Databricks and configuring access to GCP to talk to Cloud Composer (Airflow).
+
 ```yaml
 name: SQLMesh Bot
 run-name: 🚀SQLMesh Bot 🚀
diff --git a/sqlmesh/core/console.py b/sqlmesh/core/console.py
index 765744b3..a42a604b 100644
--- a/sqlmesh/core/console.py
+++ b/sqlmesh/core/console.py
@@ -1406,16 +1406,16 @@ class MarkdownConsole(CaptureTerminalConsole):
         ignored_snapshot_ids = ignored_snapshot_ids or set()
         if context_diff.is_new_environment:
             self._print(
-                f"**New environment `{context_diff.environment}` will be created from `{context_diff.create_from}`**\n\n"
+                f"**New environment `{context_diff.environment}` will be created from `{context_diff.create_from}`**\n"
             )
             if not context_diff.has_snapshot_changes:
                 return
 
         if not context_diff.has_changes:
-            self._print(f"**No differences when compared to `{context_diff.environment}`**\n\n")
+            self._print(f"**No differences when compared to `{context_diff.environment}`**\n")
             return
 
-        self._print(f"**Summary of differences against `{context_diff.environment}`:**\n\n")
+        self._print(f"**Summary of differences against `{context_diff.environment}`:**\n")
 
         added_snapshots = {
             context_diff.snapshots[s_id]
@@ -1424,21 +1424,19 @@ class MarkdownConsole(CaptureTerminalConsole):
         }
         added_snapshot_models = {s for s in added_snapshots if s.is_model}
         if added_snapshot_models:
-            self._print(f"**Added Models:**\n")
-            for snapshot in added_snapshot_models:
+            self._print(f"\n**Added Models:**")
+            for snapshot in sorted(added_snapshot_models):
                 self._print(
-                    f"- {snapshot.display_name(environment_naming_info, default_catalog)}\n"
+                    f"- `{snapshot.display_name(environment_naming_info, default_catalog)}`"
                 )
-            self._print("\n")
 
         added_snapshot_audits = {s for s in added_snapshots if s.is_audit}
         if added_snapshot_audits:
-            self._print(f"**Added Standalone Audits:**\n")
-            for snapshot in added_snapshot_audits:
+            self._print(f"\n**Added Standalone Audits:**")
+            for snapshot in sorted(added_snapshot_audits):
                 self._print(
-                    f"- {snapshot.display_name(environment_naming_info, default_catalog)}\n"
+                    f"- `{snapshot.display_name(environment_naming_info, default_catalog)}`"
                 )
-            self._print("\n")
 
         removed_snapshot_table_infos = {
             snapshot_table_info
@@ -1447,21 +1445,19 @@ class MarkdownConsole(CaptureTerminalConsole):
         }
         removed_model_snapshot_table_infos = {s for s in removed_snapshot_table_infos if s.is_model}
         if removed_model_snapshot_table_infos:
-            self._print(f"**Removed Models:**\n")
-            for snapshot_table_info in removed_model_snapshot_table_infos:
+            self._print(f"\n**Removed Models:**")
+            for snapshot_table_info in sorted(removed_model_snapshot_table_infos):
                 self._print(
-                    f"- {snapshot_table_info.display_name(environment_naming_info, default_catalog)}\n"
+                    f"- `{snapshot_table_info.display_name(environment_naming_info, default_catalog)}`"
                 )
-            self._print("\n")
 
         removed_audit_snapshot_table_infos = {s for s in removed_snapshot_table_infos if s.is_audit}
         if removed_audit_snapshot_table_infos:
-            self._print(f"**Removed Standalone Audits:**\n")
-            for snapshot_table_info in removed_audit_snapshot_table_infos:
+            self._print(f"\n**Removed Standalone Audits:**")
+            for snapshot_table_info in sorted(removed_audit_snapshot_table_infos):
                 self._print(
-                    f"- {snapshot_table_info.display_name(environment_naming_info, default_catalog)}\n"
+                    f"- `{snapshot_table_info.display_name(environment_naming_info, default_catalog)}`"
                 )
-            self._print("\n")
 
         modified_snapshots = {
             current_snapshot
@@ -1480,43 +1476,39 @@ class MarkdownConsole(CaptureTerminalConsole):
                 elif context_diff.metadata_updated(snapshot.name):
                     metadata_modified.append(snapshot)
             if directly_modified:
-                self._print(f"**Directly Modified:**\n")
-                for snapshot in directly_modified:
+                self._print(f"\n**Directly Modified:**")
+                for snapshot in sorted(directly_modified):
                     self._print(
-                        f"- `{snapshot.display_name(environment_naming_info, default_catalog)}`\n"
+                        f"- `{snapshot.display_name(environment_naming_info, default_catalog)}`"
                     )
                     if not no_diff:
-                        self._print(f"```diff\n{context_diff.text_diff(snapshot.name)}\n```\n")
-                self._print("\n")
+                        self._print(f"```diff\n{context_diff.text_diff(snapshot.name)}\n```")
             if indirectly_modified:
-                self._print(f"**Indirectly Modified:**\n")
-                for snapshot in indirectly_modified:
+                self._print(f"\n**Indirectly Modified:**")
+                for snapshot in sorted(indirectly_modified):
                     self._print(
-                        f"- `{snapshot.display_name(environment_naming_info, default_catalog)}`\n"
+                        f"- `{snapshot.display_name(environment_naming_info, default_catalog)}`"
                     )
-                self._print("\n")
             if metadata_modified:
-                self._print(f"**Metadata Updated:**\n")
-                for snapshot in metadata_modified:
+                self._print(f"\n**Metadata Updated:**")
+                for snapshot in sorted(metadata_modified):
                     self._print(
-                        f"- `{snapshot.display_name(environment_naming_info, default_catalog)}`\n"
+                        f"- `{snapshot.display_name(environment_naming_info, default_catalog)}`"
                     )
-                self._print("\n")
         if ignored_snapshot_ids:
-            self._print(f"**Ignored Models (Expected Plan Start):**\n")
-            for s_id in ignored_snapshot_ids:
+            self._print(f"\n**Ignored Models (Expected Plan Start):**")
+            for s_id in sorted(ignored_snapshot_ids):
                 snapshot = context_diff.snapshots[s_id]
                 self._print(
-                    f"- `{snapshot.display_name(environment_naming_info, default_catalog)}` ({snapshot.get_latest(start_date(snapshot, context_diff.snapshots.values()))})\n"
+                    f"- `{snapshot.display_name(environment_naming_info, default_catalog)}` ({snapshot.get_latest(start_date(snapshot, context_diff.snapshots.values()))})"
                 )
-            self._print("\n")
 
     def _show_missing_dates(self, plan: Plan, default_catalog: t.Optional[str]) -> None:
         """Displays the models with missing dates."""
         missing_intervals = plan.missing_intervals
         if not missing_intervals:
             return
-        self._print("**Models needing backfill (missing dates):**\n\n")
+        self._print("\n**Models needing backfill (missing dates):**")
         for missing in missing_intervals:
             snapshot = plan.context_diff.snapshots[missing.snapshot_id]
             if not snapshot.is_model:
@@ -1527,9 +1519,8 @@ class MarkdownConsole(CaptureTerminalConsole):
                 preview_modifier = " (**preview**)"
 
             self._print(
-                f"* `{snapshot.display_name(plan.environment_naming_info, default_catalog)}`: {missing.format_intervals(snapshot.node.interval_unit)}{preview_modifier}\n"
+                f"* `{snapshot.display_name(plan.environment_naming_info, default_catalog)}`: {missing.format_intervals(snapshot.node.interval_unit)}{preview_modifier}"
             )
-        self._print("\n")
 
     def _show_categorized_snapshots(self, plan: Plan, default_catalog: t.Optional[str]) -> None:
         context_diff = plan.context_diff
diff --git a/sqlmesh/core/engine_adapter/databricks.py b/sqlmesh/core/engine_adapter/databricks.py
index 56430d98..f20082e4 100644
--- a/sqlmesh/core/engine_adapter/databricks.py
+++ b/sqlmesh/core/engine_adapter/databricks.py
@@ -6,7 +6,6 @@ import typing as t
 import pandas as pd
 from sqlglot import exp
 
-from sqlmesh.core.engine_adapter.mixins import RenameTableFullTargetNameMixin
 from sqlmesh.core.engine_adapter.shared import (
     CatalogSupport,
     DataObject,
@@ -30,7 +29,7 @@ logger = logging.getLogger(__name__)
         "_get_data_objects": CatalogSupport.REQUIRES_SET_CATALOG,
     }
 )
-class DatabricksEngineAdapter(SparkEngineAdapter, RenameTableFullTargetNameMixin):
+class DatabricksEngineAdapter(SparkEngineAdapter):
     DIALECT = "databricks"
     INSERT_OVERWRITE_STRATEGY = InsertOverwriteStrategy.INSERT_OVERWRITE
     SUPPORTS_CLONING = True
diff --git a/sqlmesh/core/engine_adapter/mixins.py b/sqlmesh/core/engine_adapter/mixins.py
index 309f54fd..403229ed 100644
--- a/sqlmesh/core/engine_adapter/mixins.py
+++ b/sqlmesh/core/engine_adapter/mixins.py
@@ -246,18 +246,3 @@ class NonTransactionalTruncateMixin(EngineAdapter):
         if self._connection_pool.is_transaction_active:
             return self.execute(exp.Delete(this=exp.to_table(table_name)))
         super()._truncate_table(table_name)
-
-
-class RenameTableFullTargetNameMixin(EngineAdapter):
-    def _rename_table(
-        self,
-        old_table_name: TableName,
-        new_table_name: TableName,
-    ) -> None:
-        old_table = exp.to_table(old_table_name)
-        new_table = exp.to_table(new_table_name)
-        if not new_table.db and old_table.db:
-            # In MySQL and Snowflake you need to provide the full target table name.
-            # Therefore we use the old schema name if one is not provided explicitly in the new name.
-            new_table.set("db", old_table.args["db"])
-        return super()._rename_table(old_table, new_table)
diff --git a/sqlmesh/core/engine_adapter/mysql.py b/sqlmesh/core/engine_adapter/mysql.py
index 98b33986..51963ce0 100644
--- a/sqlmesh/core/engine_adapter/mysql.py
+++ b/sqlmesh/core/engine_adapter/mysql.py
@@ -10,7 +10,6 @@ from sqlmesh.core.engine_adapter.mixins import (
     LogicalMergeMixin,
     NonTransactionalTruncateMixin,
     PandasNativeFetchDFSupportMixin,
-    RenameTableFullTargetNameMixin,
 )
 from sqlmesh.core.engine_adapter.shared import (
     CommentCreationTable,
@@ -31,7 +30,6 @@ class MySQLEngineAdapter(
     LogicalMergeMixin,
     PandasNativeFetchDFSupportMixin,
     NonTransactionalTruncateMixin,
-    RenameTableFullTargetNameMixin,
 ):
     DEFAULT_BATCH_SIZE = 200
     DIALECT = "mysql"
diff --git a/sqlmesh/core/engine_adapter/snowflake.py b/sqlmesh/core/engine_adapter/snowflake.py
index 0982393b..f089cb0d 100644
--- a/sqlmesh/core/engine_adapter/snowflake.py
+++ b/sqlmesh/core/engine_adapter/snowflake.py
@@ -10,10 +10,7 @@ from sqlglot.optimizer.normalize_identifiers import normalize_identifiers
 from sqlglot.optimizer.qualify_columns import quote_identifiers
 
 from sqlmesh.core.dialect import to_schema
-from sqlmesh.core.engine_adapter.mixins import (
-    GetCurrentCatalogFromFunctionMixin,
-    RenameTableFullTargetNameMixin,
-)
+from sqlmesh.core.engine_adapter.mixins import GetCurrentCatalogFromFunctionMixin
 from sqlmesh.core.engine_adapter.shared import (
     CatalogSupport,
     DataObject,
@@ -35,7 +32,7 @@ if t.TYPE_CHECKING:
         "drop_schema": CatalogSupport.REQUIRES_SET_CATALOG,
     }
 )
-class SnowflakeEngineAdapter(GetCurrentCatalogFromFunctionMixin, RenameTableFullTargetNameMixin):
+class SnowflakeEngineAdapter(GetCurrentCatalogFromFunctionMixin):
     DIALECT = "snowflake"
     SUPPORTS_MATERIALIZED_VIEWS = True
     SUPPORTS_MATERIALIZED_VIEW_SCHEMA = True
diff --git a/sqlmesh/core/plan/definition.py b/sqlmesh/core/plan/definition.py
index 141b859e..2aad228c 100644
--- a/sqlmesh/core/plan/definition.py
+++ b/sqlmesh/core/plan/definition.py
@@ -17,7 +17,12 @@ from sqlmesh.core.snapshot import (
     merge_intervals,
     missing_intervals,
 )
-from sqlmesh.core.snapshot.definition import Interval, SnapshotId, format_intervals
+from sqlmesh.core.snapshot.definition import (
+    Interval,
+    SnapshotId,
+    SnapshotTableInfo,
+    format_intervals,
+)
 from sqlmesh.utils.date import TimeLike, now, to_datetime, to_timestamp
 from sqlmesh.utils.pydantic import PydanticModel
 
@@ -123,7 +128,7 @@ class Plan(PydanticModel, frozen=True):
         }
 
     @cached_property
-    def modified_snapshots(self) -> t.Dict[SnapshotId, Snapshot]:
+    def modified_snapshots(self) -> t.Dict[SnapshotId, t.Union[Snapshot, SnapshotTableInfo]]:
         """Returns the modified (either directly or indirectly) snapshots."""
         return {
             **{s_id: self.context_diff.snapshots[s_id] for s_id in sorted(self.directly_modified)},
@@ -132,6 +137,7 @@ class Plan(PydanticModel, frozen=True):
                 for downstream_s_ids in self.indirectly_modified.values()
                 for s_id in sorted(downstream_s_ids)
             },
+            **self.context_diff.removed_snapshots,
         }
 
     @property
diff --git a/sqlmesh/core/plan/evaluator.py b/sqlmesh/core/plan/evaluator.py
index 5a17ebbc..3458f3ac 100644
--- a/sqlmesh/core/plan/evaluator.py
+++ b/sqlmesh/core/plan/evaluator.py
@@ -99,9 +99,8 @@ class BuiltInPlanEvaluator(PlanEvaluator):
                 after_promote_snapshots = all_names - before_promote_snapshots
                 deployability_index_for_evaluation = DeployabilityIndex.all_deployable()
 
-            update_intervals_for_new_snapshots(plan.new_snapshots, self.state_sync)
-
             self._push(plan, deployability_index_for_creation)
+            update_intervals_for_new_snapshots(plan.new_snapshots, self.state_sync)
             self._restate(plan)
             self._backfill(
                 plan,
diff --git a/sqlmesh/core/renderer.py b/sqlmesh/core/renderer.py
index e94f6918..a0b6457c 100644
--- a/sqlmesh/core/renderer.py
+++ b/sqlmesh/core/renderer.py
@@ -100,12 +100,10 @@ class BaseExpressionRenderer:
             return self._cache
 
         if self._model_fqn and "this_model" not in kwargs:
-            if snapshots:
-                kwargs["this_model"] = self._to_table_mapping(
-                    [snapshots[self._model_fqn]], deployability_index
-                )
-            else:
-                kwargs["this_model"] = self._model_fqn
+            kwargs["this_model"] = self._to_table_mapping(
+                [snapshots[self._model_fqn]] if snapshots and self._model_fqn in snapshots else [],
+                deployability_index,
+            ).get(self._model_fqn, self._model_fqn)
 
         expressions = [self._expression]
 
diff --git a/sqlmesh/integrations/github/cicd/controller.py b/sqlmesh/integrations/github/cicd/controller.py
index 7c418ad1..8c242e53 100644
--- a/sqlmesh/integrations/github/cicd/controller.py
+++ b/sqlmesh/integrations/github/cicd/controller.py
@@ -22,7 +22,13 @@ from sqlmesh.core.console import SNAPSHOT_CHANGE_CATEGORY_STR, MarkdownConsole
 from sqlmesh.core.context import Context
 from sqlmesh.core.environment import Environment
 from sqlmesh.core.plan import Plan, PlanBuilder
-from sqlmesh.core.snapshot.definition import Snapshot, SnapshotId, format_intervals
+from sqlmesh.core.snapshot.definition import (
+    Snapshot,
+    SnapshotChangeCategory,
+    SnapshotId,
+    SnapshotTableInfo,
+    format_intervals,
+)
 from sqlmesh.core.user import User
 from sqlmesh.integrations.github.cicd.config import GithubCICDBotConfig
 from sqlmesh.utils import word_characters_only
@@ -426,9 +432,13 @@ class GithubController:
         return bot_config
 
     @property
-    def modified_snapshots(self) -> t.Dict[SnapshotId, Snapshot]:
+    def modified_snapshots(self) -> t.Dict[SnapshotId, t.Union[Snapshot, SnapshotTableInfo]]:
         return self.prod_plan_with_gaps.modified_snapshots
 
+    @property
+    def removed_snapshots(self) -> t.Set[SnapshotId]:
+        return set(self.prod_plan_with_gaps.context_diff.removed_snapshots)
+
     @classmethod
     def _append_output(cls, key: str, value: str) -> None:
         """
@@ -442,13 +452,19 @@ class GithubController:
         try:
             # Clear out any output that might exist from prior steps
             self._console.clear_captured_outputs()
-            self._console._show_categorized_snapshots(plan, self._context.default_catalog)
-            catagorized_snapshots = self._console.consume_captured_output()
+            self._console.show_model_difference_summary(
+                context_diff=plan.context_diff,
+                environment_naming_info=plan.environment_naming_info,
+                default_catalog=self._context.default_catalog,
+                no_diff=False,
+                ignored_snapshot_ids=plan.ignored,
+            )
+            difference_summary = self._console.consume_captured_output()
             self._console._show_missing_dates(plan, self._context.default_catalog)
             missing_dates = self._console.consume_captured_output()
-            if not catagorized_snapshots and not missing_dates:
+            if not difference_summary and not missing_dates:
                 return "No changes to apply."
-            return f"{catagorized_snapshots}\n{missing_dates}"
+            return f"{difference_summary}\n{missing_dates}"
         except PlanError as e:
             return f"Plan failed to generate. Check for pending or unresolved changes. Error: {e}"
 
@@ -759,22 +775,32 @@ class GithubController:
                         # We don't want to display indirect non-breaking since to users these are effectively no-op changes
                         if modified_snapshot.is_indirect_non_breaking:
                             continue
-                        model_name = modified_snapshot.node.name
-                        change_category = (
-                            "Uncategorized"
-                            if not modified_snapshot.change_category
-                            else SNAPSHOT_CHANGE_CATEGORY_STR[modified_snapshot.change_category]
-                        )
-                        intervals = (
-                            modified_snapshot.dev_intervals
-                            if modified_snapshot.is_forward_only
-                            else modified_snapshot.intervals
-                        )
-                        interval_output = (
-                            format_intervals(intervals, modified_snapshot.node.interval_unit)
-                            if intervals
-                            else "N/A"
-                        )
+                        if modified_snapshot.snapshot_id in self.removed_snapshots:
+                            # This will be an FQN since we don't have access to node name from a snapshot table info
+                            # which is what a removed snapshot is
+                            model_name = modified_snapshot.name
+                            change_category = SNAPSHOT_CHANGE_CATEGORY_STR[
+                                SnapshotChangeCategory.BREAKING
+                            ]
+                            interval_output = "REMOVED"
+                        else:
+                            assert isinstance(modified_snapshot, Snapshot)
+                            model_name = modified_snapshot.node.name
+                            change_category = (
+                                "Uncategorized"
+                                if not modified_snapshot.change_category
+                                else SNAPSHOT_CHANGE_CATEGORY_STR[modified_snapshot.change_category]
+                            )
+                            intervals = (
+                                modified_snapshot.dev_intervals
+                                if modified_snapshot.is_forward_only
+                                else modified_snapshot.intervals
+                            )
+                            interval_output = (
+                                format_intervals(intervals, modified_snapshot.node.interval_unit)
+                                if intervals
+                                else "N/A"
+                            )
                         body_rows.append(
                             [
                                 h("td", model_name),
@@ -865,8 +891,6 @@ class GithubController:
             title = conclusion_to_title.get(
                 conclusion, f"Got an unexpected conclusion: {conclusion.value}"
             )
-            if conclusion.is_success and summary:
-                summary = "**Preview of Prod Plan**\n" + summary
             return conclusion, title, summary
 
         self._update_check_handler(
diff --git a/sqlmesh/migrations/v0049_replace_identifier_with_version_in_seeds_table.py b/sqlmesh/migrations/v0049_replace_identifier_with_version_in_seeds_table.py
index 5b0fb958..186b5f78 100644
--- a/sqlmesh/migrations/v0049_replace_identifier_with_version_in_seeds_table.py
+++ b/sqlmesh/migrations/v0049_replace_identifier_with_version_in_seeds_table.py
@@ -54,4 +54,4 @@ def migrate(state_sync, **kwargs):  # type: ignore
 
     engine_adapter.insert_append(new_seeds_table, query)
     engine_adapter.drop_table(seeds_table)
-    engine_adapter.rename_table(new_seeds_table, "_seeds")
+    engine_adapter.rename_table(new_seeds_table, seeds_table)


Test patch:
diff --git a/tests/core/engine_adapter/test_databricks.py b/tests/core/engine_adapter/test_databricks.py
index ad946b0b..63308e43 100644
--- a/tests/core/engine_adapter/test_databricks.py
+++ b/tests/core/engine_adapter/test_databricks.py
@@ -103,18 +103,3 @@ def test_get_current_database(make_mocked_engine_adapter: t.Callable):
 
     assert adapter.get_current_database() == "test_database"
     assert to_sql_calls(adapter) == ["SELECT CURRENT_DATABASE()"]
-
-
-def test_rename_table(make_mocked_engine_adapter: t.Callable):
-    adapter = make_mocked_engine_adapter(DatabricksEngineAdapter)
-
-    adapter.rename_table("test_schema.old_name", "new_name")
-    adapter.rename_table("test_schema.old_name", "new_test_schema.new_name")
-    adapter.rename_table("old_name", "new_name")
-
-    sql_calls = to_sql_calls(adapter)
-    assert sql_calls == [
-        "ALTER TABLE `test_schema`.`old_name` RENAME TO `test_schema`.`new_name`",
-        "ALTER TABLE `test_schema`.`old_name` RENAME TO `new_test_schema`.`new_name`",
-        "ALTER TABLE `old_name` RENAME TO `new_name`",
-    ]
diff --git a/tests/core/engine_adapter/test_mysql.py b/tests/core/engine_adapter/test_mysql.py
index 348e5752..e0f07de1 100644
--- a/tests/core/engine_adapter/test_mysql.py
+++ b/tests/core/engine_adapter/test_mysql.py
@@ -62,18 +62,3 @@ def test_comments(make_mocked_engine_adapter: t.Callable, mocker: MockerFixture)
         f"ALTER TABLE `test_table` COMMENT = '{truncated_table_comment}'",
         f"ALTER TABLE `test_table` MODIFY `a` INT COMMENT '{truncated_column_comment}'",
     ]
-
-
-def test_rename_table(make_mocked_engine_adapter: t.Callable, mocker: MockerFixture):
-    adapter = make_mocked_engine_adapter(MySQLEngineAdapter)
-
-    adapter.rename_table("test_schema.old_name", "new_name")
-    adapter.rename_table("test_schema.old_name", "new_test_schema.new_name")
-    adapter.rename_table("old_name", "new_name")
-
-    sql_calls = to_sql_calls(adapter)
-    assert sql_calls == [
-        "ALTER TABLE `test_schema`.`old_name` RENAME TO `test_schema`.`new_name`",
-        "ALTER TABLE `test_schema`.`old_name` RENAME TO `new_test_schema`.`new_name`",
-        "ALTER TABLE `old_name` RENAME TO `new_name`",
-    ]
diff --git a/tests/core/engine_adapter/test_snowflake.py b/tests/core/engine_adapter/test_snowflake.py
index b8e5b87c..c2425b44 100644
--- a/tests/core/engine_adapter/test_snowflake.py
+++ b/tests/core/engine_adapter/test_snowflake.py
@@ -143,18 +143,3 @@ def test_comments(make_mocked_engine_adapter: t.Callable, mocker: MockerFixture)
         """COMMENT ON TABLE "test_table" IS 'table description'""",
         """ALTER TABLE "test_table" ALTER COLUMN "a" COMMENT 'a column description'""",
     ]
-
-
-def test_rename_table(make_mocked_engine_adapter: t.Callable, mocker: MockerFixture):
-    adapter = make_mocked_engine_adapter(SnowflakeEngineAdapter)
-
-    adapter.rename_table("test_schema.old_name", "new_name")
-    adapter.rename_table("test_schema.old_name", "new_test_schema.new_name")
-    adapter.rename_table("old_name", "new_name")
-
-    sql_calls = to_sql_calls(adapter)
-    assert sql_calls == [
-        'ALTER TABLE "test_schema"."old_name" RENAME TO "test_schema"."new_name"',
-        'ALTER TABLE "test_schema"."old_name" RENAME TO "new_test_schema"."new_name"',
-        'ALTER TABLE "old_name" RENAME TO "new_name"',
-    ]
diff --git a/tests/core/test_model.py b/tests/core/test_model.py
index f18dd875..46b47a02 100644
--- a/tests/core/test_model.py
+++ b/tests/core/test_model.py
@@ -36,7 +36,7 @@ from sqlmesh.core.model.common import parse_expression
 from sqlmesh.core.model.kind import _model_kind_validator
 from sqlmesh.core.model.seed import CsvSettings
 from sqlmesh.core.node import IntervalUnit, _Node
-from sqlmesh.core.snapshot import SnapshotChangeCategory
+from sqlmesh.core.snapshot import Snapshot, SnapshotChangeCategory
 from sqlmesh.utils.date import TimeLike, to_datetime, to_ds, to_timestamp
 from sqlmesh.utils.errors import ConfigError, SQLMeshError
 from sqlmesh.utils.jinja import JinjaMacroRegistry, MacroInfo
@@ -3810,6 +3810,15 @@ def test_this_model() -> None:
     assert model.render_pre_statements()[0].sql() == """COPY "db"."table" TO 'a'"""
     assert model.render_post_statements()[0].sql() == """COPY "db"."table" TO 'b'"""
 
+    snapshot = Snapshot.from_node(model, nodes={})
+    assert (
+        model.render_query_or_raise(
+            start="2020-01-01",
+            snapshots={snapshot.name: snapshot},
+        ).sql()
+        == '''SELECT '"db"."table"' AS "x"'''
+    )
+
 
 def test_macros_in_model_statement(sushi_context, assert_exp_eq):
     expressions = d.parse(
diff --git a/tests/integrations/github/cicd/test_github_commands.py b/tests/integrations/github/cicd/test_github_commands.py
index 503c5f81..1d480adb 100644
--- a/tests/integrations/github/cicd/test_github_commands.py
+++ b/tests/integrations/github/cicd/test_github_commands.py
@@ -141,8 +141,7 @@ def test_run_all_success_with_approvers_approved(
 <details>
   <summary>Prod Plan Being Applied</summary>
 
-
-**Models needing backfill (missing dates):**"""
+**New environment `prod` will be created from `prod`**"""
     )
     with open(github_output_file, "r", encoding="utf-8") as f:
         output = f.read()
@@ -268,8 +267,7 @@ def test_run_all_success_with_approvers_approved_merge_delete(
 <details>
   <summary>Prod Plan Being Applied</summary>
 
-
-**Models needing backfill (missing dates):**"""
+**New environment `prod` will be created from `prod`**"""
     )
     with open(github_output_file, "r", encoding="utf-8") as f:
         output = f.read()
@@ -907,8 +905,7 @@ def test_prod_update_failure(
 <details>
   <summary>Prod Plan Being Applied</summary>
 
-
-**Models needing backfill (missing dates):**"""
+**New environment `prod` will be created from `prod`**"""
     )
 
     with open(github_output_file, "r", encoding="utf-8") as f:
@@ -1101,8 +1098,7 @@ def test_comment_command_deploy_prod(
 <details>
   <summary>Prod Plan Being Applied</summary>
 
-
-**Models needing backfill (missing dates):**"""
+**New environment `prod` will be created from `prod`**"""
     )
 
     with open(github_output_file, "r", encoding="utf-8") as f:
diff --git a/tests/integrations/github/cicd/test_integration.py b/tests/integrations/github/cicd/test_integration.py
index bcb80e7f..b8d469d4 100644
--- a/tests/integrations/github/cicd/test_integration.py
+++ b/tests/integrations/github/cicd/test_integration.py
@@ -154,7 +154,12 @@ def test_merge_pr_has_non_breaking_change(
     assert GithubCheckStatus(prod_plan_preview_checks_runs[1]["status"]).is_in_progress
     assert GithubCheckStatus(prod_plan_preview_checks_runs[2]["status"]).is_completed
     assert GithubCheckConclusion(prod_plan_preview_checks_runs[2]["conclusion"]).is_success
-    expected_prod_plan_summary = """```diff
+    expected_prod_plan_summary = """**Summary of differences against `prod`:**
+
+
+**Directly Modified:**
+- `sushi.waiter_revenue_by_day`
+```diff
 --- 
 
 +++ 
@@ -172,20 +177,12 @@ def test_merge_pr_has_non_breaking_change(
    ON o.id = oi.order_id AND o.event_date = oi.event_date
 ```
 
-```
-
-Directly Modified: sushi.waiter_revenue_by_day (Non-breaking)
-└── Indirectly Modified Children:
-    └── sushi.top_waiters (Indirect Non-breaking)
-
-```
+**Indirectly Modified:**
+- `sushi.top_waiters`
 
 """
     assert prod_plan_preview_checks_runs[2]["output"]["title"] == "Prod Plan Preview"
-    assert (
-        prod_plan_preview_checks_runs[2]["output"]["summary"]
-        == "**Preview of Prod Plan**\n" + expected_prod_plan_summary
-    )
+    assert prod_plan_preview_checks_runs[2]["output"]["summary"] == expected_prod_plan_summary
 
     assert "SQLMesh - Prod Environment Synced" in controller._check_run_mapping
     prod_checks_runs = controller._check_run_mapping["SQLMesh - Prod Environment Synced"].all_kwargs
@@ -353,7 +350,12 @@ def test_merge_pr_has_non_breaking_change_diff_start(
     assert GithubCheckStatus(prod_plan_preview_checks_runs[2]["status"]).is_completed
     assert GithubCheckConclusion(prod_plan_preview_checks_runs[2]["conclusion"]).is_success
     assert prod_plan_preview_checks_runs[2]["output"]["title"] == "Prod Plan Preview"
-    expected_prod_plan = """```diff
+    expected_prod_plan = """**Summary of differences against `prod`:**
+
+
+**Directly Modified:**
+- `sushi.waiter_revenue_by_day`
+```diff
 --- 
 
 +++ 
@@ -371,26 +373,14 @@ def test_merge_pr_has_non_breaking_change_diff_start(
    ON o.id = oi.order_id AND o.event_date = oi.event_date
 ```
 
-```
-
-Directly Modified: sushi.waiter_revenue_by_day (Non-breaking)
-└── Indirectly Modified Children:
-    └── sushi.top_waiters (Indirect Non-breaking)
+**Indirectly Modified:**
+- `sushi.top_waiters`
 
-```
 
 **Models needing backfill (missing dates):**
-
-
 * `sushi.waiter_revenue_by_day`: 2022-12-25 - 2022-12-28
-
-
-
 """
-    assert (
-        prod_plan_preview_checks_runs[2]["output"]["summary"]
-        == "**Preview of Prod Plan**\n" + expected_prod_plan
-    )
+    assert prod_plan_preview_checks_runs[2]["output"]["summary"] == expected_prod_plan
 
     assert "SQLMesh - Prod Environment Synced" in controller._check_run_mapping
     prod_checks_runs = controller._check_run_mapping["SQLMesh - Prod Environment Synced"].all_kwargs
@@ -701,12 +691,9 @@ def test_merge_pr_has_no_changes(
     assert GithubCheckStatus(prod_plan_preview_checks_runs[1]["status"]).is_in_progress
     assert GithubCheckStatus(prod_plan_preview_checks_runs[2]["status"]).is_completed
     assert GithubCheckConclusion(prod_plan_preview_checks_runs[2]["conclusion"]).is_success
-    expected_prod_plan_summary = "No changes to apply."
+    expected_prod_plan_summary = "**No differences when compared to `prod`**\n\n\n"
     assert prod_plan_preview_checks_runs[2]["output"]["title"] == "Prod Plan Preview"
-    assert (
-        prod_plan_preview_checks_runs[2]["output"]["summary"]
-        == "**Preview of Prod Plan**\n" + expected_prod_plan_summary
-    )
+    assert prod_plan_preview_checks_runs[2]["output"]["summary"] == expected_prod_plan_summary
 
     assert "SQLMesh - Prod Environment Synced" in controller._check_run_mapping
     prod_checks_runs = controller._check_run_mapping["SQLMesh - Prod Environment Synced"].all_kwargs
@@ -867,7 +854,12 @@ def test_no_merge_since_no_deploy_signal(
     assert GithubCheckStatus(prod_plan_preview_checks_runs[1]["status"]).is_in_progress
     assert GithubCheckStatus(prod_plan_preview_checks_runs[2]["status"]).is_completed
     assert GithubCheckConclusion(prod_plan_preview_checks_runs[2]["conclusion"]).is_success
-    expected_prod_plan = """```diff
+    expected_prod_plan = """**Summary of differences against `prod`:**
+
+
+**Directly Modified:**
+- `sushi.waiter_revenue_by_day`
+```diff
 --- 
 
 +++ 
@@ -885,20 +877,12 @@ def test_no_merge_since_no_deploy_signal(
    ON o.id = oi.order_id AND o.event_date = oi.event_date
 ```
 
-```
-
-Directly Modified: sushi.waiter_revenue_by_day (Non-breaking)
-└── Indirectly Modified Children:
-    └── sushi.top_waiters (Indirect Non-breaking)
-
-```
+**Indirectly Modified:**
+- `sushi.top_waiters`
 
 """
     assert prod_plan_preview_checks_runs[2]["output"]["title"] == "Prod Plan Preview"
-    assert (
-        prod_plan_preview_checks_runs[2]["output"]["summary"]
-        == "**Preview of Prod Plan**\n" + expected_prod_plan
-    )
+    assert prod_plan_preview_checks_runs[2]["output"]["summary"] == expected_prod_plan
 
     assert "SQLMesh - Prod Environment Synced" in controller._check_run_mapping
     prod_checks_runs = controller._check_run_mapping["SQLMesh - Prod Environment Synced"].all_kwargs
@@ -1050,7 +1034,12 @@ def test_no_merge_since_no_deploy_signal_no_approvers_defined(
     assert GithubCheckStatus(prod_plan_preview_checks_runs[1]["status"]).is_in_progress
     assert GithubCheckStatus(prod_plan_preview_checks_runs[2]["status"]).is_completed
     assert GithubCheckConclusion(prod_plan_preview_checks_runs[2]["conclusion"]).is_success
-    expected_prod_plan = """```diff
+    expected_prod_plan = """**Summary of differences against `prod`:**
+
+
+**Directly Modified:**
+- `sushi.waiter_revenue_by_day`
+```diff
 --- 
 
 +++ 
@@ -1068,27 +1057,15 @@ def test_no_merge_since_no_deploy_signal_no_approvers_defined(
    ON o.id = oi.order_id AND o.event_date = oi.event_date
 ```
 
-```
+**Indirectly Modified:**
+- `sushi.top_waiters`
 
-Directly Modified: sushi.waiter_revenue_by_day (Non-breaking)
-└── Indirectly Modified Children:
-    └── sushi.top_waiters (Indirect Non-breaking)
-
-```
 
 **Models needing backfill (missing dates):**
-
-
 * `sushi.waiter_revenue_by_day`: 2022-12-25 - 2022-12-29
-
-
-
 """
     assert prod_plan_preview_checks_runs[2]["output"]["title"] == "Prod Plan Preview"
-    assert (
-        prod_plan_preview_checks_runs[2]["output"]["summary"]
-        == "**Preview of Prod Plan**\n" + expected_prod_plan
-    )
+    assert prod_plan_preview_checks_runs[2]["output"]["summary"] == expected_prod_plan
 
     assert "SQLMesh - Prod Environment Synced" not in controller._check_run_mapping
     assert "SQLMesh - Has Required Approval" not in controller._check_run_mapping
@@ -1226,7 +1203,12 @@ def test_deploy_comment_pre_categorized(
     assert GithubCheckStatus(prod_plan_preview_checks_runs[1]["status"]).is_in_progress
     assert GithubCheckStatus(prod_plan_preview_checks_runs[2]["status"]).is_completed
     assert GithubCheckConclusion(prod_plan_preview_checks_runs[2]["conclusion"]).is_success
-    expected_prod_plan = """```diff
+    expected_prod_plan = """**Summary of differences against `prod`:**
+
+
+**Directly Modified:**
+- `sushi.waiter_revenue_by_day`
+```diff
 --- 
 
 +++ 
@@ -1244,20 +1226,12 @@ def test_deploy_comment_pre_categorized(
    ON o.id = oi.order_id AND o.event_date = oi.event_date
 ```
 
-```
-
-Directly Modified: sushi.waiter_revenue_by_day (Non-breaking)
-└── Indirectly Modified Children:
-    └── sushi.top_waiters (Indirect Non-breaking)
-
-```
+**Indirectly Modified:**
+- `sushi.top_waiters`
 
 """
     assert prod_plan_preview_checks_runs[2]["output"]["title"] == "Prod Plan Preview"
-    assert (
-        prod_plan_preview_checks_runs[2]["output"]["summary"]
-        == "**Preview of Prod Plan**\n" + expected_prod_plan
-    )
+    assert prod_plan_preview_checks_runs[2]["output"]["summary"] == expected_prod_plan
 
     assert "SQLMesh - Prod Environment Synced" in controller._check_run_mapping
     prod_checks_runs = controller._check_run_mapping["SQLMesh - Prod Environment Synced"].all_kwargs
@@ -1564,7 +1538,12 @@ def test_overlapping_changes_models(
     assert GithubCheckStatus(prod_plan_preview_checks_runs[1]["status"]).is_in_progress
     assert GithubCheckStatus(prod_plan_preview_checks_runs[2]["status"]).is_completed
     assert GithubCheckConclusion(prod_plan_preview_checks_runs[2]["conclusion"]).is_success
-    expected_prod_plan_summary = """```diff
+    expected_prod_plan_summary = """**Summary of differences against `prod`:**
+
+
+**Directly Modified:**
+- `sushi.customers`
+```diff
 --- 
 
 +++ 
@@ -1581,14 +1560,7 @@ def test_overlapping_changes_models(
  LEFT JOIN current_marketing AS m
    ON o.customer_id = m.customer_id
 ```
-
-```
-
-Directly Modified: sushi.customers (Non-breaking)
-└── Indirectly Modified Children:
-    └── sushi.waiter_as_customer_by_day (Indirect Breaking)
-
-```
+- `sushi.waiter_names`
 ```diff
 --- 
 
@@ -1602,20 +1574,12 @@ Directly Modified: sushi.customers (Non-breaking)
 +10,Trey
 ```
 
-```
-
-Directly Modified: sushi.waiter_names (Breaking)
-└── Indirectly Modified Children:
-    └── sushi.waiter_as_customer_by_day (Indirect Breaking)
-
-```
+**Indirectly Modified:**
+- `sushi.waiter_as_customer_by_day`
 
 """
     assert prod_plan_preview_checks_runs[2]["output"]["title"] == "Prod Plan Preview"
-    assert (
-        prod_plan_preview_checks_runs[2]["output"]["summary"]
-        == "**Preview of Prod Plan**\n" + expected_prod_plan_summary
-    )
+    assert prod_plan_preview_checks_runs[2]["output"]["summary"] == expected_prod_plan_summary
 
     assert "SQLMesh - Prod Environment Synced" in controller._check_run_mapping
     prod_checks_runs = controller._check_run_mapping["SQLMesh - Prod Environment Synced"].all_kwargs
@@ -1678,7 +1642,7 @@ Directly Modified: sushi.waiter_names (Breaking)
 
 
 @freeze_time("2023-01-01 15:00:00")
-def test_capture_console_errors(
+def test_pr_delete_model(
     github_client,
     make_controller,
     make_mock_check_run,
@@ -1688,7 +1652,7 @@ def test_capture_console_errors(
     mocker: MockerFixture,
 ):
     """
-    PR with a non-breaking change and auto-categorization will be backfilled, merged, and deployed to prod
+    PR with a removed model and auto-categorization will be backfilled, merged, and deployed to prod
 
     Scenario:
     - PR is not merged
@@ -1735,10 +1699,10 @@ def test_capture_console_errors(
     controller._context.users = [
         User(username="test", github_username="test_github", roles=[UserRole.REQUIRED_APPROVER])
     ]
-    # Make a non-breaking change
-    model = controller._context.get_model("sushi.waiter_revenue_by_day").copy()
-    model.query.select(exp.alias_("1", "new_col"), copy=False)
-    controller._context.upsert_model(model)
+    # Remove a model
+    model = controller._context.get_model("sushi.top_waiters").copy()
+    del controller._context._models[model.fqn]
+    controller._context.dag = controller._context.dag.prune(*controller._context._models.keys())
 
     github_output_file = tmp_path / "github_output.txt"
 
@@ -1769,42 +1733,45 @@ def test_capture_console_errors(
     assert GithubCheckStatus(pr_checks_runs[0]["status"]).is_queued
     assert GithubCheckStatus(pr_checks_runs[1]["status"]).is_in_progress
     assert GithubCheckStatus(pr_checks_runs[2]["status"]).is_completed
-    assert not GithubCheckConclusion(pr_checks_runs[2]["conclusion"]).is_success
+    assert GithubCheckConclusion(pr_checks_runs[2]["conclusion"]).is_success
     assert pr_checks_runs[2]["output"]["title"] == "PR Virtual Data Environment: hello_world_2"
-    assert pr_checks_runs[2]["output"]["summary"].startswith(
-        """:x: Failed to create or update PR Environment `hello_world_2`.
-**Errors:**
-```
-FAILED processing snapshot "memory"."sushi"."waiter_revenue_by_day"
-"""
+    assert (
+        pr_checks_runs[2]["output"]["summary"]
+        == """<table><thead><tr><th colspan="3">PR Environment Summary</th></tr><tr><th>Model</th><th>Change Type</th><th>Dates Loaded</th></tr></thead><tbody><tr><td>"memory"."sushi"."top_waiters"</td><td>Breaking</td><td>REMOVED</td></tr></tbody></table>"""
     )
 
+    expected_prod_plan_summary = """**Summary of differences against `prod`:**
+
+
+**Removed Models:**
+- `sushi.top_waiters`
+
+"""
+
     assert "SQLMesh - Prod Plan Preview" in controller._check_run_mapping
     prod_plan_preview_checks_runs = controller._check_run_mapping[
         "SQLMesh - Prod Plan Preview"
     ].all_kwargs
-    assert len(prod_plan_preview_checks_runs) == 2
+    assert len(prod_plan_preview_checks_runs) == 3
     assert GithubCheckStatus(prod_plan_preview_checks_runs[0]["status"]).is_queued
-    assert GithubCheckStatus(prod_plan_preview_checks_runs[1]["status"]).is_completed
-    assert GithubCheckConclusion(prod_plan_preview_checks_runs[1]["conclusion"]).is_skipped
-    assert (
-        prod_plan_preview_checks_runs[1]["output"]["title"]
-        == "Skipped generating prod plan preview since PR was not synchronized"
-    )
-    assert (
-        prod_plan_preview_checks_runs[1]["output"]["summary"]
-        == "Skipped generating prod plan preview since PR was not synchronized"
-    )
+    assert GithubCheckStatus(prod_plan_preview_checks_runs[1]["status"]).is_in_progress
+    assert GithubCheckStatus(prod_plan_preview_checks_runs[2]["status"]).is_completed
+    assert GithubCheckConclusion(prod_plan_preview_checks_runs[2]["conclusion"]).is_success
+    assert prod_plan_preview_checks_runs[2]["output"]["title"] == "Prod Plan Preview"
+    assert prod_plan_preview_checks_runs[2]["output"]["summary"] == expected_prod_plan_summary
 
     assert "SQLMesh - Prod Environment Synced" in controller._check_run_mapping
     prod_checks_runs = controller._check_run_mapping["SQLMesh - Prod Environment Synced"].all_kwargs
-    assert len(prod_checks_runs) == 2
+    assert len(prod_checks_runs) == 3
     assert GithubCheckStatus(prod_checks_runs[0]["status"]).is_queued
-    assert GithubCheckStatus(prod_checks_runs[1]["status"]).is_completed
-    assert GithubCheckConclusion(prod_checks_runs[1]["conclusion"]).is_skipped
-    skip_reason = "Skipped Deploying to Production because the PR environment was not updated"
-    assert prod_checks_runs[1]["output"]["title"] == skip_reason
-    assert prod_checks_runs[1]["output"]["summary"] == skip_reason
+    assert GithubCheckStatus(prod_checks_runs[1]["status"]).is_in_progress
+    assert GithubCheckStatus(prod_checks_runs[2]["status"]).is_completed
+    assert GithubCheckConclusion(prod_checks_runs[2]["conclusion"]).is_success
+    assert prod_checks_runs[2]["output"]["title"] == "Deployed to Prod"
+    assert (
+        prod_checks_runs[2]["output"]["summary"]
+        == "**Generated Prod Plan**\n" + expected_prod_plan_summary
+    )
 
     assert "SQLMesh - Has Required Approval" in controller._check_run_mapping
     approval_checks_runs = controller._check_run_mapping[
@@ -1828,13 +1795,25 @@ FAILED processing snapshot "memory"."sushi"."waiter_revenue_by_day"
 
     assert len(get_environment_objects(controller, "hello_world_2")) == 0
 
-    assert not mock_pull_request.merge.called
+    assert mock_pull_request.merge.called
 
-    assert len(created_comments) == 0
+    assert len(created_comments) == 1
+    assert (
+        created_comments[0].body
+        == f"""**SQLMesh Bot Info**
+- PR Virtual Data Environment: hello_world_2
+<details>
+  <summary>Prod Plan Being Applied</summary>
+
+{expected_prod_plan_summary}
+</details>
+
+"""
+    )
 
     with open(github_output_file, "r", encoding="utf-8") as f:
         output = f.read()
         assert (
             output
-            == "run_unit_tests=success\nhas_required_approval=success\npr_environment_name=hello_world_2\npr_environment_synced=failure\nprod_plan_preview=skipped\nprod_environment_synced=skipped\n"
+            == "run_unit_tests=success\nhas_required_approval=success\ncreated_pr_environment=true\npr_environment_name=hello_world_2\npr_environment_synced=success\nprod_plan_preview=success\nprod_environment_synced=success\n"
         )