Allow realtime get to read from translog (#48843)

The realtime GET API currently has erratic performance in case where a document is accessed
that has just been indexed but not refreshed yet, as the implementation will currently force an
internal refresh in that case. Refreshing can be an expensive operation, and also will block the
thread that executes the GET operation, blocking other GETs to be processed. In case of
frequent access of recently indexed documents, this can lead to a refresh storm and terrible
GET performance.

While older versions of Elasticsearch (2.x and older) did not trigger refreshes and instead opted
to read from the translog in case of realtime GET API or update API, this was removed in 5.0
(#20102) to avoid inconsistencies between values that were returned from the translog and
those returned by the index. This was partially reverted in 6.3 (#29264) to allow _update and
upsert to read from the translog again as it was easier to guarantee consistency for these, and
also brought back more predictable performance characteristics of this API. Calls to the realtime
GET API, however, would still always do a refresh if necessary to return consistent results. This
means that users that were calling realtime GET APIs to coordinate updates on client side
(realtime GET + CAS for conditional index of updated doc) would still see very erratic
performance.

This PR (together with #48707) resolves the inconsistencies between reading from translog and
index. In particular it fixes the inconsistencies that happen when requesting stored fields, which
were not available when reading from translog. In case where stored fields are requested, this
PR will reparse the _source from the translog and derive the stored fields to be returned. With
this, it changes the realtime GET API to allow reading from the translog again, avoid refresh
storms and blocking the GET threadpool, and provide overall much better and predictable
performance for this API.

docs/reference/docs/get.asciidoc

@@ -35,11 +35,11 @@ that it exists.
 ===== Realtime
 
 By default, the get API is realtime, and is not affected by the refresh
-rate of the index (when data will become visible for search). If a document
-has been updated but is not yet refreshed, the get API will issue a refresh
-call in-place to make the document visible. This will also make other documents
-changed since the last refresh visible. In order to disable realtime GET,
-one can set the `realtime` parameter to `false`.
+rate of the index (when data will become visible for search). In case where
+stored fields are requested (see `stored_fields` parameter) and the document
+has been updated but is not yet refreshed, the get API will have to parse
+and analyze the source to extract the stored fields. In order to disable
+realtime GET, the `realtime` parameter can be set to `false`.
 
 [float]
 [[get-source-filtering]]

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

@@ -35,6 +35,7 @@
 import org.apache.lucene.index.Terms;
 import org.apache.lucene.util.Bits;
 import org.apache.lucene.util.BytesRef;
+import org.elasticsearch.common.util.set.Sets;
 import org.elasticsearch.index.mapper.IdFieldMapper;
 import org.elasticsearch.index.mapper.RoutingFieldMapper;
 import org.elasticsearch.index.mapper.SourceFieldMapper;
@@ -44,11 +45,12 @@
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 import java.util.Collections;
+import java.util.Set;
 
 /**
  * Internal class that mocks a single doc read from the transaction log as a leaf reader.
  */
-final class TranslogLeafReader extends LeafReader {
+public final class TranslogLeafReader extends LeafReader {
 
     private final Translog.Index operation;
     private static final FieldInfo FAKE_SOURCE_FIELD
@@ -60,6 +62,7 @@ final class TranslogLeafReader extends LeafReader {
     private static final FieldInfo FAKE_ID_FIELD
         = new FieldInfo(IdFieldMapper.NAME, 3, false, false, false, IndexOptions.NONE, DocValuesType.NONE, -1, Collections.emptyMap(),
         0, 0, 0, false);
+    public static Set<String> ALL_FIELD_NAMES = Sets.newHashSet(FAKE_SOURCE_FIELD.name, FAKE_ROUTING_FIELD.name, FAKE_ID_FIELD.name);
 
     TranslogLeafReader(Translog.Index operation) {
         this.operation = operation;
@@ -161,7 +164,7 @@ public void document(int docID, StoredFieldVisitor visitor) throws IOException {
             BytesRef bytesRef = Uid.encodeId(operation.id());
             final byte[] id = new byte[bytesRef.length];
             System.arraycopy(bytesRef.bytes, bytesRef.offset, id, 0, bytesRef.length);
-            visitor.stringField(FAKE_ID_FIELD, id);
+            visitor.binaryField(FAKE_ID_FIELD, id);
         }
     }
 

server/src/main/java/org/elasticsearch/index/fieldvisitor/CustomFieldsVisitor.java

@@ -20,7 +20,6 @@
 
 import org.apache.lucene.index.FieldInfo;
 
-import java.io.IOException;
 import java.util.Set;
 
 /**
@@ -39,7 +38,7 @@ public CustomFieldsVisitor(Set<String> fields, boolean loadSource) {
     }
 
     @Override
-    public Status needsField(FieldInfo fieldInfo) throws IOException {
+    public Status needsField(FieldInfo fieldInfo) {
         if (super.needsField(fieldInfo) == Status.YES) {
             return Status.YES;
         }

server/src/main/java/org/elasticsearch/index/fieldvisitor/FieldsVisitor.java

@@ -32,7 +32,6 @@
 import org.elasticsearch.index.mapper.SourceFieldMapper;
 import org.elasticsearch.index.mapper.Uid;
 
-import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.HashMap;
@@ -72,7 +71,7 @@ public FieldsVisitor(boolean loadSource, String sourceFieldName) {
     }
 
     @Override
-    public Status needsField(FieldInfo fieldInfo) throws IOException {
+    public Status needsField(FieldInfo fieldInfo) {
         if (requiredFields.remove(fieldInfo.name)) {
             return Status.YES;
         }
@@ -108,42 +107,54 @@ public void postProcess(MapperService mapperService) {
     }
 
     @Override
-    public void binaryField(FieldInfo fieldInfo, byte[] value) throws IOException {
+    public void binaryField(FieldInfo fieldInfo, byte[] value) {
+        binaryField(fieldInfo, new BytesRef(value));
+    }
+
+    public void binaryField(FieldInfo fieldInfo, BytesRef value) {
         if (sourceFieldName.equals(fieldInfo.name)) {
             source = new BytesArray(value);
         } else if (IdFieldMapper.NAME.equals(fieldInfo.name)) {
-            id = Uid.decodeId(value);
+            id = Uid.decodeId(value.bytes, value.offset, value.length);
         } else {
-            addValue(fieldInfo.name, new BytesRef(value));
+            addValue(fieldInfo.name, value);
         }
     }
 
     @Override
-    public void stringField(FieldInfo fieldInfo, byte[] bytes) throws IOException {
+    public void stringField(FieldInfo fieldInfo, byte[] bytes) {
+        assert IdFieldMapper.NAME.equals(fieldInfo.name) == false : "_id field must go through binaryField";
+        assert sourceFieldName.equals(fieldInfo.name) == false : "source field must go through binaryField";
         final String value = new String(bytes, StandardCharsets.UTF_8);
         addValue(fieldInfo.name, value);
     }
 
     @Override
-    public void intField(FieldInfo fieldInfo, int value) throws IOException {
+    public void intField(FieldInfo fieldInfo, int value) {
         addValue(fieldInfo.name, value);
     }
 
     @Override
-    public void longField(FieldInfo fieldInfo, long value) throws IOException {
+    public void longField(FieldInfo fieldInfo, long value) {
         addValue(fieldInfo.name, value);
     }
 
     @Override
-    public void floatField(FieldInfo fieldInfo, float value) throws IOException {
+    public void floatField(FieldInfo fieldInfo, float value) {
         addValue(fieldInfo.name, value);
     }
 
     @Override
-    public void doubleField(FieldInfo fieldInfo, double value) throws IOException {
+    public void doubleField(FieldInfo fieldInfo, double value) {
         addValue(fieldInfo.name, value);
     }
 
+    public void objectField(FieldInfo fieldInfo, Object object) {
+        assert IdFieldMapper.NAME.equals(fieldInfo.name) == false : "_id field must go through binaryField";
+        assert sourceFieldName.equals(fieldInfo.name) == false : "source field must go through binaryField";
+        addValue(fieldInfo.name, object);
+    }
+
     public BytesReference source() {
         return source;
     }