feat(ingestion): add a prefix to LF tags to distinguish from regular tags.

metadata-ingestion/src/datahub/api/entities/external/lake_formation_external_entites.py

@@ -10,7 +10,7 @@
 # Tag search using the workspace search UI is supported only for tables, views, and table columns.
 # Tag search requires exact term matching.
 # https://learn.microsoft.com/en-us/azure/databricks/database-objects/tags#constraint
-from typing import Any, Dict, Optional
+from typing import Any, Dict, Optional, Union
 
 from pydantic import validator
 from typing_extensions import ClassVar
@@ -49,6 +49,22 @@ class LakeFormationTag(ExternalTag):
     key: LakeFormationTagKeyText
     value: Optional[LakeFormationTagValueText] = None
     catalog: Optional[str] = None
+    prefix: Optional[str] = None
+
+    def __init__(
+        self,
+        key: Optional[Union[str, LakeFormationTagKeyText]] = None,
+        value: Optional[Union[str, LakeFormationTagValueText]] = None,
+        prefix: Optional[str] = None,
+        **data: Any,
+    ) -> None:
+        if prefix:
+            key = f"{prefix}/{key}"
+        super().__init__(
+            key=key,
+            value=value,
+            **data,
+        )
 
     # Pydantic v1 validators
     @validator("key", pre=True)

metadata-ingestion/src/datahub/ingestion/source/aws/glue.py

@@ -185,6 +185,11 @@ class GlueSourceConfig(
         description="When True, extracts Lake Formation tags directly assigned to Glue tables/databases. Note: Tags inherited from databases or other parent resources are excluded.",
     )
 
+    lakeformation_tags_prefix: Optional[str] = Field(
+        default=None,
+        description="Adds prefix to Lake Formation tags assigned to Glue tables/databases to distinguish Lake Formation tags from regular tags.",
+    )
+
     profiling: GlueProfilingConfig = Field(
         default_factory=GlueProfilingConfig,
         description="Configs to ingest data profiles from glue table",
@@ -399,6 +404,7 @@ def get_database_lf_tags(
                         key=tag_key,
                         value=tag_value,
                         catalog=catalog_id,
+                        prefix=self.source_config.lakeformation_tags_prefix,
                     )
                     tags.append(t)
             return tags
@@ -442,6 +448,7 @@ def get_table_lf_tags(
                         key=tag_key,
                         value=tag_value,
                         catalog=catalog_id,
+                        prefix=self.source_config.lakeformation_tags_prefix,
                     )
                     tags.append(t)
             return tags