improvements

Author: izeigerman

sqlmesh/core/plan/common.py

@@ -4,6 +4,9 @@
 
 
 def should_force_rebuild(old: Snapshot, new: Snapshot) -> bool:
+    if new.is_view:
+        # View models always need to be rebuilt to reflect updated upstream dependencies.
+        return True
     if old.virtual_environment_mode != new.virtual_environment_mode:
         # If the virtual environment mode has changed, then we need to rebuild
         return True

sqlmesh/core/plan/stages.py

@@ -304,12 +304,30 @@ def build(self, plan: EvaluatablePlan) -> t.List[PlanStage]:
                 )
             )
         else:
+            snapshots_to_create = self._get_snapshots_to_create(plan, snapshots)
             stages.append(
                 PhysicalLayerSchemaCreationStage(
-                    snapshots=self._get_snapshots_to_create(plan, snapshots),
-                    deployability_index=deployability_index,
+                    snapshots=snapshots_to_create, deployability_index=deployability_index
                 )
             )
+            forward_only_snapshots_to_create = []
+            if plan.is_dev:
+                for snapshot in snapshots_to_create:
+                    if snapshot.is_forward_only:
+                        forward_only_snapshots_to_create.append(snapshot)
+            if forward_only_snapshots_to_create:
+                stages.append(
+                    PhysicalLayerUpdateStage(
+                        snapshots=forward_only_snapshots_to_create,
+                        all_snapshots=snapshots,
+                        snapshots_with_missing_intervals={
+                            s.snapshot_id
+                            for s in snapshots_to_intervals
+                            if plan.is_selected_for_backfill(s.name)
+                        },
+                        deployability_index=deployability_index,
+                    )
+                )
 
         audit_only_snapshots = self._get_audit_only_snapshots(new_snapshots)
         if audit_only_snapshots:

sqlmesh/core/scheduler.py

@@ -627,16 +627,18 @@ def _dag(
                 # Add a separate node for table creation in case when there multiple concurrent
                 # evaluation nodes.
                 create_node = CreateNode(snapshot_name=snapshot.name)
+                dag.add(create_node, upstream_dependencies)
 
             for i, interval in enumerate(intervals):
                 node = EvaluateNode(snapshot_name=snapshot.name, interval=interval, batch_index=i)
-                dag.add(node, upstream_dependencies)
-
-                if len(intervals) > 1:
-                    dag.add(DummyNode(snapshot_name=snapshot.name), [node])
 
                 if create_node:
                     dag.add(node, [create_node])
+                else:
+                    dag.add(node, upstream_dependencies)
+
+                if len(intervals) > 1:
+                    dag.add(DummyNode(snapshot_name=snapshot.name), [node])
 
                 if batch_concurrency and i >= batch_concurrency:
                     batch_idx_to_wait_for = i - batch_concurrency

sqlmesh/core/snapshot/definition.py

@@ -1383,12 +1383,11 @@ def requires_schema_migration_in_prod(self) -> bool:
         return (
             self.is_paused
             and self.is_model
-            and not self.is_symbolic
+            and self.is_materialized
             and (
                 (self.previous_version and self.previous_version.version == self.version)
                 or self.model.forward_only
                 or bool(self.model.physical_version)
-                or self.is_view
                 or not self.virtual_environment_mode.is_full
             )
         )
@@ -1588,7 +1587,9 @@ def create(
                     # Similarly, if the model depends on past and the start date is not aligned with the
                     # model's start, we should consider this snapshot non-deployable.
                     this_deployable = False
-                    if not snapshot.is_paused or snapshot.is_indirect_non_breaking:
+                    if not snapshot.is_paused or (
+                        snapshot.is_indirect_non_breaking and snapshot.intervals
+                    ):
                         # This snapshot represents what's currently deployed in prod.
                         representative_shared_version_ids.add(node)
 

sqlmesh/core/snapshot/evaluator.py

@@ -731,7 +731,7 @@ def _evaluate_snapshot(
         ):
             adapter.execute(model.render_pre_statements(**render_statements_kwargs))
 
-            if not target_table_exists:
+            if not target_table_exists or (model.is_seed and not snapshot.intervals):
                 if (
                     not is_snapshot_deployable
                     and snapshot.is_forward_only

tests/core/test_integration.py

@@ -488,12 +488,6 @@ def test_full_history_restatement_model_regular_plan_preview_enabled(
     waiter_as_customer_snapshot = context.get_snapshot(
         "sushi.waiter_as_customer_by_day", raise_if_missing=True
     )
-    count_customers_active_snapshot = context.get_snapshot(
-        "sushi.count_customers_active", raise_if_missing=True
-    )
-    count_customers_inactive_snapshot = context.get_snapshot(
-        "sushi.count_customers_inactive", raise_if_missing=True
-    )
 
     plan = context.plan_builder("dev", skip_tests=True, enable_preview=True).build()
 
@@ -1461,6 +1455,18 @@ def test_indirect_non_breaking_downstream_of_forward_only(init_and_plan_context:
     plan = context.plan_builder("prod", skip_tests=True).build()
     assert plan.start == to_timestamp("2023-01-01")
     assert plan.missing_intervals == [
+        SnapshotIntervals(
+            snapshot_id=top_waiter_snapshot.snapshot_id,
+            intervals=[
+                (to_timestamp("2023-01-01"), to_timestamp("2023-01-02")),
+                (to_timestamp("2023-01-02"), to_timestamp("2023-01-03")),
+                (to_timestamp("2023-01-03"), to_timestamp("2023-01-04")),
+                (to_timestamp("2023-01-04"), to_timestamp("2023-01-05")),
+                (to_timestamp("2023-01-05"), to_timestamp("2023-01-06")),
+                (to_timestamp("2023-01-06"), to_timestamp("2023-01-07")),
+                (to_timestamp("2023-01-07"), to_timestamp("2023-01-08")),
+            ],
+        ),
         SnapshotIntervals(
             snapshot_id=non_breaking_snapshot.snapshot_id,
             intervals=[
@@ -1485,8 +1491,9 @@ def test_indirect_non_breaking_downstream_of_forward_only(init_and_plan_context:
 
 @time_machine.travel("2023-01-08 15:00:00 UTC")
 def test_breaking_only_impacts_immediate_children(init_and_plan_context: t.Callable):
-    context, plan = init_and_plan_context("examples/sushi")
-    context.apply(plan)
+    context, _ = init_and_plan_context("examples/sushi")
+    context.upsert_model(context.get_model("sushi.top_waiters").copy(update={"kind": FullKind()}))
+    context.plan("prod", skip_tests=True, auto_apply=True, no_prompts=True)
 
     breaking_model = context.get_model("sushi.orders")
     breaking_model = breaking_model.copy(update={"stamp": "force new version"})
@@ -2206,9 +2213,7 @@ def test_indirect_non_breaking_view_model_non_representative_snapshot(
     context.upsert_model(add_projection_to_model(t.cast(SqlModel, forward_only_model)))
     forward_only_model_snapshot_id = context.get_snapshot(forward_only_model_name).snapshot_id
     full_downstream_model_snapshot_id = context.get_snapshot(full_downstream_model_name).snapshot_id
-    full_downstream_model_2_snapshot_id = context.get_snapshot(
-        view_downstream_model_name
-    ).snapshot_id
+    view_downstream_model_snapshot_id = context.get_snapshot(view_downstream_model_name).snapshot_id
     dev_plan = context.plan("dev", auto_apply=True, no_prompts=True, enable_preview=False)
     assert (
         dev_plan.snapshots[forward_only_model_snapshot_id].change_category
@@ -2219,7 +2224,7 @@ def test_indirect_non_breaking_view_model_non_representative_snapshot(
         == SnapshotChangeCategory.INDIRECT_NON_BREAKING
     )
     assert (
-        dev_plan.snapshots[full_downstream_model_2_snapshot_id].change_category
+        dev_plan.snapshots[view_downstream_model_snapshot_id].change_category
         == SnapshotChangeCategory.INDIRECT_NON_BREAKING
     )
     assert not dev_plan.missing_intervals
@@ -2238,9 +2243,7 @@ def test_indirect_non_breaking_view_model_non_representative_snapshot(
     new_full_downstream_model = load_sql_based_model(new_full_downstream_model_expressions)
     context.upsert_model(new_full_downstream_model)
     full_downstream_model_snapshot_id = context.get_snapshot(full_downstream_model_name).snapshot_id
-    full_downstream_model_2_snapshot_id = context.get_snapshot(
-        view_downstream_model_name
-    ).snapshot_id
+    view_downstream_model_snapshot_id = context.get_snapshot(view_downstream_model_name).snapshot_id
     dev_plan = context.plan(
         "dev",
         categorizer_config=CategorizerConfig.all_full(),
@@ -2253,12 +2256,12 @@ def test_indirect_non_breaking_view_model_non_representative_snapshot(
         == SnapshotChangeCategory.BREAKING
     )
     assert (
-        dev_plan.snapshots[full_downstream_model_2_snapshot_id].change_category
+        dev_plan.snapshots[view_downstream_model_snapshot_id].change_category
         == SnapshotChangeCategory.INDIRECT_BREAKING
     )
     assert len(dev_plan.missing_intervals) == 2
     assert dev_plan.missing_intervals[0].snapshot_id == full_downstream_model_snapshot_id
-    assert dev_plan.missing_intervals[1].snapshot_id == full_downstream_model_2_snapshot_id
+    assert dev_plan.missing_intervals[1].snapshot_id == view_downstream_model_snapshot_id
 
     # Check that the representative view hasn't been created yet.
     assert not context.engine_adapter.table_exists(
@@ -2272,9 +2275,7 @@ def test_indirect_non_breaking_view_model_non_representative_snapshot(
     # Finally, make a non-breaking change to the full model in the same dev environment.
     context.upsert_model(add_projection_to_model(t.cast(SqlModel, new_full_downstream_model)))
     full_downstream_model_snapshot_id = context.get_snapshot(full_downstream_model_name).snapshot_id
-    full_downstream_model_2_snapshot_id = context.get_snapshot(
-        view_downstream_model_name
-    ).snapshot_id
+    view_downstream_model_snapshot_id = context.get_snapshot(view_downstream_model_name).snapshot_id
     dev_plan = context.plan(
         "dev",
         categorizer_config=CategorizerConfig.all_full(),
@@ -2287,10 +2288,13 @@ def test_indirect_non_breaking_view_model_non_representative_snapshot(
         == SnapshotChangeCategory.NON_BREAKING
     )
     assert (
-        dev_plan.snapshots[full_downstream_model_2_snapshot_id].change_category
+        dev_plan.snapshots[view_downstream_model_snapshot_id].change_category
         == SnapshotChangeCategory.INDIRECT_NON_BREAKING
     )
 
+    # Deploy changes to prod
+    context.plan("prod", auto_apply=True, no_prompts=True)
+
     # Check that the representative view has been created.
     assert context.engine_adapter.table_exists(
         context.get_snapshot(view_downstream_model_name).table_name()