Fix duplicate strings in SearchHit serialization

The map key is always the field name. We exploited this fact in the get results but not in
search hits, leading to a lot of duplicate strings in many heap dumps.
We could do much better here since the names are generally coming out of a know limited set of names,
but as a first step lets at least align the get- and search-responses and non-trivial amount of bytes
in a number of use-cases. Plus, having a single string instance is faster on lookup etc. and saves on CPU
also.

Author: original-brownbear

modules/percolator/src/main/java/org/elasticsearch/percolator/PercolatorMatchedSlotSubFetchPhase.java

@@ -100,7 +100,7 @@ public void process(HitContext hitContext) throws IOException {
                     IntStream slots = convertTopDocsToSlots(topDocs, pc.rootDocsBySlot);
                     // _percolator_document_slot fields are document fields and should be under "fields" section in a hit
                     List<Object> docSlots = slots.boxed().collect(Collectors.toList());
-                    hitContext.hit().setDocumentField(fieldName, new DocumentField(fieldName, docSlots));
+                    hitContext.hit().setDocumentField(new DocumentField(fieldName, docSlots));
 
                     // Add info what sub-queries of percolator query matched this each percolated document
                     if (fetchContext.getSearchExecutionContext().hasNamedQueries()) {
@@ -120,7 +120,7 @@ public void process(HitContext hitContext) throws IOException {
                                 matchedQueries.add(match.getName());
                             }
                             String matchedFieldName = fieldName + "_" + docSlots.get(i) + "_matched_queries";
-                            hitContext.hit().setDocumentField(matchedFieldName, new DocumentField(matchedFieldName, matchedQueries));
+                            hitContext.hit().setDocumentField(new DocumentField(matchedFieldName, matchedQueries));
                         }
                     }
                 }

server/src/internalClusterTest/java/org/elasticsearch/search/aggregations/metrics/TopHitsIT.java

@@ -1327,8 +1327,7 @@ public void setNextReader(LeafReaderContext ctx) {
                         public void process(FetchSubPhase.HitContext hitContext) {
                             leafSearchLookup.setDocument(hitContext.docId());
                             FieldLookup fieldLookup = leafSearchLookup.fields().get("text");
-                            hitContext.hit()
-                                .setDocumentField("text_stored_lookup", new DocumentField("text_stored_lookup", fieldLookup.getValues()));
+                            hitContext.hit().setDocumentField(new DocumentField("text_stored_lookup", fieldLookup.getValues()));
                         }
 
                         @Override

server/src/internalClusterTest/java/org/elasticsearch/search/fetch/FetchSubPhasePluginIT.java

@@ -137,7 +137,7 @@ private void hitExecute(FetchContext context, HitContext hitContext) throws IOEx
             DocumentField hitField = hitContext.hit().getFields().get(NAME);
             if (hitField == null) {
                 hitField = new DocumentField(NAME, new ArrayList<>(1));
-                hitContext.hit().setDocumentField(NAME, hitField);
+                hitContext.hit().setDocumentField(hitField);
             }
             Terms terms = hitContext.reader().termVectors().get(hitContext.docId(), field);
             if (terms != null) {

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

@@ -233,6 +233,7 @@ static TransportVersion def(int id) {
     public static final TransportVersion SEARCH_INCREMENTAL_TOP_DOCS_NULL = def(9_058_0_00);
     public static final TransportVersion COMPRESS_DELAYABLE_WRITEABLE = def(9_059_0_00);
     public static final TransportVersion SYNONYMS_REFRESH_PARAM = def(9_060_0_00);
+    public static final TransportVersion DOC_FIELDS_AS_LIST = def(9_061_0_00);
 
     /*
      * STOP! READ THIS FIRST! No, really,

server/src/main/java/org/elasticsearch/common/document/DocumentField.java

@@ -24,6 +24,7 @@
 import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 
 import static org.elasticsearch.common.xcontent.XContentParserUtils.ensureExpectedToken;
@@ -70,6 +71,15 @@ public DocumentField(String name, List<Object> values, List<Object> ignoredValue
             : "DocumentField can't have both lookup fields and values";
     }
 
+    /**
+     * Read map of document fields written via {@link StreamOutput#writeMapValues(Map)}.
+     * @param in stream input
+     * @return map of {@link DocumentField} keyed by {@link DocumentField#getName()}
+     */
+    public static Map<String, DocumentField> readFieldsFromMapValues(StreamInput in) throws IOException {
+        return in.readMapValues(DocumentField::new, DocumentField::getName);
+    }
+
     /**
      * The name of the field.
      */

server/src/main/java/org/elasticsearch/index/get/GetResult.java

@@ -74,8 +74,8 @@ public GetResult(StreamInput in) throws IOException {
             if (source.length() == 0) {
                 source = null;
             }
-            documentFields = in.readMapValues(DocumentField::new, DocumentField::getName);
-            metaFields = in.readMapValues(DocumentField::new, DocumentField::getName);
+            documentFields = DocumentField.readFieldsFromMapValues(in);
+            metaFields = DocumentField.readFieldsFromMapValues(in);
         } else {
             metaFields = Collections.emptyMap();
             documentFields = Collections.emptyMap();

server/src/main/java/org/elasticsearch/search/SearchHit.java

@@ -216,8 +216,15 @@ public static SearchHit readFrom(StreamInput in, boolean pooled) throws IOExcept
         if (in.readBoolean()) {
             explanation = readExplanation(in);
         }
-        final Map<String, DocumentField> documentFields = in.readMap(DocumentField::new);
-        final Map<String, DocumentField> metaFields = in.readMap(DocumentField::new);
+        final Map<String, DocumentField> documentFields;
+        final Map<String, DocumentField> metaFields;
+        if (in.getTransportVersion().onOrAfter(TransportVersions.DOC_FIELDS_AS_LIST)) {
+            documentFields = DocumentField.readFieldsFromMapValues(in);
+            metaFields = DocumentField.readFieldsFromMapValues(in);
+        } else {
+            documentFields = in.readMap(DocumentField::new);
+            metaFields = in.readMap(DocumentField::new);
+        }
         Map<String, HighlightField> highlightFields = in.readMapValues(HighlightField::new, HighlightField::name);
         highlightFields = highlightFields.isEmpty() ? null : unmodifiableMap(highlightFields);
 
@@ -322,8 +329,13 @@ public void writeTo(StreamOutput out) throws IOException {
             out.writeBoolean(true);
             writeExplanation(out, explanation);
         }
-        out.writeMap(documentFields, StreamOutput::writeWriteable);
-        out.writeMap(metaFields, StreamOutput::writeWriteable);
+        if (out.getTransportVersion().onOrAfter(TransportVersions.DOC_FIELDS_AS_LIST)) {
+            out.writeMapValues(documentFields);
+            out.writeMapValues(metaFields);
+        } else {
+            out.writeMap(documentFields, StreamOutput::writeWriteable);
+            out.writeMap(metaFields, StreamOutput::writeWriteable);
+        }
         if (highlightFields == null) {
             out.writeVInt(0);
         } else {
@@ -502,9 +514,9 @@ public DocumentField field(String fieldName) {
     /*
     * Adds a new DocumentField to the map in case both parameters are not null.
     * */
-    public void setDocumentField(String fieldName, DocumentField field) {
-        if (fieldName == null || field == null) return;
-        this.documentFields.put(fieldName, field);
+    public void setDocumentField(DocumentField field) {
+        if (field == null) return;
+        this.documentFields.put(field.getName(), field);
     }
 
     public void addDocumentFields(Map<String, DocumentField> docFields, Map<String, DocumentField> metaFields) {

server/src/main/java/org/elasticsearch/search/fetch/subphase/FetchDocValuesPhase.java

@@ -76,7 +76,7 @@ public void process(HitContext hit) throws IOException {
                         hitField = new DocumentField(f.field, new ArrayList<>(2));
                         // even if we request a doc values of a meta-field (e.g. _routing),
                         // docValues fields will still be document fields, and put under "fields" section of a hit.
-                        hit.hit().setDocumentField(f.field, hitField);
+                        hit.hit().setDocumentField(hitField);
                     }
                     List<Object> ignoredValues = new ArrayList<>();
                     hitField.getValues().addAll(f.fetcher.fetchValues(hit.source(), hit.docId(), ignoredValues));

server/src/main/java/org/elasticsearch/search/fetch/subphase/ScriptFieldsPhase.java

@@ -75,7 +75,7 @@ public void process(HitContext hitContext) {
                         }
                         hitField = new DocumentField(scriptFieldName, values);
                         // script fields are never meta-fields
-                        hitContext.hit().setDocumentField(scriptFieldName, hitField);
+                        hitContext.hit().setDocumentField(hitField);
                     }
                 }
             }

server/src/main/java/org/elasticsearch/search/fetch/subphase/StoredFieldsPhase.java

@@ -95,8 +95,7 @@ public void process(HitContext hitContext) {
                 Map<String, List<Object>> loadedFields = hitContext.loadedFields();
                 for (StoredField storedField : storedFields) {
                     if (storedField.hasValue(loadedFields)) {
-                        hitContext.hit()
-                            .setDocumentField(storedField.name, new DocumentField(storedField.name, storedField.process(loadedFields)));
+                        hitContext.hit().setDocumentField(new DocumentField(storedField.name, storedField.process(loadedFields)));
                     }
                 }
             }