fix: allow an explicit --execution-time to extend the plan end past the prod frontier

sqlmesh/core/context.py

@@ -1721,6 +1721,35 @@ def plan_builder(
                 execution_time or now(),
             )
 
+            execution_time_ts = to_timestamp(execution_time) if execution_time is not None else None
+            if (
+                execution_time_ts is not None
+                and end is None
+                and default_end is not None
+                and execution_time_ts > default_end
+            ):
+                # An explicitly provided execution time acts as the plan's effective "now", so the
+                # default end is allowed to extend past the recorded prod frontier instead of being
+                # capped by it. This mirrors how an explicit `end` causes the per-model interval end
+                # caps to be dropped entirely in PlanBuilder.build() (see `self.override_end`,
+                # sqlmesh/core/plan/builder.py line 206). Raising (rather than dropping) the caps is
+                # safe here because `plan --run` already runs with no caps at all.
+                #
+                # Note this raises every entry already present in max_interval_end_per_model (i.e.
+                # every model in this dict, which _get_max_interval_end_per_model above has already
+                # scoped down to backfill_models/its ancestors when a selection is in effect) - not
+                # just modified or explicitly selected models within that scope. That's intentional,
+                # not an oversight. It's what makes a plain, unscoped `sqlmesh plan --execution-time X`
+                # in prod report the same missing intervals as `sqlmesh plan --run --execution-time X`
+                # would at the same simulated time. Narrowing this further would reintroduce a `plan`
+                # vs `plan --run` mismatch for models within that scope.
+                default_end = execution_time_ts
+                execution_time_dt = to_datetime(execution_time_ts)
+                max_interval_end_per_model = {
+                    model_fqn: max(interval_end, execution_time_dt)
+                    for model_fqn, interval_end in max_interval_end_per_model.items()
+                }
+
             # Refresh snapshot intervals to ensure that they are up to date with values reflected in the max_interval_end_per_model.
             self.state_sync.refresh_snapshot_intervals(context_diff.snapshots.values())
 

tests/core/test_context.py

@@ -1671,6 +1671,176 @@ def test_plan_start_ahead_of_end(copy_to_temp_path):
         context.close()
 
 
+@pytest.mark.slow
+def test_plan_execution_time_ahead_of_prod_frontier(copy_to_temp_path):
+    """An explicitly provided `execution_time` should be able to extend the plan's default end
+    past the recorded prod frontier, since it represents the plan's effective "now". Without
+    this, an explicit execution_time beyond the last applied interval is silently ignored and
+    the plan reports no changes/no backfill even though new intervals are due.
+
+    See: https://github.com/SQLMesh/sqlmesh/issues/5640
+    """
+    path = copy_to_temp_path("examples/sushi")
+    with time_machine.travel("2024-01-02 00:00:00 UTC"):
+        context = Context(paths=path, gateway="duckdb_persistent")
+        context.plan("prod", no_prompts=True, auto_apply=True)
+        assert all(
+            i == to_timestamp("2024-01-02")
+            for i in context.state_sync.max_interval_end_per_model("prod").values()
+        )
+        context.close()
+
+    # No model changes, but a subsequent plan explicitly passes an execution_time that is ahead
+    # of the recorded prod frontier (2024-01-02). This mimics running
+    # `sqlmesh plan --execution-time '2024-01-05'` a few days later.
+    with time_machine.travel("2024-01-06 00:00:00 UTC"):
+        context = Context(paths=path, gateway="duckdb_persistent")
+        plan = context.plan_builder("prod", execution_time="2024-01-05").build()
+        assert plan.requires_backfill
+        assert to_timestamp(plan.end) == to_timestamp("2024-01-05")
+        context.apply(plan)
+        # The seed model isn't backfilled on a cron schedule like the other models, so its
+        # recorded interval end doesn't advance to the new execution time (same exclusion as
+        # test_plan_seed_model_excluded_from_default_end).
+        max_ends = context.state_sync.max_interval_end_per_model("prod")
+        assert all(
+            i == to_timestamp("2024-01-05")
+            for fqn, i in max_ends.items()
+            if "waiter_names" not in fqn
+        )
+        context.close()
+
+    # Sanity check that the downward clamp is unaffected: an explicit execution_time that is
+    # *behind* the recorded prod frontier should still clamp the default end down to it (this is
+    # already covered for the dev case by test_plan_execution_time_start_end).
+    with time_machine.travel("2024-01-07 00:00:00 UTC"):
+        context = Context(paths=path, gateway="duckdb_persistent")
+        plan = context.plan_builder("prod", execution_time="2024-01-04").build()
+        assert not plan.requires_backfill
+        assert to_timestamp(plan.end) == to_timestamp("2024-01-04")
+        context.close()
+
+
+def _write_daily_and_weekly_model_project(tmp_path: Path) -> None:
+    """Minimal 2-model project used to exercise the interaction between execution_time and
+    multiple, differently-cadenced models' recorded prod frontiers. A daily and a weekly model
+    are used so that, after an initial backfill, they end up with different recorded interval
+    ends (the weekly model's frontier lags the daily model's), which is what the original bug
+    report's project topology looked like.
+    """
+    (tmp_path / "models").mkdir()
+    (tmp_path / "config.yaml").write_text(
+        """
+model_defaults:
+    dialect: duckdb
+"""
+    )
+    (tmp_path / "models" / "daily_model.sql").write_text(
+        """
+MODEL (
+  name daily_model,
+  kind INCREMENTAL_BY_TIME_RANGE (
+    time_column start_dt
+  ),
+  start '2024-01-01',
+  cron '@daily'
+);
+
+select @start_ds as start_ds, @end_ds as end_ds, @start_dt as start_dt, @end_dt as end_dt;
+"""
+    )
+    (tmp_path / "models" / "weekly_model.sql").write_text(
+        """
+MODEL (
+  name weekly_model,
+  kind INCREMENTAL_BY_TIME_RANGE (
+    time_column start_dt
+  ),
+  start '2024-01-01',
+  cron '@weekly'
+);
+
+select @start_ds as start_ds, @end_ds as end_ds, @start_dt as start_dt, @end_dt as end_dt;
+"""
+    )
+
+
+def _missing_intervals_by_name(plan: Plan) -> t.Dict[str, t.Tuple[t.Tuple[int, int], ...]]:
+    return {si.snapshot_id.name: tuple(si.merged_intervals) for si in plan.missing_intervals}
+
+
+def test_plan_execution_time_ahead_of_prod_frontier_matches_run_for_all_models(tmp_path: Path):
+    """Locks in that raising `max_interval_end_per_model` for an explicitly provided
+    `execution_time` sweeps in *every* model with a recorded prod frontier, not just
+    modified/selected ones. This is intentional, not an oversight: it's what makes a plain,
+    unscoped `sqlmesh plan --execution-time X` in prod report the exact same missing intervals
+    that `sqlmesh plan --run --execution-time X` would report at the same simulated time - the
+    parity the original bug report asks for (https://github.com/SQLMesh/sqlmesh/issues/5640,
+    which cites `--run` as already having the correct behavior). A future change that "scopes"
+    the raise down to fewer models would silently break this `plan`/`plan --run` parity and
+    should fail this test.
+    """
+    _write_daily_and_weekly_model_project(tmp_path)
+    context = Context(paths=tmp_path)
+
+    # Catch both models up, but to different frontiers: the weekly model's cadence means its
+    # last fully-elapsed interval (2024-01-14) is a day behind the daily model's (2024-01-15).
+    context.plan(auto_apply=True, no_prompts=True, execution_time="2024-01-15 00:00:01")
+    max_ends = context.state_sync.max_interval_end_per_model("prod")
+    assert max_ends['"daily_model"'] == to_timestamp("2024-01-15")
+    assert max_ends['"weekly_model"'] == to_timestamp("2024-01-14")
+
+    # A plain, unscoped prod plan (no model changes, no --select-model/--backfill-model, no
+    # restatement) with execution_time set well ahead of both frontiers and no explicit end.
+    execution_time = "2024-01-25 00:00:01"
+    plan = context.plan_builder("prod", execution_time=execution_time).build()
+    assert plan.requires_backfill
+    assert to_timestamp(plan.end) == to_timestamp(execution_time)
+
+    missing = _missing_intervals_by_name(plan)
+    # Both models show missing intervals, even though only the daily model's cadence would
+    # naturally put it "due" first - the weekly model is swept in too.
+    assert set(missing) == {'"daily_model"', '"weekly_model"'}
+
+    # An equivalent `plan --run` at the same execution_time computes missing intervals with no
+    # caps at all. If it matches exactly, that confirms the plain-plan raise reproduces the
+    # `--run` result rather than under- or over-shooting it.
+    run_plan = context.plan_builder("prod", execution_time=execution_time, run=True).build()
+    assert run_plan.requires_backfill
+    assert _missing_intervals_by_name(run_plan) == missing
+
+
+def test_plan_execution_time_ahead_of_prod_frontier_with_explicit_end(tmp_path: Path):
+    """When the user provides an explicit `end` alongside `execution_time`, the new
+    `end is None` guard means the per-model interval end caps are never raised towards
+    `execution_time` in the first place - the plan's end is exactly the explicit end, and
+    the pre-existing `PlanBuilder.override_end` behavior (which drops the per-model caps
+    entirely once `end` is explicit, see sqlmesh/core/plan/builder.py) takes over instead, same
+    as it did before this fix. This locks in that combining explicit `end` with a far-future
+    `execution_time` cannot make the backfill silently jump past the requested end.
+    """
+    _write_daily_and_weekly_model_project(tmp_path)
+    context = Context(paths=tmp_path)
+    context.plan(auto_apply=True, no_prompts=True, execution_time="2024-01-15 00:00:01")
+
+    # Explicit start/end is only allowed for dev plans (or prod plans with restatements), so use
+    # a dev plan here; the guard being exercised doesn't depend on which of those it is.
+    plan = context.plan_builder(
+        "dev",
+        execution_time="2024-01-30 00:00:01",
+        end="2024-01-21",
+        include_unmodified=True,
+    ).build()
+    assert plan.requires_backfill
+    # The plan's end matches the explicit end exactly - it is not raised towards execution_time.
+    assert to_timestamp(plan.end) == to_timestamp("2024-01-21")
+
+    # Missing intervals for both models stop at the explicit end and never reach anywhere near
+    # the much-later execution_time.
+    for si in plan.missing_intervals:
+        assert si.merged_intervals[-1][1] <= to_timestamp("2024-01-22")
+
+
 @pytest.mark.slow
 def test_plan_seed_model_excluded_from_default_end(copy_to_temp_path: t.Callable):
     path = copy_to_temp_path("examples/sushi")
@@ -3254,7 +3424,12 @@ def test_plan_min_intervals(tmp_path: Path):
     plan = context.plan(execution_time=current_time)
 
     assert to_datetime(plan.start) == to_datetime("2020-01-01 00:00:00")
-    assert to_datetime(plan.end) == to_datetime("2020-02-01 00:00:00")
+    # the explicitly provided execution_time is 1 second past the day-aligned frontier of the
+    # other, already-caught-up models, so the plan end now matches it exactly instead of being
+    # capped at that frontier (see https://github.com/SQLMesh/sqlmesh/issues/5640). This
+    # doesn't change which intervals are missing below since they're still bucketed by each
+    # model's own cron.
+    assert to_datetime(plan.end) == to_datetime("2020-02-01 00:00:01")
     assert to_datetime(plan.execution_time) == to_datetime("2020-02-01 00:00:01")
 
     def _get_missing_intervals(plan: Plan, name: str) -> t.List[t.Tuple[datetime, datetime]]: