Fetch meta fields in FetchFieldsPhase using ValueFetcher

docs/reference/search/profile.asciidoc

@@ -197,6 +197,17 @@ The API returns the following result:
             "stored_fields": ["_id", "_routing", "_source"]
           },
           "children": [
+            {
+              "type" : "FetchFieldsPhase",
+              "description" : "",
+              "time_in_nanos" : 238762,
+              "breakdown" : {
+                "process_count" : 5,
+                "process" : 227914,
+                "next_reader" : 10848,
+                "next_reader_count" : 1
+              }
+            },
             {
               "type": "FetchSourcePhase",
               "description": "",
@@ -1043,6 +1054,17 @@ And here is the fetch profile:
             "stored_fields": ["_id", "_routing", "_source"]
           },
           "children": [
+            {
+              "type" : "FetchFieldsPhase",
+              "description" : "",
+              "time_in_nanos" : 238762,
+              "breakdown" : {
+                "process_count" : 5,
+                "process" : 227914,
+                "next_reader" : 10848,
+                "next_reader_count" : 1
+              }
+            },
             {
               "type": "FetchSourcePhase",
               "description": "",

rest-api-spec/src/yamlRestTest/resources/rest-api-spec/test/search/370_profile.yml

@@ -22,8 +22,8 @@ setup:
 ---
 fetch fields:
   - skip:
-      version: ' - 8.5.99'
-      reason:  stored fields phase added in 8.6
+      version: ' - 8.13.99'
+      reason: fetch fields and stored_fields using ValueFetcher
 
   - do:
       search:
@@ -44,17 +44,21 @@ fetch fields:
   - match: { profile.shards.0.fetch.debug.stored_fields: [_id, _routing, _source] }
   - length: { profile.shards.0.fetch.children: 2 }
   - match: { profile.shards.0.fetch.children.0.type: FetchFieldsPhase }
-  - match: { profile.shards.0.fetch.children.1.type: StoredFieldsPhase }
   - gt: { profile.shards.0.fetch.children.0.breakdown.next_reader_count: 0 }
   - gt: { profile.shards.0.fetch.children.0.breakdown.next_reader: 0 }
   - gt: { profile.shards.0.fetch.children.0.breakdown.next_reader_count: 0 }
   - gt: { profile.shards.0.fetch.children.0.breakdown.next_reader: 0 }
+  - match: { profile.shards.0.fetch.children.1.type: StoredFieldsPhase }
+  - gt: { profile.shards.0.fetch.children.1.breakdown.next_reader_count: 0 }
+  - gt: { profile.shards.0.fetch.children.1.breakdown.next_reader: 0 }
+  - gt: { profile.shards.0.fetch.children.1.breakdown.next_reader_count: 0 }
+  - gt: { profile.shards.0.fetch.children.1.breakdown.next_reader: 0 }
 
 ---
 fetch source:
   - skip:
-      version: ' - 8.5.99'
-      reason:  stored fields phase added in 8.6
+      version: ' - 8.13.99'
+      reason: fetch fields and stored_fields using ValueFetcher
 
   - do:
       search:
@@ -71,20 +75,21 @@ fetch source:
   - gt: { profile.shards.0.fetch.breakdown.load_stored_fields_count: 0 }
   - gt: { profile.shards.0.fetch.breakdown.load_stored_fields: 0 }
   - match: { profile.shards.0.fetch.debug.stored_fields: [_id, _routing, _source] }
-  - length: { profile.shards.0.fetch.children: 2 }
-  - match: { profile.shards.0.fetch.children.0.type: FetchSourcePhase }
-  - gt: { profile.shards.0.fetch.children.0.breakdown.next_reader_count: 0 }
-  - gt: { profile.shards.0.fetch.children.0.breakdown.next_reader: 0 }
-  - gt: { profile.shards.0.fetch.children.0.breakdown.next_reader_count: 0 }
-  - gt: { profile.shards.0.fetch.children.0.breakdown.next_reader: 0 }
-  - match: { profile.shards.0.fetch.children.0.debug.fast_path: 1 }
-  - match: { profile.shards.0.fetch.children.1.type: StoredFieldsPhase }
+  - length: { profile.shards.0.fetch.children: 3 }
+  - match: { profile.shards.0.fetch.children.0.type: FetchFieldsPhase }
+  - match: { profile.shards.0.fetch.children.1.type: FetchSourcePhase }
+  - gt: { profile.shards.0.fetch.children.1.breakdown.next_reader_count: 0 }
+  - gt: { profile.shards.0.fetch.children.1.breakdown.next_reader: 0 }
+  - gt: { profile.shards.0.fetch.children.1.breakdown.next_reader_count: 0 }
+  - gt: { profile.shards.0.fetch.children.1.breakdown.next_reader: 0 }
+  - match: { profile.shards.0.fetch.children.1.debug.fast_path: 1 }
+  - match: { profile.shards.0.fetch.children.2.type: StoredFieldsPhase }
 
 ---
 fetch nested source:
   - skip:
-      version: ' - 8.5.99'
-      reason:  stored fields phase added in 8.6
+      version: ' - 8.13.99'
+      reason: fetch fields and stored_fields using ValueFetcher
 
   - do:
       indices.create:
@@ -135,24 +140,25 @@ fetch nested source:
   - gt: { profile.shards.0.fetch.breakdown.load_stored_fields_count: 0 }
   - gt: { profile.shards.0.fetch.breakdown.load_stored_fields: 0 }
   - match: { profile.shards.0.fetch.debug.stored_fields: [_id, _routing, _source] }
-  - length: { profile.shards.0.fetch.children: 3 }
-  - match: { profile.shards.0.fetch.children.0.type: FetchSourcePhase }
-  - gt: { profile.shards.0.fetch.children.0.breakdown.next_reader_count: 0 }
-  - gt: { profile.shards.0.fetch.children.0.breakdown.next_reader: 0 }
-  - gt: { profile.shards.0.fetch.children.0.breakdown.next_reader_count: 0 }
-  - gt: { profile.shards.0.fetch.children.0.breakdown.next_reader: 0 }
-  - match: { profile.shards.0.fetch.children.1.type: InnerHitsPhase }
+  - length: { profile.shards.0.fetch.children: 4 }
+  - match: { profile.shards.0.fetch.children.0.type: FetchFieldsPhase }
+  - match: { profile.shards.0.fetch.children.1.type: FetchSourcePhase }
   - gt: { profile.shards.0.fetch.children.1.breakdown.next_reader_count: 0 }
   - gt: { profile.shards.0.fetch.children.1.breakdown.next_reader: 0 }
   - gt: { profile.shards.0.fetch.children.1.breakdown.next_reader_count: 0 }
   - gt: { profile.shards.0.fetch.children.1.breakdown.next_reader: 0 }
-  - match: { profile.shards.0.fetch.children.2.type: StoredFieldsPhase }
+  - match: { profile.shards.0.fetch.children.2.type: InnerHitsPhase }
+  - gt: { profile.shards.0.fetch.children.2.breakdown.next_reader_count: 0 }
+  - gt: { profile.shards.0.fetch.children.2.breakdown.next_reader: 0 }
+  - gt: { profile.shards.0.fetch.children.2.breakdown.next_reader_count: 0 }
+  - gt: { profile.shards.0.fetch.children.2.breakdown.next_reader: 0 }
+  - match: { profile.shards.0.fetch.children.3.type: StoredFieldsPhase }
 
 ---
 disabling stored fields removes fetch sub phases:
   - skip:
-      version: ' - 7.15.99'
-      reason:  fetch profiling implemented in 7.16.0
+      version: ' - 8.13.99'
+      reason:  fetch fields and stored_fields using ValueFetcher
 
   - do:
       search:
@@ -163,7 +169,8 @@ disabling stored fields removes fetch sub phases:
 
   - match: { hits.hits.0._index: test }
   - match: { profile.shards.0.fetch.debug.stored_fields: [] }
-  - is_false: profile.shards.0.fetch.children
+  - length: { profile.shards.0.fetch.children: 1 }
+  - match: { profile.shards.0.fetch.children.0.type: FetchFieldsPhase }
 
 ---
 dfs knn vector profiling:

server/src/main/java/org/elasticsearch/index/mapper/NestedValueFetcher.java

@@ -51,13 +51,13 @@ public List<Object> fetchValues(Source source, int doc, List<Object> includedVal
         for (Object entry : nestedValues) {
             // add this one entry only to the stub and use this as source lookup
             stub.put(nestedFieldName, entry);
-            Map<String, DocumentField> fetchResult = nestedFieldFetcher.fetch(
+            FieldFetcher.DocAndMetaFields fetchResult = nestedFieldFetcher.fetch(
                 Source.fromMap(filteredSource, source.sourceContentType()),
                 doc
             );
 
             Map<String, Object> nestedEntry = new HashMap<>();
-            for (DocumentField field : fetchResult.values()) {
+            for (DocumentField field : fetchResult.documentFields().values()) {
                 List<Object> fetchValues = field.getValues();
                 if (fetchValues.isEmpty() == false) {
                     String keyInNestedMap = field.getName().substring(nestedFieldPath.length() + 1);

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

@@ -10,29 +10,69 @@
 
 import org.apache.lucene.index.LeafReaderContext;
 import org.elasticsearch.common.document.DocumentField;
-import org.elasticsearch.search.SearchHit;
+import org.elasticsearch.index.mapper.IgnoredFieldMapper;
+import org.elasticsearch.index.mapper.LegacyTypeFieldMapper;
+import org.elasticsearch.index.mapper.RoutingFieldMapper;
 import org.elasticsearch.search.fetch.FetchContext;
 import org.elasticsearch.search.fetch.FetchSubPhase;
 import org.elasticsearch.search.fetch.FetchSubPhaseProcessor;
+import org.elasticsearch.search.fetch.StoredFieldsContext;
 import org.elasticsearch.search.fetch.StoredFieldsSpec;
 
 import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
 import java.util.Map;
+import java.util.stream.Stream;
 
 /**
  * A fetch sub-phase for high-level field retrieval. Given a list of fields, it
  * retrieves the field values through the relevant {@link org.elasticsearch.index.mapper.ValueFetcher}
  * and returns them as document fields.
  */
 public final class FetchFieldsPhase implements FetchSubPhase {
+
+    private static final List<String> ROOT_LEVEL_METADATA_FIELD_NAMES = List.of(
+        IgnoredFieldMapper.NAME,
+        RoutingFieldMapper.NAME,
+        LegacyTypeFieldMapper.NAME
+    );
+
+    private static final List<FieldAndFormat> ROOT_LEVEL_METADATA_FIELDS = ROOT_LEVEL_METADATA_FIELD_NAMES.stream()
+        .map(field -> new FieldAndFormat(field, null))
+        .toList();
+
+    public static boolean isMetadataField(final String field) {
+        return ROOT_LEVEL_METADATA_FIELD_NAMES.stream().anyMatch(f -> f.equals(field));
+    }
+
+    private static <T> List<T> emptyListIfNull(final List<T> theList) {
+        return theList == null ? Collections.emptyList() : theList;
+    }
+
     @Override
     public FetchSubPhaseProcessor getProcessor(FetchContext fetchContext) {
-        FetchFieldsContext fetchFieldsContext = fetchContext.fetchFieldsContext();
-        if (fetchFieldsContext == null) {
-            return null;
-        }
+        final FetchFieldsContext fetchFieldsContext = fetchContext.fetchFieldsContext();
+        final StoredFieldsContext storedFieldsContext = fetchContext.storedFieldsContext();
+
+        boolean fetchStoredFields = storedFieldsContext != null && storedFieldsContext.fetchFields();
+        final List<FieldAndFormat> storedFields = storedFieldsContext == null
+            ? Collections.emptyList()
+            : emptyListIfNull(storedFieldsContext.fieldNames()).stream()
+                .map(storedField -> new FieldAndFormat(storedField, null))
+                .distinct()
+                .toList();
+        final List<FieldAndFormat> storedFieldsIncludingDefaultMetadataFields = fetchStoredFields
+            ? Stream.concat(ROOT_LEVEL_METADATA_FIELDS.stream(), storedFields.stream()).distinct().toList()
+            : Collections.emptyList();
 
-        FieldFetcher fieldFetcher = FieldFetcher.create(fetchContext.getSearchExecutionContext(), fetchFieldsContext.fields());
+        final List<FieldAndFormat> fetchFields = fetchFieldsContext == null
+            ? Collections.emptyList()
+            : emptyListIfNull(fetchFieldsContext.fields());
+        final FieldFetcher fieldFetcher = FieldFetcher.create(
+            fetchContext.getSearchExecutionContext(),
+            Stream.concat(fetchFields.stream(), storedFieldsIncludingDefaultMetadataFields.stream()).toList()
+        );
 
         return new FetchSubPhaseProcessor() {
             @Override
@@ -47,11 +87,10 @@ public StoredFieldsSpec storedFieldsSpec() {
 
             @Override
             public void process(HitContext hitContext) throws IOException {
-                Map<String, DocumentField> documentFields = fieldFetcher.fetch(hitContext.source(), hitContext.docId());
-                SearchHit hit = hitContext.hit();
-                for (Map.Entry<String, DocumentField> entry : documentFields.entrySet()) {
-                    hit.setDocumentField(entry.getKey(), entry.getValue());
-                }
+                final FieldFetcher.DocAndMetaFields fields = fieldFetcher.fetch(hitContext.source(), hitContext.docId());
+                final Map<String, DocumentField> documentFields = fields.documentFields();
+                final Map<String, DocumentField> metadataFields = fields.metadataFields();
+                hitContext.hit().addDocumentFields(documentFields, metadataFields);
             }
         };
     }

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

@@ -35,7 +35,7 @@
  */
 public class FieldFetcher {
 
-    private record ResolvedField(String field, String matchingPattern, MappedFieldType ft, String format) {}
+    private record ResolvedField(String field, String matchingPattern, MappedFieldType ft, String format, boolean isMetadataField) {}
 
     /**
      * Build a FieldFetcher for a given search context and collection of fields and formats
@@ -58,7 +58,9 @@ public static FieldFetcher create(SearchExecutionContext context, Collection<Fie
                 if (context.isMetadataField(field) && matchingPattern != null) {
                     continue;
                 }
-                resolvedFields.add(new ResolvedField(field, matchingPattern, ft, fieldAndFormat.format));
+                resolvedFields.add(
+                    new ResolvedField(field, matchingPattern, ft, fieldAndFormat.format, FetchFieldsPhase.isMetadataField(field))
+                );
             }
         }
 
@@ -127,7 +129,7 @@ private static Map<String, FieldContext> buildFieldContexts(
             if (nestedScope.equals(scope)) {
                 // These are fields in the current scope, so add them directly to the output map
                 for (ResolvedField ff : fieldsByNestedMapper.get(nestedScope)) {
-                    output.put(ff.field, new FieldContext(ff.field, buildValueFetcher(context, ff)));
+                    output.put(ff.field, new FieldContext(ff.field, buildValueFetcher(context, ff), ff.isMetadataField));
                 }
             } else {
                 // don't create nested fetchers if no children have been requested as part of the fields
@@ -147,7 +149,7 @@ private static Map<String, FieldContext> buildFieldContexts(
                         unmappedFetchPatterns
                     );
                     NestedValueFetcher nvf = new NestedValueFetcher(scope, new FieldFetcher(scopedFields, unmappedFieldFetcher));
-                    output.put(scope, new FieldContext(scope, nvf));
+                    output.put(scope, new FieldContext(scope, nvf, FetchFieldsPhase.isMetadataField(scope)));
                 }
             }
         }
@@ -161,25 +163,39 @@ private static Map<String, FieldContext> buildFieldContexts(
     private FieldFetcher(Map<String, FieldContext> fieldContexts, UnmappedFieldFetcher unmappedFieldFetcher) {
         this.fieldContexts = fieldContexts;
         this.unmappedFieldFetcher = unmappedFieldFetcher;
-        this.storedFieldsSpec = StoredFieldsSpec.build(fieldContexts.values(), fc -> fc.valueFetcher.storedFieldsSpec());
+        this.storedFieldsSpec = StoredFieldsSpec.build(
+            fieldContexts.values().stream().filter(f -> f.isMetadataField == false).toList(),
+            fc -> fc.valueFetcher.storedFieldsSpec()
+        );
     }
 
     public StoredFieldsSpec storedFieldsSpec() {
         return storedFieldsSpec;
     }
 
-    public Map<String, DocumentField> fetch(Source source, int doc) throws IOException {
+    /**
+     * Collects document and metadata fields as two separate maps
+     * mapping the field name to the actual {@link DocumentField}.
+     */
+    public record DocAndMetaFields(Map<String, DocumentField> documentFields, Map<String, DocumentField> metadataFields) {}
+
+    public DocAndMetaFields fetch(Source source, int doc) throws IOException {
         Map<String, DocumentField> documentFields = new HashMap<>();
-        for (FieldContext context : fieldContexts.values()) {
-            String field = context.fieldName;
-            ValueFetcher valueFetcher = context.valueFetcher;
-            final DocumentField docField = valueFetcher.fetchDocumentField(field, source, doc);
-            if (docField != null) {
-                documentFields.put(field, docField);
+        Map<String, DocumentField> metadataFields = new HashMap<>();
+        for (FieldContext fieldContext : fieldContexts.values()) {
+            String field = fieldContext.fieldName;
+            ValueFetcher valueFetcher = fieldContext.valueFetcher;
+            final DocumentField value = valueFetcher.fetchDocumentField(field, source, doc);
+            if (value != null) {
+                if (fieldContext.isMetadataField) {
+                    metadataFields.put(field, value);
+                } else {
+                    documentFields.put(field, value);
+                }
             }
         }
         unmappedFieldFetcher.collectUnmapped(documentFields, source);
-        return documentFields;
+        return new DocAndMetaFields(documentFields, metadataFields);
     }
 
     public void setNextReader(LeafReaderContext readerContext) {
@@ -188,5 +204,5 @@ public void setNextReader(LeafReaderContext readerContext) {
         }
     }
 
-    private record FieldContext(String fieldName, ValueFetcher valueFetcher) {}
+    private record FieldContext(String fieldName, ValueFetcher valueFetcher, boolean isMetadataField) {}
 }