don't constrain backfill models to preview start

Signed-off-by: Jesse Hodges <hodges.jesse@gmail.com>

Author: gjesse

sqlmesh/core/plan/builder.py

@@ -185,10 +185,17 @@ def __init__(
         self._explain = explain
 
         self._start = start
-        if not self._start and (
-            self._forward_only_preview_needed or self._non_forward_only_preview_needed
-        ):
-            self._start = default_start or yesterday_ds()
+        if not self._start and self._forward_only_preview_needed:
+            self._preview_start = self._preview_start or default_start or yesterday_ds()
+            # Only use the preview fallback as the plan start for implicit selection.
+            # None means no explicit selection; an empty set intentionally backfills no models.
+            if self._backfill_models is None:
+                self._start = self._preview_start
+
+        if not self._start and self._non_forward_only_preview_needed:
+            # Do not bind explicit non-preview backfills to the short preview range.
+            if self._backfill_models is None:
+                self._start = default_start or yesterday_ds()
 
         self._plan_id: str = random_id()
         self._model_fqn_to_snapshot = {s.name: s for s in self._context_diff.snapshots.values()}

tests/core/test_plan.py

@@ -2986,6 +2986,132 @@ def _make_forward_only_preview_context_diff(make_snapshot):
     return context_diff, new_snapshot
 
 
+def test_forward_only_preview_start_does_not_limit_regular_backfill(make_snapshot):
+    context_diff, preview_new_snapshot = _make_forward_only_preview_context_diff(make_snapshot)
+
+    normal_old_snapshot = make_snapshot(
+        SqlModel(
+            name="normal",
+            query=parse_one("select 1, ds"),
+            dialect="duckdb",
+            kind=dict(name=ModelKindName.INCREMENTAL_BY_TIME_RANGE, time_column="ds"),
+            start="2025-01-01",
+        )
+    )
+    normal_old_snapshot.categorize_as(SnapshotChangeCategory.BREAKING)
+    normal_new_snapshot = make_snapshot(
+        SqlModel(
+            name="normal",
+            query=parse_one("select 2, ds"),
+            dialect="duckdb",
+            kind=dict(name=ModelKindName.INCREMENTAL_BY_TIME_RANGE, time_column="ds"),
+            start="2025-01-01",
+        )
+    )
+    normal_new_snapshot.previous_versions = normal_old_snapshot.all_versions
+
+    context_diff.modified_snapshots[normal_new_snapshot.name] = (
+        normal_new_snapshot,
+        normal_old_snapshot,
+    )
+    context_diff.snapshots[normal_new_snapshot.snapshot_id] = normal_new_snapshot
+    context_diff.new_snapshots[normal_new_snapshot.snapshot_id] = normal_new_snapshot
+
+    plan = PlanBuilder(
+        context_diff,
+        default_start="2025-01-02",
+        end="2025-01-03",
+        preview_min_intervals=1,
+        backfill_models={normal_new_snapshot.name, preview_new_snapshot.name},
+        is_dev=True,
+        enable_preview=True,
+    ).build()
+
+    missing_intervals = {i.snapshot_id: i.intervals for i in plan.missing_intervals}
+
+    assert plan.provided_start is None
+    assert missing_intervals == {
+        normal_new_snapshot.snapshot_id: [
+            (to_timestamp("2025-01-01"), to_timestamp("2025-01-02")),
+            (to_timestamp("2025-01-02"), to_timestamp("2025-01-03")),
+            (to_timestamp("2025-01-03"), to_timestamp("2025-01-04")),
+        ],
+        preview_new_snapshot.snapshot_id: [
+            (to_timestamp("2025-01-02"), to_timestamp("2025-01-03")),
+            (to_timestamp("2025-01-03"), to_timestamp("2025-01-04")),
+        ],
+    }
+
+
+def test_non_forward_only_preview_start_does_not_limit_explicit_backfill(make_snapshot):
+    old_snapshot = make_snapshot(
+        SqlModel(
+            name="a",
+            query=parse_one("select 1, ds"),
+            dialect="duckdb",
+            kind=IncrementalByTimeRangeKind(
+                time_column="ds",
+                auto_restatement_cron="@daily",
+            ),
+            start="2025-01-01",
+        )
+    )
+    old_snapshot.categorize_as(SnapshotChangeCategory.BREAKING)
+    new_snapshot = make_snapshot(
+        SqlModel(
+            name="a",
+            query=parse_one("select 2, ds"),
+            dialect="duckdb",
+            kind=IncrementalByTimeRangeKind(
+                time_column="ds",
+                auto_restatement_cron="@daily",
+            ),
+            start="2025-01-01",
+        )
+    )
+    new_snapshot.previous_versions = old_snapshot.all_versions
+
+    context_diff = ContextDiff(
+        environment="test_environment",
+        is_new_environment=True,
+        is_unfinalized_environment=False,
+        normalize_environment_name=True,
+        create_from="prod",
+        create_from_env_exists=True,
+        added=set(),
+        removed_snapshots={},
+        snapshots={new_snapshot.snapshot_id: new_snapshot},
+        new_snapshots={new_snapshot.snapshot_id: new_snapshot},
+        modified_snapshots={old_snapshot.name: (new_snapshot, old_snapshot)},
+        previous_plan_id=None,
+        previously_promoted_snapshot_ids=set(),
+        previous_finalized_snapshots=None,
+        previous_gateway_managed_virtual_layer=False,
+        gateway_managed_virtual_layer=False,
+        environment_statements=[],
+    )
+
+    plan = PlanBuilder(
+        context_diff,
+        default_start="2025-01-02",
+        end="2025-01-03",
+        backfill_models={new_snapshot.name},
+        is_dev=True,
+    ).build()
+
+    assert plan.provided_start is None
+    assert plan.missing_intervals == [
+        SnapshotIntervals(
+            snapshot_id=new_snapshot.snapshot_id,
+            intervals=[
+                (to_timestamp("2025-01-01"), to_timestamp("2025-01-02")),
+                (to_timestamp("2025-01-02"), to_timestamp("2025-01-03")),
+                (to_timestamp("2025-01-03"), to_timestamp("2025-01-04")),
+            ],
+        )
+    ]
+
+
 @time_machine.travel("2026-06-02 00:00:00 UTC")
 def test_forward_only_preview_uses_preview_start(make_snapshot):
     context_diff, new_snapshot = _make_forward_only_preview_context_diff(make_snapshot)