Deduplicate Strings in REST Bulk Request Parsing (#56506)

We can save a little memory here since these strings might live for quite
a while on the coordinating node.

Author: original-brownbear

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

@@ -37,8 +37,11 @@
 
 import java.io.IOException;
 import java.io.InputStream;
+import java.util.HashMap;
+import java.util.Map;
 import java.util.function.BiConsumer;
 import java.util.function.Consumer;
+import java.util.function.Function;
 
 import static org.elasticsearch.index.seqno.SequenceNumbers.UNASSIGNED_PRIMARY_TERM;
 
@@ -115,6 +118,10 @@ public void parse(
         int line = 0;
         int from = 0;
         byte marker = xContent.streamSeparator();
+        // Bulk requests can contain a lot of repeated strings for the index, pipeline and routing parameters. This map is used to
+        // deduplicate duplicate strings parsed for these parameters. While it does not prevent instantiating the duplicate strings, it
+        // reduces their lifetime to the lifetime of this parse call instead of the lifetime of the full bulk request.
+        final Map<String, String> stringDeduplicator = new HashMap<>();
         while (true) {
             int nextMarker = findNextMarker(marker, from, data);
             if (nextMarker == -1) {
@@ -174,17 +181,17 @@ public void parse(
                                 if (!allowExplicitIndex) {
                                     throw new IllegalArgumentException("explicit index in bulk is not allowed");
                                 }
-                                index = parser.text();
+                                index = stringDeduplicator.computeIfAbsent(parser.text(), Function.identity());
                             } else if (TYPE.match(currentFieldName, parser.getDeprecationHandler())) {
                                 if (errorOnType) {
                                     throw new IllegalArgumentException("Action/metadata line [" + line + "] contains an unknown parameter ["
                                         + currentFieldName + "]");
                                 }
-                                type = parser.text();
+                                type = stringDeduplicator.computeIfAbsent(parser.text(), Function.identity());
                             } else if (ID.match(currentFieldName, parser.getDeprecationHandler())) {
                                 id = parser.text();
                             } else if (ROUTING.match(currentFieldName, parser.getDeprecationHandler())) {
-                                routing = parser.text();
+                                routing = stringDeduplicator.computeIfAbsent(parser.text(), Function.identity());
                             } else if (OP_TYPE.match(currentFieldName, parser.getDeprecationHandler())) {
                                 opType = parser.text();
                             } else if (VERSION.match(currentFieldName, parser.getDeprecationHandler())) {
@@ -198,7 +205,7 @@ public void parse(
                             } else if (RETRY_ON_CONFLICT.match(currentFieldName, parser.getDeprecationHandler())) {
                                 retryOnConflict = parser.intValue();
                             } else if (PIPELINE.match(currentFieldName, parser.getDeprecationHandler())) {
-                                pipeline = parser.text();
+                                pipeline = stringDeduplicator.computeIfAbsent(parser.text(), Function.identity());
                             } else if (SOURCE.match(currentFieldName, parser.getDeprecationHandler())) {
                                 fetchSourceContext = FetchSourceContext.fromXContent(parser);
                             } else {

server/src/test/java/org/elasticsearch/action/bulk/BulkRequestParserTests.java

@@ -19,11 +19,15 @@
 
 package org.elasticsearch.action.bulk;
 
+import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.test.ESTestCase;
+import org.hamcrest.Matchers;
 
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 public class BulkRequestParserTests extends ESTestCase {
@@ -109,4 +113,20 @@ public void testTypesStillParsedForBulkMonitoring() throws IOException {
         assertTrue(parsed.get());
     }
 
+    public void testParseDeduplicatesParameterStrings() throws IOException {
+        BytesArray request = new BytesArray(
+                "{ \"index\":{ \"_index\": \"bar\", \"pipeline\": \"foo\", \"routing\": \"blub\"} }\n{}\n"
+                + "{ \"index\":{ \"_index\": \"bar\", \"pipeline\": \"foo\", \"routing\": \"blub\" } }\n{}\n");
+        BulkRequestParser parser = new BulkRequestParser(randomBoolean());
+        final List<IndexRequest> indexRequests = new ArrayList<>();
+        parser.parse(request, null, null, null, null, true, XContentType.JSON,
+                (indexRequest, type) -> indexRequests.add(indexRequest),
+                req -> fail(), req -> fail());
+        assertThat(indexRequests, Matchers.hasSize(2));
+        final IndexRequest first = indexRequests.get(0);
+        final IndexRequest second = indexRequests.get(1);
+        assertSame(first.index(), second.index());
+        assertSame(first.getPipeline(), second.getPipeline());
+        assertSame(first.routing(), second.routing());
+    }
 }