Fix: Don't expand start for full-history-restatement-only snapshots when previewing forward-only changes (#2739)

Author: izeigerman

sqlmesh/core/plan/builder.py

@@ -314,6 +314,7 @@ def is_restateable_snapshot(snapshot: Snapshot) -> bool:
 
         restatements: t.Dict[SnapshotId, Interval] = {}
         forward_only_preview_needed = self._forward_only_preview_needed
+        is_preview = False
         if not restate_models and forward_only_preview_needed:
             # Add model names for new forward-only snapshots to the restatement list
             # in order to compute previews.
@@ -322,6 +323,7 @@ def is_restateable_snapshot(snapshot: Snapshot) -> bool:
                 for s in self._context_diff.new_snapshots.values()
                 if s.is_materialized and (self._forward_only or s.model.forward_only)
             }
+            is_preview = True
 
         if not restate_models:
             return {}
@@ -358,6 +360,7 @@ def is_restateable_snapshot(snapshot: Snapshot) -> bool:
                 self._end or now(),
                 self._execution_time,
                 strict=False,
+                is_preview=is_preview,
             )
             # Since we are traversing the graph in topological order and the largest interval range is pushed down
             # the graph we just have to check our immediate parents in the graph and not the whole upstream graph.

sqlmesh/core/snapshot/definition.py

@@ -675,9 +675,21 @@ def get_removal_interval(
         execution_time: t.Optional[TimeLike] = None,
         *,
         strict: bool = True,
+        is_preview: bool = False,
     ) -> Interval:
+        """Get the interval that should be removed from the snapshot.
+
+        Args:
+            start: The start date/time of the interval to remove.
+            end: The end date/time of the interval to removed.
+            execution_time: The time the interval is being removed.
+            strict: Whether to fail when the inclusive start is the same as the exclusive end.
+            is_preview: Whether the interval needs to be removed for a preview of forward-only changes.
+                When previewing, we are not actually restating a model, but removing an interval to trigger
+                a run.
+        """
         end = execution_time or now() if self.depends_on_past else end
-        if self.full_history_restatement_only and self.intervals:
+        if not is_preview and self.full_history_restatement_only and self.intervals:
             start = self.intervals[0][0]
         return self.inclusive_exclusive(start, end, strict)