Fix: plan physical tables progress bar (#3578)

Author: treysp

sqlmesh/core/plan/evaluator.py

@@ -223,12 +223,6 @@ def _push(
             for s in snapshots.values()
             if s.is_model and not s.is_symbolic and plan.is_selected_for_backfill(s.name)
         ]
-        snapshots_to_create_count = len(snapshots_to_create)
-
-        if snapshots_to_create_count > 0:
-            self.console.start_creation_progress(
-                snapshots_to_create_count, plan.environment, self.default_catalog
-            )
 
         completed = False
         try:
@@ -237,6 +231,9 @@ def _push(
                 snapshots,
                 allow_destructive_snapshots=plan.allow_destructive_models,
                 deployability_index=deployability_index,
+                on_start=lambda x: self.console.start_creation_progress(
+                    x, plan.environment, self.default_catalog
+                ),
                 on_complete=self.console.update_creation_progress,
             )
             completed = True

sqlmesh/core/snapshot/evaluator.py

@@ -261,6 +261,7 @@ def create(
         target_snapshots: t.Iterable[Snapshot],
         snapshots: t.Dict[SnapshotId, Snapshot],
         deployability_index: t.Optional[DeployabilityIndex] = None,
+        on_start: t.Optional[t.Callable] = None,
         on_complete: t.Optional[t.Callable[[SnapshotInfoLike], None]] = None,
         allow_destructive_snapshots: t.Set[str] = set(),
     ) -> None:
@@ -270,6 +271,7 @@ def create(
             target_snapshots: Target snapshots.
             snapshots: Mapping of snapshot ID to snapshot.
             deployability_index: Determines snapshots that are deployable in the context of this creation.
+            on_start: A callback to initialize the snapshot creation progress bar.
             on_complete: A callback to call on each successfully created snapshot.
             allow_destructive_snapshots: Set of snapshots that are allowed to have destructive schema changes.
         """
@@ -326,11 +328,11 @@ def _get_data_objects(schema: exp.Table, gateway: t.Optional[str] = None) -> t.S
                         table_deployability[table_name]
                     )
                 target_deployability_flags[snapshot.name].sort()
-            elif on_complete:
-                on_complete(snapshot)
 
         if not snapshots_to_create:
             return
+        if on_start:
+            on_start(len(snapshots_to_create))
         self._create_schemas(tables_by_schema, gateway_by_schema)
         self._create_snapshots(
             snapshots_to_create,
@@ -350,7 +352,7 @@ def _create_snapshots(
         on_complete: t.Optional[t.Callable[[SnapshotInfoLike], None]],
         allow_destructive_snapshots: t.Set[str],
     ) -> None:
-        """Internal method to create tables in parrallel."""
+        """Internal method to create tables in parallel."""
         with self.concurrent_context():
             concurrent_apply_to_snapshots(
                 snapshots_to_create,

tests/cli/test_cli.py

@@ -220,7 +220,8 @@ def test_plan_restate_model(runner, tmp_path):
     assert_duckdb_test(result)
     assert "No changes to plan: project files match the `prod` environment" in result.output
     assert "sqlmesh_example.full_model evaluated in" in result.output
-    assert_backfill_success(result)
+    assert_model_batches_executed(result)
+    assert_target_env_updated(result)
 
 
 @pytest.mark.parametrize("flag", ["--skip-backfill", "--dry-run"])
@@ -348,7 +349,6 @@ def test_plan_dev_create_from_virtual(runner, tmp_path):
     )
     assert result.exit_code == 0
     assert_new_env(result, "dev2", "dev", initialize=False)
-    assert_model_versions_created(result)
     assert_target_env_updated(result)
     assert_virtual_update(result)