Issue #17012: Multi User/Team Ownership

bootstrap/sql/migrations/native/1.5.0/mysql/schemaChanges.sql

@@ -148,3 +148,41 @@ JSON_EXTRACT(json, '$.sourceConfig.config.dbtConfigSource.dbtSecurityConfig.gcpC
 JSON_EXTRACT(json, '$.sourceConfig.config.dbtConfigSource.dbtSecurityConfig.gcpConfig.externalType') OR 
 JSON_EXTRACT(json, '$.sourceConfig.config.dbtConfigSource.dbtSecurityConfig.gcpConfig.path')
 ) is NULL AND JSON_EXTRACT(json, '$.sourceConfig.config.dbtConfigSource.dbtSecurityConfig.gcpConfig') is not null;
+
+-- Update Owner Field to Owners
+DELETE from event_subscription_entity where name = 'ActivityFeedAlert';
+
+-- Update thread_entity to move previousOwner and updatedOwner to array
+UPDATE thread_entity
+SET json = JSON_SET(
+    json,
+    '$.feedInfo.entitySpecificInfo.previousOwner',
+    JSON_ARRAY(
+        JSON_EXTRACT(json, '$.feedInfo.entitySpecificInfo.previousOwner')
+    )
+)
+WHERE JSON_CONTAINS_PATH(json, 'one', '$.feedInfo.entitySpecificInfo.previousOwner')
+AND JSON_TYPE(JSON_EXTRACT(json, '$.feedInfo.entitySpecificInfo.previousOwner')) <> 'ARRAY';
+
+UPDATE thread_entity
+SET json = JSON_SET(
+    json,
+    '$.feedInfo.entitySpecificInfo.updatedOwner',
+    JSON_ARRAY(
+        JSON_EXTRACT(json, '$.feedInfo.entitySpecificInfo.updatedOwner')
+    )
+)
+WHERE JSON_CONTAINS_PATH(json, 'one', '$.feedInfo.entitySpecificInfo.updatedOwner')
+AND JSON_TYPE(JSON_EXTRACT(json, '$.feedInfo.entitySpecificInfo.updatedOwner')) <> 'ARRAY';
+
+-- Update entity_extension to move owner to array
+UPDATE entity_extension
+SET json = JSON_SET(
+    json,
+    '$.owner',
+    JSON_ARRAY(
+        JSON_EXTRACT(json, '$.owner')
+    )
+)
+WHERE JSON_CONTAINS_PATH(json, 'one', '$.owner')
+AND JSON_TYPE(JSON_EXTRACT(json, '$.owner')) <> 'ARRAY';

bootstrap/sql/migrations/native/1.5.0/postgres/schemaChanges.sql

@@ -138,3 +138,38 @@ SET json = jsonb_set(
   AND json#>>'{sourceConfig,config,dbtConfigSource,dbtSecurityConfig,gcpConfig,type}' IS NULL 
   AND json#>>'{sourceConfig,config,dbtConfigSource,dbtSecurityConfig,gcpConfig,externalType}' IS NULL 
   AND json#>>'{sourceConfig,config,dbtConfigSource,dbtSecurityConfig,gcpConfig,path}' IS NULL;
+
+-- Update Owner Field to Owners
+DELETE from event_subscription_entity where name = 'ActivityFeedAlert';
+
+-- Update thread_entity to move previousOwner and updatedOwner to array
+UPDATE thread_entity
+SET json = jsonb_set(
+    json,
+    '{feedInfo,entitySpecificInfo,previousOwner}',
+    to_jsonb(ARRAY[json->'feedInfo'->'entitySpecificInfo'->'previousOwner'])
+)
+WHERE jsonb_path_exists(json, '$.feedInfo.entitySpecificInfo.previousOwner')
+  AND jsonb_path_query_first(json, '$.feedInfo.entitySpecificInfo.previousOwner ? (@ != null)') IS NOT null
+  AND jsonb_typeof(json->'feedInfo'->'entitySpecificInfo'->'updatedOwner') <> 'array';
+
+UPDATE thread_entity
+SET json = jsonb_set(
+    json,
+    '{feedInfo,entitySpecificInfo,updatedOwner}',
+    to_jsonb(ARRAY[json->'feedInfo'->'entitySpecificInfo'->'updatedOwner'])
+)
+WHERE jsonb_path_exists(json, '$.feedInfo.entitySpecificInfo.updatedOwner')
+  AND jsonb_path_query_first(json, '$.feedInfo.entitySpecificInfo.updatedOwner ? (@ != null)') IS NOT null
+  AND jsonb_typeof(json->'feedInfo'->'entitySpecificInfo'->'updatedOwner') <> 'array';
+
+-- Update entity_extension to move owner to array
+UPDATE entity_extension
+SET json = jsonb_set(
+    json,
+    '{owner}',
+    to_jsonb(ARRAY[jsonb_path_query_first(json, '$.owner')])
+)
+WHERE jsonb_path_exists(json, '$.owner')
+  AND jsonb_path_query_first(json, '$.owner ? (@ != null)') IS NOT null
+  AND jsonb_typeof(json->'owner') <> 'array';

ingestion/src/metadata/data_insight/processor/reports/entity_report_data_processor.py

@@ -81,7 +81,9 @@ def __init__(self, metadata: OpenMetadata):
     def name(self) -> str:
         return "Entity Report Processor"
 
-    def _get_team(self, owner: EntityReference) -> Optional[str]:
+    def _get_team(  # pylint: disable=too-many-return-statements
+        self, owner: EntityReference
+    ) -> Optional[str]:
         """Get the team from an entity. We'll use this info as well to
         add info if an entity has an owner
 
@@ -91,10 +93,12 @@ def _get_team(self, owner: EntityReference) -> Optional[str]:
         Returns:
             Optional[str]
         """
-        if not owner:
+        if not owner or not owner.root:
             return None
 
         if isinstance(owner, EntityReferenceList):
+            if not owner.root:
+                return None
             return owner.root[0].name
 
         if owner.type == "team":
@@ -188,7 +192,7 @@ def refine(self, entity: Type[T]) -> None:
         data_blob_for_entity = {}
         try:
             team = (
-                self._get_team(entity.owner)
+                self._get_team(entity.owners)
                 if not isinstance(entity, User)
                 else self._get_team(entity.teams)  # type: ignore
             )

ingestion/src/metadata/data_insight/processor/reports/web_analytic_report_data_processor.py

@@ -90,7 +90,9 @@ def _pre_hook_fn(self):
         self.refine_entity_event = self._refine_entity_event()
         next(self.refine_entity_event)
 
-    def _refine_entity_event(self) -> Generator[dict, WebAnalyticEventData, None]:
+    def _refine_entity_event(  # pylint: disable=too-many-branches, too-many-statements
+        self,
+    ) -> Generator[dict, WebAnalyticEventData, None]:
         """Coroutine to process entity web analytic event
 
         Yields:
@@ -158,8 +160,11 @@ def _refine_entity_event(self) -> Generator[dict, WebAnalyticEventData, None]:
                     )
 
                 try:
-                    owner = entity.owner.name if entity.owner else None
-                    owner_id = str(entity.owner.id.root) if entity.owner else None
+                    owner = None
+                    owner_id = None
+                    if entity.owners and len(entity.owners.root) > 0:
+                        owner = entity.owners.root[0].name
+                        owner_id = str(entity.owners.root[0].id.root)
                 except AttributeError as exc:
                     owner = None
                     owner_id = None

ingestion/src/metadata/data_insight/source/metadata.py

@@ -131,7 +131,7 @@ def _iter(self, *_, **__) -> Iterable[Either[DataInsightRecord]]:
 
                 for data in (
                     producer.fetch_data(
-                        fields=["owner", "tags"], entities_cache=self.entities_cache
+                        fields=["owners", "tags"], entities_cache=self.entities_cache
                     )
                     or []
                 ):

ingestion/src/metadata/data_quality/processor/test_case_runner.py

@@ -208,7 +208,7 @@ def compare_and_create_test_cases(
                             if test_case_to_create.parameterValues
                             else None
                         ),
-                        owner=None,
+                        owners=None,
                         computePassedFailedRowCount=test_case_to_create.computePassedFailedRowCount,
                     )
                 )

ingestion/src/metadata/data_quality/source/test_suite.py

@@ -131,7 +131,7 @@ def _process_table_suite(self, table: Table) -> Iterable[Either[TableAndTests]]:
                 ),
                 displayName=f"{self.source_config.entityFullyQualifiedName.root} Test Suite",
                 description="Test Suite created from YAML processor config file",
-                owner=None,
+                owners=None,
                 executableEntityReference=self.source_config.entityFullyQualifiedName.root,
             )
             yield Either(

ingestion/src/metadata/ingestion/models/patch_request.py

@@ -85,7 +85,7 @@ class PatchedEntity(BaseModel):
     "displayName": True,
     "sourceUrl": True,
     "description": True,
-    "owner": True,
+    "owners": True,
     "tags": True,
     "sourceHash": True,
     # Table Entity Fields
@@ -143,7 +143,7 @@ class PatchedEntity(BaseModel):
     "fileFormats": True,
 }
 
-RESTRICT_UPDATE_LIST = ["description", "tags", "owner", "displayName"]
+RESTRICT_UPDATE_LIST = ["description", "tags", "owners", "displayName"]
 
 ARRAY_ENTITY_FIELDS = ["columns", "tasks", "fields"]
 

ingestion/src/metadata/ingestion/ometa/mixins/patch_mixin.py

@@ -32,6 +32,7 @@
 from metadata.generated.schema.tests.testCase import TestCase, TestCaseParameterValue
 from metadata.generated.schema.type.basic import EntityLink, Markdown
 from metadata.generated.schema.type.entityReference import EntityReference
+from metadata.generated.schema.type.entityReferenceList import EntityReferenceList
 from metadata.generated.schema.type.lifeCycle import LifeCycle
 from metadata.generated.schema.type.tagLabel import TagLabel
 from metadata.ingestion.api.models import Entity
@@ -323,7 +324,7 @@ def patch_owner(
         self,
         entity: Type[T],
         source: T,
-        owner: EntityReference = None,
+        owners: EntityReferenceList = None,
         force: bool = False,
     ) -> Optional[T]:
         """
@@ -340,20 +341,20 @@ def patch_owner(
             Updated Entity
         """
         instance: Optional[T] = self._fetch_entity_if_exists(
-            entity=entity, entity_id=source.id, fields=["owner"]
+            entity=entity, entity_id=source.id, fields=["owners"]
         )
 
         if not instance:
             return None
 
         # Don't change existing data without force
-        if instance.owner and not force:
+        if instance.owners and instance.owners.root and not force:
             # If a owner is already present and force is not passed,
             # owner will not be overridden
             return None
 
         destination = deepcopy(instance)
-        destination.owner = owner
+        destination.owners = owners
 
         return self.patch(entity=entity, source=instance, destination=destination)
 

ingestion/src/metadata/ingestion/ometa/mixins/pipeline_mixin.py

@@ -86,7 +86,7 @@ def add_task_to_pipeline(self, pipeline: Pipeline, *tasks: Task) -> Pipeline:
             startDate=pipeline.startDate,
             service=pipeline.service.fullyQualifiedName,
             tasks=all_tasks,
-            owner=pipeline.owner,
+            owners=pipeline.owners,
             tags=pipeline.tags,
         )
 
@@ -113,7 +113,7 @@ def clean_pipeline_tasks(self, pipeline: Pipeline, task_ids: List[str]) -> Pipel
             startDate=pipeline.startDate,
             service=pipeline.service.fullyQualifiedName,
             tasks=[task for task in pipeline.tasks if task.name in task_ids],
-            owner=pipeline.owner,
+            owners=pipeline.owners,
             tags=pipeline.tags,
         )
 

ingestion/src/metadata/ingestion/ometa/mixins/user_mixin.py

@@ -19,6 +19,7 @@
 from metadata.generated.schema.entity.teams.team import Team, TeamType
 from metadata.generated.schema.entity.teams.user import User
 from metadata.generated.schema.type.entityReference import EntityReference
+from metadata.generated.schema.type.entityReferenceList import EntityReferenceList
 from metadata.ingestion.api.common import T
 from metadata.ingestion.ometa.client import REST
 from metadata.utils.constants import ENTITY_REFERENCE_TYPE_MAP
@@ -118,30 +119,38 @@ def get_reference_by_email(
         from_count: int = 0,
         size: int = 1,
         fields: Optional[list] = None,
-    ) -> Optional[EntityReference]:
+    ) -> Optional[EntityReferenceList]:
         """
         Get a User or Team Entity Reference by searching by its mail
         """
         maybe_user = self._search_by_email(
             entity=User, email=email, from_count=from_count, size=size, fields=fields
         )
         if maybe_user:
-            return EntityReference(
-                id=maybe_user.id.root,
-                type=ENTITY_REFERENCE_TYPE_MAP[User.__name__],
-                name=maybe_user.name.root,
-                displayName=maybe_user.displayName,
+            return EntityReferenceList(
+                root=[
+                    EntityReference(
+                        id=maybe_user.id.root,
+                        type=ENTITY_REFERENCE_TYPE_MAP[User.__name__],
+                        name=maybe_user.name.root,
+                        displayName=maybe_user.displayName,
+                    )
+                ]
             )
 
         maybe_team = self._search_by_email(
             entity=Team, email=email, from_count=from_count, size=size, fields=fields
         )
         if maybe_team:
-            return EntityReference(
-                id=maybe_team.id.root,
-                type=ENTITY_REFERENCE_TYPE_MAP[Team.__name__],
-                name=maybe_team.name.root,
-                displayName=maybe_team.displayName,
+            return EntityReferenceList(
+                root=[
+                    EntityReference(
+                        id=maybe_team.id.root,
+                        type=ENTITY_REFERENCE_TYPE_MAP[Team.__name__],
+                        name=maybe_team.name.root,
+                        displayName=maybe_team.displayName,
+                    )
+                ]
             )
 
         return None
@@ -154,19 +163,23 @@ def get_reference_by_name(
         size: int = 1,
         fields: Optional[list] = None,
         is_owner: bool = False,
-    ) -> Optional[EntityReference]:
+    ) -> Optional[EntityReferenceList]:
         """
         Get a User or Team Entity Reference by searching by its name
         """
         maybe_user = self._search_by_name(
             entity=User, name=name, from_count=from_count, size=size, fields=fields
         )
         if maybe_user:
-            return EntityReference(
-                id=maybe_user.id.root,
-                type=ENTITY_REFERENCE_TYPE_MAP[User.__name__],
-                name=maybe_user.name.root,
-                displayName=maybe_user.displayName,
+            return EntityReferenceList(
+                root=[
+                    EntityReference(
+                        id=maybe_user.id.root,
+                        type=ENTITY_REFERENCE_TYPE_MAP[User.__name__],
+                        name=maybe_user.name.root,
+                        displayName=maybe_user.displayName,
+                    )
+                ]
             )
 
         maybe_team = self._search_by_name(
@@ -176,11 +189,15 @@ def get_reference_by_name(
             # if is_owner is True, we only want to return the team if it is a group
             if is_owner and maybe_team.teamType != TeamType.Group:
                 return None
-            return EntityReference(
-                id=maybe_team.id.root,
-                type=ENTITY_REFERENCE_TYPE_MAP[Team.__name__],
-                name=maybe_team.name.root,
-                displayName=maybe_team.displayName,
+            return EntityReferenceList(
+                root=[
+                    EntityReference(
+                        id=maybe_team.id.root,
+                        type=ENTITY_REFERENCE_TYPE_MAP[Team.__name__],
+                        name=maybe_team.name.root,
+                        displayName=maybe_team.displayName,
+                    )
+                ]
             )
 
         return None

ingestion/src/metadata/ingestion/source/dashboard/dashboard_service.py

@@ -434,7 +434,7 @@ def mark_datamodels_as_deleted(self) -> Iterable[Either[DeleteEntity]]:
 
     def get_owner_ref(  # pylint: disable=unused-argument, useless-return
         self, dashboard_details
-    ) -> Optional[EntityReference]:
+    ) -> Optional[EntityReferenceList]:
         """
         Method to process the dashboard owners
         """

ingestion/src/metadata/ingestion/source/dashboard/domodashboard/metadata.py

@@ -44,7 +44,7 @@
     Markdown,
     SourceUrl,
 )
-from metadata.generated.schema.type.entityReference import EntityReference
+from metadata.generated.schema.type.entityReferenceList import EntityReferenceList
 from metadata.ingestion.api.models import Either
 from metadata.ingestion.api.steps import InvalidSourceException
 from metadata.ingestion.ometa.ometa_api import OpenMetadata
@@ -103,7 +103,7 @@ def get_dashboard_details(self, dashboard: DomoDashboardDetails) -> dict:
 
     def get_owner_ref(
         self, dashboard_details: DomoDashboardDetails
-    ) -> Optional[EntityReference]:
+    ) -> Optional[EntityReferenceList]:
         for owner in dashboard_details.owners or []:
             try:
                 owner_details = self.client.domo.users_get(owner.id)
@@ -142,7 +142,7 @@ def yield_dashboard(
                     for chart in self.context.get().charts or []
                 ],
                 service=self.context.get().dashboard_service,
-                owner=self.get_owner_ref(dashboard_details=dashboard_details),
+                owners=self.get_owner_ref(dashboard_details=dashboard_details),
             )
             yield Either(right=dashboard_request)
             self.register_record(dashboard_request=dashboard_request)

ingestion/src/metadata/ingestion/source/dashboard/lightdash/metadata.py

@@ -129,7 +129,7 @@ def yield_dashboard(
                     for chart in self.context.get().charts or []
                 ],
                 service=self.context.get().dashboard_service,
-                owner=self.get_owner_ref(dashboard_details=dashboard_details),
+                owners=self.get_owner_ref(dashboard_details=dashboard_details),
             )
             yield dashboard_request
             self.register_record(dashboard_request=dashboard_request)