[8.15] Fix handling of bulk requests with semantic text fields and delete ops (#117233)

jimczi · web-flow · commit b954f9e3bb2b · 2024-11-21T21:59:44.000+11:00
* Fix handling of bulk requests with semantic text fields and delete ops (#116942) Previously, delete operations were not processed correctly when followed by operations containing semantic text fields. This issue caused the positions of subsequent operations in the items array to shift incorrectly by one. This PR resolves the discrepancy and includes additional tests to ensure proper behavior. * fix yml test
diff --git a/docs/changelog/116942.yaml b/docs/changelog/116942.yaml
@@ -0,0 +1,5 @@
+pr: 116942
+summary: Fix handling of bulk requests with semantic text fields and delete ops
+area: Relevance
+type: bug
+issues: []
diff --git a/x-pack/plugin/inference/src/internalClusterTest/java/org/elasticsearch/xpack/inference/action/filter/ShardBulkInferenceActionFilterIT.java b/x-pack/plugin/inference/src/internalClusterTest/java/org/elasticsearch/xpack/inference/action/filter/ShardBulkInferenceActionFilterIT.java
@@ -11,6 +11,7 @@
 import org.elasticsearch.action.bulk.BulkItemResponse;
 import org.elasticsearch.action.bulk.BulkRequestBuilder;
 import org.elasticsearch.action.bulk.BulkResponse;
+import org.elasticsearch.action.delete.DeleteRequestBuilder;
 import org.elasticsearch.action.index.IndexRequestBuilder;
 import org.elasticsearch.action.search.SearchRequest;
 import org.elasticsearch.action.search.SearchResponse;
@@ -30,8 +31,10 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.Locale;
 import java.util.Map;
+import java.util.Set;
 
 import static org.hamcrest.Matchers.equalTo;
 
@@ -86,30 +89,38 @@ public void testBulkOperations() throws Exception {
 
         int totalBulkReqs = randomIntBetween(2, 100);
         long totalDocs = 0;
+        Set<String> ids = new HashSet<>();
         for (int bulkReqs = 0; bulkReqs < totalBulkReqs; bulkReqs++) {
             BulkRequestBuilder bulkReqBuilder = client().prepareBulk();
             int totalBulkSize = randomIntBetween(1, 100);
             for (int bulkSize = 0; bulkSize < totalBulkSize; bulkSize++) {
-                String id = Long.toString(totalDocs);
+                if (ids.size() > 0 && rarely(random())) {
+                    String id = randomFrom(ids);
+                    ids.remove(id);
+                    DeleteRequestBuilder request = new DeleteRequestBuilder(client(), INDEX_NAME).setId(id);
+                    bulkReqBuilder.add(request);
+                    continue;
+                }
+                String id = Long.toString(totalDocs++);
                 boolean isIndexRequest = randomBoolean();
                 Map<String, Object> source = new HashMap<>();
                 source.put("sparse_field", isIndexRequest && rarely() ? null : randomAlphaOfLengthBetween(0, 1000));
                 source.put("dense_field", isIndexRequest && rarely() ? null : randomAlphaOfLengthBetween(0, 1000));
                 if (isIndexRequest) {
                     bulkReqBuilder.add(new IndexRequestBuilder(client()).setIndex(INDEX_NAME).setId(id).setSource(source));
-                    totalDocs++;
+                    ids.add(id);
                 } else {
                     boolean isUpsert = randomBoolean();
                     UpdateRequestBuilder request = new UpdateRequestBuilder(client()).setIndex(INDEX_NAME).setDoc(source);
-                    if (isUpsert || totalDocs == 0) {
+                    if (isUpsert || ids.size() == 0) {
                         request.setDocAsUpsert(true);
-                        totalDocs++;
                     } else {
                         // Update already existing document
-                        id = Long.toString(randomLongBetween(0, totalDocs - 1));
+                        id = randomFrom(ids);
                     }
                     request.setId(id);
                     bulkReqBuilder.add(request);
+                    ids.add(id);
                 }
             }
             BulkResponse bulkResponse = bulkReqBuilder.get();
@@ -134,7 +145,7 @@ public void testBulkOperations() throws Exception {
         SearchSourceBuilder sourceBuilder = new SearchSourceBuilder().size(0).trackTotalHits(true);
         SearchResponse searchResponse = client().search(new SearchRequest(INDEX_NAME).source(sourceBuilder)).get();
         try {
-            assertThat(searchResponse.getHits().getTotalHits().value, equalTo(totalDocs));
+            assertThat(searchResponse.getHits().getTotalHits().value, equalTo((long) ids.size()));
         } finally {
             searchResponse.decRef();
         }
diff --git a/x-pack/plugin/inference/src/main/java/org/elasticsearch/xpack/inference/InferenceFeatures.java b/x-pack/plugin/inference/src/main/java/org/elasticsearch/xpack/inference/InferenceFeatures.java
@@ -26,6 +26,10 @@ public Set<NodeFeature> getFeatures() {
 
     @Override
     public Set<NodeFeature> getTestFeatures() {
-        return Set.of(SemanticTextFieldMapper.SEMANTIC_TEXT_IN_OBJECT_FIELD_FIX, SemanticTextFieldMapper.SEMANTIC_TEXT_ZERO_SIZE_FIX);
+        return Set.of(
+            SemanticTextFieldMapper.SEMANTIC_TEXT_IN_OBJECT_FIELD_FIX,
+            SemanticTextFieldMapper.SEMANTIC_TEXT_ZERO_SIZE_FIX,
+            SemanticTextFieldMapper.SEMANTIC_TEXT_DELETE_FIX
+        );
     }
 }
diff --git a/x-pack/plugin/inference/src/main/java/org/elasticsearch/xpack/inference/action/filter/ShardBulkInferenceActionFilter.java b/x-pack/plugin/inference/src/main/java/org/elasticsearch/xpack/inference/action/filter/ShardBulkInferenceActionFilter.java
@@ -412,8 +412,8 @@ private void applyInferenceResponses(BulkItemRequest item, FieldInferenceRespons
          */
         private Map<String, List<FieldInferenceRequest>> createFieldInferenceRequests(BulkShardRequest bulkShardRequest) {
             Map<String, List<FieldInferenceRequest>> fieldRequestsMap = new LinkedHashMap<>();
-            int itemIndex = 0;
-            for (var item : bulkShardRequest.items()) {
+            for (int itemIndex = 0; itemIndex < bulkShardRequest.items().length; itemIndex++) {
+                var item = bulkShardRequest.items()[itemIndex];
                 if (item.getPrimaryResponse() != null) {
                     // item was already aborted/processed by a filter in the chain upstream (e.g. security)
                     continue;
@@ -440,6 +440,7 @@ private Map<String, List<FieldInferenceRequest>> createFieldInferenceRequests(Bu
                     // ignore delete request
                     continue;
                 }
+
                 final Map<String, Object> docMap = indexRequest.sourceAsMap();
                 for (var entry : fieldInferenceMap.values()) {
                     String field = entry.getName();
@@ -481,7 +482,6 @@ private Map<String, List<FieldInferenceRequest>> createFieldInferenceRequests(Bu
                         }
                     }
                 }
-                itemIndex++;
             }
             return fieldRequestsMap;
         }
diff --git a/x-pack/plugin/inference/src/main/java/org/elasticsearch/xpack/inference/mapper/SemanticTextFieldMapper.java b/x-pack/plugin/inference/src/main/java/org/elasticsearch/xpack/inference/mapper/SemanticTextFieldMapper.java
@@ -83,6 +83,7 @@
 public class SemanticTextFieldMapper extends FieldMapper implements InferenceFieldMapper {
     public static final NodeFeature SEMANTIC_TEXT_IN_OBJECT_FIELD_FIX = new NodeFeature("semantic_text.in_object_field_fix");
     public static final NodeFeature SEMANTIC_TEXT_ZERO_SIZE_FIX = new NodeFeature("semantic_text.zero_size_fix");
+    public static final NodeFeature SEMANTIC_TEXT_DELETE_FIX = new NodeFeature("semantic_text.delete_fix");
 
     public static final String CONTENT_TYPE = "semantic_text";
 
diff --git a/x-pack/plugin/inference/src/yamlRestTest/resources/rest-api-spec/test/inference/30_semantic_text_inference.yml b/x-pack/plugin/inference/src/yamlRestTest/resources/rest-api-spec/test/inference/30_semantic_text_inference.yml
@@ -502,3 +502,59 @@ setup:
   - match: { _source.level_1.dense_field.text: "another inference test" }
   - exists: _source.level_1.dense_field.inference.chunks.0.embeddings
   - match: { _source.level_1.dense_field.inference.chunks.0.text: "another inference test" }
+
+---
+"Deletes on bulk operation":
+  - requires:
+      cluster_features: semantic_text.delete_fix
+      reason: Delete operations are properly applied when subsequent operations include a semantic text field.
+
+  - do:
+      bulk:
+        index: test-index
+        refresh: true
+        body: |
+          {"index":{"_id": "1"}}
+          {"dense_field": ["you know, for testing", "now with chunks"]}
+          {"index":{"_id": "2"}}
+          {"dense_field": ["some more tests", "that include chunks"]}
+
+  - do:
+      search:
+        index: test-index
+        body:
+          query:
+            semantic:
+              field: dense_field
+              query: "you know, for testing"
+
+  - match: { hits.total.value: 2 }
+  - match: { hits.total.relation: eq }
+  - match: { hits.hits.0._source.dense_field.text: ["you know, for testing", "now with chunks"] }
+  - match: { hits.hits.1._source.dense_field.text: ["some more tests", "that include chunks"] }
+
+  - do:
+      bulk:
+        index: test-index
+        refresh: true
+        body: |
+          {"delete":{ "_id": "2"}}
+          {"update":{"_id": "1"}}
+          {"doc":{"dense_field": "updated text", "sparse_field": "another text"}}
+
+  - match: { errors: false }
+
+
+  - do:
+      search:
+        index: test-index
+        body:
+          query:
+            semantic:
+              field: dense_field
+              query: "you know, for testing"
+
+  - match: { hits.total.value: 1 }
+  - match: { hits.total.relation: eq }
+  - match: { hits.hits.0._source.dense_field.text: "updated text" }
+  - match: { hits.hits.0._source.sparse_field.text: "another text" }

Original file line number	Diff line number	Diff line change
`@@ -26,6 +26,10 @@ public Set<NodeFeature> getFeatures() {`
`26`	`26`
`27`	`27`	`@Override`
`28`	`28`	`public Set<NodeFeature> getTestFeatures() {`
`29`		`- return Set.of(SemanticTextFieldMapper.SEMANTIC_TEXT_IN_OBJECT_FIELD_FIX, SemanticTextFieldMapper.SEMANTIC_TEXT_ZERO_SIZE_FIX);`
	`29`	`+ return Set.of(`
	`30`	`+ SemanticTextFieldMapper.SEMANTIC_TEXT_IN_OBJECT_FIELD_FIX,`
	`31`	`+ SemanticTextFieldMapper.SEMANTIC_TEXT_ZERO_SIZE_FIX,`
	`32`	`+ SemanticTextFieldMapper.SEMANTIC_TEXT_DELETE_FIX`
	`33`	`+ );`
`30`	`34`	`}`
`31`	`35`	`}`
Original file line number	Diff line number	Diff line change
`@@ -412,8 +412,8 @@ private void applyInferenceResponses(BulkItemRequest item, FieldInferenceRespons`
`412`	`412`	`*/`
`413`	`413`	`private Map<String, List<FieldInferenceRequest>> createFieldInferenceRequests(BulkShardRequest bulkShardRequest) {`
`414`	`414`	`Map<String, List<FieldInferenceRequest>> fieldRequestsMap = new LinkedHashMap<>();`
`415`		`- int itemIndex = 0;`
`416`		`- for (var item : bulkShardRequest.items()) {`
	`415`	`+ for (int itemIndex = 0; itemIndex < bulkShardRequest.items().length; itemIndex++) {`
	`416`	`+ var item = bulkShardRequest.items()[itemIndex];`
`417`	`417`	`if (item.getPrimaryResponse() != null) {`
`418`	`418`	`// item was already aborted/processed by a filter in the chain upstream (e.g. security)`
`419`	`419`	`continue;`
`@@ -440,6 +440,7 @@ private Map<String, List<FieldInferenceRequest>> createFieldInferenceRequests(Bu`
`440`	`440`	`// ignore delete request`
`441`	`441`	`continue;`
`442`	`442`	`}`
	`443`	`+`
`443`	`444`	`final Map<String, Object> docMap = indexRequest.sourceAsMap();`
`444`	`445`	`for (var entry : fieldInferenceMap.values()) {`
`445`	`446`	`String field = entry.getName();`
`@@ -481,7 +482,6 @@ private Map<String, List<FieldInferenceRequest>> createFieldInferenceRequests(Bu`
`481`	`482`	`}`
`482`	`483`	`}`
`483`	`484`	`}`
`484`		`- itemIndex++;`
`485`	`485`	`}`
`486`	`486`	`return fieldRequestsMap;`
`487`	`487`	`}`