Change

Author: Tim-Brooks

server/src/main/java/module-info.java

@@ -54,7 +54,6 @@
     requires org.apache.lucene.queryparser;
     requires org.apache.lucene.sandbox;
     requires org.apache.lucene.suggest;
-    requires org.elasticsearch.server;
 
     exports org.elasticsearch;
     exports org.elasticsearch.action;

server/src/main/java/org/elasticsearch/TransportVersions.java

@@ -369,6 +369,7 @@ static TransportVersion def(int id) {
     public static final TransportVersion SCRIPT_RESCORER = def(9_143_0_00);
     public static final TransportVersion ESQL_LOOKUP_OPERATOR_EMITTED_ROWS = def(9_144_0_00);
     public static final TransportVersion ALLOCATION_DECISION_NOT_PREFERRED = def(9_145_0_00);
+    public static final TransportVersion SOURCE_CONTEXT = def(9_146_0_00);
 
     /*
      * STOP! READ THIS FIRST! No, really,

server/src/main/java/org/elasticsearch/action/index/IndexRequest.java

@@ -25,7 +25,6 @@
 import org.elasticsearch.cluster.metadata.ProjectMetadata;
 import org.elasticsearch.cluster.routing.IndexRouting;
 import org.elasticsearch.common.UUIDs;
-import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
@@ -39,9 +38,7 @@
 import org.elasticsearch.index.mapper.MapperService;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.ingest.IngestService;
-import org.elasticsearch.plugins.internal.XContentParserDecorator;
 import org.elasticsearch.xcontent.XContentBuilder;
-import org.elasticsearch.xcontent.XContentFactory;
 import org.elasticsearch.xcontent.XContentType;
 
 import java.io.IOException;
@@ -98,15 +95,13 @@ public class IndexRequest extends ReplicatedWriteRequest<IndexRequest> implement
     @Nullable
     private String routing;
 
-    private BytesReference source;
+    private SourceContext sourceContext = new SourceContext();
 
     private OpType opType = OpType.INDEX;
 
     private long version = Versions.MATCH_ANY;
     private VersionType versionType = VersionType.INTERNAL;
 
-    private XContentType contentType;
-
     private String pipeline;
     private String finalPipeline;
 
@@ -165,7 +160,14 @@ public IndexRequest(@Nullable ShardId shardId, StreamInput in) throws IOExceptio
         }
         id = in.readOptionalString();
         routing = in.readOptionalString();
-        source = in.readBytesReference();
+        boolean beforeSourceContext = in.getTransportVersion().before(TransportVersions.SOURCE_CONTEXT);
+        BytesReference source;
+        if (beforeSourceContext) {
+            source = in.readBytesReference();
+        } else {
+            sourceContext = new SourceContext(in);
+            source = null;
+        }
         opType = OpType.fromId(in.readByte());
         version = in.readLong();
         versionType = VersionType.fromValue(in.readByte());
@@ -174,11 +176,15 @@ public IndexRequest(@Nullable ShardId shardId, StreamInput in) throws IOExceptio
         isPipelineResolved = in.readBoolean();
         isRetry = in.readBoolean();
         autoGeneratedTimestamp = in.readLong();
-        if (in.readBoolean()) {
-            // faster than StreamInput::readEnum, do not replace we read a lot of these instances at times
-            contentType = XContentType.ofOrdinal(in.readByte());
-        } else {
-            contentType = null;
+        if (beforeSourceContext) {
+            XContentType contentType;
+            if (in.readBoolean()) {
+                // faster than StreamInput::readEnum, do not replace we read a lot of these instances at times
+                contentType = XContentType.ofOrdinal(in.readByte());
+            } else {
+                contentType = null;
+            }
+            sourceContext = new SourceContext(contentType, source, () -> {});
         }
         ifSeqNo = in.readZLong();
         ifPrimaryTerm = in.readVLong();
@@ -250,10 +256,10 @@ private static String readPipelineName(StreamInput in) throws IOException {
     @Override
     public ActionRequestValidationException validate() {
         ActionRequestValidationException validationException = super.validate();
-        if (source == null) {
+        if (sourceContext.hasSource() == false) {
             validationException = addValidationError("source is missing", validationException);
         }
-        if (contentType == null) {
+        if (sourceContext.contentType() == null) {
             validationException = addValidationError("content type is missing", validationException);
         }
         assert opType == OpType.INDEX || opType == OpType.CREATE : "unexpected op-type: " + opType;
@@ -311,7 +317,7 @@ public ActionRequestValidationException validate() {
      * source at index time
      */
     public XContentType getContentType() {
-        return contentType;
+        return sourceContext.contentType();
     }
 
     /**
@@ -412,15 +418,11 @@ public boolean isPipelineResolved() {
      * The source of the document to index, recopied to a new array if it is unsafe.
      */
     public BytesReference source() {
-        return source;
+        return sourceContext.bytes();
     }
 
     public Map<String, Object> sourceAsMap() {
-        return XContentHelper.convertToMap(source, false, contentType).v2();
-    }
-
-    public Map<String, Object> sourceAsMap(XContentParserDecorator parserDecorator) {
-        return XContentHelper.convertToMap(source, false, contentType, parserDecorator).v2();
+        return sourceContext.sourceAsMap();
     }
 
     /**
@@ -429,7 +431,8 @@ public Map<String, Object> sourceAsMap(XContentParserDecorator parserDecorator)
      * @param source The map to index
      */
     public IndexRequest source(Map<String, ?> source) throws ElasticsearchGenerationException {
-        return source(source, Requests.INDEX_CONTENT_TYPE);
+        sourceContext.source(source);
+        return this;
     }
 
     /**
@@ -438,24 +441,14 @@ public IndexRequest source(Map<String, ?> source) throws ElasticsearchGeneration
      * @param source The map to index
      */
     public IndexRequest source(Map<String, ?> source, XContentType contentType) throws ElasticsearchGenerationException {
-        try {
-            XContentBuilder builder = XContentFactory.contentBuilder(contentType);
-            builder.map(source);
-            return source(builder);
-        } catch (IOException e) {
-            throw new ElasticsearchGenerationException("Failed to generate [" + source + "]", e);
-        }
+        sourceContext.source(source, contentType);
+        return this;
     }
 
     public IndexRequest source(Map<String, ?> source, XContentType contentType, boolean ensureNoSelfReferences)
         throws ElasticsearchGenerationException {
-        try {
-            XContentBuilder builder = XContentFactory.contentBuilder(contentType);
-            builder.map(source, ensureNoSelfReferences);
-            return source(builder);
-        } catch (IOException e) {
-            throw new ElasticsearchGenerationException("Failed to generate [" + source + "]", e);
-        }
+        sourceContext.source(source, contentType, ensureNoSelfReferences);
+        return this;
     }
 
     /**
@@ -465,14 +458,16 @@ public IndexRequest source(Map<String, ?> source, XContentType contentType, bool
      * or using the {@link #source(byte[], XContentType)}.
      */
     public IndexRequest source(String source, XContentType xContentType) {
-        return source(new BytesArray(source), xContentType);
+        sourceContext.source(source, xContentType);
+        return this;
     }
 
     /**
      * Sets the content source to index.
      */
     public IndexRequest source(XContentBuilder sourceBuilder) {
-        return source(BytesReference.bytes(sourceBuilder), sourceBuilder.contentType());
+        sourceContext.source(sourceBuilder);
+        return this;
     }
 
     /**
@@ -484,7 +479,8 @@ public IndexRequest source(XContentBuilder sourceBuilder) {
      * </p>
      */
     public IndexRequest source(Object... source) {
-        return source(Requests.INDEX_CONTENT_TYPE, source);
+        sourceContext.source(source);
+        return this;
     }
 
     /**
@@ -496,57 +492,29 @@ public IndexRequest source(Object... source) {
      * </p>
      */
     public IndexRequest source(XContentType xContentType, Object... source) {
-        return source(getXContentBuilder(xContentType, source));
+        sourceContext.source(xContentType, source);
+        return this;
     }
 
-    /**
-     * Returns an XContentBuilder for the given xContentType and source array
-     * <p>
-     * <b>Note: the number of objects passed to this method as varargs must be an even
-     * number. Also the first argument in each pair (the field name) must have a
-     * valid String representation.</b>
-     * </p>
-     */
-    public static XContentBuilder getXContentBuilder(XContentType xContentType, Object... source) {
-        if (source.length % 2 != 0) {
-            throw new IllegalArgumentException("The number of object passed must be even but was [" + source.length + "]");
-        }
-        if (source.length == 2 && source[0] instanceof BytesReference && source[1] instanceof Boolean) {
-            throw new IllegalArgumentException(
-                "you are using the removed method for source with bytes and unsafe flag, the unsafe flag"
-                    + " was removed, please just use source(BytesReference)"
-            );
-        }
-        try {
-            XContentBuilder builder = XContentFactory.contentBuilder(xContentType);
-            builder.startObject();
-            // This for loop increments by 2 because the source array contains adjacent key/value pairs:
-            for (int i = 0; i < source.length; i = i + 2) {
-                String field = source[i].toString();
-                Object value = source[i + 1];
-                builder.field(field, value);
-            }
-            builder.endObject();
-            return builder;
-        } catch (IOException e) {
-            throw new ElasticsearchGenerationException("Failed to generate", e);
-        }
+    public IndexRequest sourceContext(SourceContext sourceContext) {
+        sourceContext = Objects.requireNonNull(sourceContext);
+        return this;
     }
 
     /**
      * Sets the document to index in bytes form.
      */
     public IndexRequest source(BytesReference source, XContentType xContentType) {
-        this.source = Objects.requireNonNull(source);
-        this.contentType = Objects.requireNonNull(xContentType);
+        sourceContext.source(Objects.requireNonNull(source), Objects.requireNonNull(xContentType));
         return this;
     }
 
     /**
      * Sets the document to index in bytes form.
      */
     public IndexRequest source(byte[] source, XContentType xContentType) {
-        return source(source, 0, source.length, xContentType);
+        sourceContext.source(source, xContentType);
+        return this;
     }
 
     /**
@@ -558,7 +526,8 @@ public IndexRequest source(byte[] source, XContentType xContentType) {
      * @param length The length of the data
      */
     public IndexRequest source(byte[] source, int offset, int length, XContentType xContentType) {
-        return source(new BytesArray(source, offset, length), xContentType);
+        sourceContext.source(source, offset, length, xContentType);
+        return this;
     }
 
     /**
@@ -769,7 +738,11 @@ private void writeBody(StreamOutput out) throws IOException {
         }
         out.writeOptionalString(id);
         out.writeOptionalString(routing);
-        out.writeBytesReference(source);
+        if (out.getTransportVersion().onOrAfter(TransportVersions.SOURCE_CONTEXT)) {
+            sourceContext.writeTo(out);
+        } else {
+            out.writeBytesReference(sourceContext.bytes());
+        }
         out.writeByte(opType.getId());
         out.writeLong(version);
         out.writeByte(versionType.getValue());
@@ -778,11 +751,14 @@ private void writeBody(StreamOutput out) throws IOException {
         out.writeBoolean(isPipelineResolved);
         out.writeBoolean(isRetry);
         out.writeLong(autoGeneratedTimestamp);
-        if (contentType != null) {
-            out.writeBoolean(true);
-            XContentHelper.writeTo(out, contentType);
-        } else {
-            out.writeBoolean(false);
+        if (out.getTransportVersion().before(TransportVersions.SOURCE_CONTEXT)) {
+            XContentType contentType = sourceContext.contentType();
+            if (contentType != null) {
+                out.writeBoolean(true);
+                XContentHelper.writeTo(out, contentType);
+            } else {
+                out.writeBoolean(false);
+            }
         }
         out.writeZLong(ifSeqNo);
         out.writeVLong(ifPrimaryTerm);
@@ -821,13 +797,13 @@ private void writeBody(StreamOutput out) throws IOException {
     public String toString() {
         String sSource = "_na_";
         try {
-            if (source.length() > MAX_SOURCE_LENGTH_IN_TOSTRING) {
+            if (sourceContext.byteLength() > MAX_SOURCE_LENGTH_IN_TOSTRING) {
                 sSource = "n/a, actual length: ["
-                    + ByteSizeValue.ofBytes(source.length()).toString()
+                    + ByteSizeValue.ofBytes(sourceContext.byteLength()).toString()
                     + "], max length: "
                     + ByteSizeValue.ofBytes(MAX_SOURCE_LENGTH_IN_TOSTRING).toString();
             } else {
-                sSource = XContentHelper.convertToJson(source, false);
+                sSource = XContentHelper.convertToJson(sourceContext.bytes(), false);
             }
         } catch (Exception e) {
             // ignore
@@ -862,7 +838,7 @@ public long getAutoGeneratedTimestamp() {
 
     @Override
     public long ramBytesUsed() {
-        return SHALLOW_SIZE + RamUsageEstimator.sizeOf(id) + (source == null ? 0 : source.length());
+        return SHALLOW_SIZE + RamUsageEstimator.sizeOf(id) + sourceContext.byteLength();
     }
 
     @Override
@@ -917,7 +893,7 @@ public Index getConcreteWriteIndex(IndexAbstraction ia, ProjectMetadata project)
 
     @Override
     public int route(IndexRouting indexRouting) {
-        return indexRouting.indexShard(id, routing, contentType, source);
+        return indexRouting.indexShard(id, routing, sourceContext.contentType(), sourceContext.bytes());
     }
 
     public IndexRequest setRequireAlias(boolean requireAlias) {

server/src/main/java/org/elasticsearch/action/index/IndexRequestBuilder.java

@@ -176,7 +176,7 @@ public IndexRequestBuilder setSource(Object... source) {
      * </p>
      */
     public IndexRequestBuilder setSource(XContentType xContentType, Object... source) {
-        return setSource(IndexRequest.getXContentBuilder(xContentType, source));
+        return setSource(SourceContext.getXContentBuilder(xContentType, source));
     }
 
     /**