Repository: TobikoData/sqlmesh
Pull request #2514 (2024-04-25 00:21:14)

Chore: Refactor migration script application


Patch:
diff --git a/docs/concepts/models/overview.md b/docs/concepts/models/overview.md
index 5f6b5aae..a071532f 100644
--- a/docs/concepts/models/overview.md
+++ b/docs/concepts/models/overview.md
@@ -1,6 +1,6 @@
 # Overview
 
-Models are comprised of metadata and queries that create tables and views, which can be used by other models or even outside of SQLMesh. They are defined in the `models/` directory of your SQLMesh project and live in `.sql` files.
+Models are made up of metadata and queries that create tables and views, which can be used by other models or even outside of SQLMesh. They are defined in the `models/` directory of your SQLMesh project and live in `.sql` files.
 
 SQLMesh will automatically determine the relationships among and lineage of your models by parsing SQL, so you don't have to worry about manually configuring dependencies.
 
diff --git a/docs/concepts/models/seed_models.md b/docs/concepts/models/seed_models.md
index a48afabd..08a01285 100644
--- a/docs/concepts/models/seed_models.md
+++ b/docs/concepts/models/seed_models.md
@@ -60,6 +60,10 @@ MODEL (
 
 This is useful when you want to keep all seed CSV files in a top-level directory such as `seeds/` but don't want to keep track of or manage a bunch of relative paths.
 
+### Encoding
+
+SQLMesh expects seed files to be encoded according to the [UTF-8](https://en.wikipedia.org/wiki/UTF-8) standard. Using a different encoding may lead to unexpected behavior.
+
 ## Example
 
 In this example, we use the model definition from the previous section saved in the `models/national_holidays.sql` file of the SQLMesh project.
diff --git a/docs/concepts/models/sql_models.md b/docs/concepts/models/sql_models.md
index 5b78d1ff..bd5900f1 100644
--- a/docs/concepts/models/sql_models.md
+++ b/docs/concepts/models/sql_models.md
@@ -53,11 +53,13 @@ UNCACHE TABLE countries;
 ```
 
 ### `MODEL` DDL
+
 The `MODEL` DDL is used to specify metadata about the model such as its name, [kind](./model_kinds.md), owner, cron, and others. This should be the first statement in your SQL-based model's file.
 
 Refer to `MODEL` [properties](./overview.md#properties) for the full list of allowed properties.
 
 ### Optional pre/post-statements
+
 Optional pre/post-statements allow you to execute SQL commands before and after a model runs, respectively.
 
 For example, post/post-statements might modify settings or create indexes. However, be careful not to run any statement that could conflict with the execution of another statement if the models run concurrently, such as creating a physical table.
@@ -65,6 +67,7 @@ For example, post/post-statements might modify settings or create indexes. Howev
 Pre/post-statements are evaluated twice: when a model's table is created and when its query logic is evaluated. Since executing such statements more than once can have unintended side-effects, it is also possible to [conditionally execute](../macros/sqlmesh_macros.md#if) them depending on SQLMesh's [runtime stage](../macros/macro_variables.md#predefined-variables).
 
 ### The model query
+
 The model must contain a standalone query, which can be a single `SELECT` expression, or multiple `SELECT` expressions combined with the `UNION`, `INTERSECT`, or `EXCEPT` operators. The result of this query will be used to populate the model's table or view.
 
 ## Python-based definition
@@ -113,6 +116,7 @@ The `@model` decorator is the Python equivalent of the `MODEL` DDL. In addition
 **Note:** All of the [metadata](./overview.md#properties) field names are the same as those in the `MODEL` DDL.
 
 ## Automatic dependencies
+
 SQLMesh parses your SQL, so it understands what the code does and how it relates to other models. There is no need for you to manually specify dependencies to other models with special tags or commands.
 
 For example, consider a model with this query:
@@ -131,14 +135,21 @@ External dependencies not defined in SQLMesh are also supported. SQLMesh can eit
 Although automatic dependency detection works most of the time, there may be specific cases for which you want to define dependencies manually. You can do so in the `MODEL` DDL with the [dependencies property](./overview.md#properties).
 
 ## Conventions
+
 SQLMesh encourages explicitly specifying the data types of a model's columns through casting. This allows SQLMesh to understand the data types in your models, and it prevents incorrect type inference. SQLMesh supports the casting format `<column name>::<data type>` in models of any SQL dialect.
 
 ### Explicit SELECTs
+
 Although `SELECT *` is convenient, it is dangerous because a model's results can change due to external factors (e.g., an upstream source adding or removing a column). In general, we encourage listing out every column you need or using [`create_external_models`](../../reference/cli.md#create_external_models) to capture the schema of an external data source.
 
 If you select from an external source, `SELECT *` will prevent SQLMesh from performing some optimization steps and from determining upstream column-level lineage. Use an [`external` model kind](./model_kinds.md#external) to enable optimizations and upstream column-level lineage for external sources.
 
+### Encoding
+
+SQLMesh expects files containing SQL models to be encoded according to the [UTF-8](https://en.wikipedia.org/wiki/UTF-8) standard. Using a different encoding may lead to unexpected behavior.
+
 ## Transpilation
+
 SQLMesh leverages [SQLGlot](https://github.com/tobymao/sqlglot) to parse and transpile SQL. Therefore, you can write your SQL in any supported dialect and transpile it into another supported dialect.
 
 You can also use advanced syntax that may not be available in your engine of choice. For example, `x::int` is equivalent to `CAST(x as INT)`, but is only supported in some dialects. SQLGlot allows you to use this feature regardless of what engine you're using.
@@ -146,6 +157,7 @@ You can also use advanced syntax that may not be available in your engine of cho
 Additionally, you won't have to worry about minor formatting differences such as trailing commas, as SQLGlot will remove them at parse time.
 
 ## Macros
+
 Although standard SQL is very powerful, complex data systems often require running SQL queries with dynamic components such as date filters. For example, you may want to change the date ranges in a `between` statement so that you can get the latest batch of data. SQLMesh provides these dates automatically through [macro variables](../macros/macro_variables.md).
 
 Additionally, large queries can be difficult to read and maintain. In order to make queries more compact, SQLMesh supports a powerful [macro syntax](../macros/overview.md) as well as [Jinja](https://jinja.palletsprojects.com/en/3.1.x/), allowing you to write macros that make your SQL queries easier to manage.
diff --git a/examples/sushi/hooks/__init__.py b/examples/sushi/hooks/__init__.py
deleted file mode 100644
index e69de29b..00000000
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/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/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/core/state_sync/engine_adapter.py b/sqlmesh/core/state_sync/engine_adapter.py
index 82bc6a9f..e584cec1 100644
--- a/sqlmesh/core/state_sync/engine_adapter.py
+++ b/sqlmesh/core/state_sync/engine_adapter.py
@@ -917,23 +917,13 @@ class EngineAdapterStateSync(CommonStateSyncMixin, StateSync):
     ) -> None:
         """Migrate the state sync to the latest SQLMesh / SQLGlot version."""
         versions = self.get_versions(validate=False)
-        migrations = MIGRATIONS[versions.schema_version :]
 
-        if (
-            not migrations
-            and major_minor(SQLGLOT_VERSION) == versions.minor_sqlglot_version
-            and major_minor(SQLMESH_VERSION) == versions.minor_sqlmesh_version
-        ):
-            return
+        try:
+            migrate_rows = self._apply_migrations(default_catalog, skip_backup)
 
-        migrate_rows = migrations or major_minor(SQLGLOT_VERSION) != versions.minor_sqlglot_version
-        if not skip_backup and migrate_rows:
-            self._backup_state()
+            if not migrate_rows and major_minor(SQLMESH_VERSION) == versions.minor_sqlmesh_version:
+                return
 
-        try:
-            for migration in migrations:
-                logger.info(f"Applying migration {migration}")
-                migration.migrate(self, default_catalog=default_catalog)
             if migrate_rows:
                 self._migrate_rows(promoted_snapshots_only)
                 # Cleanup plan DAGs since we currently don't migrate snapshot records that are in there.
@@ -992,6 +982,24 @@ class EngineAdapterStateSync(CommonStateSyncMixin, StateSync):
                         backup_name, exp.select("*").from_(table), exists=False
                     )
 
+    def _apply_migrations(
+        self,
+        default_catalog: t.Optional[str],
+        skip_backup: bool,
+    ) -> bool:
+        versions = self.get_versions(validate=False)
+        migrations = MIGRATIONS[versions.schema_version :]
+
+        migrate_rows = migrations or major_minor(SQLGLOT_VERSION) != versions.minor_sqlglot_version
+        if not skip_backup and migrate_rows:
+            self._backup_state()
+
+        for migration in migrations:
+            logger.info(f"Applying migration {migration}")
+            migration.migrate(self, default_catalog=default_catalog)
+
+        return bool(migrate_rows)
+
     def _migrate_rows(self, promoted_snapshots_only: bool) -> None:
         logger.info("Fetching environments")
         environments = self.get_environments()
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(


Test patch:
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"
         )