more test fixes

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

Author: gjesse

sqlmesh/core/plan/builder.py

@@ -188,17 +188,16 @@ def __init__(
         self._start = start
         if not self._start and self._forward_only_preview_needed:
             self._preview_start = self._preview_start or default_start or yesterday_ds()
-            # Keep the forward-only preview window separate from the plan start.
-            # The plan start bounds regular backfills, so it should continue to
-            # use the state-derived default start when one is available.
-            # None means no explicit selection; an empty set intentionally backfills no models.
-            if self._backfill_models is None:
+            # If a separate preview start was provided, don't let it shorten the
+            # plan start for regular backfills. Fallback preview starts preserve
+            # the previous preview behavior of using default_start or yesterday.
+            if self._preview_start_provided and not self._skip_backfill:
                 self._start = default_start or yesterday_ds()
+            else:
+                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._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

@@ -3077,132 +3077,6 @@ def test_forward_only_preview_start_follows_implicit_plan_start_updates(make_sna
     }
 
 
-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)