Fix: Set updated_at when updating a snapshot record in state (#1526)

Author: izeigerman

sqlmesh/core/state_sync/common.py

@@ -145,11 +145,17 @@ def promote(self, environment: Environment, no_gaps: bool = False) -> PromotionR
             table_infos -= set(existing_environment.promoted_snapshots)
 
         self._update_environment(environment)
+
         removed = (
             list(existing_table_infos.values())
             if environment_suffix_target_changed
             else [existing_table_infos[name] for name in missing_models]
         )
+        if removed and not environment_suffix_target_changed:
+            for removed_snapshot in self._get_snapshots(removed).values():
+                # Update the updated_at attribute.
+                self._update_snapshot(removed_snapshot)
+
         return PromotionResult(
             added=sorted(table_infos),
             removed=removed,

sqlmesh/core/state_sync/engine_adapter.py

@@ -303,6 +303,7 @@ def _update_environment(self, environment: Environment) -> None:
         )
 
     def _update_snapshot(self, snapshot: Snapshot) -> None:
+        snapshot.updated_ts = now_timestamp()
         self.engine_adapter.update_table(
             self.snapshots_table,
             {"snapshot": snapshot.json()},

tests/core/test_state_sync.py

@@ -437,6 +437,8 @@ def test_promote_snapshots(state_sync: EngineAdapterStateSync, make_snapshot: t.
     assert not promotion_result.removed
     assert not promotion_result.removed_environment_naming_info
 
+    prev_snapshot_c_updated_ts = snapshot_c.updated_ts
+
     promotion_result = promote_snapshots(
         state_sync,
         [snapshot_a, snapshot_b],
@@ -446,6 +448,10 @@ def test_promote_snapshots(state_sync: EngineAdapterStateSync, make_snapshot: t.
     assert set(promotion_result.removed) == {snapshot_c.table_info}
     assert promotion_result.removed_environment_naming_info
     assert promotion_result.removed_environment_naming_info.suffix_target.is_schema
+    assert (
+        state_sync.get_snapshots([snapshot_c])[snapshot_c.snapshot_id].updated_ts
+        > prev_snapshot_c_updated_ts
+    )
 
     snapshot_d = make_snapshot(
         SqlModel(