Fixes 35383: Respect dbt source physical identifiers

ingestion/src/metadata/ingestion/source/database/dbt/metadata.py

@@ -808,15 +808,13 @@ def add_dbt_sources(self, key: str, manifest_node, dbt_objects: DbtObjects) -> N
         # A freshness check tests the source table itself. Sources are graph roots with no
         # depends_on, so resolving upstreams the way a dbt test node does always yields an
         # empty list and leaves the test case with no table to attach to.
-        source_fqn = fqn.build(
-            self.metadata,
-            entity_type=Table,
-            service_name=self.config.serviceName,
+        source_fqn = self._build_table_fqn(
             database_name=get_corrected_name(manifest_node.database),
             schema_name=get_corrected_name(manifest_node.schema_),
             table_name=get_source_physical_name(manifest_node),
+            literal_table_name=True,
         )
-        table_entity = self._get_table_entity(table_fqn=source_fqn) if source_fqn else None
+        table_entity = self._get_table_entity(table_fqn=source_fqn, literal_table_name=True) if source_fqn else None
         if not table_entity:
             logger.warning(f"Table entity not found for dbt source {key}, skipping its freshness test")
             return
@@ -836,9 +834,33 @@ def add_dbt_sources(self, key: str, manifest_node, dbt_objects: DbtObjects) -> N
         # instead of sniffing the parsed object downstream
         self.context.get().dbt_tests[key + "_freshness"][DbtCommonEnum.IS_FRESHNESS.value] = True
 
-    def _search_table(self, fqn_search_string: str | None) -> Table | None:
+    def _build_table_fqn(
+        self,
+        database_name: str | None,
+        schema_name: str | None,
+        table_name: str,
+        *,
+        literal_table_name: bool = False,
+    ) -> str | None:
+        return fqn.build(
+            self.metadata,
+            entity_type=Table,
+            service_name=self.config.serviceName,
+            database_name=(database_name or "*") if literal_table_name else database_name,
+            schema_name=(schema_name or "*") if literal_table_name else schema_name,
+            table_name=table_name,
+            skip_es_search=literal_table_name,
+        )
+
+    def _search_table(self, fqn_search_string: str | None, *, literal_table_name: bool = False) -> Table | None:
         if not fqn_search_string:
             return None
+        if literal_table_name:
+            # A source identifier can contain search metacharacters; choosing the first
+            # wildcard match would attach its metadata and lineage to an arbitrary table.
+            parts = fqn.split(fqn_search_string)
+            parts[-1] = parts[-1].replace("\\", "\\\\").replace("*", "\\*").replace("?", "\\?")
+            fqn_search_string = fqn._build(*parts)
         table_entities = get_entity_from_es_result(
             entity_list=self.metadata.es_search_from_fqn(
                 entity_type=Table,
@@ -854,7 +876,7 @@ def _search_table(self, fqn_search_string: str | None) -> Table | None:
         logger.debug("Found table entities from %s: %d entities", fqn_search_string, len(table_entities))
         return next(iter(filter(None, table_entities)), None) if table_entities else None
 
-    def _get_table_entity_in_mapped_database(self, table_fqn: str) -> Table | None:
+    def _get_table_entity_in_mapped_database(self, table_fqn: str, *, literal_table_name: bool = False) -> Table | None:
         """
         The manifest names a node's database the way the dbt adapter sees it: dbt-trino
         writes the Trino catalog, while the BigQuery connector ingests the same table under
@@ -875,9 +897,10 @@ def _get_table_entity_in_mapped_database(self, table_fqn: str) -> Table | None:
                     service_name=service_name,
                     database_name=mapped_database,
                     schema_name=schema_name,
-                    table_name=table_name,
+                    table_name=fqn.unquote_name(table_name) if literal_table_name else table_name,
                     skip_es_search=True,
-                )
+                ),
+                literal_table_name=literal_table_name,
             )
             if table_entity:
                 return table_entity
@@ -891,9 +914,11 @@ def _get_table_entity_in_mapped_database(self, table_fqn: str) -> Table | None:
         )
         return None
 
-    def _get_table_entity(self, table_fqn) -> Table | None:
+    def _get_table_entity(self, table_fqn, *, literal_table_name: bool = False) -> Table | None:
         try:
-            table_entity = self._get_table_entity_in_mapped_database(table_fqn) or self._search_table(table_fqn)
+            table_entity = self._get_table_entity_in_mapped_database(
+                table_fqn, literal_table_name=literal_table_name
+            ) or self._search_table(table_fqn, literal_table_name=literal_table_name)
             if table_entity:
                 logger.debug(f"Using Table Entity: {table_entity.fullyQualifiedName.root}with id {table_entity.id}")
                 return table_entity
@@ -910,9 +935,10 @@ def _get_table_entity(self, table_fqn) -> Table | None:
                     service_name="*",
                     database_name=database_name,
                     schema_name=schema_name,
-                    table_name=table_name,
+                    table_name=fqn.unquote_name(table_name) if literal_table_name else table_name,
+                    skip_es_search=literal_table_name,
                 )
-                table_entity = self._search_table(table_fqn)
+                table_entity = self._search_table(table_fqn, literal_table_name=literal_table_name)
                 if table_entity:
                     return table_entity
 
@@ -1002,7 +1028,11 @@ def yield_data_models(self, dbt_objects: DbtObjects) -> Iterable[Either[DataMode
                         logger.debug(f"Skipping DBT node: {key}.")
                         continue
 
-                    model_name = get_dbt_model_name(manifest_node)
+                    model_name = (
+                        get_source_physical_name(manifest_node)
+                        if resource_type == DbtCommonEnum.SOURCE.value
+                        else get_dbt_model_name(manifest_node)
+                    )
 
                     # snapshots can redirect output to a different schema/database via config.target_schema/target_database
                     if resource_type == "snapshot":
@@ -1041,19 +1071,20 @@ def yield_data_models(self, dbt_objects: DbtObjects) -> Iterable[Either[DataMode
                             )
                             or []
                         )
-                    table_fqn = fqn.build(
-                        self.metadata,
-                        entity_type=Table,
-                        service_name=self.config.serviceName,
+                    table_fqn = self._build_table_fqn(
                         database_name=get_corrected_name(node_database),
                         schema_name=get_corrected_name(node_schema),
                         table_name=model_name,
+                        literal_table_name=resource_type == DbtCommonEnum.SOURCE.value,
                     )
 
                     dbt_compiled_query = get_dbt_compiled_query(manifest_node)
                     dbt_raw_query = get_dbt_raw_query(manifest_node)
 
-                    if table_entity := self._get_table_entity(table_fqn=table_fqn):
+                    if table_entity := self._get_table_entity(
+                        table_fqn=table_fqn,
+                        literal_table_name=resource_type == DbtCommonEnum.SOURCE.value,
+                    ):
                         logger.debug(
                             f"Using Table Entity for datamodel: {table_entity.fullyQualifiedName.root}"
                             f"with id {table_entity.id}"
@@ -1144,13 +1175,16 @@ def parse_upstream_nodes_with_names(self, manifest_entities, dbt_node):
             for node in dbt_node.depends_on.nodes:
                 try:
                     parent_node = manifest_entities[node]
-                    table_name = get_dbt_model_name(parent_node)
-
                     parent_resource_type = getattr(
                         parent_node.resource_type,
                         "value",
                         parent_node.resource_type,
                     )
+                    table_name = (
+                        get_source_physical_name(parent_node)
+                        if parent_resource_type == DbtCommonEnum.SOURCE.value
+                        else get_dbt_model_name(parent_node)
+                    )
                     if parent_resource_type == "snapshot":
                         parent_location = get_snapshot_effective_schema_and_database(parent_node)
                         parent_database = parent_location.database
@@ -1172,17 +1206,22 @@ def parse_upstream_nodes_with_names(self, manifest_entities, dbt_node):
                     if check_ephemeral_node(parent_node):
                         upstream_nodes.extend(self.parse_upstream_nodes_with_names(manifest_entities, parent_node))
                     else:
-                        parent_fqn = fqn.build(
-                            self.metadata,
-                            entity_type=Table,
-                            service_name=self.config.serviceName,
+                        parent_fqn = self._build_table_fqn(
                             database_name=get_corrected_name(parent_database),
                             schema_name=get_corrected_name(parent_schema),
                             table_name=table_name,
+                            literal_table_name=parent_resource_type == DbtCommonEnum.SOURCE.value,
                         )
 
                         # check if the parent table exists in OM before adding it to the upstream list
-                        table_entity = self._get_table_entity(table_fqn=parent_fqn) if parent_fqn else None
+                        table_entity = (
+                            self._get_table_entity(
+                                table_fqn=parent_fqn,
+                                literal_table_name=parent_resource_type == DbtCommonEnum.SOURCE.value,
+                            )
+                            if parent_fqn
+                            else None
+                        )
                         # catalogMapping and searchAcrossDatabases can resolve the parent under another
                         # database or service, so lineage and tests follow the table that was found
                         if parent_fqn and table_entity:

ingestion/tests/unit/resources/datasets/manifest_source_identifier.json

@@ -0,0 +1,71 @@
+{
+  "metadata": {
+    "dbt_schema_version": "https://schemas.getdbt.com/dbt/manifest/v8.json",
+    "dbt_version": "1.3.0",
+    "generated_at": "2026-10-17T00:00:00.000000Z"
+  },
+  "sources": {
+    "source.jaffle_shop.raw.orders": {
+      "database": "RAW_DB",
+      "schema": "RAW",
+      "name": "orders",
+      "identifier": "raw_orders",
+      "resource_type": "source",
+      "package_name": "jaffle_shop",
+      "path": "models/sources.yml",
+      "original_file_path": "models/sources.yml",
+      "unique_id": "source.jaffle_shop.raw.orders",
+      "fqn": ["jaffle_shop", "raw", "orders"],
+      "source_name": "raw",
+      "source_description": "Raw order data",
+      "loader": "",
+      "description": "Orders from the warehouse source",
+      "columns": {
+        "order_id": {
+          "name": "order_id",
+          "description": "Source order identifier",
+          "data_type": "BIGINT"
+        }
+      },
+      "freshness": {
+        "warn_after": {"count": 12, "period": "hour"},
+        "error_after": {"count": 24, "period": "hour"}
+      },
+      "meta": {
+        "openmetadata": {
+          "domain": "Finance"
+        }
+      }
+    }
+  },
+  "nodes": {
+    "model.jaffle_shop.orders": {
+      "database": "RAW_DB",
+      "schema": "RAW",
+      "name": "orders",
+      "alias": "stg_orders",
+      "resource_type": "model",
+      "package_name": "jaffle_shop",
+      "path": "orders.sql",
+      "original_file_path": "models/orders.sql",
+      "unique_id": "model.jaffle_shop.orders",
+      "fqn": ["jaffle_shop", "orders"],
+      "checksum": {"name": "sha256", "checksum": "source-identifier-fixture"},
+      "config": {"materialized": "table"},
+      "depends_on": {
+        "macros": [],
+        "nodes": ["source.jaffle_shop.raw.orders"]
+      },
+      "raw_code": "select order_id from {{ source('raw', 'orders') }}",
+      "compiled_code": "select order_id from RAW_DB.RAW.raw_orders"
+    }
+  },
+  "exposures": {},
+  "metrics": {},
+  "macros": {},
+  "docs": {},
+  "selectors": {},
+  "disabled": {},
+  "parent_map": {},
+  "child_map": {}
+}