Add Owner for DBT (#6872)

* Added owner field in DataModel

* Added user in backend

* review comments addressed

Co-authored-by: Onkar Ravgan <onkarravgan@Onkars-MacBook-Pro.local>

catalog-rest-service/src/main/java/org/openmetadata/catalog/jdbi3/TableRepository.java

@@ -625,6 +625,12 @@ public class TableRepository extends EntityRepository<Table> {
     if (nullOrEmpty(table.getDescription())) {
       table.setDescription(dataModel.getDescription());
     }
+
+    // Carry forward the table owner from the model to table entity, if empty
+    if (table.getOwner() == null) {
+      storeOwner(table, dataModel.getOwner());
+    }
+
     // Carry forward the column description from the model to table columns, if empty
     for (Column modelColumn : listOrEmpty(dataModel.getColumns())) {
       Column stored =
@@ -640,7 +646,9 @@ public class TableRepository extends EntityRepository<Table> {
       }
     }
     dao.update(table.getId(), JsonUtils.pojoToJson(table));
-    setFields(table, Fields.EMPTY_FIELDS);
+
+    setFields(table, new Fields(List.of(FIELD_OWNER), FIELD_OWNER));
+
     return table;
   }
 

catalog-rest-service/src/main/resources/json/schema/entity/data/table.json

@@ -628,6 +628,11 @@
             "type": "string"
           }
         },
+        "owner": {
+          "description": "Owner of this Model.",
+          "$ref": "../../type/entityReference.json",
+          "default": null
+        },
         "columns": {
           "description": "Columns from the schema defined during modeling. In case of DBT, the metadata here comes from `schema.yaml`.",
           "type": "array",

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

@@ -21,6 +21,7 @@ from metadata.generated.schema.entity.data.table import (
     ModelType,
     Table,
 )
+from metadata.generated.schema.entity.teams.user import User
 from metadata.generated.schema.type.entityLineage import EntitiesEdge
 from metadata.generated.schema.type.entityReference import EntityReference
 from metadata.ingestion.ometa.ometa_api import OpenMetadata
@@ -77,6 +78,14 @@ class DBTMixin:
                     schema = mnode["schema"] if mnode["schema"] else "default"
                     raw_sql = mnode.get("raw_sql", "")
                     description = mnode.get("description")
+                    dbt_user_name = cnode["metadata"].get("owner")
+                    user_name = f"*{dbt_user_name}*"
+                    user_fqn = fqn.build(
+                        self.metadata, entity_type=User, user_name=user_name
+                    )
+                    owner = self.metadata.get_entity_reference(
+                        entity=User, fqn=user_fqn
+                    )
                     model = DataModel(
                         modelType=ModelType.DBT,
                         description=description if description else None,
@@ -85,6 +94,7 @@ class DBTMixin:
                         sql=mnode.get("compiled_sql", raw_sql),
                         columns=columns,
                         upstream=upstream_nodes,
+                        owner=owner,
                     )
                     model_fqn = fqn.build(
                         self.metadata,

ingestion/src/metadata/utils/fqn.py

@@ -33,6 +33,7 @@ from metadata.generated.schema.entity.data.location import Location
 from metadata.generated.schema.entity.data.pipeline import Pipeline
 from metadata.generated.schema.entity.data.table import Column, DataModel, Table
 from metadata.generated.schema.entity.tags.tagCategory import Tag
+from metadata.generated.schema.entity.teams.user import User
 from metadata.ingestion.ometa.ometa_api import OpenMetadata
 from metadata.utils.dispatch import class_register
 from metadata.utils.elasticsearch import get_entity_from_es_result
@@ -280,6 +281,36 @@ def _(
     return _build(service_name, database_name, schema_name, table_name, column_name)
 
 
+@fqn_build_registry.add(User)
+def _(
+    metadata: OpenMetadata,
+    *,
+    user_name: str,
+    fetch_multiple_entities: bool = False,
+) -> Union[Optional[str], Optional[List[str]]]:
+    """
+    Building logic for User
+    :param metadata: OMeta client
+    :param user_name: User name
+    :return:
+    """
+
+    fqn_search_string = _build(user_name)
+
+    es_result = metadata.es_search_from_fqn(
+        entity_type=User,
+        fqn_search_string=fqn_search_string,
+    )
+    entity: Optional[Union[User, List[User]]] = get_entity_from_es_result(
+        entity_list=es_result, fetch_multiple_entities=fetch_multiple_entities
+    )
+    if not entity:
+        return None
+    if fetch_multiple_entities:
+        return [str(user.fullyQualifiedName.__root__) for user in entity]
+    return str(entity.fullyQualifiedName.__root__)
+
+
 def split_table_name(table_name: str) -> Dict[str, Optional[str]]:
     """
     Given a table name, try to extract database, schema and