Add executed pipelines to bulk api response (#100031)

This change allows users to pass a new list_executed_pipelines parameter
to the bulk API, which results in an executed_pipelines list being returned.

client/client-benchmark-noop-api-plugin/src/main/java/org/elasticsearch/plugin/noop/action/bulk/RestNoopBulkAction.java

@@ -54,6 +54,7 @@ public RestChannelConsumer prepareRequest(final RestRequest request, final NodeC
         String defaultRouting = request.param("routing");
         String defaultPipeline = request.param("pipeline");
         Boolean defaultRequireAlias = request.paramAsBoolean("require_alias", null);
+        Boolean defaultListExecutedPipelines = request.paramAsBoolean("list_executed_pipelines", null);
 
         String waitForActiveShards = request.param("wait_for_active_shards");
         if (waitForActiveShards != null) {
@@ -68,6 +69,7 @@ public RestChannelConsumer prepareRequest(final RestRequest request, final NodeC
             null,
             defaultPipeline,
             defaultRequireAlias,
+            defaultListExecutedPipelines,
             true,
             request.getXContentType(),
             request.getRestApiVersion()

docs/changelog/100031.yaml

@@ -0,0 +1,5 @@
+pr: 100031
+summary: Add executed pipelines to bulk api response
+area: Indices APIs
+type: enhancement
+issues: []

docs/reference/docs/bulk.asciidoc

@@ -244,6 +244,11 @@ on.
 [[docs-bulk-api-query-params]]
 ==== {api-query-parms-title}
 
+`list_executed_pipelines`::
+(Optional, Boolean) If `true`, the response will include the ingest pipelines that
+were executed for each `index` or ``create`.
+Defaults to `false`.
+
 include::{es-repo-dir}/rest-api/common-parms.asciidoc[tag=pipeline]
 
 include::{es-repo-dir}/rest-api/common-parms.asciidoc[tag=refresh]
@@ -292,6 +297,8 @@ include::{es-repo-dir}/rest-api/common-parms.asciidoc[tag=bulk-index-ds]
 
 include::{es-repo-dir}/rest-api/common-parms.asciidoc[tag=bulk-id]
 
+include::{es-repo-dir}/rest-api/common-parms.asciidoc[tag=bulk-list-executed-pipelines]
+
 include::{es-repo-dir}/rest-api/common-parms.asciidoc[tag=bulk-require-alias]
 
 include::{es-repo-dir}/rest-api/common-parms.asciidoc[tag=bulk-dynamic-templates]
@@ -321,6 +328,8 @@ include::{es-repo-dir}/rest-api/common-parms.asciidoc[tag=bulk-index]
 
 include::{es-repo-dir}/rest-api/common-parms.asciidoc[tag=bulk-id]
 
+include::{es-repo-dir}/rest-api/common-parms.asciidoc[tag=bulk-list-executed-pipelines]
+
 include::{es-repo-dir}/rest-api/common-parms.asciidoc[tag=bulk-require-alias]
 
 include::{es-repo-dir}/rest-api/common-parms.asciidoc[tag=bulk-dynamic-templates]

docs/reference/rest-api/common-parms.asciidoc

@@ -689,6 +689,12 @@ tag::require-alias[]
 Defaults to `false`.
 end::require-alias[]
 
+tag::bulk-list-executed-pipelines[]
+`list_executed_pipelines`::
+(Optional, Boolean) If `true`, the response will include the ingest pipelines that
+were executed. Defaults to `false`.
+end::bulk-list-executed-pipelines[]
+
 tag::bulk-dynamic-templates[]
 `dynamic_templates`::
 (Optional, map)

modules/ingest-common/src/yamlRestTest/resources/rest-api-spec/test/ingest/70_bulk.yml

@@ -220,3 +220,29 @@ teardown:
         index: test_index
         id: 1
   - match: { _source:  { my_field: "upper" } }
+
+---
+"Test bulk request with list_executed_pipelines":
+
+  - do:
+      bulk:
+        list_executed_pipelines: true
+        body:
+          - index:
+              _index: test_index
+              _id:    test_id1
+              pipeline: pipeline1
+          - f1: v1
+          - index:
+              _index: test_index
+              _id:    test_id2
+          - f1: v2
+          - index:
+              _index: test_index
+              _id:    test_id2
+              pipeline: fake_pipeline
+          - f1: v2
+
+  - match: { items.0.index.executed_pipelines: ["pipeline1"] }
+  - match: { items.1.index.executed_pipelines: [] }
+  - match: { items.2.index.executed_pipelines: null }

rest-api-spec/src/main/resources/rest-api-spec/api/bulk.json

@@ -79,6 +79,10 @@
       "require_alias": {
         "type": "boolean",
         "description": "Sets require_alias for all incoming documents. Defaults to unset (false)"
+      },
+      "list_executed_pipelines": {
+        "type": "boolean",
+        "description": "Sets list_executed_pipelines for all incoming documents. Defaults to unset (false)"
       }
     },
     "body":{

server/src/main/java/org/elasticsearch/TransportVersions.java

@@ -141,6 +141,8 @@ static TransportVersion def(int id) {
     public static final TransportVersion NODE_STATS_HTTP_ROUTE_STATS_ADDED = def(8_516_00_0);
     public static final TransportVersion INCLUDE_SHARDS_STATS_ADDED = def(8_517_00_0);
     public static final TransportVersion BUILD_QUALIFIER_SEPARATED = def(8_518_00_0);
+    public static final TransportVersion PIPELINES_IN_BULK_RESPONSE_ADDED = def(8_519_00_0);
+
     /*
      * STOP! READ THIS FIRST! No, really,
      *        ____ _____ ___  ____  _        ____  _____    _    ____    _____ _   _ ___ ____    _____ ___ ____  ____ _____ _

server/src/main/java/org/elasticsearch/action/DocWriteRequest.java

@@ -39,6 +39,9 @@ public interface DocWriteRequest<T> extends IndicesRequest, Accountable {
     // Flag set for disallowing index auto creation for an individual write request.
     String REQUIRE_ALIAS = "require_alias";
 
+    // Flag indicating that the list of executed pipelines should be returned in the request
+    String LIST_EXECUTED_PIPELINES = "list_executed_pipelines";
+
     /**
      * Set the index for this request
      * @return the Request

server/src/main/java/org/elasticsearch/action/bulk/BulkPrimaryExecutionContext.java

@@ -11,6 +11,7 @@
 import org.elasticsearch.action.DocWriteRequest;
 import org.elasticsearch.action.DocWriteResponse;
 import org.elasticsearch.action.delete.DeleteResponse;
+import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.action.index.IndexResponse;
 import org.elasticsearch.action.support.replication.ReplicationResponse;
 import org.elasticsearch.action.support.replication.TransportWriteAction;
@@ -19,6 +20,7 @@
 import org.elasticsearch.index.translog.Translog;
 
 import java.util.Arrays;
+import java.util.List;
 
 /**
  * This is a utility class that holds the per request state needed to perform bulk operations on the primary.
@@ -231,13 +233,20 @@ public void markOperationAsExecuted(Engine.Result result) {
                 final DocWriteResponse response;
                 if (result.getOperationType() == Engine.Operation.TYPE.INDEX) {
                     Engine.IndexResult indexResult = (Engine.IndexResult) result;
+                    List<String> executedPipelines;
+                    if (docWriteRequest instanceof IndexRequest indexRequest) {
+                        executedPipelines = indexRequest.getExecutedPipelines();
+                    } else {
+                        executedPipelines = null;
+                    }
                     response = new IndexResponse(
                         primary.shardId(),
                         indexResult.getId(),
                         result.getSeqNo(),
                         result.getTerm(),
                         indexResult.getVersion(),
-                        indexResult.isCreated()
+                        indexResult.isCreated(),
+                        executedPipelines
                     );
                 } else if (result.getOperationType() == Engine.Operation.TYPE.DELETE) {
                     Engine.DeleteResult deleteResult = (Engine.DeleteResult) result;

server/src/main/java/org/elasticsearch/action/bulk/BulkProcessor.java

@@ -438,7 +438,7 @@ public BulkProcessor add(
         lock.lock();
         try {
             ensureOpen();
-            bulkRequest.add(data, defaultIndex, null, null, defaultPipeline, null, true, xContentType, RestApiVersion.current());
+            bulkRequest.add(data, defaultIndex, null, null, defaultPipeline, null, null, true, xContentType, RestApiVersion.current());
             bulkRequestToExecute = newBulkRequestIfNeeded();
         } finally {
             lock.unlock();

server/src/main/java/org/elasticsearch/action/bulk/BulkRequest.java

@@ -232,15 +232,15 @@ public BulkRequest add(byte[] data, int from, int length, @Nullable String defau
      * Adds a framed data in binary format
      */
     public BulkRequest add(BytesReference data, @Nullable String defaultIndex, XContentType xContentType) throws IOException {
-        return add(data, defaultIndex, null, null, null, null, true, xContentType, RestApiVersion.current());
+        return add(data, defaultIndex, null, null, null, null, null, true, xContentType, RestApiVersion.current());
     }
 
     /**
      * Adds a framed data in binary format
      */
     public BulkRequest add(BytesReference data, @Nullable String defaultIndex, boolean allowExplicitIndex, XContentType xContentType)
         throws IOException {
-        return add(data, defaultIndex, null, null, null, null, allowExplicitIndex, xContentType, RestApiVersion.current());
+        return add(data, defaultIndex, null, null, null, null, null, allowExplicitIndex, xContentType, RestApiVersion.current());
 
     }
 
@@ -251,6 +251,7 @@ public BulkRequest add(
         @Nullable FetchSourceContext defaultFetchSourceContext,
         @Nullable String defaultPipeline,
         @Nullable Boolean defaultRequireAlias,
+        @Nullable Boolean defaultListExecutedPipelines,
         boolean allowExplicitIndex,
         XContentType xContentType,
         RestApiVersion restApiVersion
@@ -265,6 +266,7 @@ public BulkRequest add(
             defaultFetchSourceContext,
             pipeline,
             requireAlias,
+            defaultListExecutedPipelines,
             allowExplicitIndex,
             xContentType,
             (indexRequest, type) -> internalAdd(indexRequest),

server/src/main/java/org/elasticsearch/action/bulk/BulkRequestParser.java

@@ -61,6 +61,7 @@ public final class BulkRequestParser {
     private static final ParseField IF_SEQ_NO = new ParseField("if_seq_no");
     private static final ParseField IF_PRIMARY_TERM = new ParseField("if_primary_term");
     private static final ParseField REQUIRE_ALIAS = new ParseField(DocWriteRequest.REQUIRE_ALIAS);
+    private static final ParseField LIST_EXECUTED_PIPELINES = new ParseField(DocWriteRequest.LIST_EXECUTED_PIPELINES);
     private static final ParseField DYNAMIC_TEMPLATES = new ParseField("dynamic_templates");
 
     // TODO: Remove this parameter once the BulkMonitoring endpoint has been removed
@@ -126,6 +127,7 @@ public void parse(
         @Nullable FetchSourceContext defaultFetchSourceContext,
         @Nullable String defaultPipeline,
         @Nullable Boolean defaultRequireAlias,
+        @Nullable Boolean defaultListExecutedPipelines,
         boolean allowExplicitIndex,
         XContentType xContentType,
         BiConsumer<IndexRequest, String> indexRequestConsumer,
@@ -207,6 +209,7 @@ public void parse(
                 int retryOnConflict = 0;
                 String pipeline = defaultPipeline;
                 boolean requireAlias = defaultRequireAlias != null && defaultRequireAlias;
+                boolean listExecutedPipelines = defaultListExecutedPipelines != null && defaultListExecutedPipelines;
                 Map<String, String> dynamicTemplates = Map.of();
 
                 // at this stage, next token can either be END_OBJECT (and use default index and type, with auto generated id)
@@ -260,6 +263,8 @@ public void parse(
                                 fetchSourceContext = FetchSourceContext.fromXContent(parser);
                             } else if (REQUIRE_ALIAS.match(currentFieldName, parser.getDeprecationHandler())) {
                                 requireAlias = parser.booleanValue();
+                            } else if (LIST_EXECUTED_PIPELINES.match(currentFieldName, parser.getDeprecationHandler())) {
+                                listExecutedPipelines = parser.booleanValue();
                             } else {
                                 throw new IllegalArgumentException(
                                     "Action/metadata line [" + line + "] contains an unknown parameter [" + currentFieldName + "]"
@@ -343,7 +348,8 @@ public void parse(
                                     .setIfPrimaryTerm(ifPrimaryTerm)
                                     .source(sliceTrimmingCarriageReturn(data, from, nextMarker, xContentType), xContentType)
                                     .setDynamicTemplates(dynamicTemplates)
-                                    .setRequireAlias(requireAlias),
+                                    .setRequireAlias(requireAlias)
+                                    .setListExecutedPipelines(listExecutedPipelines),
                                 type
                             );
                         } else {
@@ -358,7 +364,8 @@ public void parse(
                                     .setIfPrimaryTerm(ifPrimaryTerm)
                                     .source(sliceTrimmingCarriageReturn(data, from, nextMarker, xContentType), xContentType)
                                     .setDynamicTemplates(dynamicTemplates)
-                                    .setRequireAlias(requireAlias),
+                                    .setRequireAlias(requireAlias)
+                                    .setListExecutedPipelines(listExecutedPipelines),
                                 type
                             );
                         }
@@ -374,7 +381,8 @@ public void parse(
                                 .setIfPrimaryTerm(ifPrimaryTerm)
                                 .source(sliceTrimmingCarriageReturn(data, from, nextMarker, xContentType), xContentType)
                                 .setDynamicTemplates(dynamicTemplates)
-                                .setRequireAlias(requireAlias),
+                                .setRequireAlias(requireAlias)
+                                .setListExecutedPipelines(listExecutedPipelines),
                             type
                         );
                     } else if ("update".equals(action)) {
@@ -410,7 +418,7 @@ public void parse(
                         }
                         IndexRequest upsertRequest = updateRequest.upsertRequest();
                         if (upsertRequest != null) {
-                            upsertRequest.setPipeline(pipeline);
+                            upsertRequest.setPipeline(pipeline).setListExecutedPipelines(listExecutedPipelines);
                         }
 
                         updateRequestConsumer.accept(updateRequest);

server/src/main/java/org/elasticsearch/action/index/IndexRequest.java

@@ -42,6 +42,9 @@
 
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
 import java.util.Locale;
 import java.util.Map;
 import java.util.Objects;
@@ -103,6 +106,16 @@ public class IndexRequest extends ReplicatedWriteRequest<IndexRequest> implement
     private boolean isPipelineResolved;
 
     private boolean requireAlias;
+    /**
+     * This indicates whether the response to this request ought to list the ingest pipelines that were executed on the document
+     */
+    private boolean listExecutedPipelines;
+    /**
+     * This holds the names of the ingest pipelines that have been executed on the document for this request. This is not meant to be set by
+     * the creator of the request -- pipelines are added here at runtime as they are executed.
+     */
+    @Nullable
+    private List<String> executedPipelines = null;
 
     /**
      * Value for {@link #getAutoGeneratedTimestamp()} if the document has an external
@@ -167,6 +180,15 @@ public IndexRequest(@Nullable ShardId shardId, StreamInput in) throws IOExceptio
         if (in.getTransportVersion().onOrAfter(PIPELINES_HAVE_RUN_FIELD_ADDED)) {
             pipelinesHaveRun = in.readBoolean();
         }
+        if (in.getTransportVersion().onOrAfter(TransportVersions.PIPELINES_IN_BULK_RESPONSE_ADDED)) {
+            this.listExecutedPipelines = in.readBoolean();
+            if (listExecutedPipelines) {
+                List<String> possiblyImmutableExecutedPipelines = in.readOptionalCollectionAsList(StreamInput::readString);
+                this.executedPipelines = possiblyImmutableExecutedPipelines == null
+                    ? null
+                    : new ArrayList<>(possiblyImmutableExecutedPipelines);
+            }
+        }
     }
 
     public IndexRequest() {
@@ -727,6 +749,12 @@ private void writeBody(StreamOutput out) throws IOException {
         if (out.getTransportVersion().onOrAfter(PIPELINES_HAVE_RUN_FIELD_ADDED)) {
             out.writeBoolean(pipelinesHaveRun);
         }
+        if (out.getTransportVersion().onOrAfter(TransportVersions.PIPELINES_IN_BULK_RESPONSE_ADDED)) {
+            out.writeBoolean(listExecutedPipelines);
+            if (listExecutedPipelines) {
+                out.writeOptionalCollection(executedPipelines, StreamOutput::writeString);
+            }
+        }
     }
 
     @Override
@@ -797,6 +825,15 @@ public IndexRequest setRequireAlias(boolean requireAlias) {
         return this;
     }
 
+    public IndexRequest setListExecutedPipelines(boolean listExecutedPipelines) {
+        this.listExecutedPipelines = listExecutedPipelines;
+        return this;
+    }
+
+    public boolean getListExecutedPipelines() {
+        return listExecutedPipelines;
+    }
+
     /**
      * Specifies a map from the full path of field names to the name of dynamic mapping templates
      */
@@ -830,4 +867,33 @@ public void setPipelinesHaveRun() {
     public boolean pipelinesHaveRun() {
         return pipelinesHaveRun;
     }
+
+    /**
+     * Adds the pipeline to the list of executed pipelines, if listExecutedPipelines is true
+     * @param pipeline
+     */
+    public void addPipeline(String pipeline) {
+        if (listExecutedPipelines) {
+            if (executedPipelines == null) {
+                executedPipelines = new ArrayList<>();
+            }
+            executedPipelines.add(pipeline);
+        }
+    }
+
+    /**
+     * This returns the list of pipelines executed on the document for this request. If listExecutedPipelines is false, the response will be
+     * null, even if pipelines were executed. If listExecutedPipelines is true but no pipelines were executed, the list will be empty.
+     * @return
+     */
+    @Nullable
+    public List<String> getExecutedPipelines() {
+        if (listExecutedPipelines == false) {
+            return null;
+        } else if (executedPipelines == null) { // The client has asked to list pipelines, but none have been executed
+            return List.of();
+        } else {
+            return Collections.unmodifiableList(executedPipelines);
+        }
+    }
 }