[dagster-dbt] Fix issue with non-portable partial-parse (#33074)

## Summary & Motivation

Kinda a gnarly fix but the quick explanation here is that for seeds
specifically, dbt serializes information about the absolute path to a
given seed file inside the partial_parse.msgpack file that gets created
when you run `dbt parse`.

This information gets used when dbt attempts to load that seed file in a
run, meaning if you generated the manifest in a different system than
the one where you execute the command, seed execution can fail because
the path will be incorrect.

This strips out the seed information from the partial parse file, which
basically forces dbt to recalculate that information every time a dbt
command runs. This is thankfully very fast for seeds (it's just a
reference to the seed files), and so doesn't introduce latency in the
way that doing this for models would.

## How I Tested These Changes

## Changelog

> Insert changelog entry or delete this section.

Author: OwenKephart

python_modules/libraries/dagster-dbt/dagster_dbt/dbt_project.py

@@ -132,6 +132,42 @@ def _prepare_manifest(self, project: "DbtProject") -> None:
             .wait()
         )
 
+        # Remove seed entries from partial_parse to force re-parsing at runtime.
+        # This ensures seeds get correct root_path based on current project location.
+        self._invalidate_seeds_in_partial_parse(project)
+
+    def _invalidate_seeds_in_partial_parse(self, project: "DbtProject") -> None:
+        """Remove seed entries from partial_parse.msgpack to force re-parsing.
+
+        Seeds contain root_path which is an absolute path from build time. When state
+        is generated in one environment (e.g., CI/CD) and used in another (e.g., deployed
+        container), the root_path points to the wrong location and seed loading fails.
+
+        By removing seed entries from the cache, dbt will re-parse them at runtime with
+        the correct current project path. Models keep their cached data for fast loading.
+        """
+        import msgpack
+
+        partial_parse_path = project.project_dir / project.target_path / "partial_parse.msgpack"
+        if not partial_parse_path.exists():
+            return
+
+        with open(partial_parse_path, "rb") as f:
+            data = msgpack.unpack(f, raw=False, strict_map_key=False)
+
+        # Remove seed nodes
+        seed_node_ids = [k for k in data.get("nodes", {}).keys() if k.startswith("seed.")]
+        for seed_id in seed_node_ids:
+            del data["nodes"][seed_id]
+
+        # Remove seed file entries (CSVs)
+        seed_file_ids = [k for k in data.get("files", {}).keys() if k.lower().endswith(".csv")]
+        for file_id in seed_file_ids:
+            del data["files"][file_id]
+
+        with open(partial_parse_path, "wb") as f:
+            msgpack.pack(data, f)
+
 
 @record_custom
 class DbtProject(IHaveNew):

python_modules/libraries/dagster-dbt/dagster_dbt_tests/core/test_project.py

@@ -1,10 +1,11 @@
 import multiprocessing
 
+import msgpack
 import pytest
 from dagster._core.test_utils import environ
 from dagster._utils.test import copy_directory
 from dagster_dbt.dbt_manifest import validate_manifest
-from dagster_dbt.dbt_project import DbtProject
+from dagster_dbt.dbt_project import DagsterDbtProjectPreparer, DbtProject
 
 from dagster_dbt_tests.dbt_projects import test_jaffle_shop_path
 
@@ -53,3 +54,83 @@ def test_concurrent_processes(project_dir):
             assert proc.exitcode == 0
 
         assert my_project.manifest_path.exists()
+
+
+def test_invalidate_seeds_in_partial_parse(project_dir) -> None:
+    """Test that seed entries are removed from partial_parse.msgpack after preparation.
+
+    Seeds contain root_path which is an absolute path from build time. When state
+    is generated in one environment (CI/CD) and used in another (deployed container),
+    the root_path points to the wrong location and seed loading fails.
+
+    By removing seed entries from the cache, dbt will re-parse them at runtime with
+    the correct current project path. Models keep their cached data for fast loading.
+    """
+    my_project = DbtProject(project_dir)
+    partial_parse_path = my_project.project_dir / my_project.target_path / "partial_parse.msgpack"
+
+    # Prepare the project which generates partial_parse.msgpack
+    with environ({"DAGSTER_IS_DEV_CLI": "1"}):
+        my_project.prepare_if_dev()
+
+    assert partial_parse_path.exists(), "partial_parse.msgpack should exist after preparation"
+
+    # Read the partial_parse and verify seeds were removed
+    with open(partial_parse_path, "rb") as f:
+        data = msgpack.unpack(f, raw=False, strict_map_key=False)
+
+    # Check that no seed nodes remain
+    seed_node_ids = [k for k in data.get("nodes", {}).keys() if k.startswith("seed.")]
+    assert len(seed_node_ids) == 0, f"Expected no seed nodes, but found: {seed_node_ids}"
+
+    # Check that no seed files remain (CSVs)
+    seed_file_ids = [k for k in data.get("files", {}).keys() if k.lower().endswith(".csv")]
+    assert len(seed_file_ids) == 0, f"Expected no seed files, but found: {seed_file_ids}"
+
+    # Check that model nodes are preserved
+    model_node_ids = [k for k in data.get("nodes", {}).keys() if k.startswith("model.")]
+    assert len(model_node_ids) > 0, "Model nodes should be preserved"
+
+    # Check that model files are preserved
+    model_file_ids = [k for k in data.get("files", {}).keys() if ".sql" in k.lower()]
+    assert len(model_file_ids) > 0, "Model files should be preserved"
+
+
+def test_invalidate_seeds_handles_missing_partial_parse() -> None:
+    """Test that _invalidate_seeds_in_partial_parse handles missing file gracefully."""
+    with copy_directory(test_jaffle_shop_path) as project_dir:
+        my_project = DbtProject(project_dir)
+        partial_parse_path = (
+            my_project.project_dir / my_project.target_path / "partial_parse.msgpack"
+        )
+
+        # Ensure partial_parse doesn't exist
+        if partial_parse_path.exists():
+            partial_parse_path.unlink()
+
+        # Should not raise an error
+        preparer = DagsterDbtProjectPreparer()
+        preparer._invalidate_seeds_in_partial_parse(my_project)  # noqa: SLF001
+
+
+def test_seed_command_succeeds_after_invalidation() -> None:
+    """Test that dbt seed command works after seed entries are invalidated.
+
+    This is the end-to-end validation that removing seed entries from
+    partial_parse.msgpack allows dbt to correctly re-parse and load seeds.
+    """
+    from dagster_dbt import DbtCliResource
+
+    with copy_directory(test_jaffle_shop_path) as project_dir:
+        my_project = DbtProject(project_dir)
+
+        # Prepare the project (which invalidates seeds in partial_parse)
+        with environ({"DAGSTER_IS_DEV_CLI": "1"}):
+            my_project.prepare_if_dev()
+
+        # Run dbt seed - this should succeed because dbt will re-parse
+        # the seed files with correct paths
+        dbt = DbtCliResource(project_dir=my_project)
+        result = dbt.cli(["seed"]).wait()
+
+        assert result.is_successful(), "dbt seed failed"