Fetch meta fields in FetchFieldsPhase using ValueFetcher (#106325)

Here we extract the logic to populate metadata fields such as _ignored, _routing, _size and the deprecated _type into FetchFieldsPhase so that we can use the ValueFetcher interface to retrieve field values. This allows us to fetch values no matter if the Mapper uses stored or doc values.

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": "",

modules/parent-join/src/yamlRestTest/resources/rest-api-spec/test/30_inner_hits.yml

@@ -120,8 +120,8 @@ teardown:
 ---
 profile fetch:
   - 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:
@@ -141,15 +141,20 @@ profile fetch:
   - 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 }
+  - length: { profile.shards.0.fetch.children: 4 }
+  - match: { profile.shards.0.fetch.children.0.type: FetchFieldsPhase }
   - 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 }
+  - 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 }

rest-api-spec/src/yamlRestTest/resources/rest-api-spec/test/get/120_stored_fields_ignored.yml

@@ -0,0 +1,159 @@
+---
+"_ignored field through get api using stored_fields":
+  - do:
+      indices.create:
+        index: test
+        body:
+          mappings:
+            properties:
+              keyword:
+                type: keyword
+                ignore_above: 5
+              ip:
+                type: ip
+                ignore_malformed: true
+              value:
+                type: long
+                ignore_malformed: true
+
+  - do:
+      index:
+        index:   test
+        id:      1
+        refresh: true
+        body:
+          keyword: foo
+          ip: 192.168.0.1
+          value: 23
+  - do:
+      index:
+        index:   test
+        id:      2
+        refresh: true
+        body:
+          keyword: foobar
+          ip: garbage
+          value: missing
+  - do:
+      index:
+        index:   test
+        id:      3
+        refresh: true
+        body:
+          keyword:
+            - foo
+            - bar
+            - foobar
+          ip:
+            - 10.10.1.1
+            - 192.8.1.2
+            - 199.199.300.999
+          value:
+            - 1
+            - 2
+            - ops
+
+  - do:
+      get:
+        index: test
+        id:    1
+
+  - match: {_index: "test"}
+  - match: {_id: "1"}
+  - match: {_version: 1}
+  - match: {found: true}
+  - match:
+      _source:
+        keyword: foo
+        ip: 192.168.0.1
+        value: 23
+
+  - is_false: fields
+
+  - do:
+      get:
+        index: test
+        id: 2
+  - match: { _index: "test" }
+  - match: { _id: "2" }
+  - match: { _version: 1 }
+  - match: { found: true }
+  - match:
+      _source:
+        ip: garbage
+        keyword: foobar
+        value: missing
+
+  - is_false: fields
+
+  - do:
+      get:
+        index: test
+        id: 3
+  - match: { _index: "test" }
+  - match: { _id: "3" }
+  - match: { _version: 1 }
+  - match: { found: true }
+  - match:
+      _source:
+        ip:
+          - 10.10.1.1
+          - 192.8.1.2
+          - 199.199.300.999
+        keyword:
+          - foo
+          - bar
+          - foobar
+        value:
+          - 1
+          - 2
+          - ops
+
+  - is_false: fields
+
+  - do:
+      get:
+        index: test
+        id: 1
+        stored_fields:
+          - _ignored
+
+  - match: { _index: "test" }
+  - match: { _id: "1" }
+  - match: { _version: 1 }
+  - match: { found: true }
+  - match: { _ignored: null}
+
+  - do:
+      get:
+        index: test
+        id: 2
+        stored_fields:
+          - _ignored
+
+  - match: { _index: "test" }
+  - match: { _id: "2" }
+  - match: { _version: 1 }
+  - match: { found: true }
+  - match:
+      _ignored:
+        - ip
+        - keyword
+        - value
+
+  - do:
+      get:
+        index: test
+        id: 3
+        stored_fields:
+          - _ignored
+
+  - match: { _index: "test" }
+  - match: { _id: "3" }
+  - match: { _version: 1 }
+  - match: { found: true }
+  - match:
+      _ignored:
+        - ip
+        - keyword
+        - value

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 }
+  - 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.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 }
+  - 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
+      reason: fetch profiling implemented in 7.16.0
 
   - do:
       search:

rest-api-spec/src/yamlRestTest/resources/rest-api-spec/test/search/520_fetch_fields.yml

@@ -0,0 +1,159 @@
+---
+setup:
+  - do:
+      indices.create:
+        index:  test
+        body:
+          settings:
+            index.number_of_shards: 1
+          mappings:
+            properties:
+              stored_keyword:
+                type: keyword
+                store: true
+              keyword:
+                type: keyword
+              stored_value:
+                type: integer
+                store: true
+              value:
+                type: integer
+              ignored_keyword:
+                type: keyword
+                ignore_above: 3
+              ignored_value:
+                type: integer
+                ignore_malformed: true
+
+  - do:
+      index:
+        index:   test
+        id:      "1"
+        refresh: true
+        body:
+          stored_keyword: "stored_keyword_value"
+          keyword: "keyword_value"
+          stored_value: 10
+          value: 100
+          ignored_keyword: "foobar"
+          ignored_value: foobar
+
+---
+fetch stored fields:
+
+  - do:
+      search:
+        index: test
+        body:
+          stored_fields: [ stored_keyword, stored_value, keyword, value ]
+
+  - match: { hits.total.value: 1 }
+  - match: { hits.hits.0.fields.stored_keyword.0: "stored_keyword_value" }
+  - match: { hits.hits.0.fields.stored_value.0: 10 }
+  - match: { hits.hits.0.fields.keyword: null }
+  - match: { hits.hits.0.fields.value: null }
+
+---
+fetch fields:
+
+  - do:
+      search:
+        index: test
+        body:
+          fields: [ stored_keyword, stored_value, keyword, value ]
+
+  - match: { hits.total.value: 1 }
+  - match: { hits.hits.0.fields.stored_keyword.0: "stored_keyword_value" }
+  - match: { hits.hits.0.fields.stored_value.0: 10 }
+  - match: { hits.hits.0.fields.keyword.0: "keyword_value" }
+  - match: { hits.hits.0.fields.value.0: 100 }
+
+---
+fetch fields and stored fields:
+
+  - do:
+      search:
+        index: test
+        body:
+          fields: [ keyword, stored_value ]
+          stored_fields: [ stored_keyword, value ]
+
+  - match: { hits.total.value: 1 }
+  - match: { hits.hits.0.fields.stored_keyword.0: "stored_keyword_value" }
+  - match: { hits.hits.0.fields.stored_value.0: 10 }
+  - match: { hits.hits.0.fields.keyword.0: "keyword_value" }
+  - match: { hits.hits.0.fields.value: null }
+
+---
+fetch _ignored via stored_fields:
+
+  - do:
+      search:
+        index: test
+        body:
+          stored_fields: [ _ignored ]
+
+  - match: { hits.total.value: 1 }
+  - match: { hits.hits.0.fields._ignored: null }
+  - match: { hits.hits.0._ignored.0: "ignored_keyword" }
+  - match: { hits.hits.0._ignored.1: "ignored_value" }
+
+---
+fetch _ignored via fields:
+
+  - do:
+      search:
+        index: test
+        body:
+          fields: [ _ignored ]
+
+  - match: { hits.total.value: 1 }
+  - match: { hits.hits.0.fields._ignored.0: "ignored_keyword" }
+  - match: { hits.hits.0.fields._ignored.1: "ignored_value" }
+  - match: { hits.hits.0._ignored.0: "ignored_keyword" }
+  - match: { hits.hits.0._ignored.1: "ignored_value" }
+
+---
+fetch _seq_no via stored_fields:
+
+  - do:
+      search:
+        index: test
+        body:
+          stored_fields: [ _seq_no ]
+
+  - match: { hits.total.value: 1 }
+  - match: { hits.hits.0.fields._seq_no: null }
+  - match: { hits.hits.0._seq_no: null }
+
+---
+fetch _seq_no via fields:
+
+  - do:
+      catch: "request"
+      search:
+        index: test
+        body:
+          fields: [ _seq_no ]
+
+  # This should be `unauthorized` (401) or `forbidden` (403) or at least `bad request` (400)
+  # while instead it is mapped to an `internal_server_error (500)`
+  - match: { status: 500 }
+  - match: { error.root_cause.0.type: unsupported_operation_exception }
+
+---
+fetch fields with none stored_fields:
+  - skip:
+      version: " - 7.99.99"
+      reason: "from illegal_argument_exception to action_request_validation_exception"
+
+  - do:
+      catch: "bad_request"
+      search:
+        index: test
+        body:
+          stored_fields: _none_
+          fields: [stored_keyword, keyword, stored_value, value, ignored_keyword, ignored_value, _ignored]
+
+  - match: { status: 400 }
+  - match: { error.root_cause.0.type: action_request_validation_exception }

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

@@ -10,14 +10,25 @@ package org.elasticsearch.search.fetch.subphase;
 
 import org.apache.lucene.index.LeafReaderContext;
 import org.elasticsearch.common.document.DocumentField;
-import org.elasticsearch.search.SearchHit;
+import org.elasticsearch.index.mapper.IdFieldMapper;
+import org.elasticsearch.index.mapper.IgnoredFieldMapper;
+import org.elasticsearch.index.mapper.LegacyTypeFieldMapper;
+import org.elasticsearch.index.mapper.MappedFieldType;
+import org.elasticsearch.index.mapper.RoutingFieldMapper;
+import org.elasticsearch.index.mapper.SourceFieldMapper;
+import org.elasticsearch.index.query.SearchExecutionContext;
 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.HashSet;
+import java.util.List;
 import java.util.Map;
+import java.util.Set;
 
 /**
  * A fetch sub-phase for high-level field retrieval. Given a list of fields, it
@@ -25,33 +36,77 @@ import java.util.Map;
  * and returns them as document fields.
  */
 public final class FetchFieldsPhase implements FetchSubPhase {
+
+    private static final List<FieldAndFormat> DEFAULT_METADATA_FIELDS = List.of(
+        new FieldAndFormat(IgnoredFieldMapper.NAME, null),
+        new FieldAndFormat(RoutingFieldMapper.NAME, null),
+        new FieldAndFormat(LegacyTypeFieldMapper.NAME, null)
+    );
+
     @Override
     public FetchSubPhaseProcessor getProcessor(FetchContext fetchContext) {
-        FetchFieldsContext fetchFieldsContext = fetchContext.fetchFieldsContext();
-        if (fetchFieldsContext == null) {
+        final FetchFieldsContext fetchFieldsContext = fetchContext.fetchFieldsContext();
+        final StoredFieldsContext storedFieldsContext = fetchContext.storedFieldsContext();
+
+        boolean fetchStoredFields = storedFieldsContext != null && storedFieldsContext.fetchFields();
+        if (fetchFieldsContext == null && fetchStoredFields == false) {
             return null;
         }
 
-        FieldFetcher fieldFetcher = FieldFetcher.create(fetchContext.getSearchExecutionContext(), fetchFieldsContext.fields());
+        final SearchExecutionContext searchExecutionContext = fetchContext.getSearchExecutionContext();
+        final FieldFetcher fieldFetcher = fetchFieldsContext == null ? null
+            : fetchFieldsContext.fields() == null ? null
+            : fetchFieldsContext.fields().isEmpty() ? null
+            : FieldFetcher.create(searchExecutionContext, fetchFieldsContext.fields());
 
+        final FieldFetcher metadataFieldFetcher;
+        if (storedFieldsContext != null
+            && storedFieldsContext.fieldNames() != null
+            && storedFieldsContext.fieldNames().isEmpty() == false) {
+            final Set<FieldAndFormat> metadataFields = new HashSet<>(DEFAULT_METADATA_FIELDS);
+            for (final String storedField : storedFieldsContext.fieldNames()) {
+                final Set<String> matchingFieldNames = searchExecutionContext.getMatchingFieldNames(storedField);
+                for (final String matchingFieldName : matchingFieldNames) {
+                    if (SourceFieldMapper.NAME.equals(matchingFieldName) || IdFieldMapper.NAME.equals(matchingFieldName)) {
+                        continue;
+                    }
+                    final MappedFieldType fieldType = searchExecutionContext.getFieldType(matchingFieldName);
+                    // NOTE: checking if the field is stored is required for backward compatibility reasons and to make
+                    // sure we also handle here stored fields requested via `stored_fields`, which was previously a
+                    // responsibility of StoredFieldsPhase.
+                    if (searchExecutionContext.isMetadataField(matchingFieldName) && fieldType.isStored()) {
+                        metadataFields.add(new FieldAndFormat(matchingFieldName, null));
+                    }
+                }
+            }
+            metadataFieldFetcher = FieldFetcher.create(searchExecutionContext, metadataFields);
+        } else {
+            metadataFieldFetcher = FieldFetcher.create(searchExecutionContext, DEFAULT_METADATA_FIELDS);
+        }
         return new FetchSubPhaseProcessor() {
             @Override
             public void setNextReader(LeafReaderContext readerContext) {
-                fieldFetcher.setNextReader(readerContext);
+                if (fieldFetcher != null) {
+                    fieldFetcher.setNextReader(readerContext);
+                }
+                metadataFieldFetcher.setNextReader(readerContext);
             }
 
             @Override
             public StoredFieldsSpec storedFieldsSpec() {
-                return fieldFetcher.storedFieldsSpec();
+                if (fieldFetcher != null) {
+                    return fieldFetcher.storedFieldsSpec();
+                }
+                return StoredFieldsSpec.NO_REQUIREMENTS;
             }
 
             @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 Map<String, DocumentField> fields = fieldFetcher != null
+                    ? fieldFetcher.fetch(hitContext.source(), hitContext.docId())
+                    : Collections.emptyMap();
+                final Map<String, DocumentField> metadataFields = metadataFieldFetcher.fetch(hitContext.source(), hitContext.docId());
+                hitContext.hit().addDocumentFields(fields, metadataFields);
             }
         };
     }

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

@@ -11,10 +11,7 @@ package org.elasticsearch.search.fetch.subphase;
 import org.apache.lucene.index.LeafReaderContext;
 import org.elasticsearch.common.document.DocumentField;
 import org.elasticsearch.index.mapper.IdFieldMapper;
-import org.elasticsearch.index.mapper.IgnoredFieldMapper;
-import org.elasticsearch.index.mapper.LegacyTypeFieldMapper;
 import org.elasticsearch.index.mapper.MappedFieldType;
-import org.elasticsearch.index.mapper.RoutingFieldMapper;
 import org.elasticsearch.index.mapper.SourceFieldMapper;
 import org.elasticsearch.index.query.SearchExecutionContext;
 import org.elasticsearch.search.fetch.FetchContext;
@@ -25,7 +22,6 @@ import org.elasticsearch.search.fetch.StoredFieldsSpec;
 
 import java.util.ArrayList;
 import java.util.Collection;
-import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
@@ -37,7 +33,7 @@ import java.util.Set;
 public class StoredFieldsPhase implements FetchSubPhase {
 
     /** Associates a field name with a mapped field type and whether or not it is a metadata field */
-    private record StoredField(String name, MappedFieldType ft, boolean isMetadataField) {
+    private record StoredField(String name, MappedFieldType ft) {
 
         /** Processes a set of stored fields using field type information */
         List<Object> process(Map<String, List<Object>> loadedFields) {
@@ -54,13 +50,6 @@ public class StoredFieldsPhase implements FetchSubPhase {
 
     }
 
-    private static final List<StoredField> METADATA_FIELDS = List.of(
-        new StoredField("_routing", RoutingFieldMapper.FIELD_TYPE, true),
-        new StoredField("_ignored", IgnoredFieldMapper.FIELD_TYPE, true),
-        // pre-6.0 indexes can return a _type field, this will be valueless in modern indexes and ignored
-        new StoredField("_type", LegacyTypeFieldMapper.FIELD_TYPE, true)
-    );
-
     @Override
     public FetchSubPhaseProcessor getProcessor(FetchContext fetchContext) {
         StoredFieldsContext storedFieldsContext = fetchContext.storedFieldsContext();
@@ -69,7 +58,7 @@ public class StoredFieldsPhase implements FetchSubPhase {
         }
 
         // build the StoredFieldsSpec and a list of StoredField records to process
-        List<StoredField> storedFields = new ArrayList<>(METADATA_FIELDS);
+        List<StoredField> storedFields = new ArrayList<>();
         Set<String> fieldsToLoad = new HashSet<>();
         if (storedFieldsContext.fieldNames() != null) {
             SearchExecutionContext sec = fetchContext.getSearchExecutionContext();
@@ -82,10 +71,10 @@ public class StoredFieldsPhase implements FetchSubPhase {
                         continue;
                     }
                     MappedFieldType ft = sec.getFieldType(fieldName);
-                    if (ft.isStored() == false) {
+                    if (ft.isStored() == false || sec.isMetadataField(fieldName)) {
                         continue;
                     }
-                    storedFields.add(new StoredField(fieldName, ft, sec.isMetadataField(ft.name())));
+                    storedFields.add(new StoredField(fieldName, ft));
                     fieldsToLoad.add(ft.name());
                 }
             }
@@ -101,19 +90,12 @@ public class StoredFieldsPhase implements FetchSubPhase {
             @Override
             public void process(HitContext hitContext) {
                 Map<String, List<Object>> loadedFields = hitContext.loadedFields();
-                Map<String, DocumentField> docFields = new HashMap<>();
-                Map<String, DocumentField> metaFields = new HashMap<>();
                 for (StoredField storedField : storedFields) {
                     if (storedField.hasValue(loadedFields)) {
-                        DocumentField df = new DocumentField(storedField.name, storedField.process(loadedFields));
-                        if (storedField.isMetadataField) {
-                            metaFields.put(storedField.name, df);
-                        } else {
-                            docFields.put(storedField.name, df);
-                        }
+                        hitContext.hit()
+                            .setDocumentField(storedField.name, new DocumentField(storedField.name, storedField.process(loadedFields)));
                     }
                 }
-                hitContext.hit().addDocumentFields(docFields, metaFields);
             }
 
             @Override