FIX - OpenMetadata ops bulk deploy via API (#24508)

Author: pmbrull

openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/EntityTimeSeriesDAO.java

@@ -400,6 +400,35 @@ default void deleteBeforeTimestamp(String entityFQNHash, String extension, Long
     deleteBeforeTimestamp(getTimeSeriesTableName(), entityFQNHash, extension, timestamp);
   }
 
+  @ConnectionAwareSqlUpdate(
+      value =
+          "DELETE FROM <table> "
+              + "WHERE entityFQNHash = :entityFQNHash "
+              + "AND extension = :extension "
+              + "<mysqlCond>",
+      connectionType = MYSQL)
+  @ConnectionAwareSqlUpdate(
+      value =
+          "DELETE FROM <table> "
+              + "WHERE entityFQNHash = :entityFQNHash "
+              + "AND extension = :extension "
+              + "<psqlCond>",
+      connectionType = POSTGRES)
+  void deleteExtensionByKeyInternal(
+      @Define("table") String table,
+      @Bind("value") String value,
+      @BindFQN("entityFQNHash") String entityFQNHash,
+      @Bind("extension") String extension,
+      @Define("mysqlCond") String mysqlCond,
+      @Define("psqlCond") String psqlCond);
+
+  default void deleteExtensionByKey(String key, String value, String entityFQN, String extension) {
+    String mysqlCond = String.format("AND JSON_UNQUOTE(JSON_EXTRACT(json, '$.%s')) = :value", key);
+    String psqlCond = String.format("AND json->>'%s' = :value", key);
+    deleteExtensionByKeyInternal(
+        getTimeSeriesTableName(), value, entityFQN, extension, mysqlCond, psqlCond);
+  }
+
   @SqlQuery(
       "SELECT json FROM <table> where entityFQNHash = :entityFQNHash and extension = :extension "
           + " AND timestamp >= :startTs and timestamp <= :endTs ORDER BY timestamp DESC")

openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/IngestionPipelineRepository.java

@@ -480,6 +480,20 @@ public PipelineStatus getPipelineStatus(String ingestionPipelineFQN, UUID pipeli
         PipelineStatus.class);
   }
 
+  @Transaction
+  public IngestionPipeline deletePipelineStatusByRunId(UUID ingestionPipelineId, UUID runId) {
+    IngestionPipeline ingestionPipeline = find(ingestionPipelineId, Include.NON_DELETED);
+    daoCollection
+        .entityExtensionTimeSeriesDao()
+        .deleteExtensionByKey(
+            RUN_ID_EXTENSION_KEY,
+            runId.toString(),
+            ingestionPipeline.getFullyQualifiedName(),
+            PIPELINE_STATUS_EXTENSION);
+    setFieldsInternal(ingestionPipeline, Fields.EMPTY_FIELDS);
+    return ingestionPipeline;
+  }
+
   /**
    * Handles entity updated from PUT and POST operation.
    */

openmetadata-service/src/main/java/org/openmetadata/service/resources/services/ingestionpipelines/IngestionPipelineResource.java

@@ -1214,6 +1214,34 @@ public PipelineStatus getPipelineStatus(
     return repository.getPipelineStatus(fqn, runId);
   }
 
+  @DELETE
+  @Path("/{id}/pipelineStatus/{runId}")
+  @Operation(
+      operationId = "deletePipelineStatusByRunId",
+      summary = "Delete pipeline status by run ID",
+      description =
+          "Delete a specific pipeline status by its run ID for the given ingestion pipeline.",
+      responses = {
+        @ApiResponse(responseCode = "204", description = "Pipeline status deleted successfully"),
+        @ApiResponse(
+            responseCode = "404",
+            description = "Ingestion Pipeline or Pipeline Status not found")
+      })
+  public Response deletePipelineStatusByRunId(
+      @Context UriInfo uriInfo,
+      @Context SecurityContext securityContext,
+      @Parameter(description = "Id of the Ingestion Pipeline", schema = @Schema(type = "UUID"))
+          @PathParam("id")
+          UUID id,
+      @Parameter(description = "Run ID of the pipeline status", schema = @Schema(type = "UUID"))
+          @PathParam("runId")
+          UUID runId) {
+    OperationContext operationContext = new OperationContext(entityType, MetadataOperation.DELETE);
+    authorizer.authorize(securityContext, operationContext, getResourceContextById(id));
+    repository.deletePipelineStatusByRunId(id, runId);
+    return Response.noContent().build();
+  }
+
   @DELETE
   @Path("/{id}/pipelineStatus")
   @Operation(