Fix needless copying during X-Content parsing (elastic#103516)

A couple new spots made it into the codebase where we needless stream
from an array backed `BytesReference`. We can be a little faster and
cache-efficient by not doing that. I added a new utilty that avoids
compress check to XContentHelper and dried up some logic around this
topic that way.

modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/JsonProcessor.java

@@ -8,8 +8,6 @@
 
 package org.elasticsearch.ingest.common;
 
-import org.elasticsearch.common.bytes.BytesArray;
-import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.core.Strings;
 import org.elasticsearch.ingest.AbstractProcessor;
 import org.elasticsearch.ingest.ConfigurationUtils;
@@ -20,7 +18,6 @@
 import org.elasticsearch.xcontent.json.JsonXContent;
 
 import java.io.IOException;
-import java.io.InputStream;
 import java.util.Locale;
 import java.util.Map;
 
@@ -90,10 +87,11 @@ public ConflictStrategy getAddToRootConflictStrategy() {
     }
 
     public static Object apply(Object fieldValue, boolean allowDuplicateKeys, boolean strictJsonParsing) {
-        BytesReference bytesRef = fieldValue == null ? new BytesArray("null") : new BytesArray(fieldValue.toString());
         try (
-            InputStream stream = bytesRef.streamInput();
-            XContentParser parser = JsonXContent.jsonXContent.createParser(XContentParserConfiguration.EMPTY, stream)
+            XContentParser parser = JsonXContent.jsonXContent.createParser(
+                XContentParserConfiguration.EMPTY,
+                fieldValue == null ? "null" : fieldValue.toString()
+            )
         ) {
             parser.allowDuplicateKeys(allowDuplicateKeys);
             XContentParser.Token token = parser.nextToken();

modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/ScriptProcessor.java

@@ -10,6 +10,7 @@
 
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.xcontent.LoggingDeprecationHandler;
+import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.ingest.AbstractProcessor;
 import org.elasticsearch.ingest.IngestDocument;
@@ -25,7 +26,6 @@
 import org.elasticsearch.xcontent.XContentType;
 import org.elasticsearch.xcontent.json.JsonXContent;
 
-import java.io.InputStream;
 import java.util.Arrays;
 import java.util.Map;
 
@@ -108,9 +108,11 @@ public ScriptProcessor create(
         ) throws Exception {
             try (
                 XContentBuilder builder = XContentBuilder.builder(JsonXContent.jsonXContent).map(config);
-                InputStream stream = BytesReference.bytes(builder).streamInput();
-                XContentParser parser = XContentType.JSON.xContent()
-                    .createParser(XContentParserConfiguration.EMPTY.withDeprecationHandler(LoggingDeprecationHandler.INSTANCE), stream)
+                XContentParser parser = XContentHelper.createParserNotCompressed(
+                    XContentParserConfiguration.EMPTY.withDeprecationHandler(LoggingDeprecationHandler.INSTANCE),
+                    BytesReference.bytes(builder),
+                    XContentType.JSON
+                )
             ) {
                 Script script = Script.parse(parser);
 

modules/reindex/src/main/java/org/elasticsearch/reindex/Reindexer.java

@@ -36,6 +36,7 @@
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.lucene.uid.Versions;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.index.IndexMode;
 import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.index.VersionType;
@@ -60,7 +61,6 @@
 import org.elasticsearch.xcontent.XContentType;
 
 import java.io.IOException;
-import java.io.InputStream;
 import java.io.UncheckedIOException;
 import java.util.ArrayList;
 import java.util.List;
@@ -321,8 +321,11 @@ protected RequestWrapper<IndexRequest> buildRequest(ScrollableHitSource.Hit doc)
             if (mainRequestXContentType != null && doc.getXContentType() != mainRequestXContentType) {
                 // we need to convert
                 try (
-                    InputStream stream = doc.getSource().streamInput();
-                    XContentParser parser = sourceXContentType.xContent().createParser(XContentParserConfiguration.EMPTY, stream);
+                    XContentParser parser = XContentHelper.createParserNotCompressed(
+                        XContentParserConfiguration.EMPTY,
+                        doc.getSource(),
+                        sourceXContentType
+                    );
                     XContentBuilder builder = XContentBuilder.builder(mainRequestXContentType.xContent())
                 ) {
                     parser.nextToken();

server/src/main/java/org/elasticsearch/action/search/MultiSearchRequest.java

@@ -19,6 +19,7 @@
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.logging.DeprecationLogger;
+import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.core.RestApiVersion;
 import org.elasticsearch.rest.action.search.RestMultiSearchAction;
 import org.elasticsearch.rest.action.search.RestSearchAction;
@@ -33,7 +34,6 @@
 
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
-import java.io.InputStream;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Locale;
@@ -239,8 +239,11 @@ public static void readMultiLineFormat(
             // now parse the action
             if (nextMarker - from > 0) {
                 try (
-                    InputStream stream = data.slice(from, nextMarker - from).streamInput();
-                    XContentParser parser = xContent.createParser(parserConfig, stream)
+                    XContentParser parser = XContentHelper.createParserNotCompressed(
+                        parserConfig,
+                        data.slice(from, nextMarker - from),
+                        xContent.type()
+                    )
                 ) {
                     Map<String, Object> source = parser.map();
                     Object expandWildcards = null;
@@ -301,8 +304,13 @@ public static void readMultiLineFormat(
             if (nextMarker == -1) {
                 break;
             }
-            BytesReference bytes = data.slice(from, nextMarker - from);
-            try (InputStream stream = bytes.streamInput(); XContentParser parser = xContent.createParser(parserConfig, stream)) {
+            try (
+                XContentParser parser = XContentHelper.createParserNotCompressed(
+                    parserConfig,
+                    data.slice(from, nextMarker - from),
+                    xContent.type()
+                )
+            ) {
                 consumer.accept(searchRequest, parser);
             }
             // move pointers

server/src/main/java/org/elasticsearch/cluster/metadata/AliasValidator.java

@@ -24,7 +24,6 @@
 import org.elasticsearch.xcontent.XContentParserConfiguration;
 
 import java.io.IOException;
-import java.io.InputStream;
 import java.util.Optional;
 import java.util.function.Function;
 
@@ -142,14 +141,11 @@ public static void validateAliasFilter(
         assert searchExecutionContext != null;
 
         try (
-            InputStream inputStream = filter.streamInput();
-            XContentParser parser = XContentFactory.xContentType(inputStream)
-                .xContent()
-                .createParser(
-                    XContentParserConfiguration.EMPTY.withRegistry(xContentRegistry)
-                        .withDeprecationHandler(LoggingDeprecationHandler.INSTANCE),
-                    filter.streamInput()
-                )
+            XContentParser parser = XContentHelper.createParserNotCompressed(
+                XContentParserConfiguration.EMPTY.withRegistry(xContentRegistry).withDeprecationHandler(LoggingDeprecationHandler.INSTANCE),
+                filter,
+                XContentHelper.xContentType(filter)
+            )
         ) {
             validateAliasFilter(parser, searchExecutionContext);
         } catch (Exception e) {

server/src/main/java/org/elasticsearch/cluster/metadata/DataStream.java

@@ -28,6 +28,7 @@
 import org.elasticsearch.common.time.DateFormatter;
 import org.elasticsearch.common.time.DateFormatters;
 import org.elasticsearch.common.util.FeatureFlag;
+import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.core.Tuple;
@@ -38,7 +39,6 @@
 import org.elasticsearch.xcontent.ConstructingObjectParser;
 import org.elasticsearch.xcontent.ParseField;
 import org.elasticsearch.xcontent.ToXContentObject;
-import org.elasticsearch.xcontent.XContent;
 import org.elasticsearch.xcontent.XContentBuilder;
 import org.elasticsearch.xcontent.XContentParser;
 import org.elasticsearch.xcontent.XContentParserConfiguration;
@@ -1129,8 +1129,7 @@ private static Instant getTimeStampFromRaw(Object rawTimestamp) {
     }
 
     private static Instant getTimestampFromParser(BytesReference source, XContentType xContentType) {
-        XContent xContent = xContentType.xContent();
-        try (XContentParser parser = xContent.createParser(TS_EXTRACT_CONFIG, source.streamInput())) {
+        try (XContentParser parser = XContentHelper.createParserNotCompressed(TS_EXTRACT_CONFIG, source, xContentType)) {
             ensureExpectedToken(XContentParser.Token.START_OBJECT, parser.nextToken(), parser);
             ensureExpectedToken(XContentParser.Token.FIELD_NAME, parser.nextToken(), parser);
             return switch (parser.nextToken()) {

server/src/main/java/org/elasticsearch/common/xcontent/XContentHelper.java

@@ -74,10 +74,26 @@ public static XContentParser createParser(XContentParserConfiguration config, By
             final XContentType contentType = XContentFactory.xContentType(compressedInput);
             return XContentFactory.xContent(contentType).createParser(config, compressedInput);
         } else {
-            return XContentFactory.xContent(xContentType(bytes)).createParser(config, bytes.streamInput());
+            return createParserNotCompressed(config, bytes, xContentType(bytes));
         }
     }
 
+    /**
+     * Same as {@link #createParser(XContentParserConfiguration, BytesReference, XContentType)} but only supports uncompressed
+     * {@code bytes}.
+     */
+    public static XContentParser createParserNotCompressed(
+        XContentParserConfiguration config,
+        BytesReference bytes,
+        XContentType xContentType
+    ) throws IOException {
+        XContent xContent = xContentType.xContent();
+        if (bytes.hasArray()) {
+            return xContent.createParser(config, bytes.array(), bytes.arrayOffset(), bytes.length());
+        }
+        return xContent.createParser(config, bytes.streamInput());
+    }
+
     /**
      * Creates a parser for the bytes provided
      * @deprecated use {@link #createParser(XContentParserConfiguration, BytesReference, XContentType)}
@@ -111,10 +127,7 @@ public static XContentParser createParser(XContentParserConfiguration config, By
             return XContentFactory.xContent(xContentType).createParser(config, compressedInput);
         } else {
             // TODO now that we have config we make a method on bytes to do this building wihout needing this check everywhere
-            if (bytes.hasArray()) {
-                return xContentType.xContent().createParser(config, bytes.array(), bytes.arrayOffset(), bytes.length());
-            }
-            return xContentType.xContent().createParser(config, bytes.streamInput());
+            return createParserNotCompressed(config, bytes, xContentType);
         }
     }
 
@@ -337,20 +350,8 @@ public static String convertToJson(BytesReference bytes, boolean reformatJson, b
             return bytes.utf8ToString();
         }
 
-        if (bytes.hasArray()) {
-            try (
-                XContentParser parser = XContentFactory.xContent(xContentType)
-                    .createParser(XContentParserConfiguration.EMPTY, bytes.array(), bytes.arrayOffset(), bytes.length())
-            ) {
-                return toJsonString(prettyPrint, parser);
-            }
-        } else {
-            try (
-                InputStream stream = bytes.streamInput();
-                XContentParser parser = XContentFactory.xContent(xContentType).createParser(XContentParserConfiguration.EMPTY, stream)
-            ) {
-                return toJsonString(prettyPrint, parser);
-            }
+        try (var parser = createParserNotCompressed(XContentParserConfiguration.EMPTY, bytes, xContentType)) {
+            return toJsonString(prettyPrint, parser);
         }
     }
 
@@ -746,7 +747,7 @@ public static void writeTo(StreamOutput out, XContentType xContentType) throws I
     public static XContentParser mapToXContentParser(XContentParserConfiguration config, Map<String, ?> source) {
         try (XContentBuilder builder = XContentFactory.contentBuilder(XContentType.JSON)) {
             builder.map(source);
-            return XContentFactory.xContent(builder.contentType()).createParser(config, Strings.toString(builder));
+            return createParserNotCompressed(config, BytesReference.bytes(builder), builder.contentType());
         } catch (IOException e) {
             throw new ElasticsearchGenerationException("Failed to generate [" + source + "]", e);
         }

server/src/main/java/org/elasticsearch/gateway/PersistedClusterStateService.java

@@ -61,6 +61,7 @@
 import org.elasticsearch.common.util.Maps;
 import org.elasticsearch.common.xcontent.ChunkedToXContent;
 import org.elasticsearch.common.xcontent.LoggingDeprecationHandler;
+import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.core.Assertions;
 import org.elasticsearch.core.CheckedConsumer;
 import org.elasticsearch.core.CheckedFunction;
@@ -668,8 +669,7 @@ public OnDiskStateMetadata loadOnDiskStateMetadataFromUserData(Map<String, Strin
     }
 
     private <T> T readXContent(BytesReference bytes, CheckedFunction<XContentParser, T, IOException> reader) throws IOException {
-        final XContentParser parser = XContentFactory.xContent(XContentType.SMILE).createParser(parserConfig, bytes.streamInput());
-        try {
+        try (XContentParser parser = XContentHelper.createParserNotCompressed(parserConfig, bytes, XContentType.SMILE)) {
             return reader.apply(parser);
         } catch (Exception e) {
             throw new CorruptStateException(e);

server/src/main/java/org/elasticsearch/indices/IndicesService.java

@@ -64,6 +64,7 @@
 import org.elasticsearch.common.util.concurrent.EsThreadPoolExecutor;
 import org.elasticsearch.common.util.iterable.Iterables;
 import org.elasticsearch.common.xcontent.LoggingDeprecationHandler;
+import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.core.AbstractRefCounted;
 import org.elasticsearch.core.CheckedConsumer;
 import org.elasticsearch.core.CheckedFunction;
@@ -138,14 +139,12 @@
 import org.elasticsearch.search.query.QueryPhase;
 import org.elasticsearch.search.query.QuerySearchResult;
 import org.elasticsearch.threadpool.ThreadPool;
-import org.elasticsearch.xcontent.XContentFactory;
 import org.elasticsearch.xcontent.XContentParser;
 import org.elasticsearch.xcontent.XContentParserConfiguration;
 import org.elasticsearch.xcontent.XContentType;
 
 import java.io.Closeable;
 import java.io.IOException;
-import java.io.InputStream;
 import java.io.UncheckedIOException;
 import java.nio.file.Files;
 import java.util.ArrayList;
@@ -1654,8 +1653,7 @@ public AliasFilter buildAliasFilter(ClusterState state, String index, Set<String
          * of dependencies we pass in a function that can perform the parsing. */
         CheckedFunction<BytesReference, QueryBuilder, IOException> filterParser = bytes -> {
             try (
-                InputStream inputStream = bytes.streamInput();
-                XContentParser parser = XContentFactory.xContentType(inputStream).xContent().createParser(parserConfig, inputStream)
+                XContentParser parser = XContentHelper.createParserNotCompressed(parserConfig, bytes, XContentHelper.xContentType(bytes))
             ) {
                 return parseTopLevelQuery(parser);
             }

server/src/main/java/org/elasticsearch/repositories/blobstore/ChecksumBlobStoreFormat.java

@@ -25,6 +25,7 @@
 import org.elasticsearch.common.io.Streams;
 import org.elasticsearch.common.lucene.store.IndexOutputOutputStream;
 import org.elasticsearch.common.xcontent.LoggingDeprecationHandler;
+import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.common.xcontent.XContentParserUtils;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.gateway.CorruptStateException;
@@ -33,6 +34,7 @@
 import org.elasticsearch.xcontent.XContentBuilder;
 import org.elasticsearch.xcontent.XContentFactory;
 import org.elasticsearch.xcontent.XContentParser;
+import org.elasticsearch.xcontent.XContentParserConfiguration;
 import org.elasticsearch.xcontent.XContentType;
 
 import java.io.FilterInputStream;
@@ -144,15 +146,23 @@ public T deserialize(String repoName, NamedXContentRegistry namedXContentRegistr
                 BytesReference bytesReference = Streams.readFully(wrappedStream);
                 deserializeMetaBlobInputStream.verifyFooter();
                 try (
-                    XContentParser parser = XContentType.SMILE.xContent()
-                        .createParser(namedXContentRegistry, LoggingDeprecationHandler.INSTANCE, bytesReference.streamInput())
+                    XContentParser parser = XContentHelper.createParserNotCompressed(
+                        XContentParserConfiguration.EMPTY.withRegistry(namedXContentRegistry)
+                            .withDeprecationHandler(LoggingDeprecationHandler.INSTANCE),
+                        bytesReference,
+                        XContentType.SMILE
+                    )
                 ) {
                     result = reader.apply(repoName, parser);
                     XContentParserUtils.ensureExpectedToken(null, parser.nextToken(), parser);
                 } catch (Exception e) {
                     try (
-                        XContentParser parser = XContentType.SMILE.xContent()
-                            .createParser(namedXContentRegistry, LoggingDeprecationHandler.INSTANCE, bytesReference.streamInput())
+                        XContentParser parser = XContentHelper.createParserNotCompressed(
+                            XContentParserConfiguration.EMPTY.withRegistry(namedXContentRegistry)
+                                .withDeprecationHandler(LoggingDeprecationHandler.INSTANCE),
+                            bytesReference,
+                            XContentType.SMILE
+                        )
                     ) {
                         result = fallbackReader.apply(repoName, parser);
                         XContentParserUtils.ensureExpectedToken(null, parser.nextToken(), parser);

server/src/main/java/org/elasticsearch/rest/RestRequest.java

@@ -15,6 +15,7 @@
 import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.unit.ByteSizeValue;
+import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.core.Booleans;
 import org.elasticsearch.core.CheckedConsumer;
 import org.elasticsearch.core.Nullable;
@@ -31,7 +32,6 @@
 import org.elasticsearch.xcontent.XContentType;
 
 import java.io.IOException;
-import java.io.InputStream;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
@@ -541,12 +541,7 @@ public final XContentParser contentOrSourceParamParser() throws IOException {
     public final void withContentOrSourceParamParserOrNull(CheckedConsumer<XContentParser, IOException> withParser) throws IOException {
         if (hasContentOrSourceParam()) {
             Tuple<XContentType, BytesReference> tuple = contentOrSourceParam();
-            BytesReference content = tuple.v2();
-            XContentType xContentType = tuple.v1();
-            try (
-                InputStream stream = content.streamInput();
-                XContentParser parser = xContentType.xContent().createParser(parserConfig, stream)
-            ) {
+            try (XContentParser parser = XContentHelper.createParserNotCompressed(parserConfig, tuple.v2(), tuple.v1())) {
                 withParser.accept(parser);
             }
         } else {