feat: Add Reverse Metadata Agent pipeline type support

ingestion/src/metadata/cmd.py

@@ -29,6 +29,7 @@
 from metadata.cli.ingest_dbt import run_ingest_dbt
 from metadata.cli.lineage import run_lineage
 from metadata.cli.profile import run_profiler
+from metadata.cli.reverse_metadata import run_reverse_metadata
 from metadata.cli.scaffold import (
     AUTH_CHOICES,
     CAPABILITY_CHOICES,
@@ -53,6 +54,7 @@ class MetadataCommands(Enum):
     LINEAGE = "lineage"
     APP = "app"
     AUTO_CLASSIFICATION = "classify"
+    REVERSE_METADATA = "reverse_metadata"
     SCAFFOLD_CONNECTOR = "scaffold-connector"
 
 
@@ -65,6 +67,7 @@ class MetadataCommands(Enum):
     MetadataCommands.TEST.value: run_test,
     MetadataCommands.APP.value: run_app,
     MetadataCommands.AUTO_CLASSIFICATION.value: run_classification,
+    MetadataCommands.REVERSE_METADATA.value: run_reverse_metadata,
 }
 
 
@@ -165,6 +168,12 @@ def get_parser(args: Optional[List[str]] = None):
             help="Workflow for running auto classification",
         )
     )
+    create_common_config_parser_args(
+        sub_parser.add_parser(
+            MetadataCommands.REVERSE_METADATA.value,
+            help="Reverse Metadata Workflow",
+        )
+    )
     webhook_args(
         sub_parser.add_parser(
             MetadataCommands.WEBHOOK.value,

openmetadata-spec/src/main/java/org/openmetadata/sdk/PipelineServiceClientInterface.java

@@ -65,7 +65,9 @@ public interface PipelineServiceClientInterface {
           PipelineType.DATA_INSIGHT.toString(),
           "data_insight_task",
           PipelineType.APPLICATION.toString(),
-          "application_task");
+          "application_task",
+          PipelineType.REVERSE_METADATA.toString(),
+          "reverse_metadata_task");
 
   URL validateServiceURL(String serviceURL);
 

openmetadata-spec/src/main/resources/json/schema/entity/services/ingestionPipelines/ingestionPipeline.json

@@ -11,7 +11,7 @@
       "description": "Type of Pipeline - metadata, usage",
       "type": "string",
       "javaType": "org.openmetadata.schema.entity.services.ingestionPipelines.PipelineType",
-      "enum": ["metadata", "usage", "lineage", "profiler", "autoClassification", "TestSuite", "dataInsight", "elasticSearchReindex", "dbt", "application"]
+      "enum": ["metadata", "usage", "lineage", "profiler", "autoClassification", "TestSuite", "dataInsight", "elasticSearchReindex", "dbt", "application", "reverseMetadata"]
     },
     "pipelineStatus": {
       "type": "object",

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

@@ -0,0 +1,80 @@
+{
+    "$id": "https://open-metadata.org/schema/metadataIngestion/reverseMetadataPipeline.json",
+    "$schema": "http://json-schema.org/draft-07/schema#",
+    "title": "ReverseMetadataPipeline",
+    "javaType": "org.openmetadata.schema.metadataIngestion.ReverseMetadataPipeline",
+    "description": "Reverse Metadata Agent Pipeline Configuration. Controls which metadata changes (descriptions, tags, owners) are propagated back to source systems.",
+    "type": "object",
+    "definitions": {
+        "reverseMetadataConfigType": {
+            "description": "Reverse Metadata Pipeline type",
+            "type": "string",
+            "enum": [
+                "ReverseMetadata"
+            ],
+            "default": "ReverseMetadata"
+        }
+    },
+    "properties": {
+        "type": {
+            "description": "Pipeline type",
+            "$ref": "#/definitions/reverseMetadataConfigType",
+            "default": "ReverseMetadata"
+        },
+        "syncDescriptions": {
+            "description": "Enable syncing description changes back to the source system.",
+            "type": "boolean",
+            "default": true,
+            "title": "Sync Descriptions"
+        },
+        "syncTags": {
+            "description": "Enable syncing tag changes back to the source system.",
+            "type": "boolean",
+            "default": true,
+            "title": "Sync Tags"
+        },
+        "syncOwners": {
+            "description": "Enable syncing owner changes back to the source system.",
+            "type": "boolean",
+            "default": false,
+            "title": "Sync Owners"
+        },
+        "databaseFilterPattern": {
+            "description": "Regex to only include/exclude databases that match the pattern.",
+            "$ref": "../type/filterPattern.json#/definitions/filterPattern",
+            "title": "Database Filter Pattern"
+        },
+        "schemaFilterPattern": {
+            "description": "Regex to only include/exclude schemas that match the pattern.",
+            "$ref": "../type/filterPattern.json#/definitions/filterPattern",
+            "title": "Schema Filter Pattern"
+        },
+        "tableFilterPattern": {
+            "description": "Regex to only include/exclude tables that match the pattern.",
+            "$ref": "../type/filterPattern.json#/definitions/filterPattern",
+            "title": "Table Filter Pattern"
+        },
+        "useFqnForFiltering": {
+            "description": "Regex will be applied on fully qualified name (e.g service_name.db_name.schema_name.table_name) instead of raw name (e.g. table_name).",
+            "type": "boolean",
+            "default": false,
+            "title": "Use FQN For Filtering"
+        },
+        "service": {
+            "description": "Service to apply reverse metadata operations on.",
+            "$ref": "../type/entityReference.json"
+        },
+        "operations": {
+            "description": "List of operations to be performed on the service. Populated at runtime by the handler.",
+            "type": "array",
+            "items": {
+                "$ref": "reverseIngestionPipeline.json#/definitions/operation"
+            },
+            "default": []
+        }
+    },
+    "additionalProperties": false,
+    "required": [
+        "type"
+    ]
+}

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

@@ -68,6 +68,9 @@
             },
             {
               "$ref": "reverseIngestionPipeline.json"
+            },
+            {
+              "$ref": "reverseMetadataPipeline.json"
             }
           ]
         }