Fix: Support a proper migration of seed models

Author: izeigerman

sqlmesh/core/snapshot/definition.py

@@ -1480,7 +1480,7 @@ def expiration_ts(self) -> int:
     @property
     def supports_schema_migration_in_prod(self) -> bool:
         """Returns whether or not this snapshot supports schema migration when deployed to production."""
-        return self.is_paused and self.is_model and not self.is_symbolic and not self.is_seed
+        return self.is_paused and self.is_model and not self.is_symbolic
 
     @property
     def requires_schema_migration_in_prod(self) -> bool:

sqlmesh/core/snapshot/evaluator.py

@@ -1092,7 +1092,7 @@ def _migrate_snapshot(
                 **render_kwargs
             )
 
-            if table_exists:
+            if table_exists and not snapshot.is_seed:
                 self._migrate_target_table(
                     target_table_name=target_table_name,
                     snapshot=snapshot,
@@ -1105,14 +1105,19 @@ def _migrate_snapshot(
                     run_pre_post_statements=True,
                 )
             else:
+                if table_exists:
+                    assert snapshot.is_seed
+                    logger.info("Deleting the existing seed table '%s'", target_table_name)
+                    adapter.drop_table(target_table_name)
+
                 self._execute_create(
                     snapshot=snapshot,
                     table_name=snapshot.table_name(is_deployable=True),
                     is_table_deployable=True,
                     deployability_index=deployability_index,
                     create_render_kwargs=render_kwargs,
                     rendered_physical_properties=rendered_physical_properties,
-                    dry_run=True,
+                    dry_run=not snapshot.is_seed,
                 )
 
     # Retry in case when the table is migrated concurrently from another plan application

tests/core/test_integration.py

@@ -3232,6 +3232,62 @@ def test_virtual_environment_mode_dev_only_model_change_standalone_audit(
     context.apply(plan)
 
 
+@time_machine.travel("2023-01-08 15:00:00 UTC")
+def test_virtual_environment_mode_dev_only_seed_model_change_schema(
+    init_and_plan_context: t.Callable,
+):
+    context, plan = init_and_plan_context(
+        "examples/sushi", config="test_config_virtual_environment_mode_dev_only"
+    )
+    context.apply(plan)
+
+    new_csv = []
+    with open(context.path / "seeds" / "waiter_names.csv", "r") as fd:
+        is_header = True
+        for idx, line in enumerate(fd):
+            line = line.strip()
+            if not line:
+                continue
+            if is_header:
+                new_csv.append(line + ",new_column")
+                is_header = False
+            else:
+                new_csv.append(line + f",v{idx}")
+
+    with open(context.path / "seeds" / "waiter_names.csv", "w") as fd:
+        fd.write("\n".join(new_csv))
+
+    context.load()
+
+    downstream_model = context.get_model("sushi.waiter_as_customer_by_day")
+    downstream_model_kind = downstream_model.kind.dict()
+    downstream_model_kwargs = {
+        **downstream_model.dict(),
+        "kind": {
+            **downstream_model_kind,
+            "on_destructive_change": "allow",
+        },
+        "audits": [],
+        # Use the new column
+        "query": "SELECT '2023-01-07' AS event_date, new_column AS new_column FROM sushi.waiter_names",
+    }
+    context.upsert_model(SqlModel.parse_obj(downstream_model_kwargs))
+
+    context.plan("dev", auto_apply=True, no_prompts=True, skip_tests=True, enable_preview=True)
+
+    assert (
+        context.engine_adapter.fetchone(
+            "SELECT COUNT(*) FROM sushi__dev.waiter_as_customer_by_day"
+        )[0]
+        == len(new_csv) - 1
+    )
+
+    # Deploy to prod
+    context.clear_caches()
+    context.plan("prod", auto_apply=True, no_prompts=True, skip_tests=True)
+    assert "new_column" in context.engine_adapter.columns("sushi.waiter_as_customer_by_day")
+
+
 @time_machine.travel("2023-01-08 15:00:00 UTC")
 def test_restatement_plan_ignores_changes(init_and_plan_context: t.Callable):
     context, plan = init_and_plan_context("examples/sushi")