[dagster-dbt] Fix #33856 - incorrect asset key translation when generating column lineage and using DbtProjectComponent (#33857)

## Summary

Fixes [#33856](#33856) —
column lineage emitted by `.fetch_column_metadata()` referenced
default-derived `AssetKey`s instead of the translated keys actually
present in the asset graph when assets were defined via
`DBTProjectComponent`.

`_build_column_lineage_metadata()` resolved upstream dependencies by
calling `dagster_dbt_translator.get_asset_key(parent_resource_props)`.
But `DbtProjectComponentTranslator` (and any translator that customizes
translation via `get_asset_spec`) does **not** override `get_asset_key`
— it overrides `get_asset_spec`. As a result, every upstream
`TableColumnDep` pointed at a key that didn't exist in the graph, and
column lineage failed to resolve.

## Fix

Switch the upstream key resolution to
`dagster_dbt_translator.get_asset_spec(manifest, parent_unique_id,
project).key`. This is already the canonical pattern in the same file
(`_to_model_events` at `dbt_cli_event.py:401`) and in
`asset_utils.py:823`.

Chose this over an `isinstance(translator,
DbtProjectComponentTranslator)` branch because:

1. **Consistency** — matches how the materialized asset's own key is
resolved a few lines away. Using a different mechanism for upstream
resolution is what caused this drift bug in the first place.
2. **Generality** — works for any translator subclass that customizes
via `get_asset_spec` (the documented extension point), not just
`DbtProjectComponentTranslator`.
3. **No layering violation** — avoids importing a component-layer class
into core event code.
4. **Default path preserved** — `DagsterDbtTranslator.get_asset_spec()`
internally calls `self.get_asset_key()`, so users who only override
`get_asset_key` are unaffected.

To make this work, `project: DbtProject | None` is plumbed through
`_build_column_lineage_metadata` → `_get_lineage_metadata` →
`_get_materialization_metadata`. `_to_model_events` already had
`project` available, and `DbtCliInvocation.project` is the source at the
`_fetch_column_metadata` call site.

### Out of scope

`_to_observation_events_for_test` (`dbt_cli_event.py:498`) and
`freshness_builder.py` use the same `get_asset_key()` pattern and likely
have the same bug for `DbtProjectComponent` users. Kept this PR tight to
the reported column-lineage issue; those are follow-ups.

## Test plan

- [x] New regression test
`test_column_lineage_uses_get_asset_spec_for_upstream_keys` in
`test_columns_metadata.py` — uses a translator that overrides only
`get_asset_spec` (mimicking `DbtProjectComponentTranslator`'s pattern)
and asserts every upstream key in the emitted `TableColumnLineage`
carries the translated `renamed/...` prefix.
- [x] Verified the new test **fails** without the fix (`AssertionError:
Upstream key AssetKey(['raw_source_customers']) in column lineage was
not translated via get_asset_spec`) and passes with it.
- [x] All 12 duckdb-based column-lineage tests in
`test_columns_metadata.py` pass locally (snowflake/bigquery variants
skipped — require cloud credentials).
- [x] `make ruff` clean.

Author: stevenayers

python_modules/libraries/dagster-dbt/dagster_dbt/core/dbt_cli_event.py

@@ -65,6 +65,7 @@ def _build_column_lineage_metadata(
     manifest: Mapping[str, Any],
     dagster_dbt_translator: DagsterDbtTranslator,
     target_path: Path | None,
+    project: DbtProject | None,
 ) -> dict[str, Any]:
     """Process the lineage metadata for a dbt CLI event.
 
@@ -200,7 +201,9 @@ def _build_column_lineage_metadata(
             # Add the column dependency.
             column_deps.add(
                 TableColumnDep(
-                    asset_key=dagster_dbt_translator.get_asset_key(parent_resource_props),
+                    asset_key=dagster_dbt_translator.get_asset_spec(
+                        manifest, parent_resource_props["unique_id"], project
+                    ).key,
                     column_name=parent_column_name,
                 )
             )
@@ -342,6 +345,7 @@ def _get_lineage_metadata(
         translator: DagsterDbtTranslator,
         manifest: Mapping[str, Any],
         target_path: Path | None,
+        project: DbtProject | None,
     ) -> Mapping[str, Any]:
         try:
             column_data = self._event_history_metadata.get("columns", {})
@@ -362,6 +366,7 @@ def _get_lineage_metadata(
                     manifest=manifest,
                     dagster_dbt_translator=translator,
                     target_path=target_path,
+                    project=project,
                 )
         except Exception as e:
             logger.warning(
@@ -378,10 +383,11 @@ def _get_materialization_metadata(
         translator: DagsterDbtTranslator,
         manifest: Mapping[str, Any],
         target_path: Path | None,
+        project: DbtProject | None,
     ) -> dict[str, Any]:
         return {
             **self._get_default_metadata(manifest),
-            **self._get_lineage_metadata(translator, manifest, target_path),
+            **self._get_lineage_metadata(translator, manifest, target_path, project),
         }
 
     def _to_model_events(
@@ -393,7 +399,9 @@ def _to_model_events(
         project: DbtProject | None,
     ) -> Iterator[Output | AssetMaterialization]:
         asset_key = dagster_dbt_translator.get_asset_spec(manifest, self._unique_id, project).key
-        metadata = self._get_materialization_metadata(dagster_dbt_translator, manifest, target_path)
+        metadata = self._get_materialization_metadata(
+            dagster_dbt_translator, manifest, target_path, project
+        )
         if context and context.has_assets_def:
             yield Output(
                 value=None, output_name=asset_key.to_python_identifier(), metadata=metadata

python_modules/libraries/dagster-dbt/dagster_dbt/core/dbt_event_iterator.py

@@ -127,6 +127,7 @@ def _fetch_column_metadata(
                     manifest=invocation.manifest,
                     dagster_dbt_translator=invocation.dagster_dbt_translator,
                     target_path=invocation.target_path,
+                    project=invocation.project,
                 )
 
             except Exception as e:

python_modules/libraries/dagster-dbt/dagster_dbt_tests/core/dbt_packages/test_columns_metadata.py

@@ -2,6 +2,7 @@
 import os
 import shutil
 import subprocess
+from collections.abc import Mapping
 from pathlib import Path
 from typing import Any, cast
 
@@ -11,6 +12,7 @@
     AssetExecutionContext,
     AssetKey,
     AssetSelection,
+    AssetSpec,
     TableColumn,
     TableColumnDep,
     TableColumnLineage,
@@ -21,6 +23,8 @@
 from dagster._core.definitions.metadata.table import TableColumnConstraints
 from dagster_dbt.asset_decorator import dbt_assets
 from dagster_dbt.core.resource import DbtCliResource
+from dagster_dbt.dagster_dbt_translator import DagsterDbtTranslator
+from dagster_dbt.dbt_project import DbtProject
 from pytest_mock import MockFixture
 from sqlglot import Dialect
 
@@ -294,6 +298,81 @@ def my_dbt_assets(context: AssetExecutionContext, dbt: DbtCliResource):
     )
 
 
+@pytest.mark.parametrize(
+    "use_fetch_column_metadata",
+    [True, False],
+    ids=["adapter_path", "native_event_history_path"],
+)
+def test_column_lineage_uses_get_asset_spec_for_upstream_keys(
+    test_metadata_manifest: dict[str, Any],
+    use_fetch_column_metadata: bool,
+) -> None:
+    """Regression test for https://github.com/dagster-io/dagster/issues/33856.
+
+    Column lineage upstream asset keys must be resolved via
+    ``translator.get_asset_spec(...).key`` so that translators which customize
+    translation only by overriding ``get_asset_spec`` (e.g.
+    ``DbtProjectComponentTranslator``) produce lineage entries that point at the
+    *translated* keys actually present in the asset graph, rather than at the
+    default-derived keys returned by ``get_asset_key``.
+
+    Exercised against both lineage-building paths through
+    ``_build_column_lineage_metadata``:
+
+    - ``adapter_path``: the post-run adapter-querying thread invoked when the
+      user calls ``.fetch_column_metadata()``
+      (``dbt_event_iterator._fetch_column_metadata``).
+    - ``native_event_history_path``: the path driven by dbt's structured event
+      history when ``has_column_lineage_metadata`` is ``True``
+      (``dbt_cli_event._get_lineage_metadata``).
+    """
+
+    class SpecOverrideTranslator(DagsterDbtTranslator):
+        def get_asset_spec(
+            self,
+            manifest: Mapping[str, Any],
+            unique_id: str,
+            project: DbtProject | None,
+        ) -> AssetSpec:
+            spec = super().get_asset_spec(manifest, unique_id, project)
+            return spec.replace_attributes(key=AssetKey(["renamed", *spec.key.path]))
+
+    translator = SpecOverrideTranslator()
+
+    @dbt_assets(manifest=test_metadata_manifest, dagster_dbt_translator=translator)
+    def my_dbt_assets(context: AssetExecutionContext, dbt: DbtCliResource):
+        cli_invocation = dbt.cli(["build"], context=context).stream()
+        if use_fetch_column_metadata:
+            cli_invocation = cli_invocation.fetch_column_metadata()
+        yield from cli_invocation
+
+    result = materialize(
+        [my_dbt_assets],
+        resources={"dbt": DbtCliResource(project_dir=os.fspath(test_metadata_path))},
+    )
+    assert result.success
+
+    upstream_keys_in_lineage: set[AssetKey] = set()
+    for event in result.get_asset_materialization_events():
+        lineage = TableMetadataSet.extract(event.materialization.metadata).column_lineage
+        if lineage is None:
+            continue
+        for col_deps in lineage.deps_by_column.values():
+            for dep in col_deps:
+                upstream_keys_in_lineage.add(dep.asset_key)
+
+    # We need at least one lineage entry for the assertion to be meaningful.
+    assert upstream_keys_in_lineage, (
+        "Expected at least one column lineage entry in the materialization metadata"
+    )
+    # Every upstream key referenced in column lineage must be the translated key
+    # ("renamed/..."), not the default-derived key.
+    for key in upstream_keys_in_lineage:
+        assert key.path[0] == "renamed", (
+            f"Upstream key {key} in column lineage was not translated via get_asset_spec"
+        )
+
+
 @pytest.mark.parametrize(
     "use_experimental_fetch_column_schema",
     [True, False],