fix(ingest/dbt): always encode tag urns (#10799)

datahub-project · Jul 16, 2024 · a484d32 · a484d32
1 parent d7b5d90
commit a484d32
Show file tree

Hide file tree

Showing 3 changed files with 10 additions and 7 deletions.
diff --git a/metadata-ingestion/src/datahub/emitter/mce_builder.py b/metadata-ingestion/src/datahub/emitter/mce_builder.py
@@ -476,7 +476,7 @@ def get_or_add_aspect(mce: MetadataChangeEventClass, default: Aspect) -> Aspect:
 
 def make_global_tag_aspect_with_tag_list(tags: List[str]) -> GlobalTagsClass:
     return GlobalTagsClass(
-        tags=[TagAssociationClass(f"urn:li:tag:{tag}") for tag in tags]
+        tags=[TagAssociationClass(make_tag_urn(tag)) for tag in tags]
     )
 
 

diff --git a/metadata-ingestion/src/datahub/ingestion/source/dbt/dbt_common.py b/metadata-ingestion/src/datahub/ingestion/source/dbt/dbt_common.py
@@ -1483,7 +1483,7 @@ def get_patched_mce(self, mce):
             transformed_tag_list = self.get_transformed_tags_by_prefix(
                 tag_aspect.tags,
                 mce.proposedSnapshot.urn,
-                mce_builder.make_tag_urn(self.config.tag_prefix),
+                tag_prefix=self.config.tag_prefix,
             )
             tag_aspect.tags = transformed_tag_list
 
@@ -1874,16 +1874,19 @@ def get_transformed_tags_by_prefix(
         self,
         new_tags: List[TagAssociationClass],
         entity_urn: str,
-        tags_prefix_filter: str,
+        tag_prefix: str,
     ) -> List[TagAssociationClass]:
         tag_set = {new_tag.tag for new_tag in new_tags}
 
         if self.ctx.graph:
             existing_tags_class = self.ctx.graph.get_tags(entity_urn)
             if existing_tags_class and existing_tags_class.tags:
-                for exiting_tag in existing_tags_class.tags:
-                    if not exiting_tag.tag.startswith(tags_prefix_filter):
-                        tag_set.add(exiting_tag.tag)
+                for existing_tag in existing_tags_class.tags:
+                    if tag_prefix and existing_tag.tag.startswith(
+                        mce_builder.make_tag_urn(tag_prefix)
+                    ):
+                        continue
+                    tag_set.add(existing_tag.tag)
         return [TagAssociationClass(tag) for tag in sorted(tag_set)]
 
     # This method attempts to read-modify and return the glossary terms of a dataset.

diff --git a/metadata-ingestion/tests/unit/test_dbt_source.py b/metadata-ingestion/tests/unit/test_dbt_source.py
@@ -148,7 +148,7 @@ def test_dbt_source_patching_tags():
         ["new_non_dbt", "dbt:new_dbt"]
     )
     transformed_tags = source.get_transformed_tags_by_prefix(
-        new_tag_aspect.tags, "urn:li:dataset:dummy", "urn:li:tag:dbt:"
+        new_tag_aspect.tags, "urn:li:dataset:dummy", "dbt:"
     )
     expected_tags = {
         "urn:li:tag:new_non_dbt",