Support system ingest pipelines for bulk update operations #18277

q-andy · 2025-05-13T16:48:08Z

Description

Support system ingest pipelines for bulk update operations

Update Request Type Classification

Introduce a method to expose all child index requests associated with an update operation

Pipeline Resolution Enhancement

Use resolveSystemIngestPipeline to enable resolving only the system ingest pipeline while setting the others to NOOP
Based on update request fields, we extract the update request children and conditionally resolve ALL pipelines, resolve ONLY system ingest pipelines, or no pipelines at all.

Slot Management

Introduce innerSlot to track individual child index requests within anupdate operation
Use innerslot to map pipeline execution results back to the correct child request using (slot, innerSlot) pairs
Maintain proper error handling and response mapping for both parent and child operations to their original bulk request slot

Related Issues

Resolves #18276

Check List

Functionality includes testing.
API changes companion pull request created, if applicable.
Public documentation issue/PR created, if applicable.

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

Signed-off-by: Andy Qin <[email protected]>

github-actions · 2025-05-13T16:59:15Z

❌ Gradle check result for f8169f3: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

github-actions · 2025-05-13T18:22:12Z

❌ Gradle check result for 94653c1: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

Signed-off-by: Andy Qin <[email protected]>

github-actions · 2025-05-13T19:19:18Z

❌ Gradle check result for 959b59f: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

server/src/main/java/org/opensearch/action/bulk/TransportBulkAction.java

bzhangam · 2025-05-13T20:02:08Z

server/src/main/java/org/opensearch/action/bulk/TransportBulkAction.java

+                // This means the pipeline is only executed on the partial doc, which may not contain all fields.
+                // System ingest pipelines and processors should handle these cases individually.
+                boolean indexRequestHasPipeline = false;
+                switch (updateRequest.getType()) {


Can we simplify the logic as:

If(updateRequest.upsertRequest()!=null){ indexRequestHasPipeline |= ingestService.resolvePipelines(actionRequest, updateRequest.upsertRequest(), metadata); } if(updateRequest.doc() != null){ if(updateRequest.docAsUpsert()){ indexRequestHasPipeline |= ingestService.resolvePipelines(actionRequest, updateRequest.doc(), metadata); }else{ indexRequestHasPipeline |= ingestService. resolveSystemIngestPipeline(actionRequest, updateRequest.doc(), metadata); } }

Seems currently we don't care about the script and it just complicate the logic.

makes sense

bzhangam · 2025-05-13T20:13:15Z

server/src/main/java/org/opensearch/action/bulk/TransportBulkAction.java

+                BulkItemResponse.Failure failure = new BulkItemResponse.Failure(
+                    indexRequest.index(),
+                    indexRequest.id(),
+                    e,


Let's say our update request has doc and upsert. The getIndexWriteRequest function will return upsert. What if the upsert is processed successfully but we failed to process the doc? In that case we will attach the exception to the wrong request.

So I think we need to pass the inner slot to this function to handle the update request properly. If both doc and upsert failed ideally I think we should return both failures. But it will require significant changes so we can just take one for now. But we should map the failure to the request properly for the update request case.

Similar comment to the markItemAsDropped function.

I think in case it is should be okay to use getIndexWriteRequest because the failed request is used to create a single BulkItemResponse.Failure using indexRequest.index(), indexRequest.id(), and indexRequest.opType(). Since the doc and upsert are from the same parent UpdateRequest, they should share the same index, id, and opType already as well so the same information would be retrieved either child request. Looking at BulkItemResponse.Failure the failure information does not contain detailed information mapping to a specific index request other than those above shared values as well.

Agree that ideally we should be able to surface both exceptions to the user in the case where both fail but it would require significant refactoring, maybe it's a task we could take on for #17742. From user perspective though, regardless if one or both fail, they need to retry the request anyways.

bzhangam · 2025-05-13T20:17:55Z

server/src/main/java/org/opensearch/ingest/IngestService.java

-                    onCompletion.accept(originalThread, null);
+            List<IndexRequest> childIndexRequests = new ArrayList<>();
+            if (actionRequest instanceof UpdateRequest updateRequest) {
+                childIndexRequests.addAll(updateRequest.getChildIndexRequests());


Should we consider to add this logic to getIndexWriteRequest and let it return a list of requests?

I considered this as well, it requires some minor refactoring wherever we call getIndexWriteRequest to make it work (changing null checks to isEmpty checks and extracting the inner element of the list). The way I see it, UpdateRequests (and the case where an update has two child index requests) are the edge case that we handle separately, otherwise its fine to call getIndexWriteRequest directly. So its okay to have it separate.

bzhangam · 2025-05-13T20:21:37Z

server/src/main/java/org/opensearch/ingest/IngestService.java

-            if (IngestService.NOOP_PIPELINE_NAME.equals(systemPipelineId) == false) {
-                pipelinesInfoList.add(new IngestPipelineInfo(systemPipelineId, IngestPipelineType.SYSTEM_FINAL));
-            }
+                if (pipelineId != null && IngestService.NOOP_PIPELINE_NAME.equals(pipelineId) == false) {


No need to check pipelineId != null since null doesn't equal NOOP_PIPELINE_NAME.

In the case where pipelineId = null then:

IngestService.NOOP_PIPELINE_NAME.equals(null) evaluates to false

false == false evaluates to true which enters the block
So then null pipeline is added to the pipelinesInfoList which causes an NPE later on.

The null check avoids this NPE, so I think it's needed

ic I think you are right.

server/src/main/java/org/opensearch/ingest/IngestService.java

bzhangam · 2025-05-13T20:41:59Z

server/src/main/java/org/opensearch/ingest/IngestService.java

@@ -1196,7 +1242,8 @@ private Pipeline getPipeline(
            // In very edge case it is possible the cache is invalidated after we resolve the
            // pipeline. So try to resolve the system ingest pipeline again here.
            if (indexPipeline == null) {
-                final String newPipelineId = resolveSystemIngestPipeline(actionRequest, indexRequest, state.metadata());
+                resolveSystemIngestPipeline(actionRequest, indexRequest, state.metadata());
+                final String newPipelineId = indexRequest.getSystemIngestPipeline();
                // set it as NOOP to avoid duplicated execution after we switch back to the write thread
                indexRequest.setSystemIngestPipeline(NOOP_PIPELINE_NAME);
                indexPipeline = systemIngestPipelineCache.getSystemIngestPipeline(newPipelineId);


I think I have an issue in getSystemIngestPipeline that it should not mark the input as NonNull. Could you help remove it?

Could you explain a little more? What's the issue and which input is being marked non-null?

In getSystemIngestPipeline I mark the pipeline id as non-null but I think it's ok to allow to pass a null. Remove the non-null check can make it more resilient.

got it, fixed in latest revision

github-actions · 2025-05-14T19:26:22Z

❌ Gradle check result for b2c9db0: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

github-actions · 2025-05-14T20:32:18Z

❌ Gradle check result for fbafb82: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

github-actions · 2025-05-14T21:14:47Z

❌ Gradle check result for 2341708: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

Signed-off-by: Andy Qin <[email protected]>

github-actions · 2025-05-14T22:52:45Z

✅ Gradle check result for f5aa904: SUCCESS

codecov · 2025-05-14T22:53:10Z

Codecov Report

Attention: Patch coverage is 94.61538% with 7 lines in your changes missing coverage. Please review.

Project coverage is 72.53%. Comparing base (58c281f) to head (f5aa904).
Report is 14 commits behind head on main.

Files with missing lines	Patch %	Lines
...main/java/org/opensearch/ingest/IngestService.java	94.52%	1 Missing and 3 partials ⚠️
...rg/opensearch/action/bulk/TransportBulkAction.java	92.85%	1 Missing and 2 partials ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##               main   #18277      +/-   ##
============================================
+ Coverage     72.44%   72.53%   +0.09%     
- Complexity    67256    67397     +141     
============================================
  Files          5488     5488              
  Lines        310994   311063      +69     
  Branches      45212    45228      +16     
============================================
+ Hits         225288   225624     +336     
+ Misses        67297    67088     -209     
+ Partials      18409    18351      -58

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

Support system ingest pipelines for bulk update operations

f8169f3

Signed-off-by: Andy Qin <[email protected]>

q-andy requested a review from a team as a code owner May 13, 2025 16:48

github-actions bot added enhancement Enhancement or improvement to existing feature or request Indexing Indexing, Bulk Indexing and anything related to indexing labels May 13, 2025

Spotless

959b59f

Signed-off-by: Andy Qin <[email protected]>

q-andy force-pushed the system-pipeline-bulk-update-2 branch from 94653c1 to 959b59f Compare May 13, 2025 18:41

opensearch-ci-bot mentioned this pull request May 13, 2025

[AUTOCUT] Gradle Check Flaky Test Report for SearchServiceTests #18238

Closed

bzhangam reviewed May 13, 2025

View reviewed changes

server/src/main/java/org/opensearch/action/bulk/TransportBulkAction.java Outdated Show resolved Hide resolved

bzhangam reviewed May 13, 2025

View reviewed changes

server/src/main/java/org/opensearch/ingest/IngestService.java Outdated Show resolved Hide resolved

bzhangam reviewed May 13, 2025

View reviewed changes

opensearch-ci-bot mentioned this pull request May 14, 2025

[AUTOCUT] Gradle Check Flaky Test Report for QueueResizableOpenSearchThreadPoolExecutorTests #14297

Closed

q-andy force-pushed the system-pipeline-bulk-update-2 branch from b2c9db0 to fbafb82 Compare May 14, 2025 20:21

q-andy force-pushed the system-pipeline-bulk-update-2 branch from fbafb82 to 2341708 Compare May 14, 2025 21:04

Refactor UpdateRequest type checking

f5aa904

Signed-off-by: Andy Qin <[email protected]>

q-andy force-pushed the system-pipeline-bulk-update-2 branch from 2341708 to f5aa904 Compare May 14, 2025 21:39

opensearch-ci-bot mentioned this pull request May 15, 2025

[AUTOCUT] Gradle Check Flaky Test Report for SimpleSecureNetty4TransportTests #18071

Open

Support system ingest pipelines for bulk update operations #18277

Are you sure you want to change the base?

Support system ingest pipelines for bulk update operations #18277

Uh oh!

Conversation

q-andy commented May 13, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Description

Related Issues

Check List

Uh oh!

github-actions bot commented May 13, 2025

Uh oh!

github-actions bot commented May 13, 2025

Uh oh!

github-actions bot commented May 13, 2025

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

q-andy May 14, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

github-actions bot commented May 14, 2025

Uh oh!

github-actions bot commented May 14, 2025

Uh oh!

github-actions bot commented May 14, 2025

Uh oh!

github-actions bot commented May 14, 2025

Uh oh!

codecov bot commented May 14, 2025

Codecov Report

Uh oh!

Uh oh!

q-andy commented May 13, 2025 •

edited

Loading

q-andy May 14, 2025 •

edited

Loading