Add DEMOTING runtime stage

Author: treysp

docs/concepts/macros/macro_variables.md

@@ -132,7 +132,8 @@ SQLMesh provides additional predefined variables used to modify model behavior b
     * 'loading' - The project is being loaded into SQLMesh's runtime context.
     * 'creating' - The model tables are being created.
     * 'evaluating' - The model query logic is being evaluated.
-    * 'promoting' - The model is being promoted in the target environment (virtual layer update).
+    * 'promoting' - The model is being promoted in the target environment (view created during virtual layer update).
+    * 'demoting' - The model is being demoted in the target environment (view dropped during virtual layer update).
     * 'auditing' - The audit is being run.
     * 'testing' - The model query logic is being evaluated in the context of a unit test.
 * @gateway - A string value containing the name of the current [gateway](../../guides/connections.md).

sqlmesh/core/macros.py

@@ -67,6 +67,7 @@ class RuntimeStage(Enum):
     CREATING = "creating"
     EVALUATING = "evaluating"
     PROMOTING = "promoting"
+    DEMOTING = "demoting"
     AUDITING = "auditing"
     TESTING = "testing"
     BEFORE_ALL = "before_all"

sqlmesh/core/snapshot/evaluator.py

@@ -277,7 +277,7 @@ def demote(
         self,
         target_snapshots: t.Iterable[Snapshot],
         environment_naming_info: EnvironmentNamingInfo,
-        snapshots: t.Dict[SnapshotId, Snapshot],
+        snapshots: t.Optional[t.Dict[SnapshotId, Snapshot]] = None,
         deployability_index: t.Optional[DeployabilityIndex] = None,
         on_complete: t.Optional[t.Callable[[SnapshotInfoLike], None]] = None,
     ) -> None:
@@ -1017,7 +1017,7 @@ def _demote_snapshot(
         self,
         snapshot: Snapshot,
         environment_naming_info: EnvironmentNamingInfo,
-        snapshots: t.Dict[SnapshotId, Snapshot],
+        snapshots: t.Optional[t.Dict[SnapshotId, Snapshot]],
         deployability_index: t.Optional[DeployabilityIndex],
         on_complete: t.Optional[t.Callable[[SnapshotInfoLike], None]],
     ) -> None:
@@ -1037,7 +1037,7 @@ def _demote_snapshot(
                         engine_adapter=adapter,
                         snapshots=snapshots,
                         deployability_index=deployability_index,
-                        runtime_stage=RuntimeStage.PROMOTING,
+                        runtime_stage=RuntimeStage.DEMOTING,
                     )
                 ),
             ):

tests/core/test_snapshot_evaluator.py

@@ -320,7 +320,7 @@ def test_demote(mocker: MockerFixture, adapter_mock, make_snapshot):
     snapshot = make_snapshot(model)
     snapshot.categorize_as(SnapshotChangeCategory.BREAKING)
 
-    evaluator.demote([snapshot], EnvironmentNamingInfo(name="test_env"), snapshots={})
+    evaluator.demote([snapshot], EnvironmentNamingInfo(name="test_env"))
 
     adapter_mock.transaction.assert_called()
     adapter_mock.session.assert_called()
@@ -2711,7 +2711,7 @@ def test_standalone_audit(mocker: MockerFixture, adapter_mock, make_snapshot):
     adapter_mock.session.assert_not_called()
 
     # Demote
-    evaluator.demote([snapshot], EnvironmentNamingInfo(name="test_env"), snapshots={})
+    evaluator.demote([snapshot], EnvironmentNamingInfo(name="test_env"))
 
     adapter_mock.assert_not_called()
     adapter_mock.transaction.assert_not_called()
@@ -3854,7 +3854,7 @@ def test_multiple_engine_creation(snapshot: Snapshot, adapters, make_snapshot):
     assert len(post_calls) == 1
     assert post_calls[0].sql(dialect="postgres") == expected_call
 
-    evaluator.demote([snapshot_2], EnvironmentNamingInfo(name="test_env"), snapshots={})
+    evaluator.demote([snapshot_2], EnvironmentNamingInfo(name="test_env"))
     engine_adapters["secondary"].drop_view.assert_not_called()
     engine_adapters["default"].drop_view.assert_called_once_with(
         "test_schema__test_env.test_model",
@@ -4100,7 +4100,7 @@ def model_with_statements(context, **kwargs):
     engine_adapters["default"].get_catalog_type.assert_not_called()
     assert len(engine_adapters["secondary"].get_catalog_type.call_args_list) == 2
 
-    evaluator.demote([snapshot], environment_naming_info, snapshots={})
+    evaluator.demote([snapshot], environment_naming_info)
     engine_adapters["default"].drop_view.assert_called_once_with(
         "db__test_env.multi_engine_test_model",
         cascade=False,
@@ -4114,7 +4114,7 @@ def model_with_statements(context, **kwargs):
     assert view_args[0][0][0] == "db__test_env.multi_engine_test_model"
 
     # Similarly for demotion
-    evaluator.demote([snapshot], environment_naming_info_gw, snapshots={})
+    evaluator.demote([snapshot], environment_naming_info_gw)
     engine_adapters["secondary"].drop_view.assert_called_once_with(
         "db__test_env.multi_engine_test_model",
         cascade=False,
@@ -4172,7 +4172,7 @@ def test_multiple_engine_virtual_layer(snapshot: Snapshot, adapters, make_snapsh
     assert view_args_secondary[0][0][0] == "test_schema__test_env.test_model"
 
     # Demotion will follow with the same pattern
-    evaluator.demote([snapshot_2, snapshot], environment_naming_info, snapshots={})
+    evaluator.demote([snapshot_2, snapshot], environment_naming_info)
     engine_adapters["default"].drop_view.assert_called_once_with(
         "db__test_env.model",
         cascade=False,