Fix: Support run flag in the plan command (#4043)

Author: izeigerman

sqlmesh/core/context.py

@@ -1388,17 +1388,20 @@ def plan_builder(
             # This ensures that no models outside the impacted sub-DAG(s) will be backfilled unexpectedly.
             backfill_models = modified_model_names or None
 
-        max_interval_end_per_model = self._get_max_interval_end_per_model(
-            snapshots, backfill_models
-        )
-        # If no end date is specified, use the max interval end from prod
-        # to prevent unintended evaluation of the entire DAG.
-        default_start, default_end = self._get_plan_default_start_end(
-            snapshots, max_interval_end_per_model, backfill_models, modified_model_names
-        )
+        max_interval_end_per_model = None
+        default_start, default_end = None, None
+        if not run:
+            max_interval_end_per_model = self._get_max_interval_end_per_model(
+                snapshots, backfill_models
+            )
+            # If no end date is specified, use the max interval end from prod
+            # to prevent unintended evaluation of the entire DAG.
+            default_start, default_end = self._get_plan_default_start_end(
+                snapshots, max_interval_end_per_model, backfill_models, modified_model_names
+            )
 
-        # 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())
+            # 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())
 
         return self.PLAN_BUILDER_TYPE(
             context_diff=context_diff,

tests/core/test_integration.py

@@ -1512,6 +1512,44 @@ def test_run_with_select_models(
         }
 
 
+@time_machine.travel("2023-01-08 15:00:00 UTC")
+def test_plan_with_run(
+    init_and_plan_context: t.Callable,
+):
+    context, plan = init_and_plan_context("examples/sushi")
+    context.apply(plan)
+
+    model = context.get_model("sushi.waiter_revenue_by_day")
+    context.upsert_model(add_projection_to_model(t.cast(SqlModel, model)))
+
+    with time_machine.travel("2023-01-09 00:00:00 UTC"):
+        plan = context.plan(run=True)
+        assert plan.has_changes
+        assert plan.missing_intervals
+
+        context.apply(plan)
+
+        snapshots = context.state_sync.state_sync.get_snapshots(context.snapshots.values())
+        assert {s.name: s.intervals[0][1] for s in snapshots.values() if s.intervals} == {
+            '"memory"."sushi"."waiter_revenue_by_day"': to_timestamp("2023-01-09"),
+            '"memory"."sushi"."order_items"': to_timestamp("2023-01-09"),
+            '"memory"."sushi"."orders"': to_timestamp("2023-01-09"),
+            '"memory"."sushi"."items"': to_timestamp("2023-01-09"),
+            '"memory"."sushi"."customer_revenue_lifetime"': to_timestamp("2023-01-09"),
+            '"memory"."sushi"."customer_revenue_by_day"': to_timestamp("2023-01-09"),
+            '"memory"."sushi"."latest_order"': to_timestamp("2023-01-09"),
+            '"memory"."sushi"."waiter_names"': to_timestamp("2023-01-08"),
+            '"memory"."sushi"."raw_marketing"': to_timestamp("2023-01-09"),
+            '"memory"."sushi"."marketing"': to_timestamp("2023-01-09"),
+            '"memory"."sushi"."waiter_as_customer_by_day"': to_timestamp("2023-01-09"),
+            '"memory"."sushi"."top_waiters"': to_timestamp("2023-01-09"),
+            '"memory"."raw"."demographics"': to_timestamp("2023-01-09"),
+            "assert_item_price_above_zero": to_timestamp("2023-01-09"),
+            '"memory"."sushi"."active_customers"': to_timestamp("2023-01-09"),
+            '"memory"."sushi"."customers"': to_timestamp("2023-01-09"),
+        }
+
+
 @time_machine.travel("2023-01-08 15:00:00 UTC")
 def test_run_with_select_models_no_auto_upstream(
     init_and_plan_context: t.Callable,