Fix "dbt found two resources" error with multiple snapshot blocks in …

…one file (#4773) (#4877)

* Fix handling of multiple snapshot blocks in partial parsing

* Update tests for partial parsing snapshots

.changes/unreleased/Fixes-20220316-143959.yaml

@@ -0,0 +1,7 @@
+kind: Fixes
+body: Fix partial parsing bug with multiple snapshot blocks
+time: 2022-03-16T14:39:59.16756-04:00
+custom:
+  Author: gshank
+  Issue: "4771"
+  PR: "4773"

core/dbt/parser/partial.py

@@ -272,10 +272,10 @@ def update_mssat_in_saved(self, new_source_file, old_source_file):
         if self.already_scheduled_for_parsing(old_source_file):
             return
 
-        # These files only have one node.
-        unique_id = None
+        # These files only have one node except for snapshots
+        unique_ids = []
         if old_source_file.nodes:
-            unique_id = old_source_file.nodes[0]
+            unique_ids = old_source_file.nodes
         else:
             # It's not clear when this would actually happen.
             # Logging in case there are other associated errors.
@@ -286,7 +286,7 @@ def update_mssat_in_saved(self, new_source_file, old_source_file):
         self.deleted_manifest.files[file_id] = old_source_file
         self.saved_files[file_id] = deepcopy(new_source_file)
         self.add_to_pp_files(new_source_file)
-        if unique_id:
+        for unique_id in unique_ids:
             self.remove_node_in_saved(new_source_file, unique_id)
 
     def remove_node_in_saved(self, source_file, unique_id):
@@ -358,7 +358,7 @@ def remove_mssat_file(self, source_file):
         if not source_file.nodes:
             fire_event(PartialParsingMissingNodes(file_id=source_file.file_id))
             return
-        # There is generally only 1 node for SQL files, except for macros
+        # There is generally only 1 node for SQL files, except for macros and snapshots
         for unique_id in source_file.nodes:
             self.remove_node_in_saved(source_file, unique_id)
             self.schedule_referencing_nodes_for_parsing(unique_id)

test/integration/068_partial_parsing_tests/test-files/snapshot.sql

@@ -13,3 +13,17 @@ select * from {{ ref('orders') }}
 
 {% endsnapshot %}
 
+{% snapshot orders2_snapshot %}
+
+{{
+    config(
+      target_schema=schema,
+      strategy='check',
+      unique_key='id',
+      check_cols=['order_date'],
+    )
+}}
+
+select * from {{ ref('orders') }}
+
+{% endsnapshot %}

test/integration/068_partial_parsing_tests/test-files/snapshot2.sql

@@ -1,3 +1,4 @@
+- add a comment
 {% snapshot orders_snapshot %}
 
 {{
@@ -8,7 +9,22 @@
       check_cols=['status'],
     )
 }}
+
 select * from {{ ref('orders') }}
 
 {% endsnapshot %}
 
+{% snapshot orders2_snapshot %}
+
+{{
+    config(
+      target_schema=schema,
+      strategy='check',
+      unique_key='id',
+      check_cols=['order_date'],
+    )
+}}
+
+select * from {{ ref('orders') }}
+
+{% endsnapshot %}

test/integration/068_partial_parsing_tests/test_partial_parsing.py

@@ -503,10 +503,12 @@ def test_postgres_pp_snapshots(self):
         manifest = get_manifest()
         snapshot_id = 'snapshot.test.orders_snapshot'
         self.assertIn(snapshot_id, manifest.nodes)
+        snapshot2_id = 'snapshot.test.orders2_snapshot'
+        self.assertIn(snapshot2_id, manifest.nodes)
 
         # run snapshot
         results = self.run_dbt(["--partial-parse", "snapshot"])
-        self.assertEqual(len(results), 1)
+        self.assertEqual(len(results), 2)
 
         # modify snapshot
         self.copy_file('test-files/snapshot2.sql', 'snapshots/snapshot.sql')