Update SchemaSourceFile, partial parser

core/dbt/contracts/files.py

@@ -192,6 +192,7 @@ class SchemaSourceFile(BaseSourceFile):
     sources: List[str] = field(default_factory=list)
     exposures: List[str] = field(default_factory=list)
     metrics: List[str] = field(default_factory=list)
+    snapshots: List[str] = field(default_factory=list)
     # The following field will no longer be used. Leaving
     # here to avoid breaking existing projects. To be removed
     # later if possible.

core/dbt/contracts/graph/manifest.py

@@ -59,6 +59,7 @@
     SeedNode,
     SemanticModel,
     SingularTestNode,
+    SnapshotNode,
     SourceDefinition,
     UnitTestDefinition,
     UnitTestFileFixture,
@@ -1600,12 +1601,14 @@ def add_node(self, source_file: AnySourceFile, node: ManifestNode, test_from=Non
             if isinstance(node, GenericTestNode):
                 assert test_from
                 source_file.add_test(node.unique_id, test_from)
-            if isinstance(node, Metric):
+            elif isinstance(node, Metric):
                 source_file.metrics.append(node.unique_id)
-            if isinstance(node, Exposure):
+            elif isinstance(node, Exposure):
                 source_file.exposures.append(node.unique_id)
-            if isinstance(node, Group):
+            elif isinstance(node, Group):
                 source_file.groups.append(node.unique_id)
+            elif isinstance(node, SnapshotNode):
+                source_file.snapshots.append(node.unique_id)
         elif isinstance(source_file, FixtureSourceFile):
             pass
         else:

core/dbt/parser/partial.py

@@ -658,10 +658,14 @@ def handle_schema_file_changes(self, schema_file, saved_yaml_dict, new_yaml_dict
             key_diff = self.get_diff_for(dict_key, saved_yaml_dict, new_yaml_dict)
             if key_diff["changed"]:
                 for elem in key_diff["changed"]:
+                    if dict_key == "snapshots" and "relation" in elem:
+                        self.delete_yaml_snapshot(schema_file, elem)
                     self.delete_schema_mssa_links(schema_file, dict_key, elem)
                     self.merge_patch(schema_file, dict_key, elem, True)
             if key_diff["deleted"]:
                 for elem in key_diff["deleted"]:
+                    if dict_key == "snapshots" and "relation" in elem:
+                        self.delete_yaml_snapshot(schema_file, elem)
                     self.delete_schema_mssa_links(schema_file, dict_key, elem)
             if key_diff["added"]:
                 for elem in key_diff["added"]:
@@ -673,6 +677,8 @@ def handle_schema_file_changes(self, schema_file, saved_yaml_dict, new_yaml_dict
                         continue
                     elem = self.get_schema_element(new_yaml_dict[dict_key], name)
                     if elem:
+                        if dict_key == "snapshots" and "relation" in elem:
+                            self.delete_yaml_snapshot(schema_file, elem)
                         self.delete_schema_mssa_links(schema_file, dict_key, elem)
                         self.merge_patch(schema_file, dict_key, elem, True)
 
@@ -868,6 +874,19 @@ def remove_tests(self, schema_file, dict_key, name):
                 self.saved_manifest.nodes.pop(test_unique_id)
         schema_file.remove_tests(dict_key, name)
 
+    def delete_yaml_snapshot(self, schema_file, snapshot_dict):
+        snapshot_name = snapshot_dict["name"]
+        snapshots = schema_file.snapshots.copy()
+        for unique_id in snapshots:
+            if unique_id in self.saved_manifest.nodes:
+                snapshot = self.saved_manifest.nodes[unique_id]
+                if snapshot.name == snapshot_name:
+                    self.saved_manifest.nodes.pop(unique_id)
+                    schema_file.snapshots.remove(unique_id)
+            elif unique_id in self.saved_manifest.disabled:
+                self.delete_disabled(unique_id, schema_file.file_id)
+                schema_file.snapshots.remove(unique_id)
+
     def delete_schema_source(self, schema_file, source_dict):
         # both patches, tests, and source nodes
         source_name = source_dict["name"]

core/dbt/parser/schemas.py

@@ -309,8 +309,9 @@ def _add_yaml_snapshot_nodes_to_manifest(
                 snapshot_node.raw_code = "select * from {{ " + snapshot["relation"] + " }}"
 
                 # Add our new node to the manifest, and note that ref lookup collections
-                # will need to be rebuilt.
-                self.manifest.add_node_nofile(snapshot_node)
+                # will need to be rebuilt. This adds the node unique_id to the "snapshots"
+                # list in the SchemaSourceFile.
+                self.manifest.add_node(block.file, snapshot_node)
                 rebuild_refs = True
 
         if rebuild_refs: