Fixes #17747: dbt update owners

ingestion/src/metadata/ingestion/source/database/dbt/dbt_service.py

@@ -122,6 +122,11 @@ class DbtServiceTopology(ServiceTopology):
                 processor="process_dbt_descriptions",
                 nullable=True,
             ),
+            NodeStage(
+                type_=DataModelLink,
+                processor="process_dbt_owners",
+                nullable=True,
+            ),
         ],
     )
     process_dbt_tests: Annotated[
@@ -293,6 +298,12 @@ def process_dbt_descriptions(self, data_model_link: DataModelLink):
         Method to process DBT descriptions using patch APIs
         """
 
+    @abstractmethod
+    def process_dbt_owners(self, data_model_link: DataModelLink):
+        """
+        Method to process DBT owners using patch APIs
+        """
+
     def get_dbt_tests(self) -> dict:
         """
         Prepare the DBT tests

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

@@ -61,6 +61,7 @@
 from metadata.ingestion.lineage.models import ConnectionTypeDialectMapper
 from metadata.ingestion.lineage.sql_lineage import get_lineage_by_query
 from metadata.ingestion.models.ometa_classification import OMetaTagAndClassification
+from metadata.ingestion.models.patch_request import PatchedEntity, PatchRequest
 from metadata.ingestion.models.table_metadata import ColumnDescription
 from metadata.ingestion.ometa.client import APIError
 from metadata.ingestion.ometa.ometa_api import OpenMetadata
@@ -890,6 +891,47 @@ def process_dbt_descriptions(self, data_model_link: DataModelLink):
                     f"to update dbt description: {exc}"
                 )
 
+    def process_dbt_owners(
+        self, data_model_link: DataModelLink
+    ) -> Iterable[Either[PatchedEntity]]:
+        """
+        Method to process DBT owners
+        """
+        table_entity: Table = data_model_link.table_entity
+        if table_entity:
+            logger.debug(
+                f"Processing DBT owners for: {table_entity.fullyQualifiedName.root}"
+            )
+            try:
+                data_model = data_model_link.datamodel
+                if (
+                    data_model.resourceType != DbtCommonEnum.SOURCE.value
+                    and self.source_config.dbtUpdateOwners
+                ):
+                    logger.debug(
+                        f"Overwriting owners with DBT owners: {table_entity.fullyQualifiedName.root}"
+                    )
+                    if data_model.owners:
+                        new_entity = deepcopy(table_entity)
+                        new_entity.owners = data_model.owners
+                        yield Either(
+                            right=PatchRequest(
+                                original_entity=table_entity,
+                                new_entity=new_entity,
+                                override_metadata=True,
+                            )
+                        )
+
+            except Exception as exc:  # pylint: disable=broad-except
+                yield Either(
+                    left=StackTraceError(
+                        name=str(table_entity.fullyQualifiedName.root),
+                        error=f"Failed to parse the node"
+                        f"{table_entity.fullyQualifiedName.root} to update dbt owner: {exc}",
+                        stackTrace=traceback.format_exc(),
+                    )
+                )
+
     def create_dbt_tests_definition(
         self, dbt_test: dict
     ) -> Iterable[Either[CreateTestDefinitionRequest]]:

openmetadata-docs/content/partials/v1.6/connectors/yaml/dbt/source-config-def.md

@@ -4,6 +4,8 @@
 
 **dbtUpdateDescriptions**: Configuration to update the description from dbt or not. If set to true descriptions from dbt will override the already present descriptions on the entity. For more details visit [here](/connectors/ingestion/workflows/dbt/ingest-dbt-descriptions)
 
+**dbtUpdateOwners**: Configuration to update the owner from dbt or not. If set to true owners from dbt will override the already present owners on the entity. For more details visit [here](/connectors/ingestion/workflows/dbt/ingest-dbt-owner)
+
 **includeTags**: true or false, to ingest tags from dbt. Default is true.
 
 **dbtClassificationName**: Custom OpenMetadata Classification name for dbt tags.

openmetadata-docs/content/partials/v1.6/connectors/yaml/dbt/source-config.md

@@ -1,5 +1,6 @@
 ```yaml {% srNumber=120 %}
       # dbtUpdateDescriptions: true or false
+      # dbtUpdateOwners: true or false
       # includeTags: true or false
       # dbtClassificationName: dbtTags
       # databaseFilterPattern:

openmetadata-docs/content/partials/v1.7/connectors/yaml/dbt/source-config-def.md

@@ -4,6 +4,8 @@
 
 **dbtUpdateDescriptions**: Configuration to update the description from dbt or not. If set to true descriptions from dbt will override the already present descriptions on the entity. For more details visit [here](/connectors/ingestion/workflows/dbt/ingest-dbt-descriptions)
 
+**dbtUpdateOwners**: Configuration to update the owner from dbt or not. If set to true owners from dbt will override the already present owners on the entity. For more details visit [here](/connectors/ingestion/workflows/dbt/ingest-dbt-owner)
+
 **includeTags**: true or false, to ingest tags from dbt. Default is true.
 
 **dbtClassificationName**: Custom OpenMetadata Classification name for dbt tags.

openmetadata-docs/content/partials/v1.7/connectors/yaml/dbt/source-config.md

@@ -1,5 +1,6 @@
 ```yaml {% srNumber=120 %}
       # dbtUpdateDescriptions: true or false
+      # dbtUpdateOwners: true or false
       # includeTags: true or false
       # dbtClassificationName: dbtTags
       # databaseFilterPattern:

openmetadata-docs/content/v1.6.x/connectors/ingestion/workflows/dbt/ingest-dbt-owner.md

@@ -138,8 +138,16 @@ After running the ingestion workflow with dbt you can see the created user or te
 
 
 
-{% note %}
+## Overriding the existing table Owners
 
-If a table already has a owner linked to it, owner from the dbt will not update the current owner.
+To establish a unified and reliable system for owners, a single source of truth is necessary. It either is directly OpenMetadata, if individuals want to go there and keep updating, or if they prefer to keep it centralized in dbt, then we can always rely on that directly.
 
-{% /note %}
+When the `Update Owners` toggle is enabled during the configuration of dbt ingestion, existing owners of tables will be overwritten with the dbt owners.
+
+If toggle is disabled during the configuration of dbt ingestion, dbt owners will only be updated for tables in OpenMetadata that currently have no owners. Existing owners will remain unchanged and will not be overwritten with dbt owners.
+
+{% image
+  src="/images/v1.6/features/ingestion/workflows/dbt/dbt-features/dbt-update-owners.webp"
+  alt="update-dbt-owners"
+  caption="Update dbt Owners"
+ /%}

openmetadata-docs/content/v1.6.x/main-concepts/metadata-standard/schemas/metadataIngestion/dbtPipeline.md

@@ -12,6 +12,7 @@ slug: /main-concepts/metadata-standard/schemas/metadataingestion/dbtpipeline
 - **`type`**: Pipeline type. Refer to *#/definitions/dbtConfigType*. Default: `DBT`.
 - **`dbtConfigSource`**: Available sources to fetch DBT catalog and manifest files.
 - **`dbtUpdateDescriptions`** *(boolean)*: Optional configuration to update the description from DBT or not. Default: `False`.
+- **`dbtUpdateOwners`** *(boolean)*: Optional configuration to update the owners from DBT or not. Default: `False`.
 - **`includeTags`** *(boolean)*: Optional configuration to toggle the tags ingestion. Default: `True`.
 - **`dbtClassificationName`** *(string)*: Custom OpenMetadata Classification name for dbt tags. Default: `dbtTags`.
 - **`schemaFilterPattern`**: Regex to only fetch tables or databases that matches the pattern. Refer to *../type/filterPattern.json#/definitions/filterPattern*.

openmetadata-docs/content/v1.7.x-SNAPSHOT/connectors/ingestion/workflows/dbt/ingest-dbt-owner.md

@@ -138,8 +138,16 @@ After running the ingestion workflow with dbt you can see the created user or te
 
 
 
-{% note %}
+## Overriding the existing table Owners
 
-If a table already has a owner linked to it, owner from the dbt will not update the current owner.
+To establish a unified and reliable system for owners, a single source of truth is necessary. It either is directly OpenMetadata, if individuals want to go there and keep updating, or if they prefer to keep it centralized in dbt, then we can always rely on that directly.
 
-{% /note %}
+When the `Update Owners` toggle is enabled during the configuration of dbt ingestion, existing owners of tables will be overwritten with the dbt owners.
+
+If toggle is disabled during the configuration of dbt ingestion, dbt owners will only be updated for tables in OpenMetadata that currently have no owners. Existing owners will remain unchanged and will not be overwritten with dbt owners.
+
+{% image
+  src="/images/v1.6/features/ingestion/workflows/dbt/dbt-features/dbt-update-owners.webp"
+  alt="update-dbt-owners"
+  caption="Update dbt Owners"
+ /%}

openmetadata-docs/content/v1.7.x-SNAPSHOT/main-concepts/metadata-standard/schemas/metadataIngestion/dbtPipeline.md

@@ -12,6 +12,7 @@ slug: /main-concepts/metadata-standard/schemas/metadataingestion/dbtpipeline
 - **`type`**: Pipeline type. Refer to *#/definitions/dbtConfigType*. Default: `DBT`.
 - **`dbtConfigSource`**: Available sources to fetch DBT catalog and manifest files.
 - **`dbtUpdateDescriptions`** *(boolean)*: Optional configuration to update the description from DBT or not. Default: `False`.
+- **`dbtUpdateOwners`** *(boolean)*: Optional configuration to update the owner from DBT or not. Default: `False`.
 - **`includeTags`** *(boolean)*: Optional configuration to toggle the tags ingestion. Default: `True`.
 - **`dbtClassificationName`** *(string)*: Custom OpenMetadata Classification name for dbt tags. Default: `dbtTags`.
 - **`schemaFilterPattern`**: Regex to only fetch tables or databases that matches the pattern. Refer to *../type/filterPattern.json#/definitions/filterPattern*.

openmetadata-spec/src/main/resources/json/schema/metadataIngestion/dbtPipeline.json

@@ -48,6 +48,11 @@
       "type": "boolean",
       "default": false
     },
+    "dbtUpdateOwners": {
+      "description": "Optional configuration to update the owners from DBT or not",
+      "type": "boolean",
+      "default": false
+    },
     "includeTags": {
       "description": "Optional configuration to toggle the tags ingestion.",
       "type": "boolean",

openmetadata-ui/src/main/resources/ui/public/locales/en-US/Database/workflows/dbt.md

@@ -346,6 +346,16 @@ If the option is disabled, only tables and columns without any existing descript
 However, if the option is enabled, descriptions for all tables and columns in the dbt manifest will be updated in OpenMetadata.
 $$
 
+$$section
+### Update Owners $(id="dbtUpdateOwners")
+
+This options updates the table owner in OpenMetadata with owners from dbt.
+
+If the option is disabled, only tables without any existing owners will have their owners updated based on the dbt manifest. 
+
+However, if the option is enabled, owners for all tables and columns in the dbt manifest will be updated in OpenMetadata.
+$$
+
 $$section
 ### Include dbt Tags $(id="includeTags")
 

openmetadata-ui/src/main/resources/ui/src/generated/metadataIngestion/dbtPipeline.ts

@@ -10,9 +10,7 @@
  *  See the License for the specific language governing permissions and
  *  limitations under the License.
  */
-
-
- /**
+/**
  * DBT Pipeline Configuration.
  */
 export interface DbtPipeline {
@@ -32,6 +30,10 @@ export interface DbtPipeline {
      * Optional configuration to update the description from DBT or not
      */
     dbtUpdateDescriptions?: boolean;
+    /**
+     * Optional configuration to update the owners from DBT or not
+     */
+    dbtUpdateOwners?: boolean;
     /**
      * Optional configuration to toggle the tags ingestion.
      */