Optimize indexing for the autogenerated ID append-only case (#20211)

If elasticsearch controls the ID values as well as the documents
version we can optimize the code that adds / appends the documents
to the index. Essentially we an skip the version lookup for all
documents unless the same document is delivered more than once.

On the lucene level we can simply call IndexWriter#addDocument instead
of #updateDocument but on the Engine level we need to ensure that we deoptimize
the case once we see the same document more than once.

This is done as follows:

1. Mark every request with a timestamp. This is done once on the first node that
receives a request and is fixed for this request. This can be even the
machine local time (see why later). The important part is that retry
requests will have the same value as the original one.

2. In the engine we make sure we keep the highest seen time stamp of "retry" requests.
This is updated while the retry request has its doc id lock. Call this `maxUnsafeAutoIdTimestamp`

3. When the engine runs an "optimized" request comes, it compares it's timestamp with the
current `maxUnsafeAutoIdTimestamp` (but doesn't update it). If the the request
timestamp is higher it is safe to execute it as optimized (no retry request with the same
timestamp has been run before). If not we fall back to "non-optimzed" mode and run the request as a retry one
and update the `maxUnsafeAutoIdTimestamp` unless it's been updated already to a higher value

Relates to #19813

Author: s1monw

buildSrc/src/main/resources/checkstyle_suppressions.xml

@@ -386,7 +386,6 @@
   <suppress files="core[/\\]src[/\\]main[/\\]java[/\\]org[/\\]elasticsearch[/\\]index[/\\]mapper[/\\]MapperService.java" checks="LineLength" />
   <suppress files="core[/\\]src[/\\]main[/\\]java[/\\]org[/\\]elasticsearch[/\\]index[/\\]mapper[/\\]Mapping.java" checks="LineLength" />
   <suppress files="core[/\\]src[/\\]main[/\\]java[/\\]org[/\\]elasticsearch[/\\]index[/\\]mapper[/\\]MetadataFieldMapper.java" checks="LineLength" />
-  <suppress files="core[/\\]src[/\\]main[/\\]java[/\\]org[/\\]elasticsearch[/\\]index[/\\]mapper[/\\]ParsedDocument.java" checks="LineLength" />
   <suppress files="core[/\\]src[/\\]main[/\\]java[/\\]org[/\\]elasticsearch[/\\]index[/\\]mapper[/\\]CompletionFieldMapper.java" checks="LineLength" />
   <suppress files="core[/\\]src[/\\]main[/\\]java[/\\]org[/\\]elasticsearch[/\\]index[/\\]mapper[/\\]LegacyDateFieldMapper.java" checks="LineLength" />
   <suppress files="core[/\\]src[/\\]main[/\\]java[/\\]org[/\\]elasticsearch[/\\]index[/\\]mapper[/\\]LegacyDoubleFieldMapper.java" checks="LineLength" />
@@ -865,7 +864,6 @@
   <suppress files="core[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]index[/\\]store[/\\]CorruptedFileIT.java" checks="LineLength" />
   <suppress files="core[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]index[/\\]store[/\\]CorruptedTranslogIT.java" checks="LineLength" />
   <suppress files="core[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]index[/\\]store[/\\]DirectoryUtilsTests.java" checks="LineLength" />
-  <suppress files="core[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]index[/\\]store[/\\]ExceptionRetryIT.java" checks="LineLength" />
   <suppress files="core[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]index[/\\]store[/\\]IndexStoreTests.java" checks="LineLength" />
   <suppress files="core[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]index[/\\]store[/\\]StoreTests.java" checks="LineLength" />
   <suppress files="core[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]index[/\\]suggest[/\\]stats[/\\]SuggestStatsIT.java" checks="LineLength" />

core/src/main/java/org/elasticsearch/action/bulk/BulkShardRequest.java

@@ -20,6 +20,7 @@
 package org.elasticsearch.action.bulk;
 
 import org.elasticsearch.action.support.replication.ReplicatedWriteRequest;
+import org.elasticsearch.action.support.replication.ReplicationRequest;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.index.shard.ShardId;
@@ -101,4 +102,15 @@ public String toString() {
         }
         return b.toString();
     }
+
+    @Override
+    public void onRetry() {
+        for (BulkItemRequest item : items) {
+            if (item.request() instanceof ReplicationRequest) {
+                // all replication requests need to be notified here as well to ie. make sure that internal optimizations are
+                // disabled see IndexRequest#canHaveDuplicates()
+                ((ReplicationRequest) item.request()).onRetry();
+            }
+        }
+    }
 }

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

@@ -72,7 +72,7 @@ public class IndexRequest extends ReplicatedWriteRequest<IndexRequest> implement
     /**
      * Operation type controls if the type of the index operation.
      */
-    public static enum OpType {
+    public enum OpType {
         /**
          * Index the source. If there an existing document with the id, it will
          * be replaced.
@@ -152,6 +152,17 @@ public static OpType fromString(String sOpType) {
 
     private String pipeline;
 
+    /**
+     * Value for {@link #getAutoGeneratedTimestamp()} if the document has an external
+     * provided ID.
+     */
+    public static final int UNSET_AUTO_GENERATED_TIMESTAMP = -1;
+
+    private long autoGeneratedTimestamp = UNSET_AUTO_GENERATED_TIMESTAMP;
+
+    private boolean isRetry = false;
+
+
     public IndexRequest() {
     }
 
@@ -202,6 +213,10 @@ public ActionRequestValidationException validate() {
             }
         }
 
+        if (opType() != OpType.INDEX && id == null) {
+            addValidationError("an id is required for a " + opType() + " operation", validationException);
+        }
+
         if (!versionType.validateVersionForWrites(version)) {
             validationException = addValidationError("illegal version value [" + version + "] for version type [" + versionType.name() + "]", validationException);
         }
@@ -216,6 +231,11 @@ public ActionRequestValidationException validate() {
             validationException = addValidationError("id is too long, must be no longer than 512 bytes but was: " +
                             id.getBytes(StandardCharsets.UTF_8).length, validationException);
         }
+
+        if (id == null && (versionType == VersionType.INTERNAL && version == Versions.MATCH_ANY) == false) {
+            validationException = addValidationError("an id must be provided if version type or value are set", validationException);
+        }
+
         return validationException;
     }
 
@@ -589,10 +609,10 @@ public void process(@Nullable MappingMetaData mappingMd, boolean allowIdGenerati
         }
 
         // generate id if not already provided and id generation is allowed
-        if (allowIdGeneration) {
-            if (id == null) {
-                id(UUIDs.base64UUID());
-            }
+        if (allowIdGeneration && id == null) {
+            assert autoGeneratedTimestamp == -1;
+            autoGeneratedTimestamp = Math.max(0, System.currentTimeMillis()); // extra paranoia
+            id(UUIDs.base64UUID());
         }
 
         // generate timestamp if not provided, we always have one post this stage...
@@ -639,6 +659,8 @@ public void readFrom(StreamInput in) throws IOException {
         version = in.readLong();
         versionType = VersionType.fromValue(in.readByte());
         pipeline = in.readOptionalString();
+        isRetry = in.readBoolean();
+        autoGeneratedTimestamp = in.readLong();
     }
 
     @Override
@@ -655,6 +677,8 @@ public void writeTo(StreamOutput out) throws IOException {
         out.writeLong(version);
         out.writeByte(versionType.getValue());
         out.writeOptionalString(pipeline);
+        out.writeBoolean(isRetry);
+        out.writeLong(autoGeneratedTimestamp);
     }
 
     @Override
@@ -667,4 +691,25 @@ public String toString() {
         }
         return "index {[" + index + "][" + type + "][" + id + "], source[" + sSource + "]}";
     }
+
+
+    /**
+     * Returns <code>true</code> if this request has been sent to a shard copy more than once.
+     */
+    public boolean isRetry() {
+        return isRetry;
+    }
+
+    @Override
+    public void onRetry() {
+        isRetry = true;
+    }
+
+    /**
+     * Returns the timestamp the auto generated ID was created or {@value #UNSET_AUTO_GENERATED_TIMESTAMP} if the
+     * document has no auto generated timestamp. This method will return a positive value iff the id was auto generated.
+     */
+    public long getAutoGeneratedTimestamp() {
+        return autoGeneratedTimestamp;
+    }
 }

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

@@ -205,15 +205,6 @@ public IndexRequestBuilder setOpType(IndexRequest.OpType opType) {
         return this;
     }
 
-    /**
-     * Sets a string representation of the {@link #setOpType(org.elasticsearch.action.index.IndexRequest.OpType)}. Can
-     * be either "index" or "create".
-     */
-    public IndexRequestBuilder setOpType(String opType) {
-        request.opType(IndexRequest.OpType.fromString(opType));
-        return this;
-    }
-
     /**
      * Set to <tt>true</tt> to force this index to use {@link org.elasticsearch.action.index.IndexRequest.OpType#CREATE}.
      */

core/src/main/java/org/elasticsearch/action/index/TransportIndexAction.java

@@ -158,7 +158,7 @@ public static Engine.Index executeIndexRequestOnReplica(IndexRequest request, In
         SourceToParse sourceToParse = SourceToParse.source(SourceToParse.Origin.REPLICA, shardId.getIndexName(), request.type(), request.id(), request.source())
                 .routing(request.routing()).parent(request.parent()).timestamp(request.timestamp()).ttl(request.ttl());
 
-        final Engine.Index operation = indexShard.prepareIndexOnReplica(sourceToParse, request.version(), request.versionType());
+        final Engine.Index operation = indexShard.prepareIndexOnReplica(sourceToParse, request.version(), request.versionType(), request.getAutoGeneratedTimestamp(), request.isRetry());
         Mapping update = operation.parsedDoc().dynamicMappingsUpdate();
         if (update != null) {
             throw new RetryOnReplicaException(shardId, "Mappings are not available on the replica yet, triggered update: " + update);
@@ -171,7 +171,7 @@ public static Engine.Index executeIndexRequestOnReplica(IndexRequest request, In
     public static Engine.Index prepareIndexOperationOnPrimary(IndexRequest request, IndexShard indexShard) {
         SourceToParse sourceToParse = SourceToParse.source(SourceToParse.Origin.PRIMARY, request.index(), request.type(), request.id(), request.source())
             .routing(request.routing()).parent(request.parent()).timestamp(request.timestamp()).ttl(request.ttl());
-        return indexShard.prepareIndexOnPrimary(sourceToParse, request.version(), request.versionType());
+        return indexShard.prepareIndexOnPrimary(sourceToParse, request.version(), request.versionType(), request.getAutoGeneratedTimestamp(), request.isRetry());
     }
 
     public static WriteResult<IndexResponse> executeIndexRequestOnPrimary(IndexRequest request, IndexShard indexShard,

core/src/main/java/org/elasticsearch/action/support/replication/ReplicationRequest.java

@@ -248,4 +248,12 @@ public String toString() {
     public String getDescription() {
         return toString();
     }
+
+    /**
+     * This method is called before this replication request is retried
+     * the first time.
+     */
+    public void onRetry() {
+        // nothing by default
+    }
 }

core/src/main/java/org/elasticsearch/action/support/replication/TransportReplicationAction.java

@@ -722,6 +722,7 @@ void retry(Exception failure) {
                 return;
             }
             setPhase(task, "waiting_for_retry");
+            request.onRetry();
             final ThreadContext.StoredContext context = threadPool.getThreadContext().newStoredContext();
             observer.waitForNextChange(new ClusterStateObserver.Listener() {
                 @Override

core/src/main/java/org/elasticsearch/common/settings/IndexScopedSettings.java

@@ -140,6 +140,7 @@ public final class IndexScopedSettings extends AbstractScopedSettings {
         PrimaryShardAllocator.INDEX_RECOVERY_INITIAL_SHARDS_SETTING,
         FsDirectoryService.INDEX_LOCK_FACTOR_SETTING,
         EngineConfig.INDEX_CODEC_SETTING,
+        EngineConfig.INDEX_OPTIMIZE_AUTO_GENERATED_IDS,
         IndexMetaData.SETTING_WAIT_FOR_ACTIVE_SHARDS,
         // validate that built-in similarities don't get redefined
         Setting.groupSetting("index.similarity.", (s) -> {

core/src/main/java/org/elasticsearch/index/engine/Engine.java

@@ -86,9 +86,6 @@
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 import java.util.function.Function;
 
-/**
- *
- */
 public abstract class Engine implements Closeable {
 
     public static final String SYNC_COMMIT_ID = "sync_id";
@@ -856,19 +853,24 @@ public long endTime() {
     public static class Index extends Operation {
 
         private final ParsedDocument doc;
+        private final long autoGeneratedIdTimestamp;
+        private final boolean isRetry;
         private boolean created;
 
-        public Index(Term uid, ParsedDocument doc, long version, VersionType versionType, Origin origin, long startTime) {
+        public Index(Term uid, ParsedDocument doc, long version, VersionType versionType, Origin origin, long startTime,
+                     long autoGeneratedIdTimestamp, boolean isRetry) {
             super(uid, version, versionType, origin, startTime);
             this.doc = doc;
+            this.isRetry = isRetry;
+            this.autoGeneratedIdTimestamp = autoGeneratedIdTimestamp;
         }
 
         public Index(Term uid, ParsedDocument doc) {
             this(uid, doc, Versions.MATCH_ANY);
-        }
+        } // TEST ONLY
 
-        public Index(Term uid, ParsedDocument doc, long version) {
-            this(uid, doc, version, VersionType.INTERNAL, Origin.PRIMARY, System.nanoTime());
+        Index(Term uid, ParsedDocument doc, long version) {
+            this(uid, doc, version, VersionType.INTERNAL, Origin.PRIMARY, System.nanoTime(), -1, false);
         }
 
         public ParsedDocument parsedDoc() {
@@ -928,6 +930,23 @@ protected int estimatedSizeInBytes() {
             return (id().length() + type().length()) * 2 + source().length() + 12;
         }
 
+        /**
+         * Returns a positive timestamp if the ID of this document is auto-generated by elasticsearch.
+         * if this property is non-negative indexing code might optimize the addition of this document
+         * due to it's append only nature.
+         */
+        public long getAutoGeneratedIdTimestamp() {
+            return autoGeneratedIdTimestamp;
+        }
+
+        /**
+         * Returns <code>true</code> if this index requests has been retried on the coordinating node and can therefor be delivered
+         * multiple times. Note: this might also be set to true if an equivalent event occurred like the replay of the transaction log
+         */
+        public boolean isRetry() {
+            return isRetry;
+        }
+
     }
 
     public static class Delete extends Operation {

core/src/main/java/org/elasticsearch/index/engine/EngineConfig.java

@@ -89,7 +89,17 @@ public final class EngineConfig {
         }
     }, Property.IndexScope, Property.NodeScope);
 
-    private TranslogConfig translogConfig;
+    /**
+     * Configures an index to optimize documents with auto generated ids for append only. If this setting is updated from <code>false</code>
+     * to <code>true</code> might not take effect immediately. In other words, disabling the optimiation will be immediately applied while
+     * re-enabling it might not be applied until the engine is in a safe state to do so. Depending on the engine implementation a change to
+     * this setting won't be reflected re-enabled optimization until the engine is restarted or the index is closed and reopened.
+     * The default is <code>true</code>
+     */
+    public static final Setting<Boolean> INDEX_OPTIMIZE_AUTO_GENERATED_IDS = Setting.boolSetting("index.optimize_auto_generated_id", true,
+        Property.IndexScope, Property.Dynamic);
+
+    private final TranslogConfig translogConfig;
     private final OpenMode openMode;
 
     /**
@@ -311,4 +321,12 @@ public enum OpenMode {
     public RefreshListeners getRefreshListeners() {
         return refreshListeners;
     }
+
+    /**
+     * Returns <code>true</code> iff auto generated IDs should be optimized inside the engine for append only.
+     * The default is <code>true</code>.
+     */
+    public boolean getOptimizeAutoGeneratedIds() {
+        return indexSettings.getValue(INDEX_OPTIMIZE_AUTO_GENERATED_IDS);
+    }
 }