Fix: duplicates on migrations (#862)

Author: tobymao

sqlmesh/core/model/seed.py

@@ -29,9 +29,7 @@ def columns_to_types(self) -> t.Dict[str, exp.DataType]:
     def column_hashes(self) -> t.Dict[str, str]:
         df = self._get_df()
         return {
-            column_name: str(  # type: ignore
-                zlib.crc32(df[column_name].to_string(index=False).encode("utf-8"))  # type: ignore
-            )
+            column_name: str(zlib.crc32(df[column_name].to_json().encode("utf-8")))
             for column_name in df.columns
         }
 

sqlmesh/core/state_sync/engine_adapter.py

@@ -449,7 +449,6 @@ def _migrate_rows(self) -> None:
         environments = self.get_environments()
 
         snapshot_mapping = {}
-        cache: t.Dict[SnapshotId, t.Dict] = {}
 
         for snapshot in all_snapshots.values():
             seen = set()
@@ -473,18 +472,9 @@ def _migrate_rows(self) -> None:
                     continue
 
                 queue.update(s.parents)
-                cached_env = cache.get(snapshot_id)
-
-                if cached_env:
-                    models.update(cached_env["models"])
-                    audits.update(cached_env["audits"])
-                else:
-                    models[s.name] = s.model
-
-                    for audit in s.audits:
-                        audits[audit.name] = audit
-
-            cache[snapshot_id] = env
+                models[s.name] = s.model
+                for audit in s.audits:
+                    audits[audit.name] = audit
 
             new_snapshot = deepcopy(snapshot)
 
@@ -496,7 +486,6 @@ def _migrate_rows(self) -> None:
                 models=models,
                 audits=audits,
             )
-
             new_snapshot.parents = tuple(
                 SnapshotId(
                     name=name,
@@ -560,7 +549,7 @@ def map_data_versions(
             }
 
         self.delete_snapshots(snapshot_mapping)
-        self._push_snapshots(snapshot_mapping.values())
+        self._push_snapshots(set(snapshot_mapping.values()), overwrite=True)
 
         updated_environments = []
 

sqlmesh/migrations/v0006_change_seed_hash.py

@@ -0,0 +1,5 @@
+"""Seed hashes moved from to_string to to_json for performance."""
+
+
+def migrate(state_sync):  # type: ignore
+    pass

tests/core/test_seed.py

@@ -43,9 +43,9 @@ def test_column_hashes():
 """
     seed = Seed(content=content)
     assert seed.column_hashes == {
-        "key": "1460551213",
-        "value": "3337736980",
-        "ds": "1377719241",
+        "key": "122302783",
+        "value": "1969959181",
+        "ds": "725407375",
     }
 
     content_column_changed = """key,value,ds
@@ -57,5 +57,5 @@ def test_column_hashes():
     seed_column_changed = Seed(content=content_column_changed)
     assert seed_column_changed.column_hashes == {
         **seed.column_hashes,
-        "ds": "646363718",
+        "ds": "3396890652",
     }

tests/core/test_snapshot.py

@@ -399,7 +399,7 @@ def test_fingerprint_seed_model():
     )
 
     expected_fingerprint = SnapshotFingerprint(
-        data_hash="4143164165",
+        data_hash="4076055159",
         metadata_hash="1120323454",
     )