From 5be5564208b765db52005b95206df939e4805cfc Mon Sep 17 00:00:00 2001 From: Vamsi-klu Date: Sat, 17 Oct 2026 12:32:34 +0000 Subject: [PATCH 1/2] fix: respect dbt source physical identifiers (#35383) --- .../ingestion/source/database/dbt/metadata.py | 13 +- .../datasets/manifest_source_identifier.json | 71 ++++ ingestion/tests/unit/test_dbt.py | 357 ++++++++++++++++++ 3 files changed, 438 insertions(+), 3 deletions(-) create mode 100644 ingestion/tests/unit/resources/datasets/manifest_source_identifier.json diff --git a/ingestion/src/metadata/ingestion/source/database/dbt/metadata.py b/ingestion/src/metadata/ingestion/source/database/dbt/metadata.py index efd764b8acc7..c4dfef7b7f30 100644 --- a/ingestion/src/metadata/ingestion/source/database/dbt/metadata.py +++ b/ingestion/src/metadata/ingestion/source/database/dbt/metadata.py @@ -1002,7 +1002,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": @@ -1144,13 +1148,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 diff --git a/ingestion/tests/unit/resources/datasets/manifest_source_identifier.json b/ingestion/tests/unit/resources/datasets/manifest_source_identifier.json new file mode 100644 index 000000000000..5e013f6e1690 --- /dev/null +++ b/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": {} +} diff --git a/ingestion/tests/unit/test_dbt.py b/ingestion/tests/unit/test_dbt.py index 283a8372f3ea..6f61d5f59fa4 100644 --- a/ingestion/tests/unit/test_dbt.py +++ b/ingestion/tests/unit/test_dbt.py @@ -5489,3 +5489,360 @@ def test_meta_domain_is_keyed_by_the_table_that_was_found(self): assert [model_str(link.table_entity.fullyQualifiedName) for link in links] == [self.BQ_MODEL] assert source.extracted_domains == {self.BQ_MODEL: "Finance"} + + +class TestDbtSourceIdentifiers: + """Source lookup uses warehouse names while dbt references keep logical names.""" + + MANIFEST = Path(__file__).parent / "resources" / "datasets" / "manifest_source_identifier.json" + SOURCE_KEY = "source.jaffle_shop.raw.orders" + MODEL_KEY = "model.jaffle_shop.orders" + PHYSICAL_FQN = "trino_svc.RAW_DB.RAW.raw_orders" + LOGICAL_FQN = "trino_svc.RAW_DB.RAW.orders" + MODEL_FQN = "trino_svc.RAW_DB.RAW.stg_orders" + + @classmethod + def _payload(cls): + with cls.MANIFEST.open(encoding="utf-8") as manifest_file: + return json.load(manifest_file) + + @staticmethod + def _source(*fqns, **config): + source = TestCatalogMapping._source(*fqns, **config) + source.metadata.get_by_name.return_value = Domain( + id=uuid.uuid4(), name="Finance", fullyQualifiedName="Finance", domainType="Aggregate", description="Finance" + ) + return source + + @classmethod + def _objects(cls, source, payload=None, **files): + source.context.get().dbt_file = DbtFiles( + dbt_manifest=payload if payload is not None else cls._payload(), **files + ) + return next(source.get_dbt_objects()) + + @staticmethod + def _links(source, objects): + results = list(source.yield_data_models(objects)) + assert all(result.left is None for result in results), [result.left for result in results if result.left] + return {model_str(result.right.table_entity.fullyQualifiedName): result.right for result in results} + + @pytest.mark.parametrize("with_catalog", [False, True]) + def test_source_metadata_attaches_to_the_physical_table(self, with_catalog): + source = self._source(self.PHYSICAL_FQN, self.MODEL_FQN) + catalog = { + "metadata": {"dbt_schema_version": "https://schemas.getdbt.com/dbt/catalog/v1.json"}, + "nodes": {}, + "sources": { + self.SOURCE_KEY: { + "metadata": {"type": "TABLE", "schema": "RAW", "name": "raw_orders"}, + "columns": { + "order_id": {"name": "order_id", "type": "BIGINT", "index": 1, "comment": "Catalog order id"} + }, + "stats": {}, + "unique_id": self.SOURCE_KEY, + } + }, + } + objects = self._objects(source, dbt_catalog=catalog if with_catalog else None) + + links = self._links(source, objects) + + assert set(links) == {self.PHYSICAL_FQN, self.MODEL_FQN} + datamodel = links[self.PHYSICAL_FQN].datamodel + assert datamodel.resourceType == "source" + assert model_str(datamodel.description) == "Orders from the warehouse source" + assert datamodel.columns[0].name.root == "order_id" + assert model_str(datamodel.columns[0].description) == ( + "Catalog order id" if with_catalog else "Source order identifier" + ) + assert objects.dbt_manifest.sources[self.SOURCE_KEY].identifier == "raw_orders" + assert objects.dbt_manifest.sources[self.SOURCE_KEY].name == "orders" + assert objects.dbt_manifest.sources[self.SOURCE_KEY].unique_id == self.SOURCE_KEY + assert not hasattr(objects.dbt_manifest.sources[self.SOURCE_KEY], "alias") + + def test_source_lineage_connects_the_physical_table_to_the_aliased_model(self): + source = self._source(self.PHYSICAL_FQN, self.MODEL_FQN) + objects = self._objects(source) + + links = self._links(source, objects) + model = links[self.MODEL_FQN] + + assert model.datamodel.upstream == [self.PHYSICAL_FQN] + source_id = links[self.PHYSICAL_FQN].table_entity.id.root + assert model.upstream_table_ids == {self.PHYSICAL_FQN: source_id} + lineage = list(source.create_dbt_lineage(model)) + assert len(lineage) == 1 + assert lineage[0].left is None + assert lineage[0].right.lineage_request.edge.fromEntity.id.root == source_id + assert lineage[0].right.lineage_request.edge.toEntity.id.root == model.table_entity.id.root + + @pytest.mark.parametrize("physical_exists", [False, True]) + def test_logical_named_decoy_does_not_receive_metadata_or_lineage(self, physical_exists): + source = self._source(self.LOGICAL_FQN, self.MODEL_FQN, *([self.PHYSICAL_FQN] if physical_exists else [])) + + links = self._links(source, self._objects(source)) + + assert self.LOGICAL_FQN not in links + assert (self.PHYSICAL_FQN in links) is physical_exists + assert links[self.MODEL_FQN].datamodel.upstream == ([self.PHYSICAL_FQN] if physical_exists else []) + assert self.LOGICAL_FQN not in links[self.MODEL_FQN].upstream_table_ids + + @pytest.mark.parametrize("resource_type", ["source", DbtCommonEnum.SOURCE]) + def test_source_resource_type_accepts_strings_and_enums(self, resource_type): + source = self._source(self.PHYSICAL_FQN, self.MODEL_FQN) + objects = self._objects(source) + node = objects.dbt_manifest.sources[self.SOURCE_KEY] + node.resource_type = resource_type + + links = self._links(source, objects) + + assert links[self.PHYSICAL_FQN].datamodel.resourceType == "source" + assert links[self.MODEL_FQN].datamodel.upstream == [self.PHYSICAL_FQN] + + @pytest.mark.parametrize( + "attributes, expected_name", + [ + ({"identifier": "raw_orders", "alias": "wrong_alias"}, "raw_orders"), + ({"identifier": "orders"}, "orders"), + ({"identifier": None, "alias": "fallback_alias"}, "fallback_alias"), + ({"identifier": "", "alias": "fallback_alias"}, "fallback_alias"), + ({"alias": "fallback_alias"}, "fallback_alias"), + ({"identifier": None}, "orders"), + ({}, "orders"), + ], + ) + def test_source_name_fallback_preserves_existing_helper_behavior(self, attributes, expected_name): + expected_fqn = f"trino_svc.RAW_DB.RAW.{expected_name}" + source = self._source(expected_fqn, self.MODEL_FQN) + objects = self._objects(source) + node = SimpleNamespace(**objects.dbt_manifest.sources[self.SOURCE_KEY].__dict__) + del node.identifier + node.__dict__.update(attributes) + objects.dbt_manifest.sources[self.SOURCE_KEY] = node + + links = self._links(source, objects) + + assert expected_fqn in links + assert links[self.MODEL_FQN].datamodel.upstream == [expected_fqn] + + @pytest.mark.parametrize("resource_type", ["model", "seed", "snapshot"]) + @pytest.mark.parametrize("alias", [None, "model_alias"]) + def test_non_sources_keep_alias_name_and_snapshot_location(self, resource_type, alias): + database = "snap_db" if resource_type == "snapshot" else "RAW_DB" + schema = "snap_schema" if resource_type == "snapshot" else "RAW" + expected_fqn = f"trino_svc.{database}.{schema}.{alias or 'orders'}" + source = self._source(expected_fqn) + objects = self._objects(source) + node = SimpleNamespace(**objects.dbt_manifest.nodes[self.MODEL_KEY].__dict__) + node.resource_type = resource_type + node.alias = alias + node.identifier = "wrong_identifier" + node.depends_on.nodes = [] + if resource_type == "snapshot": + node.config = SimpleNamespace(target_schema=schema, target_database=database, materialized="snapshot") + objects.dbt_manifest.sources = {} + objects.dbt_manifest.nodes = {self.MODEL_KEY: node} + + links = self._links(source, objects) + upstream = source.parse_upstream_nodes_with_names( + {self.MODEL_KEY: node}, SimpleNamespace(depends_on=SimpleNamespace(nodes=[self.MODEL_KEY])) + ) + + assert set(links) == {expected_fqn} + assert [parent.fqn for parent in upstream] == [expected_fqn] + assert upstream[0].name == "orders" + + @pytest.mark.parametrize( + "filter_pattern, expected", + [ + ({"includes": ["raw_orders", "stg_orders"]}, True), + ({"includes": ["orders", "stg_orders"]}, False), + ({"excludes": ["raw_orders"]}, False), + ({"excludes": ["orders"]}, True), + ], + ) + def test_source_filters_use_the_physical_name_for_metadata_and_upstreams(self, filter_pattern, expected): + from metadata.generated.schema.type.filterPattern import FilterPattern + + source = self._source(self.PHYSICAL_FQN, self.MODEL_FQN) + source.source_config.tableFilterPattern = FilterPattern(**filter_pattern) + + links = self._links(source, self._objects(source)) + + assert (self.PHYSICAL_FQN in links) is expected + assert links[self.MODEL_FQN].datamodel.upstream == ([self.PHYSICAL_FQN] if expected else []) + + @pytest.mark.parametrize( + "physical_fqn, config", + [ + ("other_svc.RAW_DB.RAW.raw_orders", {"search_across_databases": True}), + ("other_svc.mapped_db.RAW.raw_orders", {"catalog_mapping": {"RAW_DB": "mapped_db"}}), + ( + "other_svc.mapped_db.RAW.raw_orders", + {"catalog_mapping": {"raw_db": "mapped_db"}, "search_across_databases": True}, + ), + ("trino_svc.RAW_DB.RAW.raw_orders", {"catalog_mapping": {"RAW_DB": "missing_db"}}), + ], + ) + def test_physical_identifier_follows_the_entity_resolved_by_existing_lookup(self, physical_fqn, config): + source = self._source(physical_fqn, self.MODEL_FQN, self.LOGICAL_FQN, **config) + + links = self._links(source, self._objects(source)) + + assert self.LOGICAL_FQN not in links + assert links[self.MODEL_FQN].datamodel.upstream == [physical_fqn] + assert links[self.MODEL_FQN].upstream_table_ids == {physical_fqn: links[physical_fqn].table_entity.id.root} + assert source.extracted_domains == {physical_fqn: "Finance"} + assert source.context.get().table_domains[physical_fqn].fullyQualifiedName == "Finance" + + def test_logical_source_namespace_resolves_tests_when_a_model_has_the_same_name(self): + source = self._source(self.PHYSICAL_FQN, self.MODEL_FQN, "trino_svc.RAW_DB.RAW.other_orders") + objects = self._objects(source) + manifest = objects.dbt_manifest + second_source = manifest.sources[self.SOURCE_KEY].model_copy(deep=True) + second_source.source_name = "other" + second_source.identifier = "other_orders" + second_key = "source.jaffle_shop.other.orders" + second_source.unique_id = second_key + second_source.fqn = ["jaffle_shop", "other", "orders"] + manifest.sources[second_key] = second_source + test_node = SimpleNamespace( + depends_on=SimpleNamespace(nodes=[self.MODEL_KEY, second_key, self.SOURCE_KEY]), + test_metadata=SimpleNamespace(kwargs={"model": "{{ source('raw', 'orders') }}"}), + ) + parents = source.parse_upstream_nodes_with_names({**manifest.sources, **manifest.nodes}, test_node) + test = { + DbtCommonEnum.MANIFEST_NODE.value: test_node, + DbtCommonEnum.UPSTREAM.value: [parent.fqn for parent in parents], + DbtCommonEnum.UPSTREAM_BY_NAME.value: build_upstream_name_map(parents), + } + + assert test[DbtCommonEnum.UPSTREAM_BY_NAME.value] == { + "jaffle_shop.orders": self.MODEL_FQN, + "other.orders": "trino_svc.RAW_DB.RAW.other_orders", + "raw.orders": self.PHYSICAL_FQN, + "orders": None, + } + assert get_dbt_test_primary_table_fqn(test) == self.PHYSICAL_FQN + assert generate_entity_link(test) == [f"<#E::table::{self.PHYSICAL_FQN}>"] + + def test_ephemeral_model_keeps_the_physical_source_upstream(self): + source = self._source(self.PHYSICAL_FQN, self.MODEL_FQN) + objects = self._objects(source) + manifest = objects.dbt_manifest + intermediate = manifest.nodes[self.MODEL_KEY].model_copy(deep=True) + intermediate.config.materialized = "ephemeral" + intermediate_key = "model.jaffle_shop.intermediate" + intermediate.unique_id = intermediate_key + intermediate.name = "intermediate" + intermediate.alias = "intermediate" + intermediate.fqn = ["jaffle_shop", "intermediate"] + manifest.nodes[intermediate_key] = intermediate + manifest.nodes[self.MODEL_KEY].depends_on.nodes = [intermediate_key] + + links = self._links(source, objects) + + assert set(links) == {self.PHYSICAL_FQN, self.MODEL_FQN} + assert links[self.MODEL_FQN].datamodel.upstream == [self.PHYSICAL_FQN] + + def test_freshness_and_metadata_share_the_physical_table_and_keep_logical_names(self): + source = self._source(self.PHYSICAL_FQN, self.MODEL_FQN) + with TestAddDbtSourceFreshnessResults.SOURCES_FILE.open(encoding="utf-8") as sources_file: + sources = json.load(sources_file) + objects = self._objects(source, dbt_sources=sources) + + links = self._links(source, objects) + freshness = source.context.get().dbt_tests[self.SOURCE_KEY + "_freshness"] + + assert self.PHYSICAL_FQN in links + assert freshness[DbtCommonEnum.UPSTREAM.value] == [self.PHYSICAL_FQN] + assert freshness[DbtCommonEnum.MANIFEST_NODE.value].name == "orders_freshness" + assert generate_entity_link(freshness) == [f"<#E::table::{self.PHYSICAL_FQN}>"] + + def test_repeated_ingestion_preserves_logical_manifest_identity(self): + source = self._source(self.PHYSICAL_FQN, self.MODEL_FQN) + objects = self._objects(source) + + for _ in range(2): + links = self._links(source, objects) + assert set(links) == {self.PHYSICAL_FQN, self.MODEL_FQN} + assert links[self.MODEL_FQN].datamodel.upstream == [self.PHYSICAL_FQN] + node = objects.dbt_manifest.sources[self.SOURCE_KEY] + assert (node.name, node.identifier, node.unique_id) == ("orders", "raw_orders", self.SOURCE_KEY) + + @pytest.mark.parametrize("manifest_version", [4, 7, 8, 12]) + def test_source_identifier_survives_supported_manifest_parsers(self, manifest_version): + payload = self._payload() + payload["metadata"]["dbt_schema_version"] = f"https://schemas.getdbt.com/dbt/manifest/v{manifest_version}.json" + if manifest_version < 8: + for node in [*payload["sources"].values(), *payload["nodes"].values()]: + node["root_path"] = "/dbt/jaffle_shop" + model = payload["nodes"][self.MODEL_KEY] + model["compiled"] = True + if manifest_version < 7: + model["raw_sql"] = model.pop("raw_code") + model["compiled_sql"] = model.pop("compiled_code") + else: + model["language"] = "sql" + if manifest_version == 12: + payload.update(groups={}, group_map={}, saved_queries={}, semantic_models={}, unit_tests={}) + source = self._source(self.PHYSICAL_FQN, self.MODEL_FQN) + + links = self._links(source, self._objects(source, payload)) + + assert set(links) == {self.PHYSICAL_FQN, self.MODEL_FQN} + assert links[self.MODEL_FQN].datamodel.upstream == [self.PHYSICAL_FQN] + + @pytest.mark.parametrize("identifier", ["RAW_ORDERS", "raw.orders", "raw orders"]) + def test_physical_identifier_uses_existing_fqn_quoting(self, identifier): + from metadata.utils import fqn + + physical_fqn = fqn._build("trino_svc", "RAW_DB", "RAW", identifier) + source = self._source(physical_fqn, self.MODEL_FQN) + payload = self._payload() + payload["sources"][self.SOURCE_KEY]["identifier"] = identifier + + links = self._links(source, self._objects(source, payload)) + + assert physical_fqn in links + assert links[self.MODEL_FQN].datamodel.upstream == [physical_fqn] + + def test_cross_service_source_stays_unresolved_when_search_is_disabled(self): + source = self._source("other_svc.RAW_DB.RAW.raw_orders", self.MODEL_FQN, self.LOGICAL_FQN) + + links = self._links(source, self._objects(source)) + + assert set(links) == {self.MODEL_FQN} + assert links[self.MODEL_FQN].datamodel.upstream == [] + + def test_missing_dependency_does_not_drop_healthy_source_or_model(self): + source = self._source(self.PHYSICAL_FQN, self.MODEL_FQN) + objects = self._objects(source) + objects.dbt_manifest.nodes[self.MODEL_KEY].depends_on.nodes.append("source.missing") + + links = self._links(source, objects) + + assert set(links) == {self.PHYSICAL_FQN, self.MODEL_FQN} + assert links[self.MODEL_FQN].datamodel.upstream == [self.PHYSICAL_FQN] + + def test_exposure_uses_the_physical_source_upstream(self): + source = self._source(self.PHYSICAL_FQN) + source.metadata.get_by_name.return_value = Dashboard( + id=uuid.uuid4(), + name="orders_dashboard", + service=EntityReference(id=uuid.uuid4(), type="dashboardService"), + ) + objects = self._objects(source) + exposure = SimpleNamespace( + name="orders_dashboard", + type=SimpleNamespace(value="dashboard"), + meta={"open_metadata_fqn": "looker.orders_dashboard"}, + depends_on=SimpleNamespace(nodes=[self.SOURCE_KEY]), + ) + + source.add_dbt_exposure("exposure.jaffle_shop.orders_dashboard", exposure, objects.dbt_manifest.sources) + + assert source.context.get().exposures["exposure.jaffle_shop.orders_dashboard"][DbtCommonEnum.UPSTREAM] == [ + self.PHYSICAL_FQN + ] From 46a7cfb071344c0421e92ef1f958235b8c61f4be Mon Sep 17 00:00:00 2001 From: Vamsi-klu Date: Sat, 17 Oct 2026 21:52:54 +0000 Subject: [PATCH 2/2] Fixes 35383: Resolve dbt source identifiers literally --- .../ingestion/source/database/dbt/metadata.py | 78 +++++++++++++------ ingestion/tests/unit/test_dbt.py | 64 ++++++++++++++- 2 files changed, 117 insertions(+), 25 deletions(-) diff --git a/ingestion/src/metadata/ingestion/source/database/dbt/metadata.py b/ingestion/src/metadata/ingestion/source/database/dbt/metadata.py index c4dfef7b7f30..d36930c5d08d 100644 --- a/ingestion/src/metadata/ingestion/source/database/dbt/metadata.py +++ b/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 @@ -1045,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}" @@ -1179,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: diff --git a/ingestion/tests/unit/test_dbt.py b/ingestion/tests/unit/test_dbt.py index 6f61d5f59fa4..980f526b5daf 100644 --- a/ingestion/tests/unit/test_dbt.py +++ b/ingestion/tests/unit/test_dbt.py @@ -5,6 +5,7 @@ import fnmatch import json import logging +import re import uuid from copy import deepcopy from datetime import datetime @@ -2756,7 +2757,7 @@ def test_parse_upstream_nodes_snapshot_target_schema(self): expected_fqn = "dbt_test.dev.snapshots.snap_orders" - def fake_get_table_entity(table_fqn): + def fake_get_table_entity(table_fqn, **_kwargs): if table_fqn == expected_fqn: table = MagicMock() table.id.root = uuid.uuid4() @@ -5351,10 +5352,11 @@ def _source(*ingested_fqns, catalog_mapping=None, search_across_databases=False) ] def _search(*_args, fqn_search_string, **_kwargs): + search_pattern = re.sub(r"\\(.)", r"[\1]", fqn_search_string) return [ table for table in tables - if fnmatch.fnmatchcase(table.fullyQualifiedName.root.lower(), fqn_search_string.lower()) + if fnmatch.fnmatchcase(table.fullyQualifiedName.root.lower(), search_pattern.lower()) ] or None metadata = MagicMock() @@ -5846,3 +5848,61 @@ def test_exposure_uses_the_physical_source_upstream(self): assert source.context.get().exposures["exposure.jaffle_shop.orders_dashboard"][DbtCommonEnum.UPSTREAM] == [ self.PHYSICAL_FQN ] + + @pytest.mark.parametrize( + "identifier, decoy", + [ + ("raw_orders_*", "raw_orders_20261017"), + ("raw_orders_?", "raw_orders_1"), + ("raw.orders*", "raw.orders_20261017"), + (r"raw\orders*", r"raw\orders_20261017"), + (r"raw\orders", "raworders"), + ], + ) + @pytest.mark.parametrize("literal_exists", [False, True]) + @pytest.mark.parametrize( + "service, database, config", + [ + ("trino_svc", "RAW_DB", {}), + ("other_svc", "RAW_DB", {"search_across_databases": True}), + ("other_svc", "mapped_db", {"catalog_mapping": {"RAW_DB": "mapped_db"}}), + ], + ) + def test_source_identifier_is_literal_for_metadata_lineage_and_freshness( + self, identifier, decoy, literal_exists, service, database, config + ): + from metadata.utils import fqn + + physical_fqn = fqn._build(service, database, "RAW", identifier) + decoy_fqn = fqn._build(service, database, "RAW", decoy) + source = self._source(decoy_fqn, self.MODEL_FQN, *([physical_fqn] if literal_exists else []), **config) + payload = self._payload() + payload["sources"][self.SOURCE_KEY]["identifier"] = identifier + with TestAddDbtSourceFreshnessResults.SOURCES_FILE.open(encoding="utf-8") as sources_file: + sources = json.load(sources_file) + + links = self._links(source, self._objects(source, payload, dbt_sources=sources)) + + assert set(links) == {self.MODEL_FQN, *([physical_fqn] if literal_exists else [])} + assert links[self.MODEL_FQN].datamodel.upstream == ([physical_fqn] if literal_exists else []) + freshness = source.context.get().dbt_tests.get(self.SOURCE_KEY + "_freshness") + if literal_exists: + assert freshness[DbtCommonEnum.UPSTREAM.value] == [physical_fqn] + else: + assert freshness is None + + @pytest.mark.parametrize("missing_location", ["database", "schema"]) + def test_source_with_unspecified_location_still_resolves_literal_identifier(self, missing_location): + source = self._source(self.PHYSICAL_FQN, self.MODEL_FQN) + payload = self._payload() + payload["sources"][self.SOURCE_KEY][missing_location] = "None" + with TestAddDbtSourceFreshnessResults.SOURCES_FILE.open(encoding="utf-8") as sources_file: + sources = json.load(sources_file) + + links = self._links(source, self._objects(source, payload, dbt_sources=sources)) + + assert set(links) == {self.PHYSICAL_FQN, self.MODEL_FQN} + assert links[self.MODEL_FQN].datamodel.upstream == [self.PHYSICAL_FQN] + assert source.context.get().dbt_tests[self.SOURCE_KEY + "_freshness"][DbtCommonEnum.UPSTREAM.value] == [ + self.PHYSICAL_FQN + ]