Fix schema file patch collection (#12055) (#12060)

(cherry picked from commit 963251d)

Co-authored-by: Gerda Shank <gerda@dbtlabs.com>

Author: github-actions[bot]

.changes/unreleased/Fixes-20250925-212728.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: Fix errors in partial parsing when working with versioned models
+time: 2025-09-25T21:27:28.591187-04:00
+custom:
+    Author: gshank
+    Issue: "11869"

core/dbt/parser/partial.py

@@ -37,6 +37,7 @@
     "seeds": "seed",
     "snapshots": "snapshot",
     "analyses": "analysis",
+    "sources": "source",
 }
 
 
@@ -178,6 +179,10 @@ def get_parsing_files(self):
             self.add_to_saved(file_id)
         # Need to process schema files next, because the dictionaries
         # need to be in place for handling SQL file changes
+        # The reverse sort here is just to ensure that the schema file
+        # processing order test case works, because otherwise the order
+        # of processing the schema files is not guaranteed.
+        self.file_diff["changed_schema_files"].sort(reverse=True)
         for file_id in self.file_diff["changed_schema_files"]:
             self.processing_file = file_id
             self.change_schema_file(file_id)
@@ -244,6 +249,10 @@ def handle_added_schema_file(self, source_file):
                 # be properly patched
                 if "overrides" in source:
                     self.remove_source_override_target(source)
+        if "models" in source_file.pp_dict:
+            for model in source_file.pp_dict["models"]:
+                if "versions" in model:
+                    self.versioned_model_delete_schema_mssa_links(source_file, "models", model)
 
     def delete_disabled(self, unique_id, file_id):
         # This node/metric/exposure is disabled. Find it and remove it from disabled dictionary.
@@ -628,7 +637,8 @@ def change_schema_file(self, file_id):
         new_schema_file = deepcopy(self.new_files[file_id])
         saved_yaml_dict = saved_schema_file.dict_from_yaml
         new_yaml_dict = new_schema_file.dict_from_yaml
-        saved_schema_file.pp_dict = {}
+        if saved_schema_file.pp_dict is None:
+            saved_schema_file.pp_dict = {}
         self.handle_schema_file_changes(saved_schema_file, saved_yaml_dict, new_yaml_dict)
 
         # copy from new schema_file to saved_schema_file to preserve references
@@ -675,6 +685,8 @@ def handle_schema_file_changes(self, schema_file, saved_yaml_dict, new_yaml_dict
                     self.delete_schema_mssa_links(schema_file, dict_key, elem)
             if key_diff["added"]:
                 for elem in key_diff["added"]:
+                    if dict_key == "models" and "versions" in elem:
+                        self.versioned_model_delete_schema_mssa_links(schema_file, dict_key, elem)
                     self.merge_patch(schema_file, dict_key, elem, True)
             # Handle schema file updates due to env_var changes
             if dict_key in env_var_changes and dict_key in new_yaml_dict:
@@ -837,7 +849,19 @@ def delete_schema_mssa_links(self, schema_file, dict_key, elem) -> None:
             elem_name = parts[2]
             if elem_name == elem["name"]:
                 elem_unique_ids.append(unique_id)
+        self._delete_schema_mssa_links(schema_file, dict_key, elem, elem_unique_ids)
 
+    def versioned_model_delete_schema_mssa_links(self, schema_file, dict_key, elem) -> None:
+        elem_unique_ids = []
+        # We need to look up possible existing models that this new or modified patch applies to
+        unique_id = f"model.{schema_file.project_name}.{elem['name']}"
+        if unique_id in self.saved_manifest.nodes:
+            elem_unique_ids.append(unique_id)
+        if not elem_unique_ids:
+            return
+        self._delete_schema_mssa_links(schema_file, dict_key, elem, elem_unique_ids)
+
+    def _delete_schema_mssa_links(self, schema_file, dict_key, elem, elem_unique_ids):
         # remove elem node and remove unique_id from node_patches
         for elem_unique_id in elem_unique_ids:
             # might have been already removed
@@ -886,6 +910,17 @@ def remove_tests(self, schema_file, dict_key, name):
             if test_unique_id in self.saved_manifest.nodes:
                 self.saved_manifest.nodes.pop(test_unique_id)
         schema_file.remove_tests(dict_key, name)
+        # We also need to remove tests in other schema files that
+        # reference this node.
+        unique_id = f"{key_to_prefix[dict_key]}.{schema_file.project_name}.{name}"
+        if unique_id in self.saved_manifest.child_map:
+            for child_id in self.saved_manifest.child_map[unique_id]:
+                if child_id.startswith("test") and child_id in self.saved_manifest.nodes:
+                    child_test = self.saved_manifest.nodes[child_id]
+                    if child_test.attached_node:
+                        if child_test.attached_node in self.saved_manifest.nodes:
+                            attached_node = self.saved_manifest.nodes[child_test.attached_node]
+                            self.update_in_saved(attached_node.file_id)
 
     def delete_yaml_snapshot(self, schema_file, snapshot_dict):
         snapshot_name = snapshot_dict["name"]

tests/functional/partial_parsing/test_pp_schema_file_order.py

@@ -0,0 +1,270 @@
+import os
+
+import pytest
+
+from dbt.tests.util import get_manifest, rm_file, run_dbt, write_file
+
+os.environ["DBT_PP_TEST"] = "true"
+
+colors_sql = """
+    select 'green' as first, 'red' as second, 'blue' as third
+"""
+
+another_v1_sql = """
+select * from {{ ref("colors") }}
+"""
+
+another_ref_sql = """
+select * from {{ ref("another") }}
+"""
+
+colors_yml = """
+models:
+  - name: colors
+    description: "a list of colors"
+  - name: another
+    description: "another model"
+    versions:
+      - v: 1
+"""
+
+colors_alt_yml = """
+models:
+  - name: colors
+    description: "a list of colors"
+  - name: another
+    description: "YET another model"
+    versions:
+      - v: 1
+"""
+
+foo_model_sql = """
+select 1 as id
+"""
+
+another_ref_yml = """
+models:
+  - name: another_ref
+    description: "model with reference to another ref"
+  - name: foo_model
+    description: "some random model"
+"""
+
+another_ref_alt_yml = """
+models:
+  - name: another_ref
+    description: "model with reference to another ref"
+  - name: foo_model
+    description: "some random other model"
+"""
+
+
+class TestSchemaFileOrder:
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "colors.sql": colors_sql,
+            "colors.yml": colors_yml,
+            "another_v1.sql": another_v1_sql,
+            "another_ref.sql": another_ref_sql,
+            "foo_model.sql": foo_model_sql,
+            "another_ref.yml": another_ref_yml,
+        }
+
+    def test_schema_file_order(self, project):
+
+        # initial run
+        results = run_dbt(["run"])
+        assert len(results) == 4
+
+        manifest = get_manifest(project.project_root)
+        model_id = "model.test.another_ref"
+        model = manifest.nodes.get(model_id)
+        assert model.description == "model with reference to another ref"
+
+        write_file(colors_alt_yml, project.project_root, "models", "colors.yml")
+        write_file(another_ref_alt_yml, project.project_root, "models", "another_ref.yml")
+        results = run_dbt(["--partial-parse", "run"])
+        assert len(results) == 4
+        manifest = get_manifest(project.project_root)
+        model = manifest.nodes.get(model_id)
+        assert model.name == "another_ref"
+        # The description here would be '' without the bug fix
+        assert model.description == "model with reference to another ref"
+
+
+foo_sql = """
+select 1 c
+"""
+
+bar_sql = """
+select 1 c
+"""
+
+bar_with_ref_sql = """
+select * from {{ ref('foo') }}
+"""
+
+foo_v2_sql = """
+select 1 c
+"""
+
+schema_yml = """
+# models/schema.yml
+models:
+  - name: foo
+    latest_version: 1
+    versions:
+      - v: 1
+      - v: 2
+"""
+
+foo_yml = """
+# models/foo.yml
+models:
+  - name: foo
+"""
+
+bar_yml = """
+# models/bar.yml
+models:
+  - name: bar
+    columns:
+      - name: c
+        tests:
+          - relationships:
+              to: ref('foo')
+              field: c
+"""
+
+foo_alt_yml = """
+# models/foo.yml
+models:
+  - name: foo
+    latest_version: 1
+    versions:
+      - v: 1
+      - v: 2
+"""
+
+
+class TestNewVersionedSchemaFile:
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "foo.sql": foo_sql,
+            "bar.sql": bar_with_ref_sql,
+        }
+
+    def test_schema_file_order_new_versions(self, project):
+        # This tests that when a model referring to an existing model
+        # which has had a version added in a yaml file has been re-parsed
+        # in order to fix the depends_on to the correct versioned model
+
+        # initial run
+        results = run_dbt(["compile"])
+        assert len(results) == 2
+
+        write_file(foo_v2_sql, project.project_root, "models", "foo_v2.sql")
+        write_file(schema_yml, project.project_root, "models", "schema.yml")
+
+        results = run_dbt(["compile"])
+
+
+class TestMoreNewVersionedSchemaFile:
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "foo.sql": foo_sql,
+            "bar.sql": bar_sql,
+            "foo.yml": foo_yml,
+            "bar.yml": bar_yml,
+        }
+
+    def test_more_schema_file_new_versions(self, project):
+
+        # initial run
+        results = run_dbt(["compile"])
+        assert len(results) == 3
+
+        rm_file(project.project_root, "models", "foo.sql")
+        write_file(foo_sql, project.project_root, "models", "foo_v1.sql")
+        write_file(foo_sql, project.project_root, "models", "foo_v2.sql")
+        write_file(foo_alt_yml, project.project_root, "models", "foo.yml")
+
+        results = run_dbt(["compile"])
+
+
+sources_yml = """
+sources:
+  - name: top_source
+    tables:
+      - name: abcd
+      - name: efgh
+      - name: ijkl
+"""
+
+abcd_sql = """
+select * from {{ source("top_source", "abcd") }}
+"""
+
+efgh_sql = """
+select * from {{ source("top_source", "efgh") }}
+"""
+
+ijkl_sql = """
+select * from {{ source("top_source", "ijkl") }}
+"""
+
+models_yml = """
+models:
+  - name: abcd
+    description: "abcd model"
+  - name: efgh
+    description: "efgh model"
+  - name: ijkl
+    description: "ijkl model"
+"""
+
+append_sources_yml = """
+      - name: mnop
+"""
+
+append_models_yml = """
+  - name: mnop
+    description: "mnop model"
+"""
+
+mnop_sql = """
+select * from {{ source("top_source", "mnop") }}
+"""
+
+
+class TestSourcesAndSchemaFiles:
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "sources.yml": sources_yml,
+            "abcd.sql": abcd_sql,
+            "efgh.sql": efgh_sql,
+            "ijkl.sql": ijkl_sql,
+            "_models.yml": models_yml,
+        }
+
+    def test_schema_file_order_new_versions(self, project):
+
+        # initial run
+        manifest = run_dbt(["parse"])
+        assert len(manifest.nodes) == 3
+
+        write_file(models_yml + append_models_yml, project.project_root, "models", "_models.yml")
+        write_file(mnop_sql, project.project_root, "models", "mnop.sql")
+        write_file(sources_yml + append_sources_yml, project.project_root, "models", "sources.yml")
+
+        manifest = run_dbt(["parse"])
+        assert len(manifest.nodes) == 4
+
+        # Without the fix the three original nodes will all be missing the
+        # the patch updates, including description, so description will be ""
+        for node in manifest.nodes.values():
+            assert node.description == f"{node.name} model"