Allow _update and upsert to read from the transaction log

server/src/main/java/org/elasticsearch/action/explain/TransportExplainAction.java

@@ -112,13 +112,13 @@ protected ExplainResponse shardOperation(ExplainRequest request, ShardId shardId
             if (uidTerm == null) {
                 return new ExplainResponse(shardId.getIndexName(), request.type(), request.id(), false);
             }
-            result = context.indexShard().get(new Engine.Get(false, request.type(), request.id(), uidTerm));
+            result = context.indexShard().get(new Engine.Get(false, false, request.type(), request.id(), uidTerm));
             if (!result.exists()) {
                 return new ExplainResponse(shardId.getIndexName(), request.type(), request.id(), false);
             }
             context.parsedQuery(context.getQueryShardContext().toQuery(request.query()));
             context.preProcess(true);
-            int topLevelDocId = result.docIdAndVersion().docId + result.docIdAndVersion().context.docBase;
+            int topLevelDocId = result.docIdAndVersion().docId + result.docIdAndVersion().docBase;
             Explanation explanation = context.searcher().explain(context.query(), topLevelDocId);
             for (RescoreContext ctx : context.rescore()) {
                 Rescorer rescorer = ctx.rescorer();

server/src/main/java/org/elasticsearch/action/update/UpdateHelper.java

@@ -47,7 +47,6 @@
 import org.elasticsearch.script.ExecutableScript;
 import org.elasticsearch.script.Script;
 import org.elasticsearch.script.ScriptService;
-import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
 import org.elasticsearch.search.lookup.SourceLookup;
 
 import java.io.IOException;
@@ -71,9 +70,8 @@ public UpdateHelper(Settings settings, ScriptService scriptService) {
      * Prepares an update request by converting it into an index or delete request or an update response (no action).
      */
     public Result prepare(UpdateRequest request, IndexShard indexShard, LongSupplier nowInMillis) {
-        final GetResult getResult = indexShard.getService().get(request.type(), request.id(),
-                new String[]{RoutingFieldMapper.NAME, ParentFieldMapper.NAME},
-                true, request.version(), request.versionType(), FetchSourceContext.FETCH_SOURCE);
+        final GetResult getResult = indexShard.getService().getForUpdate(request.type(), request.id(), request.version(),
+            request.versionType());
         return prepare(indexShard.shardId(), request, getResult, nowInMillis);
     }
 

server/src/main/java/org/elasticsearch/common/lucene/uid/PerThreadIDVersionAndSeqNoLookup.java

@@ -100,7 +100,7 @@ public DocIdAndVersion lookupVersion(BytesRef id, LeafReaderContext context)
             if (versions.advanceExact(docID) == false) {
                 throw new IllegalArgumentException("Document [" + docID + "] misses the [" + VersionFieldMapper.NAME + "] field");
             }
-            return new DocIdAndVersion(docID, versions.longValue(), context);
+            return new DocIdAndVersion(docID, versions.longValue(), context.reader(), context.docBase);
         } else {
             return null;
         }

server/src/main/java/org/elasticsearch/common/lucene/uid/VersionsAndSeqNoResolver.java

@@ -20,6 +20,7 @@
 package org.elasticsearch.common.lucene.uid;
 
 import org.apache.lucene.index.IndexReader;
+import org.apache.lucene.index.LeafReader;
 import org.apache.lucene.index.LeafReaderContext;
 import org.apache.lucene.index.NumericDocValues;
 import org.apache.lucene.index.Term;
@@ -97,12 +98,14 @@ private VersionsAndSeqNoResolver() {
     public static class DocIdAndVersion {
         public final int docId;
         public final long version;
-        public final LeafReaderContext context;
+        public final LeafReader reader;
+        public final int docBase;
 
-        DocIdAndVersion(int docId, long version, LeafReaderContext context) {
+        public DocIdAndVersion(int docId, long version, LeafReader reader, int docBase) {
             this.docId = docId;
             this.version = version;
-            this.context = context;
+            this.reader = reader;
+            this.docBase = docBase;
         }
     }
 

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

@@ -1232,14 +1232,16 @@ public static class Get {
         private final boolean realtime;
         private final Term uid;
         private final String type, id;
+        private final boolean readFromTranslog;
         private long version = Versions.MATCH_ANY;
         private VersionType versionType = VersionType.INTERNAL;
 
-        public Get(boolean realtime, String type, String id, Term uid) {
+        public Get(boolean realtime, boolean readFromTranslog, String type, String id, Term uid) {
             this.realtime = realtime;
             this.type = type;
             this.id = id;
             this.uid = uid;
+            this.readFromTranslog = readFromTranslog;
         }
 
         public boolean realtime() {
@@ -1275,6 +1277,10 @@ public Get versionType(VersionType versionType) {
             this.versionType = versionType;
             return this;
         }
+
+        public boolean isReadFromTranslog() {
+            return readFromTranslog;
+        }
     }
 
     public static class GetResult implements Releasable {

server/src/main/java/org/elasticsearch/index/engine/InternalEngine.java

@@ -78,6 +78,7 @@
 import org.elasticsearch.threadpool.ThreadPool;
 
 import java.io.IOException;
+import java.io.UncheckedIOException;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -145,6 +146,7 @@ public class InternalEngine extends Engine {
      * being indexed/deleted.
      */
     private final AtomicLong writingBytes = new AtomicLong();
+    private final AtomicBoolean trackTranslogLocation = new AtomicBoolean(false);
 
     @Nullable
     private final String historyUUID;
@@ -558,6 +560,25 @@ public GetResult get(Get get, BiFunction<String, SearcherScope, Searcher> search
                         throw new VersionConflictEngineException(shardId, get.type(), get.id(),
                             get.versionType().explainConflictForReads(versionValue.version, get.version()));
                     }
+                    if (get.isReadFromTranslog()) {
+                        // this is only used for updates - API _GET calls will always read form a reader for consistency
+                        // the update call doesn't need the consistency since it's source only + _parent but parent can go away in 7.0
+                        if (versionValue.getLocation() != null) {
+                            try {
+                                Translog.Operation operation = translog.readOperation(versionValue.getLocation());
+                                if (operation != null) {
+                                    TranslogLeafReader reader = new TranslogLeafReader((Translog.Index) operation, engineConfig
+                                        .getIndexSettings().getIndexVersionCreated());
+                                    return new GetResult(new Searcher("realtime_get", new IndexSearcher(reader)),
+                                        new VersionsAndSeqNoResolver.DocIdAndVersion(0, ((Translog.Index) operation).version(), reader, 0));
+                                }
+                            } catch (IOException e) {
+                                throw new EngineException(shardId, "failed to read operation from translog", e);
+                            }
+                        } else {
+                            trackTranslogLocation.set(true);
+                        }
+                    }
                     refresh("realtime_get", SearcherScope.INTERNAL);
                 }
                 scope = SearcherScope.INTERNAL;
@@ -790,6 +811,10 @@ public IndexResult index(Index index) throws IOException {
                     }
                     indexResult.setTranslogLocation(location);
                 }
+                if (plan.indexIntoLucene && indexResult.hasFailure() == false) {
+                    versionMap.maybePutUnderLock(index.uid().bytes(),
+                        getVersionValue(plan.versionForIndexing, plan.seqNoForIndexing, index.primaryTerm(), indexResult.getTranslogLocation()));
+                }
                 if (indexResult.getSeqNo() != SequenceNumbers.UNASSIGNED_SEQ_NO) {
                     localCheckpointTracker.markSeqNoAsCompleted(indexResult.getSeqNo());
                 }
@@ -916,8 +941,6 @@ private IndexResult indexIntoLucene(Index index, IndexingStrategy plan)
                 assert assertDocDoesNotExist(index, canOptimizeAddDocument(index) == false);
                 index(index.docs(), indexWriter);
             }
-            versionMap.maybePutUnderLock(index.uid().bytes(),
-                new VersionValue(plan.versionForIndexing, plan.seqNoForIndexing, index.primaryTerm()));
             return new IndexResult(plan.versionForIndexing, plan.seqNoForIndexing, plan.currentNotFoundOrDeleted);
         } catch (Exception ex) {
             if (indexWriter.getTragicException() == null) {
@@ -941,6 +964,13 @@ private IndexResult indexIntoLucene(Index index, IndexingStrategy plan)
         }
     }
 
+    private VersionValue getVersionValue(long version, long seqNo, long term, Translog.Location location) {
+        if (location != null && trackTranslogLocation.get()) {
+            return new TranslogVersionValue(location, version, seqNo, term);
+        }
+        return new VersionValue(version, seqNo, term);
+    }
+
     /**
      * returns true if the indexing operation may have already be processed by this engine.
      * Note that it is OK to rarely return true even if this is not the case. However a `false`