Skip to content

Commit f5aa904

Browse files
q-andyq-andy
committed
Refactor UpdateRequest type checking
Signed-off-by: Andy Qin <[email protected]>
1 parent 959b59f commit f5aa904

File tree

5 files changed

+53
-131
lines changed

5 files changed

+53
-131
lines changed

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

Lines changed: 45 additions & 58 deletions
Original file line numberDiff line numberDiff line change
@@ -213,8 +213,6 @@ public TransportBulkAction(
213213
/**
214214
* Retrieves the {@link IndexRequest} from the provided {@link DocWriteRequest} for index or upsert actions. Upserts are
215215
* modeled as {@link IndexRequest} inside the {@link UpdateRequest}. Ignores {@link org.opensearch.action.delete.DeleteRequest}'s
216-
* Note: with the introduction of system ingest pipelines, we occasionally want to operate on all potential
217-
* child index requests of UpdateRequests. In that case, use {@link UpdateRequest#getChildIndexRequests()} instead
218216
*
219217
* @param docWriteRequest The request to find the {@link IndexRequest}
220218
* @return the found {@link IndexRequest} or {@code null} if one can not be found.
@@ -247,64 +245,9 @@ protected void doInternalExecute(Task task, BulkRequest bulkRequest, String exec
247245
final long startTime = relativeTime();
248246
final AtomicArray<BulkItemResponse> responses = new AtomicArray<>(bulkRequest.requests.size());
249247

250-
boolean hasIndexRequestsWithPipelines = false;
251248
final Metadata metadata = clusterService.state().getMetadata();
252249
final Version minNodeVersion = clusterService.state().getNodes().getMinNodeVersion();
253-
for (DocWriteRequest<?> actionRequest : bulkRequest.requests) {
254-
// Each index request needs to be evaluated, because this method also modifies the IndexRequest
255-
if (actionRequest instanceof UpdateRequest updateRequest) {
256-
// We execute all pipelines on upsert and docAsUpsert index requests (full update children)
257-
// For all other update children, we only resolve system ingest pipelines.
258-
// The cases are as follows:
259-
// 1. If regular bulk update (no upsert), execute only system pipeline on doc (upsert will be null)
260-
// 2. If regular upsert, execute all pipelines on upsert, execute only system pipeline on doc
261-
// 3. If upsert with script, execute all pipelines on upsert (doc will be null)
262-
// 4. If doc as upsert, execute all pipelines on doc (upsert will be null)
263-
// 5. For regular update with script, no pipelines will be executed on doc (there is no doc to execute)
264-
// Note that when execute pipelines on update existing docs, the full doc is NOT fetched from shard level.
265-
// This means the pipeline is only executed on the partial doc, which may not contain all fields.
266-
// System ingest pipelines and processors should handle these cases individually.
267-
boolean indexRequestHasPipeline = false;
268-
switch (updateRequest.getType()) {
269-
case NORMAL_UPDATE -> indexRequestHasPipeline |= ingestService.resolveSystemIngestPipeline(
270-
actionRequest,
271-
updateRequest.doc(),
272-
metadata
273-
);
274-
case NORMAL_UPSERT -> {
275-
indexRequestHasPipeline |= ingestService.resolveSystemIngestPipeline(actionRequest, updateRequest.doc(), metadata);
276-
indexRequestHasPipeline |= ingestService.resolvePipelines(actionRequest, updateRequest.upsertRequest(), metadata);
277-
}
278-
case UPSERT_WITH_SCRIPT -> indexRequestHasPipeline |= ingestService.resolvePipelines(
279-
actionRequest,
280-
updateRequest.upsertRequest(),
281-
metadata
282-
);
283-
case DOC_AS_UPSERT -> indexRequestHasPipeline |= ingestService.resolvePipelines(
284-
actionRequest,
285-
updateRequest.doc(),
286-
metadata
287-
);
288-
// Pure scripted updates have no child index requests, so nothing is resolved.
289-
}
290-
hasIndexRequestsWithPipelines |= indexRequestHasPipeline;
291-
} else {
292-
IndexRequest indexRequest = getIndexWriteRequest(actionRequest);
293-
// Skip resolving pipelines for delete requests
294-
if (indexRequest != null) {
295-
boolean indexRequestHasPipeline = ingestService.resolvePipelines(actionRequest, indexRequest, metadata);
296-
hasIndexRequestsWithPipelines |= indexRequestHasPipeline;
297-
}
298-
}
299-
300-
if (actionRequest instanceof IndexRequest) {
301-
IndexRequest ir = (IndexRequest) actionRequest;
302-
ir.checkAutoIdWithOpTypeCreateSupportedByVersion(minNodeVersion);
303-
if (ir.getAutoGeneratedTimestamp() != IndexRequest.UNSET_AUTO_GENERATED_TIMESTAMP) {
304-
throw new IllegalArgumentException("autoGeneratedTimestamp should not be set externally");
305-
}
306-
}
307-
}
250+
boolean hasIndexRequestsWithPipelines = resolvePipelinesForActionRequests(bulkRequest.requests, metadata, minNodeVersion);
308251

309252
if (hasIndexRequestsWithPipelines) {
310253
// this method (doExecute) will be called again, but with the bulk requests updated from the ingest node processing but
@@ -422,6 +365,50 @@ public void onRejection(Exception rejectedException) {
422365
}
423366
}
424367

368+
private boolean resolvePipelinesForActionRequests(
369+
List<DocWriteRequest<?>> docWriteRequests,
370+
Metadata metadata,
371+
Version minNodeVersion
372+
) {
373+
boolean hasIndexRequestsWithPipelines = false;
374+
for (DocWriteRequest<?> actionRequest : docWriteRequests) {
375+
// Each index request needs to be evaluated, because this method also modifies the IndexRequest
376+
if (actionRequest instanceof UpdateRequest updateRequest) {
377+
// Note that when execute pipelines on update existing docs, the full doc is NOT fetched from shard level.
378+
// This means the pipeline is only executed on the partial doc, which may not contain all fields.
379+
// System ingest pipelines and processors should handle these cases individually.
380+
boolean indexRequestHasPipeline = false;
381+
if (updateRequest.upsertRequest() != null) {
382+
indexRequestHasPipeline |= ingestService.resolvePipelines(actionRequest, updateRequest.upsertRequest(), metadata);
383+
}
384+
if (updateRequest.doc() != null) {
385+
if (updateRequest.docAsUpsert()) {
386+
indexRequestHasPipeline |= ingestService.resolvePipelines(actionRequest, updateRequest.doc(), metadata);
387+
} else {
388+
indexRequestHasPipeline |= ingestService.resolveSystemIngestPipeline(actionRequest, updateRequest.doc(), metadata);
389+
}
390+
}
391+
hasIndexRequestsWithPipelines |= indexRequestHasPipeline;
392+
} else {
393+
IndexRequest indexRequest = getIndexWriteRequest(actionRequest);
394+
// Skip resolving pipelines for delete requests
395+
if (indexRequest != null) {
396+
boolean indexRequestHasPipeline = ingestService.resolvePipelines(actionRequest, indexRequest, metadata);
397+
hasIndexRequestsWithPipelines |= indexRequestHasPipeline;
398+
}
399+
}
400+
401+
if (actionRequest instanceof IndexRequest) {
402+
IndexRequest ir = (IndexRequest) actionRequest;
403+
ir.checkAutoIdWithOpTypeCreateSupportedByVersion(minNodeVersion);
404+
if (ir.getAutoGeneratedTimestamp() != IndexRequest.UNSET_AUTO_GENERATED_TIMESTAMP) {
405+
throw new IllegalArgumentException("autoGeneratedTimestamp should not be set externally");
406+
}
407+
}
408+
}
409+
return hasIndexRequestsWithPipelines;
410+
}
411+
425412
static void prohibitAppendWritesInBackingIndices(DocWriteRequest<?> writeRequest, Metadata metadata) {
426413
IndexAbstraction indexAbstraction = metadata.getIndicesLookup().get(writeRequest.index());
427414
if (indexAbstraction == null) {

‎server/src/main/java/org/opensearch/action/update/UpdateRequest.java‎

Lines changed: 5 additions & 47 deletions
Original file line numberDiff line numberDiff line change
@@ -1013,54 +1013,12 @@ public long ramBytesUsed() {
10131013
*/
10141014
public List<IndexRequest> getChildIndexRequests() {
10151015
List<IndexRequest> childIndexRequests = new ArrayList<>();
1016-
switch (getType()) {
1017-
case NORMAL_UPDATE, DOC_AS_UPSERT -> childIndexRequests.add(doc);
1018-
case NORMAL_UPSERT -> {
1019-
childIndexRequests.add(doc);
1020-
childIndexRequests.add(upsertRequest);
1021-
}
1022-
case UPSERT_WITH_SCRIPT -> childIndexRequests.add(upsertRequest);
1023-
// Scripted updates have no child requests
1016+
if (doc != null) {
1017+
childIndexRequests.add(doc);
10241018
}
1025-
return childIndexRequests;
1026-
}
1027-
1028-
/**
1029-
* Gets the type of update request, used to determine which pipelines to resolve and execute.
1030-
* This is calculated dynamically based on what flags and fields are populated.
1031-
* Validations for invalid cases (e.g. docAsUpsert with an upsert doc) are performed at constructor/parser level.
1032-
*
1033-
* @return the type of update request
1034-
*/
1035-
public UpdateRequest.Type getType() {
1036-
if (docAsUpsert) {
1037-
return Type.DOC_AS_UPSERT;
1038-
} else if (upsertRequest != null && script != null) {
1039-
return Type.UPSERT_WITH_SCRIPT;
1040-
} else if (upsertRequest != null && script == null) {
1041-
return Type.NORMAL_UPSERT;
1042-
} else if (doc == null && script != null) {
1043-
return Type.UPDATE_WITH_SCRIPT;
1044-
} else if (doc == null && script == null) {
1045-
// This case should never occur when parsed from a bulk request, validations prevent this from occurring
1046-
return Type.EMPTY;
1019+
if (upsertRequest != null) {
1020+
childIndexRequests.add(upsertRequest);
10471021
}
1048-
1049-
// Implicit conditions: upsertRequest == null && docAsUpsert == false
1050-
// script == null is also validated during request parsing
1051-
return Type.NORMAL_UPDATE;
1052-
}
1053-
1054-
/**
1055-
* Inner enum to classify the type of update request.
1056-
*/
1057-
@PublicApi(since = "3.1.0")
1058-
public enum Type {
1059-
NORMAL_UPDATE,
1060-
NORMAL_UPSERT,
1061-
UPDATE_WITH_SCRIPT,
1062-
UPSERT_WITH_SCRIPT,
1063-
DOC_AS_UPSERT,
1064-
EMPTY
1022+
return childIndexRequests;
10651023
}
10661024
}

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

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -863,10 +863,10 @@ private void runBulkRequestInBatch(
863863
if (pipelineId != null && IngestService.NOOP_PIPELINE_NAME.equals(pipelineId) == false) {
864864
pipelinesInfoList.add(new IngestPipelineInfo(pipelineId, IngestPipelineType.DEFAULT));
865865
}
866-
if (pipelineId != null && IngestService.NOOP_PIPELINE_NAME.equals(finalPipelineId) == false) {
866+
if (finalPipelineId != null && IngestService.NOOP_PIPELINE_NAME.equals(finalPipelineId) == false) {
867867
pipelinesInfoList.add(new IngestPipelineInfo(finalPipelineId, IngestPipelineType.FINAL));
868868
}
869-
if (pipelineId != null && IngestService.NOOP_PIPELINE_NAME.equals(systemPipelineId) == false) {
869+
if (systemPipelineId != null && IngestService.NOOP_PIPELINE_NAME.equals(systemPipelineId) == false) {
870870
pipelinesInfoList.add(new IngestPipelineInfo(systemPipelineId, IngestPipelineType.SYSTEM_FINAL));
871871
}
872872

‎server/src/main/java/org/opensearch/ingest/SystemIngestPipelineCache.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -82,7 +82,7 @@ private void evictOldestAndExpiredCacheEntry() {
8282
* @param index [index_name/index_uuid]
8383
* @return cached system ingest pipeline
8484
*/
85-
public Pipeline getSystemIngestPipeline(@NonNull final String index) {
85+
public Pipeline getSystemIngestPipeline(final String index) {
8686
// Check if the cache contains a valid entry for the index
8787
final CacheEntry entry = cache.get(index);
8888
if (entry != null) {

‎server/src/test/java/org/opensearch/action/update/UpdateRequestTests.java‎

Lines changed: 0 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -670,29 +670,6 @@ public void testToString() throws IOException {
670670
);
671671
}
672672

673-
public void testGetType() {
674-
UpdateRequest request = new UpdateRequest("test", "1");
675-
assertEquals(request.getType(), UpdateRequest.Type.EMPTY);
676-
677-
IndexRequest docRequest = new IndexRequest("test").id("1");
678-
IndexRequest upsertRequest = new IndexRequest("test").id("1");
679-
680-
request = new UpdateRequest("test", "1").doc(docRequest);
681-
assertEquals(request.getType(), UpdateRequest.Type.NORMAL_UPDATE);
682-
683-
request = new UpdateRequest("test", "1").script(mockInlineScript("ctx.field = \"foo\""));
684-
assertEquals(request.getType(), UpdateRequest.Type.UPDATE_WITH_SCRIPT);
685-
686-
request = new UpdateRequest("test", "1").doc(docRequest).upsert(upsertRequest);
687-
assertEquals(request.getType(), UpdateRequest.Type.NORMAL_UPSERT);
688-
689-
request = new UpdateRequest("test", "1").doc(docRequest).upsert(upsertRequest).script(mockInlineScript("ctx.field = \"foo\""));
690-
assertEquals(request.getType(), UpdateRequest.Type.UPSERT_WITH_SCRIPT);
691-
692-
request = new UpdateRequest("test", "1").doc(docRequest).docAsUpsert(true);
693-
assertEquals(request.getType(), UpdateRequest.Type.DOC_AS_UPSERT);
694-
}
695-
696673
public void testGetChildIndexRequests() {
697674
UpdateRequest request = new UpdateRequest("test", "1");
698675
IndexRequest docRequest = new IndexRequest("test").id("1");

0 commit comments

Comments
 (0)