Fix: Reset the 'effective_from' attribute for migrated snapshots (#2313)

Author: izeigerman

sqlmesh/core/snapshot/definition.py

@@ -736,7 +736,8 @@ def merge_intervals(self, other: t.Union[Snapshot, SnapshotIntervals]) -> None:
         previous_ids = {s.snapshot_id(self.name) for s in self.previous_versions}
         if self.identifier == other.identifier or (
             # Indirect Non-Breaking snapshots share the dev table with its previous version.
-            self.is_indirect_non_breaking
+            # The same applies to migrated snapshots.
+            (self.is_indirect_non_breaking or self.migrated)
             and other.snapshot_id in previous_ids
         ):
             for start, end in other.dev_intervals:

sqlmesh/core/state_sync/engine_adapter.py

@@ -1097,6 +1097,8 @@ def _visit(
                 logger.exception("Could not compute fingerprint for %s", snapshot.snapshot_id)
                 return
 
+            # Reset the effective_from date for the new snapshot to avoid unexpected backfills.
+            new_snapshot.effective_from = None
             new_snapshot.previous_versions = snapshot.all_versions
             new_snapshot.migrated = True
             if not new_snapshot.temp_version:

tests/core/test_snapshot.py

@@ -198,7 +198,7 @@ def test_add_interval(snapshot: Snapshot, make_snapshot):
     ]
 
 
-def test_add_interval_dev(snapshot: Snapshot):
+def test_add_interval_dev(snapshot: Snapshot, make_snapshot):
     snapshot.version = "existing_version"
     snapshot.change_category = SnapshotChangeCategory.FORWARD_ONLY
 
@@ -209,6 +209,18 @@ def test_add_interval_dev(snapshot: Snapshot):
     assert snapshot.intervals == [(to_timestamp("2020-01-01"), to_timestamp("2020-01-02"))]
     assert snapshot.dev_intervals == [(to_timestamp("2020-01-02"), to_timestamp("2020-01-03"))]
 
+    new_snapshot = make_snapshot(snapshot.model)
+    new_snapshot.merge_intervals(snapshot)
+    assert new_snapshot.intervals == [(to_timestamp("2020-01-01"), to_timestamp("2020-01-02"))]
+    assert new_snapshot.dev_intervals == []
+
+    new_snapshot = make_snapshot(snapshot.model)
+    new_snapshot.previous_versions = snapshot.all_versions
+    new_snapshot.migrated = True
+    new_snapshot.merge_intervals(snapshot)
+    assert new_snapshot.intervals == [(to_timestamp("2020-01-01"), to_timestamp("2020-01-02"))]
+    assert new_snapshot.dev_intervals == [(to_timestamp("2020-01-02"), to_timestamp("2020-01-03"))]
+
 
 def test_add_interval_partial(snapshot: Snapshot, make_snapshot):
     snapshot.add_interval("2023-01-01 00:00:00", "2023-01-01 23:59:59")