Repository: TobikoData/sqlmesh (python)

Issue:
Table diffing for large seed tables takes a long time. (mssql)
**Description**
I have a seed table with 1.5m rows which are all the countries/zip codes in the world.
The initial load takes 10 minutes with the standard configuration.
When I increase the batch_size to 100.000 it goes down to 3.5 minutes.

So first issue would be to see if the default batch size makes sense?

Then when changing a row I noted that it takes around 10 minutes after "sqlmesh plan" to print the first output, then it prints all 1.5m rows which takes about 7 minutes. Then the apply takes the same amount of time than the first apply.

**Suggestion**
Can we skip the diffing on large tables? Also then maybe remove the contents in the _seeds table content column, so it does not have to be inserted twice (I assume double the upload time and more storage).

**Reproduce**
To reproduce here download the allCountries https://download.geonames.org/export/zip/
I then added the header rows
`countryCode postalCode	placeName	stateCode	stateName	countryProvinceCode	countryProvinceName	communityCode	communityName	latitude	longitude	accuracy`

and used the model
```
MODEL (
  name seed.countries_utf8,
  kind SEED (
    path '$root/seeds/allCountriesUTF8.csv',
    batch_size 100000,
    csv_settings (
      delimiter='	'
    )
  ),
  columns (
    countryCode NVARCHAR(2),
    postalCode NVARCHAR(20),
    placeName NVARCHAR(180),
    stateCode NVARCHAR(20),
    stateName NVARCHAR(100),
    countryProvinceCode NVARCHAR(20),
    countryProvinceName NVARCHAR(100),
    communityCode NVARCHAR(20),
    communityName NVARCHAR(100),
    latitude NUMERIC(9, 6),
    longitude NUMERIC(9, 6),
    accuracy INTEGER
  )
)
```

Pull request description:
Chore: Don't store the seed's content in the state
Closes #2668

We only really need to store the seed's content in order for the diffing to work at plan time. It appears that this convenience feature cannot justify the storage and compute overhead it incurs, especially for larger seed files.

This update ensures that we no longer rely on the seed's content stored in the state.

Patch:
diff --git a/sqlmesh/core/context_diff.py b/sqlmesh/core/context_diff.py
index a1455d89..2f74f7eb 100644
--- a/sqlmesh/core/context_diff.py
+++ b/sqlmesh/core/context_diff.py
@@ -119,28 +119,10 @@ class ContextDiff(PydanticModel):
             if snapshot.snapshot_id not in added
             and snapshot.fingerprint != remote_snapshot_name_to_info[snapshot.name].fingerprint
         }
-        modified_local_seed_snapshot_ids = {
-            s.snapshot_id
-            for s in snapshots.values()
-            if s.is_seed and s.name in modified_snapshot_name_to_snapshot_info
-        }
-        modified_remote_snapshot_ids = {
-            s.snapshot_id for s in modified_snapshot_name_to_snapshot_info.values()
-        }
 
-        stored = {
-            **state_reader.get_snapshots(
-                [
-                    snapshot.snapshot_id
-                    for snapshot in snapshots.values()
-                    if snapshot.snapshot_id not in modified_local_seed_snapshot_ids
-                ]
-            ),
-            **state_reader.get_snapshots(
-                modified_remote_snapshot_ids | modified_local_seed_snapshot_ids,
-                hydrate_seeds=True,
-            ),
-        }
+        stored = state_reader.get_snapshots(
+            [*snapshots.values(), *modified_snapshot_name_to_snapshot_info.values()]
+        )
 
         merged_snapshots = {}
         modified_snapshots = {}
diff --git a/sqlmesh/core/model/definition.py b/sqlmesh/core/model/definition.py
index fd109e93..74142af1 100644
--- a/sqlmesh/core/model/definition.py
+++ b/sqlmesh/core/model/definition.py
@@ -6,7 +6,6 @@ import logging
 import sys
 import types
 import typing as t
-from difflib import unified_diff
 from functools import cached_property
 from pathlib import Path
 
@@ -1135,6 +1134,7 @@ class SeedModel(_SqlBasedModel):
     kind: SeedKind
     seed: Seed
     column_hashes_: t.Optional[t.Dict[str, str]] = Field(default=None, alias="column_hashes")
+    derived_columns_to_types: t.Optional[t.Dict[str, exp.DataType]] = None
     is_hydrated: bool = True
     source_type: Literal["seed"] = "seed"
 
@@ -1147,6 +1147,11 @@ class SeedModel(_SqlBasedModel):
         execution_time: t.Optional[TimeLike] = None,
         **kwargs: t.Any,
     ) -> t.Iterator[QueryOrDF]:
+        if not self.is_hydrated:
+            return
+        yield from self.render_seed()
+
+    def render_seed(self) -> t.Iterator[QueryOrDF]:
         self._ensure_hydrated()
 
         date_columns = []
@@ -1179,29 +1184,15 @@ class SeedModel(_SqlBasedModel):
             )
             yield df
 
-    def text_diff(self, other: Node) -> str:
-        if not isinstance(other, SeedModel):
-            return super().text_diff(other)
-
-        other._ensure_hydrated()
-        self._ensure_hydrated()
-
-        return "\n".join(
-            (
-                super().text_diff(other),
-                *unified_diff(
-                    self.seed.content.split("\n"),
-                    other.seed.content.split("\n"),
-                ),
-            )
-        ).strip()
-
     @property
     def columns_to_types(self) -> t.Optional[t.Dict[str, exp.DataType]]:
         if self.columns_to_types_ is not None:
             return self.columns_to_types_
-        self._ensure_hydrated()
-        return self._reader.columns_to_types
+        if self.derived_columns_to_types is not None:
+            return self.derived_columns_to_types
+        if self.is_hydrated:
+            return self._reader.columns_to_types
+        return None
 
     @property
     def column_hashes(self) -> t.Dict[str, str]:
@@ -1254,6 +1245,9 @@ class SeedModel(_SqlBasedModel):
                 "seed": Seed(content=""),
                 "is_hydrated": False,
                 "column_hashes_": self.column_hashes,
+                "derived_columns_to_types": self.columns_to_types
+                if self.columns_to_types_ is None
+                else None,
             }
         )
 
diff --git a/sqlmesh/core/model/kind.py b/sqlmesh/core/model/kind.py
index fd5a4e8d..e58d6a53 100644
--- a/sqlmesh/core/model/kind.py
+++ b/sqlmesh/core/model/kind.py
@@ -114,7 +114,7 @@ class ModelKindMixin:
     @property
     def only_execution_time(self) -> bool:
         """Whether or not this model only cares about execution time to render."""
-        return self.is_view or self.is_full or self.is_scd_type_2
+        return self.is_view or self.is_full
 
     @property
     def full_history_restatement_only(self) -> bool:
diff --git a/sqlmesh/core/model/meta.py b/sqlmesh/core/model/meta.py
index 99bd77f7..004fac4c 100644
--- a/sqlmesh/core/model/meta.py
+++ b/sqlmesh/core/model/meta.py
@@ -183,7 +183,9 @@ class ModelMeta(_Node):
 
         return partitions
 
-    @field_validator("columns_to_types_", mode="before")
+    @field_validator(
+        "columns_to_types_", "derived_columns_to_types", mode="before", check_fields=False
+    )
     @field_validator_v1_args
     def _columns_validator(
         cls, v: t.Any, values: t.Dict[str, t.Any]
diff --git a/sqlmesh/core/scheduler.py b/sqlmesh/core/scheduler.py
index 4519adb7..f0ac453a 100644
--- a/sqlmesh/core/scheduler.py
+++ b/sqlmesh/core/scheduler.py
@@ -8,7 +8,6 @@ from datetime import datetime
 from sqlmesh.core import constants as c
 from sqlmesh.core.console import Console, get_console
 from sqlmesh.core.environment import EnvironmentNamingInfo
-from sqlmesh.core.model import SeedModel
 from sqlmesh.core.notification_target import (
     NotificationEvent,
     NotificationTargetManager,
@@ -162,11 +161,6 @@ class Scheduler:
         }
         snapshots[snapshot.name] = snapshot
 
-        if isinstance(snapshot.node, SeedModel) and not snapshot.node.is_hydrated:
-            snapshot = self.state_sync.get_snapshots([snapshot], hydrate_seeds=True)[
-                snapshot.snapshot_id
-            ]
-
         is_deployable = deployability_index.is_deployable(snapshot)
 
         wap_id = self.snapshot_evaluator.evaluate(
diff --git a/sqlmesh/core/selector.py b/sqlmesh/core/selector.py
index 63216608..54618d77 100644
--- a/sqlmesh/core/selector.py
+++ b/sqlmesh/core/selector.py
@@ -80,9 +80,7 @@ class Selector:
             )
             env_models = {
                 s.name: s.model
-                for s in self._state_reader.get_snapshots(
-                    environment_snapshot_infos, hydrate_seeds=True
-                ).values()
+                for s in self._state_reader.get_snapshots(environment_snapshot_infos).values()
                 if s.is_model
             }
 
diff --git a/sqlmesh/core/snapshot/evaluator.py b/sqlmesh/core/snapshot/evaluator.py
index 92690ad5..c83ee10a 100644
--- a/sqlmesh/core/snapshot/evaluator.py
+++ b/sqlmesh/core/snapshot/evaluator.py
@@ -43,6 +43,7 @@ from sqlmesh.core.macros import RuntimeStage
 from sqlmesh.core.model import (
     IncrementalUnmanagedKind,
     Model,
+    SeedModel,
     SCDType2ByColumnKind,
     SCDType2ByTimeKind,
     ViewKind,
@@ -856,8 +857,10 @@ def _evaluation_strategy(snapshot: SnapshotInfoLike, adapter: EngineAdapter) ->
         klass = EmbeddedStrategy
     elif snapshot.is_symbolic or snapshot.is_audit:
         klass = SymbolicStrategy
-    elif snapshot.is_full or snapshot.is_seed:
+    elif snapshot.is_full:
         klass = FullRefreshStrategy
+    elif snapshot.is_seed:
+        klass = SeedStrategy
     elif snapshot.is_incremental_by_time_range:
         klass = IncrementalByTimeRangeStrategy
     elif snapshot.is_incremental_by_unique_key:
@@ -1323,6 +1326,40 @@ class FullRefreshStrategy(MaterializableStrategy):
         self._replace_query_for_model(model, name, query_or_df)
 
 
+class SeedStrategy(MaterializableStrategy):
+    def create(
+        self,
+        snapshot: Snapshot,
+        name: str,
+        is_table_deployable: bool,
+        is_snapshot_deployable: bool,
+        **render_kwargs: t.Any,
+    ) -> None:
+        super().create(snapshot, name, is_table_deployable, is_snapshot_deployable, **render_kwargs)
+
+        # For seeds we insert data at the time of table creation.
+        if is_table_deployable:
+            model = t.cast(SeedModel, snapshot.model)
+            for index, df in enumerate(model.render_seed()):
+                if index == 0:
+                    self._replace_query_for_model(model, name, df)
+                else:
+                    self.adapter.insert_append(name, df, columns_to_types=model.columns_to_types)
+
+    def insert(
+        self,
+        snapshot: Snapshot,
+        name: str,
+        query_or_df: QueryOrDF,
+        snapshots: t.Dict[str, Snapshot],
+        deployability_index: DeployabilityIndex,
+        batch_index: int,
+        **kwargs: t.Any,
+    ) -> None:
+        # Data has already been inserted at the time of table creation.
+        pass
+
+
 class SCDType2Strategy(MaterializableStrategy):
     def create(
         self,
diff --git a/sqlmesh/core/state_sync/base.py b/sqlmesh/core/state_sync/base.py
index 2910dc0f..cc377339 100644
--- a/sqlmesh/core/state_sync/base.py
+++ b/sqlmesh/core/state_sync/base.py
@@ -86,14 +86,12 @@ class StateReader(abc.ABC):
     def get_snapshots(
         self,
         snapshot_ids: t.Optional[t.Iterable[SnapshotIdLike]],
-        hydrate_seeds: bool = False,
     ) -> t.Dict[SnapshotId, Snapshot]:
         """Bulk fetch snapshots given the corresponding snapshot ids.
 
         Args:
             snapshot_ids: Iterable of snapshot ids to get. If not provided all
                 available snapshots will be returned.
-            hydrate_seeds: Whether to hydrate seed snapshots with the content.
 
         Returns:
             A dictionary of snapshot ids to snapshots for ones that could be found.
diff --git a/sqlmesh/core/state_sync/cache.py b/sqlmesh/core/state_sync/cache.py
index e022120a..7624f7e0 100644
--- a/sqlmesh/core/state_sync/cache.py
+++ b/sqlmesh/core/state_sync/cache.py
@@ -52,12 +52,10 @@ class CachingStateSync(DelegatingStateSync):
         return snapshot
 
     def get_snapshots(
-        self,
-        snapshot_ids: t.Optional[t.Iterable[SnapshotIdLike]],
-        hydrate_seeds: bool = False,
+        self, snapshot_ids: t.Optional[t.Iterable[SnapshotIdLike]]
     ) -> t.Dict[SnapshotId, Snapshot]:
         if snapshot_ids is None:
-            return self.state_sync.get_snapshots(snapshot_ids, hydrate_seeds)
+            return self.state_sync.get_snapshots(snapshot_ids)
 
         existing = {}
         missing = set()
@@ -72,17 +70,10 @@ class CachingStateSync(DelegatingStateSync):
                 self.snapshot_cache[snapshot_id] = (False, expire_at)
                 missing.add(snapshot_id)
             elif snapshot:
-                if (
-                    hydrate_seeds
-                    and isinstance(snapshot.node, SeedModel)
-                    and not snapshot.node.is_hydrated
-                ):
-                    missing.add(snapshot_id)
-                else:
-                    existing[snapshot_id] = snapshot
+                existing[snapshot_id] = snapshot
 
         if missing:
-            existing.update(self.state_sync.get_snapshots(missing, hydrate_seeds))
+            existing.update(self.state_sync.get_snapshots(missing))
 
         for snapshot_id, snapshot in existing.items():
             cached = self._from_cache(snapshot_id, now)
diff --git a/sqlmesh/core/state_sync/common.py b/sqlmesh/core/state_sync/common.py
index b0b3d2a1..3f6610dc 100644
--- a/sqlmesh/core/state_sync/common.py
+++ b/sqlmesh/core/state_sync/common.py
@@ -90,9 +90,8 @@ class CommonStateSyncMixin(StateSync):
     def get_snapshots(
         self,
         snapshot_ids: t.Optional[t.Iterable[SnapshotIdLike]],
-        hydrate_seeds: bool = False,
     ) -> t.Dict[SnapshotId, Snapshot]:
-        return self._get_snapshots(snapshot_ids, hydrate_seeds=hydrate_seeds)
+        return self._get_snapshots(snapshot_ids)
 
     def get_environment(self, environment: str) -> t.Optional[Environment]:
         return self._get_environment(environment)
@@ -323,7 +322,6 @@ class CommonStateSyncMixin(StateSync):
         self,
         snapshot_ids: t.Optional[t.Iterable[SnapshotIdLike]] = None,
         lock_for_update: bool = False,
-        hydrate_seeds: bool = False,
         hydrate_intervals: bool = True,
     ) -> t.Dict[SnapshotId, Snapshot]:
         """Fetches specified snapshots.
@@ -331,7 +329,6 @@ class CommonStateSyncMixin(StateSync):
         Args:
             snapshot_ids: The collection of IDs of snapshots to fetch
             lock_for_update: Lock the snapshot rows for future update
-            hydrate_seeds: Whether to hydrate seed snapshots with the content.
             hydrate_intervals: Whether to hydrate result snapshots with intervals.
 
         Returns:
diff --git a/sqlmesh/core/state_sync/engine_adapter.py b/sqlmesh/core/state_sync/engine_adapter.py
index 308e54b0..8ce39089 100644
--- a/sqlmesh/core/state_sync/engine_adapter.py
+++ b/sqlmesh/core/state_sync/engine_adapter.py
@@ -41,7 +41,6 @@ from sqlmesh.core.snapshot import (
     Intervals,
     Node,
     Snapshot,
-    SnapshotChangeCategory,
     SnapshotFingerprint,
     SnapshotId,
     SnapshotIdLike,
@@ -101,7 +100,6 @@ class EngineAdapterStateSync(CommonStateSyncMixin, StateSync):
     INTERVAL_BATCH_SIZE = 1000
     SNAPSHOT_BATCH_SIZE = 1000
     SNAPSHOT_MIGRATION_BATCH_SIZE = 500
-    SNAPSHOT_SEED_MIGRATION_BATCH_SIZE = 200
 
     def __init__(
         self,
@@ -117,7 +115,6 @@ class EngineAdapterStateSync(CommonStateSyncMixin, StateSync):
         self.console = console or get_console()
         self.snapshots_table = exp.table_("_snapshots", db=self.schema)
         self.environments_table = exp.table_("_environments", db=self.schema)
-        self.seeds_table = exp.table_("_seeds", db=self.schema)
         self.intervals_table = exp.table_("_intervals", db=self.schema)
         self.plan_dags_table = exp.table_("_plan_dags", db=self.schema)
         self.versions_table = exp.table_("_versions", db=self.schema)
@@ -146,12 +143,6 @@ class EngineAdapterStateSync(CommonStateSyncMixin, StateSync):
             "previous_finalized_snapshots": exp.DataType.build("text"),
         }
 
-        self._seed_columns_to_types = {
-            "name": exp.DataType.build("text"),
-            "version": exp.DataType.build("text"),
-            "content": exp.DataType.build("text"),
-        }
-
         self._interval_columns_to_types = {
             "id": exp.DataType.build("text"),
             "created_ts": exp.DataType.build("bigint"),
@@ -210,30 +201,16 @@ class EngineAdapterStateSync(CommonStateSyncMixin, StateSync):
         if snapshots:
             self._push_snapshots(snapshots)
 
-    def _push_snapshots(
-        self, snapshots: t.Iterable[Snapshot], overwrite: bool = False, push_seeds: bool = True
-    ) -> None:
+    def _push_snapshots(self, snapshots: t.Iterable[Snapshot], overwrite: bool = False) -> None:
         if overwrite:
             snapshots = tuple(snapshots)
             self.delete_snapshots(snapshots)
 
-        seed_contents = []
         snapshots_to_store = []
 
         for snapshot in snapshots:
-            if isinstance(snapshot.node, SeedModel) and snapshot.change_category in (
-                SnapshotChangeCategory.BREAKING,
-                SnapshotChangeCategory.NON_BREAKING,
-            ):
+            if isinstance(snapshot.node, SeedModel):
                 seed_model = t.cast(SeedModel, snapshot.node)
-                if push_seeds:
-                    seed_contents.append(
-                        {
-                            "name": snapshot.name,
-                            "version": snapshot.version,
-                            "content": seed_model.seed.content,
-                        }
-                    )
                 snapshot = snapshot.copy(update={"node": seed_model.to_dehydrated()})
             snapshots_to_store.append(snapshot)
 
@@ -243,13 +220,6 @@ class EngineAdapterStateSync(CommonStateSyncMixin, StateSync):
             columns_to_types=self._snapshot_columns_to_types,
         )
 
-        if push_seeds and seed_contents:
-            self.engine_adapter.insert_append(
-                self.seeds_table,
-                pd.DataFrame(seed_contents),
-                columns_to_types=self._seed_columns_to_types,
-            )
-
     def _update_versions(
         self,
         schema_version: int = SCHEMA_VERSION,
@@ -351,10 +321,6 @@ class EngineAdapterStateSync(CommonStateSyncMixin, StateSync):
                         )
                     )
 
-        seed_deletion_candidates = [t.snapshot for t in cleanup_targets if not t.dev_table_only]
-        if seed_deletion_candidates:
-            self._delete_seeds(seed_deletion_candidates)
-
         return cleanup_targets
 
     def delete_expired_environments(self) -> t.List[Environment]:
@@ -464,7 +430,6 @@ class EngineAdapterStateSync(CommonStateSyncMixin, StateSync):
         self,
         snapshot_ids: t.Optional[t.Iterable[SnapshotIdLike]] = None,
         lock_for_update: bool = False,
-        hydrate_seeds: bool = False,
         batch_size: t.Optional[int] = None,
     ) -> t.Iterator[exp.Expression]:
         for where in (
@@ -482,19 +447,6 @@ class EngineAdapterStateSync(CommonStateSyncMixin, StateSync):
                 .from_(exp.to_table(self.snapshots_table).as_("snapshots"))
                 .where(where)
             )
-            if hydrate_seeds:
-                query = query.select(exp.column("content", table="seeds")).join(
-                    exp.to_table(self.seeds_table).as_("seeds"),
-                    on=exp.and_(
-                        exp.column("name", table="snapshots").eq(exp.column("name", table="seeds")),
-                        exp.column("version", table="snapshots").eq(
-                            exp.column("version", table="seeds")
-                        ),
-                    ),
-                    join_type="left",
-                )
-            else:
-                query = query.select(exp.Null().as_("content"))
             if lock_for_update:
                 query = query.lock(copy=False)
             yield query
@@ -503,7 +455,6 @@ class EngineAdapterStateSync(CommonStateSyncMixin, StateSync):
         self,
         snapshot_ids: t.Optional[t.Iterable[SnapshotIdLike]] = None,
         lock_for_update: bool = False,
-        hydrate_seeds: bool = False,
         hydrate_intervals: bool = True,
     ) -> t.Dict[SnapshotId, Snapshot]:
         """Fetches specified snapshots or all snapshots.
@@ -511,7 +462,6 @@ class EngineAdapterStateSync(CommonStateSyncMixin, StateSync):
         Args:
             snapshot_ids: The collection of snapshot like objects to fetch.
             lock_for_update: Lock the snapshot rows for future update
-            hydrate_seeds: Whether to hydrate seed snapshots with the content.
             hydrate_intervals: Whether to hydrate result snapshots with intervals.
 
         Returns:
@@ -521,14 +471,13 @@ class EngineAdapterStateSync(CommonStateSyncMixin, StateSync):
         duplicates: t.Dict[SnapshotId, Snapshot] = {}
         model_cache = ModelCache(self._context_path / c.CACHE)
 
-        for query in self._get_snapshots_expressions(snapshot_ids, lock_for_update, hydrate_seeds):
-            for serialized_snapshot, name, identifier, _, seed_content in self._fetchall(query):
+        for query in self._get_snapshots_expressions(snapshot_ids, lock_for_update):
+            for serialized_snapshot, name, identifier, _ in self._fetchall(query):
                 snapshot = parse_snapshot(
                     model_cache,
                     serialized_snapshot=serialized_snapshot,
                     name=name,
                     identifier=identifier,
-                    seed_content=seed_content,
                 )
                 snapshot_id = snapshot.snapshot_id
                 if snapshot_id in snapshots:
@@ -1000,7 +949,7 @@ class EngineAdapterStateSync(CommonStateSyncMixin, StateSync):
         """Rollback to the previous migration."""
         logger.info("Starting migration rollback.")
         tables = (self.snapshots_table, self.environments_table, self.versions_table)
-        optional_tables = (self.seeds_table, self.intervals_table, self.plan_dags_table)
+        optional_tables = (self.intervals_table, self.plan_dags_table)
         versions = self.get_versions(validate=False)
         if versions.schema_version == 0:
             # Clean up state tables
@@ -1028,7 +977,6 @@ class EngineAdapterStateSync(CommonStateSyncMixin, StateSync):
             self.snapshots_table,
             self.environments_table,
             self.versions_table,
-            self.seeds_table,
             self.intervals_table,
             self.plan_dags_table,
         ):
@@ -1119,7 +1067,7 @@ class EngineAdapterStateSync(CommonStateSyncMixin, StateSync):
             ]
             if new_snapshots_to_push:
                 logger.info("Pushing %s migrated snapshots", len(new_snapshots_to_push))
-                self._push_snapshots(new_snapshots_to_push, push_seeds=False)
+                self._push_snapshots(new_snapshots_to_push)
             new_snapshots.clear()
             snapshot_id_mapping.clear()
 
@@ -1253,10 +1201,6 @@ class EngineAdapterStateSync(CommonStateSyncMixin, StateSync):
             except Exception:
                 logger.warning("Failed to unpause migrated snapshots", exc_info=True)
 
-    def _delete_seeds(self, snapshots: t.Iterable[SnapshotNameVersionLike]) -> None:
-        for where in self._snapshot_name_version_filter(snapshots, alias=None):
-            self.engine_adapter.delete_from(self.seeds_table, where=where)
-
     def _snapshot_ids_exist(
         self, snapshot_ids: t.Iterable[SnapshotIdLike], table_name: exp.Table
     ) -> t.Set[SnapshotId]:
@@ -1470,7 +1414,6 @@ def parse_snapshot(
     serialized_snapshot: str,
     name: str,
     identifier: str,
-    seed_content: t.Optional[str],
 ) -> Snapshot:
     payload = json.loads(serialized_snapshot)
 
@@ -1480,9 +1423,6 @@ def parse_snapshot(
     payload["node"] = model_cache.get_or_load(f"{name}_{identifier}", loader=loader)  # type: ignore
     snapshot = Snapshot(**payload)
 
-    if seed_content and isinstance(snapshot.node, SeedModel):
-        snapshot.node = snapshot.node.to_hydrated(seed_content)
-
     return snapshot
 
 
diff --git a/sqlmesh/migrations/v0050_drop_seeds_table.py b/sqlmesh/migrations/v0050_drop_seeds_table.py
new file mode 100644
index 00000000..706fae63
--- /dev/null
+++ b/sqlmesh/migrations/v0050_drop_seeds_table.py
@@ -0,0 +1,11 @@
+"""Drop the seeds table."""
+
+
+def migrate(state_sync, **kwargs):  # type: ignore
+    engine_adapter = state_sync.engine_adapter
+
+    seeds_table = "_seeds"
+    if state_sync.schema:
+        seeds_table = f"{state_sync.schema}.{seeds_table}"
+
+    engine_adapter.drop_table(seeds_table)
diff --git a/sqlmesh/schedulers/airflow/api.py b/sqlmesh/schedulers/airflow/api.py
index b14e192c..9744efbb 100644
--- a/sqlmesh/schedulers/airflow/api.py
+++ b/sqlmesh/schedulers/airflow/api.py
@@ -123,11 +123,7 @@ def get_snapshots() -> Response:
             )
             return _success(common.SnapshotIdsResponse(snapshot_ids=existing_snapshot_ids))
 
-        hydrate_seeds = "hydrate_seeds" in request.args
-        snapshots = list(
-            state_sync.get_snapshots(snapshot_ids, hydrate_seeds=hydrate_seeds).values()
-        )
-
+        snapshots = list(state_sync.get_snapshots(snapshot_ids).values())
         return _success(common.SnapshotsResponse(snapshots=snapshots))
 
 
diff --git a/sqlmesh/schedulers/airflow/client.py b/sqlmesh/schedulers/airflow/client.py
index 9017b8e1..3bafb42b 100644
--- a/sqlmesh/schedulers/airflow/client.py
+++ b/sqlmesh/schedulers/airflow/client.py
@@ -233,11 +233,7 @@ class AirflowClient(BaseAirflowClient):
         )
         raise_for_status(response)
 
-    def get_snapshots(
-        self, snapshot_ids: t.Optional[t.List[SnapshotId]], hydrate_seeds: bool = False
-    ) -> t.List[Snapshot]:
-        flags = ["hydrate_seeds"] if hydrate_seeds else []
-
+    def get_snapshots(self, snapshot_ids: t.Optional[t.List[SnapshotId]]) -> t.List[Snapshot]:
         output = []
 
         if snapshot_ids is not None:
@@ -246,12 +242,12 @@ class AirflowClient(BaseAirflowClient):
             ):
                 output.extend(
                     common.SnapshotsResponse.parse_obj(
-                        self._get(SNAPSHOTS_PATH, *flags, ids=ids_batch)
+                        self._get(SNAPSHOTS_PATH, ids=ids_batch)
                     ).snapshots
                 )
             return output
 
-        return common.SnapshotsResponse.parse_obj(self._get(SNAPSHOTS_PATH, *flags)).snapshots
+        return common.SnapshotsResponse.parse_obj(self._get(SNAPSHOTS_PATH)).snapshots
 
     def snapshots_exist(self, snapshot_ids: t.List[SnapshotId]) -> t.Set[SnapshotId]:
         output = set()
diff --git a/sqlmesh/schedulers/airflow/operators/targets.py b/sqlmesh/schedulers/airflow/operators/targets.py
index d06c1391..76f31ed6 100644
--- a/sqlmesh/schedulers/airflow/operators/targets.py
+++ b/sqlmesh/schedulers/airflow/operators/targets.py
@@ -10,7 +10,6 @@ from sqlalchemy.orm import Session
 
 from sqlmesh.core.engine_adapter import create_engine_adapter
 from sqlmesh.core.environment import EnvironmentNamingInfo
-from sqlmesh.core.model import SeedModel
 from sqlmesh.core.snapshot import (
     DeployabilityIndex,
     Snapshot,
@@ -148,15 +147,8 @@ class SnapshotEvaluationTarget(BaseTarget[commands.EvaluateCommandPayload], Pyda
             )
 
     def _get_command_payload(self, context: Context) -> t.Optional[commands.EvaluateCommandPayload]:
-        snapshot = self.snapshot
-        if isinstance(snapshot.node, SeedModel) and not snapshot.node.is_hydrated:
-            with util.scoped_state_sync() as state_sync:
-                snapshot = state_sync.get_snapshots([snapshot], hydrate_seeds=True)[
-                    snapshot.snapshot_id
-                ]
-
         return commands.EvaluateCommandPayload(
-            snapshot=snapshot,
+            snapshot=self.snapshot,
             parent_snapshots=self.parent_snapshots,
             start=self._get_start(context),
             end=self._get_end(context),
diff --git a/sqlmesh/schedulers/airflow/state_sync.py b/sqlmesh/schedulers/airflow/state_sync.py
index 29871d23..ad997210 100644
--- a/sqlmesh/schedulers/airflow/state_sync.py
+++ b/sqlmesh/schedulers/airflow/state_sync.py
@@ -105,7 +105,6 @@ class HttpStateSync(StateSync):
     def get_snapshots(
         self,
         snapshot_ids: t.Optional[t.Iterable[SnapshotIdLike]],
-        hydrate_seeds: bool = False,
     ) -> t.Dict[SnapshotId, Snapshot]:
         """Gets multiple snapshots from the rest api.
 
@@ -115,8 +114,7 @@ class HttpStateSync(StateSync):
         on the production server.
         """
         snapshots = self._client.get_snapshots(
-            [s.snapshot_id for s in snapshot_ids] if snapshot_ids is not None else None,
-            hydrate_seeds=hydrate_seeds,
+            [s.snapshot_id for s in snapshot_ids] if snapshot_ids is not None else None
         )
         return {snapshot.snapshot_id: snapshot for snapshot in snapshots}
 


Test patch:
diff --git a/tests/core/test_integration.py b/tests/core/test_integration.py
index 7d49c319..2e9573c8 100644
--- a/tests/core/test_integration.py
+++ b/tests/core/test_integration.py
@@ -1801,7 +1801,6 @@ def test_environment_catalog_mapping(init_and_plan_context: t.Callable):
             "physical.sqlmesh._environments",
             "physical.sqlmesh._intervals",
             "physical.sqlmesh._plan_dags",
-            "physical.sqlmesh._seeds",
             "physical.sqlmesh._snapshots",
             "physical.sqlmesh._versions",
         }
@@ -1822,7 +1821,6 @@ def test_environment_catalog_mapping(init_and_plan_context: t.Callable):
             "physical.sqlmesh._environments",
             "physical.sqlmesh._intervals",
             "physical.sqlmesh._plan_dags",
-            "physical.sqlmesh._seeds",
             "physical.sqlmesh._snapshots",
             "physical.sqlmesh._versions",
         }
@@ -1843,7 +1841,6 @@ def test_environment_catalog_mapping(init_and_plan_context: t.Callable):
             "physical.sqlmesh._environments",
             "physical.sqlmesh._intervals",
             "physical.sqlmesh._plan_dags",
-            "physical.sqlmesh._seeds",
             "physical.sqlmesh._snapshots",
             "physical.sqlmesh._versions",
         }
@@ -1865,7 +1862,6 @@ def test_environment_catalog_mapping(init_and_plan_context: t.Callable):
             "physical.sqlmesh._environments",
             "physical.sqlmesh._intervals",
             "physical.sqlmesh._plan_dags",
-            "physical.sqlmesh._seeds",
             "physical.sqlmesh._snapshots",
             "physical.sqlmesh._versions",
         }
diff --git a/tests/core/test_model.py b/tests/core/test_model.py
index 928aa84c..d2f2b0ab 100644
--- a/tests/core/test_model.py
+++ b/tests/core/test_model.py
@@ -636,12 +636,17 @@ def test_seed_hydration():
 
     model = load_sql_based_model(expressions, path=Path("./examples/sushi/models/test_model.sql"))
     assert model.is_hydrated
+    assert not model.derived_columns_to_types
 
     column_hashes = model.column_hashes
 
     dehydrated_model = model.to_dehydrated()
     assert not dehydrated_model.is_hydrated
     assert dehydrated_model.column_hashes == column_hashes
+    assert dehydrated_model.derived_columns_to_types == {
+        "id": exp.DataType.build("bigint"),
+        "name": exp.DataType.build("text"),
+    }
     assert dehydrated_model.seed.content == ""
 
     hydrated_model = dehydrated_model.to_hydrated(model.seed.content)
@@ -763,31 +768,6 @@ def test_seed_marker_substitution():
     assert len(model.seed.content) > 0
 
 
-def test_seed_model_diff(tmp_path):
-    model_a_csv_path = (tmp_path / "model_a.csv").absolute()
-    model_b_csv_path = (tmp_path / "model_b.csv").absolute()
-
-    with open(model_a_csv_path, "w", encoding="utf-8") as fd:
-        fd.write(
-            """key,value
-1,value_a
-"""
-        )
-
-    with open(model_b_csv_path, "w", encoding="utf-8") as fd:
-        fd.write(
-            """key,value
-2,value_b
-"""
-        )
-
-    model_a = create_seed_model("test_db.test_model", SeedKind(path=str(model_a_csv_path)))
-    model_b = create_seed_model("test_db.test_model", SeedKind(path=str(model_b_csv_path)))
-
-    diff = model_a.text_diff(model_b)
-    assert diff.endswith("-1,value_a\n+2,value_b")
-
-
 def test_seed_pre_post_statements():
     @macro()
     def bar(**kwargs) -> None:
diff --git a/tests/core/test_snapshot_evaluator.py b/tests/core/test_snapshot_evaluator.py
index 36e5498a..f6bd4eb2 100644
--- a/tests/core/test_snapshot_evaluator.py
+++ b/tests/core/test_snapshot_evaluator.py
@@ -3,11 +3,13 @@ from unittest.mock import call, patch
 
 import logging
 import pytest
+from pathlib import Path
 from pytest_mock.plugin import MockerFixture
 from sqlglot import expressions as exp
 from sqlglot import parse, parse_one, select
 
 from sqlmesh.core.audit import ModelAudit, StandaloneAudit
+from sqlmesh.core import dialect as d
 from sqlmesh.core.dialect import schema_, to_schema
 from sqlmesh.core.engine_adapter import EngineAdapter, create_engine_adapter
 from sqlmesh.core.engine_adapter.base import MERGE_SOURCE_ALIAS, MERGE_TARGET_ALIAS
@@ -1673,6 +1675,77 @@ def test_create_incremental_by_unique_no_intervals(adapter_mock, make_snapshot):
     )
 
 
+def test_create_seed(mocker: MockerFixture, adapter_mock, make_snapshot):
+    expressions = d.parse(
+        """
+        MODEL (
+            name db.seed,
+            kind SEED (
+              path '../seeds/waiter_names.csv',
+              batch_size 5,
+            )
+        );
+    """
+    )
+
+    model = load_sql_based_model(expressions, path=Path("./examples/sushi/models/test_model.sql"))
+
+    snapshot = make_snapshot(model)
+    snapshot.categorize_as(SnapshotChangeCategory.BREAKING)
+
+    evaluator = SnapshotEvaluator(adapter_mock)
+    evaluator.create([snapshot], {})
+
+    common_create_kwargs: t.Dict[str, t.Any] = dict(
+        columns_to_types={"id": exp.DataType.build("bigint"), "name": exp.DataType.build("text")},
+        storage_format=None,
+        partitioned_by=[],
+        partition_interval_unit=IntervalUnit.DAY,
+        clustered_by=[],
+        table_properties={},
+        table_description=None,
+    )
+
+    adapter_mock.create_table.assert_has_calls(
+        [
+            call(
+                f"sqlmesh__db.db__seed__{snapshot.version}__temp",
+                column_descriptions=None,
+                **common_create_kwargs,
+            ),
+            call(
+                f"sqlmesh__db.db__seed__{snapshot.version}",
+                column_descriptions={},
+                **common_create_kwargs,
+            ),
+        ]
+    )
+
+    replace_query_calls = adapter_mock.replace_query.call_args_list
+    assert len(replace_query_calls) == 1
+
+    df = replace_query_calls[0][0][1]
+    assert list(df.itertuples(index=False)) == [
+        (0, "Toby"),
+        (1, "Tyson"),
+        (2, "Ryan"),
+        (3, "George"),
+        (4, "Chris"),
+    ]
+
+    insert_append_calls = adapter_mock.insert_append.call_args_list
+    assert len(insert_append_calls) == 1
+
+    df = insert_append_calls[0][0][1]
+    assert list(df.itertuples(index=False)) == [
+        (5, "Max"),
+        (6, "Vincent"),
+        (7, "Iaroslav"),
+        (8, "Emma"),
+        (9, "Maia"),
+    ]
+
+
 def test_standalone_audit(mocker: MockerFixture, adapter_mock, make_snapshot):
     evaluator = SnapshotEvaluator(adapter_mock)
 
diff --git a/tests/core/test_state_sync.py b/tests/core/test_state_sync.py
index c6ce1003..15bcf618 100644
--- a/tests/core/test_state_sync.py
+++ b/tests/core/test_state_sync.py
@@ -1094,18 +1094,12 @@ def test_delete_expired_snapshots_seed(
 
     state_sync.push_snapshots([snapshot])
     assert set(state_sync.get_snapshots(None)) == {snapshot.snapshot_id}
-    assert state_sync.engine_adapter.fetchall(
-        "SELECT name, version, content FROM sqlmesh._seeds"
-    ) == [
-        (snapshot.name, snapshot.version, snapshot.model.seed.content),
-    ]
 
     assert state_sync.delete_expired_snapshots() == [
         SnapshotTableCleanupTask(snapshot=snapshot.table_info, dev_table_only=False),
     ]
 
     assert not state_sync.get_snapshots(None)
-    assert not state_sync.engine_adapter.fetchall("SELECT * FROM sqlmesh._seeds")
 
 
 def test_delete_expired_snapshots_batching(
@@ -1613,7 +1607,6 @@ def test_first_migration_failure(duck_conn, mocker: MockerFixture, tmp_path) ->
     assert not state_sync.engine_adapter.table_exists(state_sync.snapshots_table)
     assert not state_sync.engine_adapter.table_exists(state_sync.environments_table)
     assert not state_sync.engine_adapter.table_exists(state_sync.versions_table)
-    assert not state_sync.engine_adapter.table_exists(state_sync.seeds_table)
     assert not state_sync.engine_adapter.table_exists(state_sync.intervals_table)
 
 
@@ -1766,20 +1759,11 @@ def test_seed_hydration(
     assert snapshot.model.is_hydrated
     assert snapshot.model.seed.content == "header\n1\n2"
 
-    stored_snapshot = state_sync.get_snapshots([snapshot.snapshot_id], hydrate_seeds=False)[
-        snapshot.snapshot_id
-    ]
+    stored_snapshot = state_sync.get_snapshots([snapshot.snapshot_id])[snapshot.snapshot_id]
     assert isinstance(stored_snapshot.model, SeedModel)
     assert not stored_snapshot.model.is_hydrated
     assert stored_snapshot.model.seed.content == ""
 
-    stored_snapshot = state_sync.get_snapshots([snapshot.snapshot_id], hydrate_seeds=True)[
-        snapshot.snapshot_id
-    ]
-    assert isinstance(stored_snapshot.model, SeedModel)
-    assert stored_snapshot.model.is_hydrated
-    assert stored_snapshot.model.seed.content == "header\n1\n2"
-
 
 def test_nodes_exist(state_sync: EngineAdapterStateSync, make_snapshot: t.Callable):
     snapshot = make_snapshot(
@@ -2224,7 +2208,6 @@ def test_snapshot_batching(state_sync, mocker, make_snapshot):
                 "a",
                 "1",
                 "1",
-                None,
             ],
             [
                 make_snapshot(
@@ -2233,7 +2216,6 @@ def test_snapshot_batching(state_sync, mocker, make_snapshot):
                 "a",
                 "2",
                 "2",
-                None,
             ],
         ],
         [
@@ -2244,7 +2226,6 @@ def test_snapshot_batching(state_sync, mocker, make_snapshot):
                 "a",
                 "3",
                 "3",
-                None,
             ],
         ],
     ]
@@ -2278,14 +2259,6 @@ def test_seed_model_metadata_update(
 
     state_sync.push_snapshots([snapshot])
 
-    seed_query = (
-        exp.select("COUNT(*)")
-        .from_("sqlmesh._seeds")
-        .where(exp.column("version").eq(snapshot.version))
-    )
-
-    assert state_sync.engine_adapter.fetchone(seed_query)[0] == 1
-
     model = model.copy(update={"owner": "jen"})
     new_snapshot = make_snapshot(model)
     new_snapshot.previous_versions = snapshot.all_versions
@@ -2295,5 +2268,4 @@ def test_seed_model_metadata_update(
     assert snapshot.version == new_snapshot.version
 
     state_sync.push_snapshots([new_snapshot])
-    assert state_sync.engine_adapter.fetchone(seed_query)[0] == 1
     assert len(state_sync.get_snapshots([new_snapshot, snapshot])) == 2
diff --git a/tests/integrations/github/cicd/test_integration.py b/tests/integrations/github/cicd/test_integration.py
index b8d469d4..c068715d 100644
--- a/tests/integrations/github/cicd/test_integration.py
+++ b/tests/integrations/github/cicd/test_integration.py
@@ -1562,16 +1562,7 @@ def test_overlapping_changes_models(
 ```
 - `sushi.waiter_names`
 ```diff
---- 
-
-+++ 
-
-@@ -9,4 +9,5 @@
 
- 7,Iaroslav
- 8,Emma
- 9,Maia
-+10,Trey
 ```
 
 **Indirectly Modified:**
diff --git a/tests/schedulers/airflow/operators/test_targets.py b/tests/schedulers/airflow/operators/test_targets.py
index 61acc04e..2c543615 100644
--- a/tests/schedulers/airflow/operators/test_targets.py
+++ b/tests/schedulers/airflow/operators/test_targets.py
@@ -129,8 +129,6 @@ def test_evaluation_target_execute_seed_model(
 
     add_interval_mock.assert_called_once_with(snapshot, interval_ds, interval_ds, is_dev=False)
 
-    get_snapshots_mock.assert_called_once_with([snapshot], hydrate_seeds=True)
-
     evaluator_evaluate_mock.assert_called_once_with(
         snapshot,
         start=interval_ds,