Fix: Improve the inference of the default plan start for new models for which cron is not aligned with interval unit boundary (#3758)

Author: izeigerman

sqlmesh/core/context.py

@@ -649,6 +649,7 @@ def run(
             True if the run was successful, False otherwise.
         """
         environment = environment or self.config.default_target_environment
+        environment = Environment.sanitize_name(environment)
         if not skip_janitor and environment.lower() == c.PROD:
             self._run_janitor()
 
@@ -1344,14 +1345,16 @@ def plan_builder(
                 ):
                     if model_name not in snapshots:
                         continue
-                    interval_unit = snapshots[model_name].node.interval_unit
+                    node = snapshots[model_name].node
+                    interval_unit = node.interval_unit
                     default_start = min(
                         default_start or sys.maxsize,
                         to_timestamp(
                             interval_unit.cron_prev(
                                 interval_unit.cron_floor(
-                                    max_interval_end_per_model.get(model_name, default_end),
-                                    estimate=True,
+                                    max_interval_end_per_model.get(
+                                        model_name, node.cron_floor(default_end)
+                                    ),
                                 ),
                                 estimate=True,
                             )

tests/core/test_integration.py

@@ -701,6 +701,72 @@ def test_cron_not_aligned_with_day_boundary(
         ]
 
 
+@time_machine.travel("2023-01-08 00:00:00 UTC")
+def test_cron_not_aligned_with_day_boundary_new_model(init_and_plan_context: t.Callable):
+    context, _ = init_and_plan_context("examples/sushi")
+
+    existing_model = context.get_model("sushi.waiter_revenue_by_day")
+    existing_model = SqlModel.parse_obj(
+        {
+            **existing_model.dict(),
+            "kind": existing_model.kind.copy(update={"forward_only": True}),
+        }
+    )
+    context.upsert_model(existing_model)
+
+    plan = context.plan_builder("prod", skip_tests=True).build()
+    context.apply(plan)
+
+    # Add a new model and make a change to a forward-only model.
+    # The cron of the new model is not aligned with the day boundary.
+    new_model = load_sql_based_model(
+        d.parse(
+            """
+        MODEL (
+            name memory.sushi.new_model,
+            kind FULL,
+            cron '0 8 * * *',
+            start '2023-01-01',
+        );
+
+        SELECT 1 AS one;
+        """
+        )
+    )
+    context.upsert_model(new_model)
+
+    existing_model = add_projection_to_model(t.cast(SqlModel, existing_model), literal=True)
+    context.upsert_model(existing_model)
+
+    plan = context.plan_builder("dev", skip_tests=True, enable_preview=True).build()
+    assert plan.missing_intervals == [
+        SnapshotIntervals(
+            snapshot_id=context.get_snapshot(
+                "memory.sushi.new_model", raise_if_missing=True
+            ).snapshot_id,
+            intervals=[(to_timestamp("2023-01-06"), to_timestamp("2023-01-07"))],
+        ),
+        SnapshotIntervals(
+            snapshot_id=context.get_snapshot(
+                "sushi.top_waiters", raise_if_missing=True
+            ).snapshot_id,
+            intervals=[
+                (to_timestamp("2023-01-06"), to_timestamp("2023-01-07")),
+                (to_timestamp("2023-01-07"), to_timestamp("2023-01-08")),
+            ],
+        ),
+        SnapshotIntervals(
+            snapshot_id=context.get_snapshot(
+                "sushi.waiter_revenue_by_day", raise_if_missing=True
+            ).snapshot_id,
+            intervals=[
+                (to_timestamp("2023-01-06"), to_timestamp("2023-01-07")),
+                (to_timestamp("2023-01-07"), to_timestamp("2023-01-08")),
+            ],
+        ),
+    ]
+
+
 @time_machine.travel("2023-01-08 00:00:00 UTC")
 def test_forward_only_monthly_model(init_and_plan_context: t.Callable):
     context, _ = init_and_plan_context("examples/sushi")