Support unmapped fields in search 'fields' option (#65386)

Currently, the 'fields' option only supports fetching mapped fields. Since
'fields' is meant to be the central place to retrieve document content, it
should allow for loading unmapped values. This change adds implementation and
tests for this feature.

Closes #63690

docs/reference/search/search-your-data/retrieve-selected-fields.asciidoc

@@ -167,7 +167,92 @@ no dedicated array type, and any field could contain multiple values. The
 a specific order. See the mapping documentation on <<array, arrays>> for more
 background.
 
+[discrete]
+[[retrieve-unmapped-fields]]
+==== Retrieving unmapped fields
+
+By default, the `fields` parameter returns only values of mapped fields. However,
+Elasticsearch allows storing fields in `_source` that are unmapped, for example by
+setting <<dynamic-field-mapping,Dynamic field mapping>> to `false` or by using an
+object field with `enabled: false`, thereby disabling parsing and indexing of its content.
+
+Fields in such an object can be retrieved from `_source` using the `include_unmapped` option
+in the `fields` section:
+
+[source,console]
+----
+PUT my-index-000001
+{
+  "mappings": {
+    "enabled": false <1>
+  }
+}
+
+PUT my-index-000001/_doc/1?refresh=true
+{
+  "user_id": "kimchy",
+  "session_data": {
+     "object": {
+       "some_field": "some_value"
+     }
+   }
+}
+
+POST my-index-000001/_search
+{
+  "fields": [
+    "user_id",
+    {
+      "field": "session_data.object.*",
+      "include_unmapped" : true <2>
+    }
+  ],
+  "_source": false
+}
+----
+
+<1> Disable all mappings.
+<2> Include unmapped fields matching this field pattern.
 
+The response will contain fields results under the  `session_data.object.*` path even if the
+fields are unmapped, but will not contain `user_id` since it is unmapped but the `include_unmapped`
+flag hasn't been set to `true` for that field pattern.
+
+[source,console-result]
+----
+{
+  "took" : 2,
+  "timed_out" : false,
+  "_shards" : {
+    "total" : 1,
+    "successful" : 1,
+    "skipped" : 0,
+    "failed" : 0
+  },
+  "hits" : {
+    "total" : {
+      "value" : 1,
+      "relation" : "eq"
+    },
+    "max_score" : 1.0,
+    "hits" : [
+      {
+        "_index" : "my-index-000001",
+        "_id" : "1",
+        "_score" : 1.0,
+        "fields" : {
+          "session_data.object.some_field": [
+            "some_value"
+          ]
+        }
+      }
+    ]
+  }
+}
+----
+// TESTRESPONSE[s/"took" : 2/"took": $body.took/]
+// TESTRESPONSE[s/"max_score" : 1.0/"max_score" : $body.hits.max_score/]
+// TESTRESPONSE[s/"_score" : 1.0/"_score" : $body.hits.hits.0._score/]
 
 [discrete]
 [[docvalue-fields]]

rest-api-spec/src/main/resources/rest-api-spec/test/search/330_fetch_fields.yml

@@ -295,3 +295,110 @@ setup:
   - is_true: hits.hits.0._id
   - match: { hits.hits.0.fields.count: [2] }
   - is_false: hits.hits.0.fields.count_without_dv
+---
+Test unmapped field:
+  -  skip:
+        version: ' - 7.99.99'
+        reason: support isn't yet backported
+  -  do:
+        indices.create:
+           index: test
+           body:
+              mappings:
+                 dynamic: false
+                 properties:
+                    f1:
+                       type: keyword
+                    f2:
+                       type: object
+                       enabled: false
+                    f3:
+                       type: object
+  -  do:
+        index:
+           index: test
+           id: 1
+           refresh: true
+           body:
+              f1: some text
+              f2:
+                 a: foo
+                 b: bar
+              f3:
+                 c: baz
+              f4: some other text
+  -  do:
+        search:
+           index: test
+           body:
+              fields:
+              - f1
+              - { "field" : "f4", "include_unmapped" : true }
+  -  match:
+        hits.hits.0.fields.f1: 
+        - some text
+  -  match:
+        hits.hits.0.fields.f4:
+        - some other text
+  -  do:
+        search:
+           index: test
+           body:
+              fields:
+              - { "field" : "f*", "include_unmapped" : true }
+  -  match:
+        hits.hits.0.fields.f1: 
+        - some text
+  -  match:
+        hits.hits.0.fields.f2\.a:
+        - foo
+  -  match:
+        hits.hits.0.fields.f2\.b:
+        - bar
+  -  match:
+        hits.hits.0.fields.f3\.c:
+        - baz
+  -  match:
+        hits.hits.0.fields.f4:
+        - some other text
+---
+Test unmapped fields inside disabled objects:
+  -  skip:
+        version: ' - 7.99.99'
+        reason: support isn't yet backported
+  -  do:
+        indices.create:
+           index: test
+           body:
+              mappings:
+                 properties:
+                    f1:
+                       type: object
+                       enabled: false
+  -  do:
+        index:
+           index: test
+           id: 1
+           refresh: true
+           body:
+              f1: 
+                 - some text
+                 - a: b
+                 -
+                   - 1
+                   - 2
+                   - 3
+  -  do:
+        search:
+           index: test
+           body:
+              fields: [ { "field" : "*", "include_unmapped" : true } ]
+  -  match:
+        hits.hits.0.fields.f1:
+        - 1
+        - 2
+        - 3
+        - some text
+  -  match:
+        hits.hits.0.fields.f1\.a:
+        - b

server/src/main/java/org/elasticsearch/action/search/ExpandSearchPhase.java

@@ -138,7 +138,7 @@ final class ExpandSearchPhase extends SearchPhase {
             }
         }
         if (options.getFetchFields() != null) {
-            options.getFetchFields().forEach(ff -> groupSource.fetchField(ff.field, ff.format));
+            options.getFetchFields().forEach(ff -> groupSource.fetchField(ff));
         }
         if (options.getDocValueFields() != null) {
             options.getDocValueFields().forEach(ff -> groupSource.docValueField(ff.field, ff.format));

server/src/main/java/org/elasticsearch/action/search/SearchRequestBuilder.java

@@ -32,6 +32,7 @@ import org.elasticsearch.search.aggregations.PipelineAggregationBuilder;
 import org.elasticsearch.search.builder.PointInTimeBuilder;
 import org.elasticsearch.search.builder.SearchSourceBuilder;
 import org.elasticsearch.search.collapse.CollapseBuilder;
+import org.elasticsearch.search.fetch.subphase.FieldAndFormat;
 import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder;
 import org.elasticsearch.search.rescore.RescorerBuilder;
 import org.elasticsearch.search.slice.SliceBuilder;
@@ -310,18 +311,18 @@ public class SearchRequestBuilder extends ActionRequestBuilder<SearchRequest, Se
      * @param name The field to load
      */
     public SearchRequestBuilder addFetchField(String name) {
-        sourceBuilder().fetchField(name, null);
+        sourceBuilder().fetchField(new FieldAndFormat(name, null, null));
         return this;
     }
 
     /**
      * Adds a field to load and return. The field must be present in the document _source.
      *
-     * @param name The field to load
-     * @param format an optional format string used when formatting values, for example a date format.
+     * @param fetchField a {@link FieldAndFormat} specifying the field pattern, optional format (for example a date format) and
+     * whether this field pattern sould also include unmapped fields
      */
-    public SearchRequestBuilder addFetchField(String name, String format) {
-        sourceBuilder().fetchField(name, format);
+    public SearchRequestBuilder addFetchField(FieldAndFormat fetchField) {
+        sourceBuilder().fetchField(fetchField);
         return this;
     }
 

server/src/main/java/org/elasticsearch/index/query/InnerHitBuilder.java

@@ -395,10 +395,20 @@ public final class InnerHitBuilder implements Writeable, ToXContentObject {
      * @param format an optional format string used when formatting values, for example a date format.
      */
     public InnerHitBuilder addFetchField(String name, @Nullable String format) {
+        return addFetchField(name, format, null);
+    }
+
+    /**
+     * Adds a field to load and return as part of the search request.
+     * @param name the field name.
+     * @param format an optional format string used when formatting values, for example a date format.
+     * @param includeUnmapped whether unmapped fields should be returned as well
+     */
+    public InnerHitBuilder addFetchField(String name, @Nullable String format, Boolean includeUnmapped) {
         if (fetchFields == null || fetchFields.isEmpty()) {
             fetchFields = new ArrayList<>();
         }
-        fetchFields.add(new FieldAndFormat(name, format));
+        fetchFields.add(new FieldAndFormat(name, format, includeUnmapped));
         return this;
     }
 

server/src/main/java/org/elasticsearch/search/aggregations/metrics/TopHitsAggregationBuilder.java

@@ -446,14 +446,14 @@ public class TopHitsAggregationBuilder extends AbstractAggregationBuilder<TopHit
     /**
      * Adds a field to load and return as part of the search request.
      */
-    public TopHitsAggregationBuilder fetchField(String field, String format) {
-        if (field == null) {
+    public TopHitsAggregationBuilder fetchField(FieldAndFormat fieldAndFormat) {
+        if (fieldAndFormat == null) {
             throw new IllegalArgumentException("[fields] must not be null: [" + name + "]");
         }
         if (fetchFields == null) {
             fetchFields = new ArrayList<>();
         }
-        fetchFields.add(new FieldAndFormat(field, format));
+        fetchFields.add(fieldAndFormat);
         return this;
     }
 
@@ -461,7 +461,7 @@ public class TopHitsAggregationBuilder extends AbstractAggregationBuilder<TopHit
      * Adds a field to load and return as part of the search request.
      */
     public TopHitsAggregationBuilder fetchField(String field) {
-        return fetchField(field, null);
+        return fetchField(new FieldAndFormat(field, null, null));
     }
 
     /**
@@ -796,7 +796,7 @@ public class TopHitsAggregationBuilder extends AbstractAggregationBuilder<TopHit
                 } else if (SearchSourceBuilder.FETCH_FIELDS_FIELD.match(currentFieldName, parser.getDeprecationHandler())) {
                     while ((token = parser.nextToken()) != XContentParser.Token.END_ARRAY) {
                         FieldAndFormat ff = FieldAndFormat.fromXContent(parser);
-                        factory.fetchField(ff.field, ff.format);
+                        factory.fetchField(ff);
                     }
                 } else if (SearchSourceBuilder.SORT_FIELD.match(currentFieldName, parser.getDeprecationHandler())) {
                     List<SortBuilder<?>> sorts = SortBuilder.fromXContent(parser);

server/src/main/java/org/elasticsearch/search/builder/SearchSourceBuilder.java

@@ -868,19 +868,18 @@ public final class SearchSourceBuilder implements Writeable, ToXContentObject, R
      * Adds a field to load and return as part of the search request.
      */
     public SearchSourceBuilder fetchField(String name) {
-        return fetchField(name, null);
+        return fetchField(new FieldAndFormat(name, null, null));
     }
 
     /**
      * Adds a field to load and return as part of the search request.
-     * @param name the field name.
-     * @param format an optional format string used when formatting values, for example a date format.
+     * @param fetchField defining the field name, optional format and optional inclusion of unmapped fields
      */
-    public SearchSourceBuilder fetchField(String name, @Nullable String format) {
+    public SearchSourceBuilder fetchField(FieldAndFormat fetchField) {
         if (fetchFields == null) {
             fetchFields = new ArrayList<>();
         }
-        fetchFields.add(new FieldAndFormat(name, format));
+        fetchFields.add(fetchField);
         return this;
     }
 

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

@@ -45,7 +45,7 @@ public class FetchDocValuesContext {
             Collection<String> fieldNames = shardContext.simpleMatchToIndexNames(field.field);
             for (String fieldName : fieldNames) {
                 if (shardContext.isFieldMapped(fieldName)) {
-                    fields.add(new FieldAndFormat(fieldName, field.format));
+                    fields.add(new FieldAndFormat(fieldName, field.format, field.includeUnmapped));
                 }
             }
         }

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

@@ -54,6 +54,7 @@ public final class FetchFieldsPhase implements FetchSubPhase {
         }
 
         FieldFetcher fieldFetcher = FieldFetcher.create(fetchContext.getQueryShardContext(), searchLookup, fetchFieldsContext.fields());
+
         return new FetchSubPhaseProcessor() {
             @Override
             public void setNextReader(LeafReaderContext readerContext) {

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

@@ -19,6 +19,7 @@
 
 package org.elasticsearch.search.fetch.subphase;
 
+import org.elasticsearch.Version;
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.ParseField;
 import org.elasticsearch.common.io.stream.StreamInput;
@@ -40,14 +41,16 @@ import java.util.Objects;
 public final class FieldAndFormat implements Writeable, ToXContentObject {
     private static final ParseField FIELD_FIELD = new ParseField("field");
     private static final ParseField FORMAT_FIELD = new ParseField("format");
+    private static final ParseField INCLUDE_UNMAPPED_FIELD = new ParseField("include_unmapped");
 
     private static final ConstructingObjectParser<FieldAndFormat, Void> PARSER =
         new ConstructingObjectParser<>("fetch_field_and_format",
-        a -> new FieldAndFormat((String) a[0], (String) a[1]));
+        a -> new FieldAndFormat((String) a[0], (String) a[1], (Boolean) a[2]));
 
     static {
         PARSER.declareString(ConstructingObjectParser.constructorArg(), FIELD_FIELD);
         PARSER.declareStringOrNull(ConstructingObjectParser.optionalConstructorArg(), FORMAT_FIELD);
+        PARSER.declareBoolean(ConstructingObjectParser.optionalConstructorArg(), INCLUDE_UNMAPPED_FIELD);
     }
 
     /**
@@ -69,6 +72,9 @@ public final class FieldAndFormat implements Writeable, ToXContentObject {
         if (format != null) {
             builder.field(FORMAT_FIELD.getPreferredName(), format);
         }
+        if (this.includeUnmapped != null) {
+            builder.field(INCLUDE_UNMAPPED_FIELD.getPreferredName(), includeUnmapped);
+        }
         builder.endObject();
         return builder;
     }
@@ -79,28 +85,44 @@ public final class FieldAndFormat implements Writeable, ToXContentObject {
     /** The format of the field, or {@code null} if defaults should be used. */
     public final String format;
 
-    /** Sole constructor. */
+    /** Whether to include unmapped fields or not. */
+    public final Boolean includeUnmapped;
+
     public FieldAndFormat(String field, @Nullable String format) {
+        this(field, format, null);
+    }
+
+    public FieldAndFormat(String field, @Nullable String format, @Nullable Boolean includeUnmapped) {
         this.field = Objects.requireNonNull(field);
         this.format = format;
+        this.includeUnmapped = includeUnmapped;
     }
 
     /** Serialization constructor. */
     public FieldAndFormat(StreamInput in) throws IOException {
         this.field = in.readString();
         format = in.readOptionalString();
+        if (in.getVersion().onOrAfter(Version.CURRENT)) {
+            this.includeUnmapped = in.readOptionalBoolean();
+        } else {
+            this.includeUnmapped = null;
+        }
     }
 
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         out.writeString(field);
         out.writeOptionalString(format);
+        if (out.getVersion().onOrAfter(Version.CURRENT)) {
+            out.writeOptionalBoolean(this.includeUnmapped);
+        }
     }
 
     @Override
     public int hashCode() {
         int h = field.hashCode();
         h = 31 * h + Objects.hashCode(format);
+        h = 31 * h + Objects.hashCode(includeUnmapped);
         return h;
     }
 
@@ -110,6 +132,6 @@ public final class FieldAndFormat implements Writeable, ToXContentObject {
             return false;
         }
         FieldAndFormat other = (FieldAndFormat) obj;
-        return field.equals(other.field) && Objects.equals(format, other.format);
+        return field.equals(other.field) && Objects.equals(format, other.format) && Objects.equals(includeUnmapped, other.includeUnmapped);
     }
 }

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

@@ -20,7 +20,10 @@
 package org.elasticsearch.search.fetch.subphase;
 
 import org.apache.lucene.index.LeafReaderContext;
+import org.apache.lucene.util.automaton.Automata;
+import org.apache.lucene.util.automaton.CharacterRunAutomaton;
 import org.elasticsearch.common.document.DocumentField;
+import org.elasticsearch.common.regex.Regex;
 import org.elasticsearch.index.mapper.MappedFieldType;
 import org.elasticsearch.index.mapper.ValueFetcher;
 import org.elasticsearch.index.query.QueryShardContext;
@@ -31,6 +34,7 @@ import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -45,9 +49,16 @@ public class FieldFetcher {
                                       Collection<FieldAndFormat> fieldAndFormats) {
 
         List<FieldContext> fieldContexts = new ArrayList<>();
+        List<String> unmappedFetchPattern = new ArrayList<>();
+        Set<String> mappedToExclude = new HashSet<>();
+        boolean includeUnmapped = false;
 
         for (FieldAndFormat fieldAndFormat : fieldAndFormats) {
             String fieldPattern = fieldAndFormat.field;
+            if (fieldAndFormat.includeUnmapped != null && fieldAndFormat.includeUnmapped) {
+                unmappedFetchPattern.add(fieldAndFormat.field);
+                includeUnmapped = true;
+            }
             String format = fieldAndFormat.format;
 
             Collection<String> concreteFields = context.simpleMatchToIndexNames(fieldPattern);
@@ -57,17 +68,34 @@ public class FieldFetcher {
                     continue;
                 }
                 ValueFetcher valueFetcher = ft.valueFetcher(context, format);
+                mappedToExclude.add(field);
                 fieldContexts.add(new FieldContext(field, valueFetcher));
             }
         }
-
-        return new FieldFetcher(fieldContexts);
+        CharacterRunAutomaton unmappedFetchAutomaton = new CharacterRunAutomaton(Automata.makeEmpty());
+        if (unmappedFetchPattern.isEmpty() == false) {
+            unmappedFetchAutomaton = new CharacterRunAutomaton(
+                Regex.simpleMatchToAutomaton(unmappedFetchPattern.toArray(new String[unmappedFetchPattern.size()]))
+            );
+        }
+        return new FieldFetcher(fieldContexts, unmappedFetchAutomaton, mappedToExclude, includeUnmapped);
     }
 
     private final List<FieldContext> fieldContexts;
-
-    private FieldFetcher(List<FieldContext> fieldContexts) {
+    private final CharacterRunAutomaton unmappedFetchAutomaton;
+    private final Set<String> mappedToExclude;
+    private final boolean includeUnmapped;
+
+    private FieldFetcher(
+        List<FieldContext> fieldContexts,
+        CharacterRunAutomaton unmappedFetchAutomaton,
+        Set<String> mappedToExclude,
+        boolean includeUnmapped
+    ) {
         this.fieldContexts = fieldContexts;
+        this.unmappedFetchAutomaton = unmappedFetchAutomaton;
+        this.mappedToExclude = mappedToExclude;
+        this.includeUnmapped = includeUnmapped;
     }
 
     public Map<String, DocumentField> fetch(SourceLookup sourceLookup, Set<String> ignoredFields) throws IOException {
@@ -85,9 +113,84 @@ public class FieldFetcher {
                 documentFields.put(field, new DocumentField(field, parsedValues));
             }
         }
+        if (this.includeUnmapped) {
+            collectUnmapped(documentFields, sourceLookup.loadSourceIfNeeded(), "", 0);
+        }
         return documentFields;
     }
 
+    private void collectUnmapped(Map<String, DocumentField> documentFields, Map<String, Object> source, String parentPath, int lastState) {
+        for (String key : source.keySet()) {
+            Object value = source.get(key);
+            String currentPath = parentPath + key;
+            int currentState = step(this.unmappedFetchAutomaton, key, lastState);
+            if (currentState == -1) {
+                // current path doesn't match any fields pattern
+                continue;
+            }
+            if (value instanceof Map) {
+                // one step deeper into source tree
+                collectUnmapped(
+                    documentFields,
+                    (Map<String, Object>) value,
+                    currentPath + ".",
+                    step(this.unmappedFetchAutomaton, ".", currentState)
+                );
+            } else if (value instanceof List) {
+                // iterate through list values
+                collectUnmappedList(documentFields, (List<?>) value, currentPath, currentState);
+            } else {
+                // we have a leaf value
+                if (this.unmappedFetchAutomaton.isAccept(currentState) && this.mappedToExclude.contains(currentPath) == false) {
+                    if (value != null) {
+                        DocumentField currentEntry = documentFields.get(currentPath);
+                        if (currentEntry == null) {
+                            List<Object> list = new ArrayList<>();
+                            list.add(value);
+                            documentFields.put(currentPath, new DocumentField(currentPath, list));
+                        } else {
+                            currentEntry.getValues().add(value);
+                        }
+                    }
+                }
+            }
+        }
+    }
+
+    private void collectUnmappedList(Map<String, DocumentField> documentFields, Iterable<?> iterable, String parentPath, int lastState) {
+        List<Object> list = new ArrayList<>();
+        for (Object value : iterable) {
+            if (value instanceof Map) {
+                collectUnmapped(
+                    documentFields,
+                    (Map<String, Object>) value,
+                    parentPath + ".",
+                    step(this.unmappedFetchAutomaton, ".", lastState)
+                );
+            } else if (value instanceof List) {
+                // weird case, but can happen for objects with "enabled" : "false"
+                collectUnmappedList(documentFields, (List<?>) value, parentPath, lastState);
+            } else if (this.unmappedFetchAutomaton.isAccept(lastState) && this.mappedToExclude.contains(parentPath) == false) {
+                list.add(value);
+            }
+        }
+        if (list.isEmpty() == false) {
+            DocumentField currentEntry = documentFields.get(parentPath);
+            if (currentEntry == null) {
+                documentFields.put(parentPath, new DocumentField(parentPath, list));
+            } else {
+                currentEntry.getValues().addAll(list);
+            }
+        }
+    }
+
+    private static int step(CharacterRunAutomaton automaton, String key, int state) {
+        for (int i = 0; state != -1 && i < key.length(); ++i) {
+            state = automaton.step(state, key.charAt(i));
+        }
+        return state;
+    }
+
     public void setNextReader(LeafReaderContext readerContext) {
         for (FieldContext field : fieldContexts) {
             field.valueFetcher.setNextReader(readerContext);

server/src/test/java/org/elasticsearch/search/fetch/subphase/FieldFetcherTests.java

@@ -21,6 +21,7 @@ package org.elasticsearch.search.fetch.subphase;
 
 import org.elasticsearch.Version;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
+import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.document.DocumentField;
@@ -35,6 +36,7 @@ import org.elasticsearch.index.query.QueryShardContext;
 import org.elasticsearch.search.lookup.SourceLookup;
 
 import java.io.IOException;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -58,7 +60,7 @@ public class FieldFetcherTests extends MapperServiceTestCase {
         List<FieldAndFormat> fieldAndFormats = List.of(
             new FieldAndFormat("field", null),
             new FieldAndFormat("object.field", null));
-        Map<String, DocumentField> fields = fetchFields(mapperService, source, fieldAndFormats);
+        Map<String, DocumentField> fields = fetchFields(mapperService, source, fieldAndFormats, null);
         assertThat(fields.size(), equalTo(2));
 
         DocumentField field = fields.get("field");
@@ -238,7 +240,8 @@ public class FieldFetcherTests extends MapperServiceTestCase {
 
         Map<String, DocumentField> fields = fetchFields(mapperService, source, List.of(
             new FieldAndFormat("field", null),
-            new FieldAndFormat("date_field", "yyyy/MM/dd")));
+            new FieldAndFormat("date_field", "yyyy/MM/dd")),
+            null);
         assertThat(fields.size(), equalTo(2));
 
         DocumentField field = fields.get("field");
@@ -393,21 +396,289 @@ public class FieldFetcherTests extends MapperServiceTestCase {
         }
     }
 
-    private static Map<String, DocumentField> fetchFields(MapperService mapperService, XContentBuilder source, String fieldPattern)
-        throws IOException {
+    public void testSimpleUnmappedFields() throws IOException {
+        MapperService mapperService = createMapperService();
+
+        XContentBuilder source = XContentFactory.jsonBuilder()
+            .startObject()
+                .field("unmapped_f1", "some text")
+                .field("unmapped_f2", "some text")
+                .field("unmapped_f3", "some text")
+                .field("something_else", "some text")
+                .nullField("null_value")
+                .startObject("object")
+                    .field("a", "foo")
+                .endObject()
+                .field("object.b", "bar")
+            .endObject();
+
+        Map<String, DocumentField> fields = fetchFields(mapperService, source, fieldAndFormatList("unmapped_f*", null, true), null);
+        assertThat(fields.size(), equalTo(3));
+        assertThat(fields.keySet(), containsInAnyOrder("unmapped_f1", "unmapped_f2", "unmapped_f3"));
+
+        fields = fetchFields(mapperService, source, fieldAndFormatList("un*1", null, true), null);
+        assertThat(fields.size(), equalTo(1));
+        assertThat(fields.keySet(), containsInAnyOrder("unmapped_f1"));
+
+        fields = fetchFields(mapperService, source, fieldAndFormatList("*thing*", null, true), null);
+        assertThat(fields.size(), equalTo(1));
+        assertThat(fields.keySet(), containsInAnyOrder("something_else"));
+
+        fields = fetchFields(mapperService, source, fieldAndFormatList("null*", null, true), null);
+        assertThat(fields.size(), equalTo(0));
+
+        fields = fetchFields(mapperService, source, fieldAndFormatList("object.a", null, true), null);
+        assertThat(fields.size(), equalTo(1));
+        assertEquals("foo", fields.get("object.a").getValues().get(0));
+
+        fields = fetchFields(mapperService, source, fieldAndFormatList("object.b", null, true), null);
+        assertThat(fields.size(), equalTo(1));
+        assertEquals("bar", fields.get("object.b").getValues().get(0));
+    }
+
+    public void testSimpleUnmappedArray() throws IOException {
+        MapperService mapperService = createMapperService();
+
+        XContentBuilder source = XContentFactory.jsonBuilder().startObject().array("unmapped_field", "foo", "bar").endObject();
+
+        Map<String, DocumentField> fields = fetchFields(mapperService, source, fieldAndFormatList("unmapped_field", null, true), null);
+        assertThat(fields.size(), equalTo(1));
+        assertThat(fields.keySet(), containsInAnyOrder("unmapped_field"));
+        DocumentField field = fields.get("unmapped_field");
+
+        assertThat(field.getValues().size(), equalTo(2));
+        assertThat(field.getValues(), hasItems("foo", "bar"));
+    }
+
+    public void testSimpleUnmappedArrayWithObjects() throws IOException {
+        MapperService mapperService = createMapperService();
+
+        XContentBuilder source = XContentFactory.jsonBuilder().startObject()
+            .startArray("unmapped_field")
+                .startObject()
+                    .field("f1", "a")
+                .endObject()
+                .startObject()
+                    .field("f2", "b")
+                .endObject()
+            .endArray()
+            .endObject();
 
-        List<FieldAndFormat> fields = List.of(new FieldAndFormat(fieldPattern, null));
-        return fetchFields(mapperService, source, fields);
+        Map<String, DocumentField> fields = fetchFields(mapperService, source, fieldAndFormatList("unmapped_field", null, true), null);
+        assertThat(fields.size(), equalTo(0));
+
+        fields = fetchFields(mapperService, source, fieldAndFormatList("unmapped_field.f*", null, true), null);
+        assertThat(fields.size(), equalTo(2));
+        assertThat(fields.get("unmapped_field.f1").getValue(), equalTo("a"));
+        assertThat(fields.get("unmapped_field.f2").getValue(), equalTo("b"));
+
+        source = XContentFactory.jsonBuilder().startObject()
+            .startArray("unmapped_field")
+                .startObject()
+                    .field("f1", "a")
+                    .array("f2", 1, 2)
+                    .array("f3", 1, 2)
+                .endObject()
+                .startObject()
+                    .field("f1", "b") // same field name, this should result in a list returned
+                    .array("f2", 3, 4)
+                    .array("f3", "foo")
+                .endObject()
+            .endArray()
+            .endObject();
+
+        fields = fetchFields(mapperService, source, fieldAndFormatList("unmapped_field.f1", null, true), null);
+        assertThat(fields.size(), equalTo(1));
+        DocumentField field = fields.get("unmapped_field.f1");
+        assertThat(field.getValues().size(), equalTo(2));
+        assertThat(field.getValues(), hasItems("a", "b"));
+
+        fields = fetchFields(mapperService, source, fieldAndFormatList("unmapped_field.f2", null, true), null);
+        assertThat(fields.size(), equalTo(1));
+        field = fields.get("unmapped_field.f2");
+        assertThat(field.getValues().size(), equalTo(4));
+        assertThat(field.getValues(), hasItems(1, 2, 3, 4));
+
+        fields = fetchFields(mapperService, source, fieldAndFormatList("unmapped_field.f3", null, true), null);
+        assertThat(fields.size(), equalTo(1));
+        field = fields.get("unmapped_field.f3");
+        assertThat(field.getValues().size(), equalTo(3));
+        assertThat(field.getValues(), hasItems(1, 2, "foo"));
     }
 
-    private static Map<String, DocumentField> fetchFields(MapperService mapperService, XContentBuilder source, List<FieldAndFormat> fields)
+    public void testUnmappedFieldsInsideObject() throws IOException {
+        XContentBuilder mapping = XContentFactory.jsonBuilder().startObject()
+            .startObject("_doc")
+            .startObject("properties")
+                .startObject("obj")
+                    .field("type", "object")
+                    .field("dynamic", "false")
+                    .startObject("properties")
+                        .startObject("f1").field("type", "keyword").endObject()
+                    .endObject()
+                .endObject()
+            .endObject()
+            .endObject()
+        .endObject();
+
+        MapperService mapperService = createMapperService(mapping);
+
+        XContentBuilder source = XContentFactory.jsonBuilder().startObject()
+            .field("obj.f1", "value1")
+            .field("obj.f2", "unmapped_value_f2")
+            .field("obj.innerObj.f3", "unmapped_value_f3")
+            .field("obj.innerObj.f4", "unmapped_value_f4")
+            .endObject();
+
+        Map<String, DocumentField> fields = fetchFields(mapperService, source, fieldAndFormatList("*", null, false), null);
+
+        // without unmapped fields this should only return "obj.f1"
+        assertThat(fields.size(), equalTo(1));
+        assertThat(fields.keySet(), containsInAnyOrder("obj.f1"));
+
+        fields = fetchFields(mapperService, source, fieldAndFormatList("*", null, true), null);
+        assertThat(fields.size(), equalTo(4));
+        assertThat(fields.keySet(), containsInAnyOrder("obj.f1", "obj.f2", "obj.innerObj.f3", "obj.innerObj.f4"));
+    }
+
+    public void testUnmappedFieldsInsideDisabledObject() throws IOException {
+        XContentBuilder mapping = XContentFactory.jsonBuilder().startObject()
+            .startObject("_doc")
+            .startObject("properties")
+                .startObject("obj")
+                    .field("type", "object")
+                    .field("enabled", "false")
+                .endObject()
+            .endObject()
+            .endObject()
+        .endObject();
+
+        MapperService mapperService = createMapperService(mapping);
+
+        XContentBuilder source = XContentFactory.jsonBuilder().startObject()
+            .startArray("obj")
+            .value("string_value")
+            .startObject()
+                .field("a", "b")
+            .endObject()
+            .startArray()
+                .value(1).value(2).value(3)
+            .endArray()
+            .endArray()
+        .endObject();
+
+        Map<String, DocumentField> fields = fetchFields(mapperService, source, fieldAndFormatList("*", null, false), null);
+        // without unmapped fields this should return nothing
+        assertThat(fields.size(), equalTo(0));
+
+        fields = fetchFields(mapperService, source, fieldAndFormatList("*", null, true), null);
+        assertThat(fields.size(), equalTo(2));
+        assertThat(fields.keySet(), containsInAnyOrder("obj", "obj.a"));
+
+        List<Object> obj = fields.get("obj").getValues();
+        assertEquals(4, obj.size());
+        assertThat(obj, hasItems("string_value", 1, 2, 3));
+
+        List<Object> innerObj = fields.get("obj.a").getValues();
+        assertEquals(1, innerObj.size());
+        assertEquals("b", fields.get("obj.a").getValue());
+    }
+
+    /**
+     * If a mapped field for some reason contains a "_source" value that is not returned by the
+     * mapped retrieval mechanism (e.g. because its malformed), we don't want to fetch it from _source.
+     */
+    public void testMappedFieldNotOverwritten() throws IOException {
+        XContentBuilder mapping = XContentFactory.jsonBuilder().startObject()
+            .startObject("_doc")
+            .startObject("properties")
+                .startObject("f1")
+                    .field("type", "integer")
+                    .field("ignore_malformed", "true")
+                .endObject()
+            .endObject()
+            .endObject()
+        .endObject();
+
+        MapperService mapperService = createMapperService(mapping);
+
+        XContentBuilder source = XContentFactory.jsonBuilder().startObject()
+            .field("f1", "malformed")
+            .endObject();
+
+        // this should not return a field bc. f1 is in the ignored fields
+        Map<String, DocumentField> fields = fetchFields(mapperService, source, List.of(new FieldAndFormat("*", null, true)), Set.of("f1"));
+        assertThat(fields.size(), equalTo(0));
+
+        // and this should neither
+        fields = fetchFields(mapperService, source, List.of(new FieldAndFormat("*", null, true)), Set.of("f1"));
+        assertThat(fields.size(), equalTo(0));
+
+        fields = fetchFields(mapperService, source, List.of(new FieldAndFormat("f1", null, true)), Set.of("f1"));
+        assertThat(fields.size(), equalTo(0));
+
+        // check this also does not overwrite with arrays
+        source = XContentFactory.jsonBuilder().startObject()
+            .array("f1", "malformed")
+            .endObject();
+
+        fields = fetchFields(mapperService, source, List.of(new FieldAndFormat("f1", null, true)), Set.of("f1"));
+        assertThat(fields.size(), equalTo(0));
+    }
+
+    public void testUnmappedFieldsWildcard() throws IOException {
+        MapperService mapperService = createMapperService();
+
+        XContentBuilder source = XContentFactory.jsonBuilder().startObject()
+            .startObject("unmapped_object")
+                .field("a", "foo")
+                .field("b", "bar")
+            .endObject()
+            .endObject();
+
+        Map<String, DocumentField> fields = fetchFields(mapperService, source, fieldAndFormatList("unmapped_object", null, true), null);
+        assertThat(fields.size(), equalTo(0));
+
+        fields = fetchFields(mapperService, source, fieldAndFormatList("unmap*object", null, true), null);
+        assertThat(fields.size(), equalTo(0));
+
+        fields = fetchFields(mapperService, source, fieldAndFormatList("unmapped_object.*", null, true), null);
+        assertThat(fields.size(), equalTo(2));
+        assertThat(fields.keySet(), containsInAnyOrder("unmapped_object.a", "unmapped_object.b"));
+
+        assertThat(fields.get("unmapped_object.a").getValue(), equalTo("foo"));
+        assertThat(fields.get("unmapped_object.b").getValue(), equalTo("bar"));
+
+        fields = fetchFields(mapperService, source, fieldAndFormatList("unmapped_object.a", null, true), null);
+        assertThat(fields.size(), equalTo(1));
+        assertThat(fields.get("unmapped_object.a").getValue(), equalTo("foo"));
+
+        fields = fetchFields(mapperService, source, fieldAndFormatList("unmapped_object.b", null, true), null);
+        assertThat(fields.size(), equalTo(1));
+        assertThat(fields.get("unmapped_object.b").getValue(), equalTo("bar"));
+    }
+
+    private List<FieldAndFormat> fieldAndFormatList(String name, String format, boolean includeUnmapped) {
+        return Collections.singletonList(new FieldAndFormat(name, format, includeUnmapped));
+    }
+
+    private Map<String, DocumentField> fetchFields(MapperService mapperService, XContentBuilder source, String fieldPattern)
         throws IOException {
+        return fetchFields(mapperService, source, fieldAndFormatList(fieldPattern, null, false), null);
+    }
+
+    private static Map<String, DocumentField> fetchFields(
+        MapperService mapperService,
+        XContentBuilder source,
+        List<FieldAndFormat> fields,
+        @Nullable Set<String> ignoreFields
+    ) throws IOException {
 
         SourceLookup sourceLookup = new SourceLookup();
         sourceLookup.setSource(BytesReference.bytes(source));
 
         FieldFetcher fieldFetcher = FieldFetcher.create(newQueryShardContext(mapperService), null, fields);
-        return fieldFetcher.fetch(sourceLookup, Set.of());
+        return fieldFetcher.fetch(sourceLookup, ignoreFields != null ? ignoreFields : Collections.emptySet());
     }
 
     public MapperService createMapperService() throws IOException {

x-pack/plugin/src/test/resources/rest-api-spec/test/flattened/10_basic.yml

@@ -150,9 +150,44 @@
       search:
         index: test
         body:
-          fields: ["flat*"]
+          fields: [ "flat*" ]
 
   - match:  { hits.total.value: 1 }
   - length: { hits.hits: 1 }
   - length: { hits.hits.0.fields: 1 }
   - match:  { hits.hits.0.fields.flattened: [ { "some_field": "some_value" } ] }
+
+---
+"Test fields option on flattened object field with include_unmapped":
+  -  skip:
+        version: ' - 7.99.99'
+        reason: support isn't yet backported
+  - do:
+      indices.create:
+        index:  test
+        body:
+          mappings:
+            properties:
+              flattened:
+                type: flattened
+
+  - do:
+      index:
+        index:  test
+        id:     1
+        body:
+          flattened:
+            some_field: some_value
+        refresh: true
+
+  - do:
+      search:
+        index: test
+        body:
+          fields: [ { "field" : "flat*", "include_unmapped" : true } ]
+
+  - match:  { hits.total.value: 1 }
+  - length: { hits.hits: 1 }
+  - length: { hits.hits.0.fields: 2 }
+  - match:  { hits.hits.0.fields.flattened: [ { "some_field": "some_value" } ] }
+  - match:  { hits.hits.0.fields.flattened\.some_field: [ "some_value" ] }