fix: partial parsing - reparse downstream nodes when adding versioning

.changes/unreleased/Fixes-20250102-140543.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: Make partial parsing reparse referencing nodes of newly versioned models.
+time: 2025-01-02T14:05:43.629959-05:00
+custom:
+    Author: d-cole
+    Issue: "8872"

core/dbt/parser/partial.py

@@ -1,6 +1,6 @@
 import os
 from copy import deepcopy
-from typing import Callable, Dict, List, MutableMapping
+from typing import Callable, Dict, List, MutableMapping, Union
 
 from dbt.constants import DEFAULT_ENV_PLACEHOLDER
 from dbt.contracts.files import (
@@ -10,6 +10,7 @@
     parse_file_type_to_parser,
 )
 from dbt.contracts.graph.manifest import Manifest
+from dbt.contracts.graph.nodes import AnalysisNode, ModelNode, SeedNode, SnapshotNode
 from dbt.events.types import PartialParsingEnabled, PartialParsingFile
 from dbt.node_types import NodeType
 from dbt_common.context import get_invocation_context
@@ -820,7 +821,7 @@ def merge_patch(self, schema_file, key, patch, new_patch=False):
 
     # For model, seed, snapshot, analysis schema dictionary keys,
     # delete the patches and tests from the patch
-    def delete_schema_mssa_links(self, schema_file, dict_key, elem):
+    def delete_schema_mssa_links(self, schema_file, dict_key, elem) -> None:
         # find elem node unique_id in node_patches
         prefix = key_to_prefix[dict_key]
         elem_unique_ids = []
@@ -841,11 +842,12 @@ def delete_schema_mssa_links(self, schema_file, dict_key, elem):
                 elem_unique_id in self.saved_manifest.nodes
                 or elem_unique_id in self.saved_manifest.disabled
             ):
+                nodes: List[Union[ModelNode, SeedNode, SnapshotNode, AnalysisNode]] = []
                 if elem_unique_id in self.saved_manifest.nodes:
-                    nodes = [self.saved_manifest.nodes.pop(elem_unique_id)]
+                    nodes = [self.saved_manifest.nodes.pop(elem_unique_id)]  # type: ignore[list-item]
                 else:
                     # The value of disabled items is a list of nodes
-                    nodes = self.saved_manifest.disabled.pop(elem_unique_id)
+                    nodes = self.saved_manifest.disabled.pop(elem_unique_id)  # type: ignore[assignment]
                 # need to add the node source_file to pp_files
                 for node in nodes:
                     file_id = node.file_id
@@ -858,9 +860,9 @@ def delete_schema_mssa_links(self, schema_file, dict_key, elem):
                     # if the node's group has changed - need to reparse all referencing nodes to ensure valid ref access
                     if node.group != elem.get("group"):
                         self.schedule_referencing_nodes_for_parsing(node.unique_id)
-                    # If the latest version has changed or a version has been removed we need to
-                    # reparse referencing nodes.
-                    if node.is_versioned:
+                    # If the latest version has changed, a version has been removed, or a version has been added,
+                    #  we need to reparse referencing nodes.
+                    if node.is_versioned or elem.get("versions"):
                         self.schedule_referencing_nodes_for_parsing(node.unique_id)
             # remove from patches
             schema_file.node_patches.remove(elem_unique_id)

tests/functional/partial_parsing/test_versioned_models.py

@@ -1,4 +1,5 @@
 import pathlib
+from typing import Dict
 
 import pytest
 
@@ -120,3 +121,39 @@ def test_pp_versioned_models(self, project):
         write_file(model_one_sql, project.project_root, "models", "model_one.sql")
         with pytest.raises(DuplicateVersionedUnversionedError):
             run_dbt(["parse"])
+
+
+model_unversioned_schema_yml = """
+models:
+    - name: model_one
+      description: "The first model"
+"""
+
+
+model_versioned_schema_yml = """
+models:
+    - name: model_one
+      description: "The first model"
+      latest_version: 1
+      versions:
+        - v: 1
+"""
+
+
+class TestAddingVersioningToModel:
+    @pytest.fixture(scope="class")
+    def models(self) -> Dict[str, str]:
+        return {
+            "model_one.sql": model_one_sql,
+            "model_one_downstream.sql": model_one_downstream_sql,
+            "schema.yml": model_unversioned_schema_yml,
+        }
+
+    def test_pp_newly_versioned_models(self, project) -> None:
+        results = run_dbt(["run"])
+        assert len(results) == 2
+
+        # update schema.yml block - model_one is now versioned
+        write_file(model_versioned_schema_yml, project.project_root, "models", "schema.yml")
+        results = run_dbt(["--partial-parse", "run"])
+        assert len(results) == 2

tests/unit/parser/test_partial.py

@@ -1,6 +1,7 @@
 import time
 from copy import deepcopy
 from typing import Dict, List
+from unittest import mock
 
 import pytest
 
@@ -188,6 +189,34 @@ def test_schedule_macro_nodes_for_parsing_basic(partial_parsing):
     }
 
 
+def test_schedule_nodes_for_parsing_versioning(partial_parsing) -> None:
+    # Modify schema file to add versioning
+    schema_file_id = "my_test://" + normalize("models/schema.yml")
+    partial_parsing.new_files[schema_file_id].checksum = FileHash.from_contents("changed")
+    partial_parsing.new_files[schema_file_id].dfy = {
+        "version": 2,
+        "models": [
+            {
+                "name": "my_model",
+                "description": "Test model",
+                "latest_version": 1,
+                "versions": [{"v": 1}],
+            },
+            {"name": "python_model", "description": "python"},
+            {"name": "not_null", "model": "test.my_test.test_my_model"},
+        ],
+    }
+    with mock.patch.object(
+        partial_parsing, "schedule_referencing_nodes_for_parsing"
+    ) as mock_schedule_referencing_nodes_for_parsing:
+        partial_parsing.build_file_diff()
+        partial_parsing.get_parsing_files()
+
+        mock_schedule_referencing_nodes_for_parsing.assert_called_once_with(
+            "model.my_test.my_model"
+        )
+
+
 class TestFileDiff:
     @pytest.fixture
     def partial_parsing(self, manifest, files):