From 77867d76f8561fc0fa2228caf7e28a92c4f7160a Mon Sep 17 00:00:00 2001
From: "github-actions[bot]"
 <41898282+github-actions[bot]@users.noreply.github.com>
Date: Tue, 25 Apr 2023 12:01:17 -0400
Subject: [PATCH] fix partial parsing of versioned models - schedule child
 nodes if latest version has been modified (#7439) (#7460)

(cherry picked from commit 2739d5f4c4d221e215e52b2c4ae0bdc6a5421176)

Co-authored-by: Michelle Ark <MichelleArk@users.noreply.github.com>
---
 .changes/unreleased/Fixes-20230424-164649.yaml      |  6 ++++++
 core/dbt/parser/partial.py                          |  3 +++
 test/unit/utils.py                                  |  6 +-----
 .../partial_parsing/test_partial_parsing.py         | 13 ++++++++++---
 4 files changed, 20 insertions(+), 8 deletions(-)
 create mode 100644 .changes/unreleased/Fixes-20230424-164649.yaml

diff --git a/.changes/unreleased/Fixes-20230424-164649.yaml b/.changes/unreleased/Fixes-20230424-164649.yaml
new file mode 100644
index 00000000000..57b6646bcfe
--- /dev/null
+++ b/.changes/unreleased/Fixes-20230424-164649.yaml
@@ -0,0 +1,6 @@
+kind: Fixes
+body: Fix partial parsing of latest_version changes for downstream references
+time: 2023-04-24T16:46:49.721231-04:00
+custom:
+  Author: MichelleArk
+  Issue: "7369"
diff --git a/core/dbt/parser/partial.py b/core/dbt/parser/partial.py
index 4a0e3067ed2..bc2aabde2e2 100644
--- a/core/dbt/parser/partial.py
+++ b/core/dbt/parser/partial.py
@@ -851,6 +851,9 @@ def delete_schema_mssa_links(self, schema_file, dict_key, elem):
                     if self.saved_files[file_id]:
                         source_file = self.saved_files[file_id]
                         self.add_to_pp_files(source_file)
+                    # if the node's latest version has changed - need to reparse all referencing nodes to ensure correct ref resolution
+                    if node.is_versioned and node.latest_version != elem.get("latest_version"):
+                        self.schedule_referencing_nodes_for_parsing(elem_unique_id)
             # remove from patches
             schema_file.node_patches.remove(elem_unique_id)
 
diff --git a/test/unit/utils.py b/test/unit/utils.py
index edd73d563cf..33e6e04a9aa 100644
--- a/test/unit/utils.py
+++ b/test/unit/utils.py
@@ -336,11 +336,7 @@ def MockNode(package, name, resource_type=None, **kwargs):
 
     version = kwargs.get("version")
     search_name = name if version is None else f"{name}.v{version}"
-    unique_id = (
-        f"{str(resource_type)}.{package}.{name}"
-        # if version is None
-        # else f"{str(resource_type)}.{package}.{name}.v{version}"
-    )
+    unique_id = f"{str(resource_type)}.{package}.{name}"
     node = mock.MagicMock(
         __class__=cls,
         resource_type=resource_type,
diff --git a/tests/functional/partial_parsing/test_partial_parsing.py b/tests/functional/partial_parsing/test_partial_parsing.py
index fdaa2ff0f9c..bf6607b1610 100644
--- a/tests/functional/partial_parsing/test_partial_parsing.py
+++ b/tests/functional/partial_parsing/test_partial_parsing.py
@@ -304,36 +304,43 @@ def models(self):
         return {
             "model_one.sql": model_one_sql,
             "model_one_v2.sql": model_one_sql,
+            "model_one_downstream.sql": model_four2_sql,
             "schema.yml": models_versions_schema_yml,
         }
 
     def test_pp_versioned_models(self, project):
         results = run_dbt(["run"])
-        assert len(results) == 2
+        assert len(results) == 3
 
         manifest = get_manifest(project.project_root)
         model_one_node = manifest.nodes["model.test.model_one.v1"]
         assert not model_one_node.is_latest_version
         model_two_node = manifest.nodes["model.test.model_one.v2"]
         assert model_two_node.is_latest_version
+        # assert unpinned ref points to latest version
+        model_one_downstream_node = manifest.nodes["model.test.model_one_downstream"]
+        assert model_one_downstream_node.depends_on.nodes == ["model.test.model_one.v2"]
 
         # update versions schema.yml block - latest_version from 2 to 1
         write_file(
             models_versions_updated_schema_yml, project.project_root, "models", "schema.yml"
         )
         results = run_dbt(["--partial-parse", "run"])
-        assert len(results) == 2
+        assert len(results) == 3
 
         manifest = get_manifest(project.project_root)
         model_one_node = manifest.nodes["model.test.model_one.v1"]
         assert model_one_node.is_latest_version
         model_two_node = manifest.nodes["model.test.model_one.v2"]
         assert not model_two_node.is_latest_version
+        # assert unpinned ref points to latest version
+        model_one_downstream_node = manifest.nodes["model.test.model_one_downstream"]
+        assert model_one_downstream_node.depends_on.nodes == ["model.test.model_one.v1"]
 
         # update versioned model
         write_file(model_two_sql, project.project_root, "models", "model_one_v2.sql")
         results = run_dbt(["--partial-parse", "run"])
-        assert len(results) == 2
+        assert len(results) == 3
 
 
 class TestSources: