Support OpenSearch delete-by-query in opensearch_api source

Is your feature request related to a problem? Please describe.

The opensearch_api source supports some OpenSearch indexing operations. Some users want to use Data Prepper to ingest data, but also be able to delete data using OpenSearch's _delete_by_query API.

This can be used to support writing events to a stream such as Kafka to support pull-based ingestion or support replication via replay.

Describe the solution you'd like

I'd like for a caller to be able to call _delete_by_query on the opensearch_api source. When this call is made, Data Prepper will execute the query itself, then generate _bulk delete requests to delete the matched documents. This effectively shifts the query portion of _delete_by_query into Data Prepper while delegating the actual deletes to the existing bulk pipeline.

The most natural solution would be for the opensearch_api to perform the query operations. This means that it must be configured with an OpenSearch cluster to query against.

For queries that match a large number of documents, the source should paginate results and write events to the buffer in batches rather than all at once. This avoids memory pressure and works with pipeline backpressure.

API

POST /{index}/_delete_by_query

With a JSON body containing the query, matching the OpenSearch _delete_by_query request format:

Converting delete-by-query to a search query

OpenSearch's own _delete_by_query internally performs a search query followed by bulk deletes. Data Prepper follows the same pattern. The query object in a _delete_by_query request uses the same Query DSL as the _search API, so the conversion is a direct extraction.

Incoming delete-by-query request to Data Prepper:

  POST /my-index/_delete_by_query
  {
    "query": {
      "range": {
        "timestamp": {
          "lt": "2025-01-01"
        }
      }
    }
  }

Search query Data Prepper executes against OpenSearch:

  POST /my-index/_search
  {
    "query": {
      "range": {
        "timestamp": {
          "lt": "2025-01-01"
        }
      }
    },
    "_source": false,
    "size": 1000
  }

The query field is passed through unchanged. Data Prepper adds "_source": false because only the _id and _index of each hit are needed to construct the delete operations — the document contents are irrelevant. The size parameter controls the batch size for pagination.

Each search hit:

  {
    "_index": "my-index",
    "_id": "abc123"
  }

Becomes an event with metadata:

{
    "opensearch_action": "delete",
    "opensearch_index": "my-index",
    "opensearch_id": "abc123"
}

The source will have a new query_cluster configuration. It must have the following:

• hosts — OpenSearch cluster endpoint(s) to query against
• username / password — Basic authentication credentials (optional)
• aws — AWS signing configuration for Amazon OpenSearch Service (optional)
• TLS/SSL settings as needed

Example pipeline configuration:

  delete-by-query-pipeline:
    source:
      opensearch_api:
        port: 9200
        query_cluster:
          hosts: ["https://opensearch-cluster:9200"]
          username: "admin"
          password: "admin"
    sink:
      - opensearch:
          hosts: ["https://opensearch-cluster:9200"]
          index: "${getMetadata(\"opensearch_index\")}"
          action: "${getMetadata(\"opensearch_action\")}"
          document_id: "${getMetadata(\"opensearch_id\")}"

Describe alternatives you've considered (Optional)

I considered using a processor to query. But this means that the delete-by-query goes into the pipeline as a command. This is not something that Data Prepper supports. Querying in the source gives the advantage of still placing events (albeit delete events) into the pipeline.

Additional context
Add any other context or screenshots about the feature request here.

Start at the opensearch_api source entry point and trace its existing indexing operations and bulk pipeline, then review how source configuration and OpenSearch requests are handled. Done means accepting _delete_by_query requests, querying the configured cluster in batches, and emitting delete events with the required metadata without loading all matches into memory.