Support unmapped fields in search 'fields' option

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 addition.

Closes elastic#63690

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

@@ -295,3 +295,69 @@ 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
+              - f4
+  -  match:
+        hits.hits.0.fields.f1: 
+        - some text
+  -  match:
+        hits.hits.0.fields.f4:
+        - some other text
+  -  do:
+        search:
+           index: test
+           body:
+              fields:
+              - f*
+  -  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

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

@@ -93,7 +93,7 @@ protected void setupInnerHitsContext(QueryShardContext queryShardContext,
             innerHitsContext.docValuesContext(docValuesContext);
         }
         if (innerHitBuilder.getFetchFields() != null) {
-            FetchFieldsContext fieldsContext = new FetchFieldsContext(innerHitBuilder.getFetchFields());
+            FetchFieldsContext fieldsContext = new FetchFieldsContext(innerHitBuilder.getFetchFields(), false);
             innerHitsContext.fetchFieldsContext(fieldsContext);
         }
         if (innerHitBuilder.getScriptFields() != null) {

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

@@ -973,7 +973,8 @@ private void parseSource(DefaultSearchContext context, SearchSourceBuilder sourc
             context.docValuesContext(docValuesContext);
         }
         if (source.fetchFields() != null) {
-            FetchFieldsContext fetchFieldsContext = new FetchFieldsContext(source.fetchFields());
+            // TODO make "includeUnmapped configurable?
+            FetchFieldsContext fetchFieldsContext = new FetchFieldsContext(source.fetchFields(), true);
             context.fetchFieldsContext(fetchFieldsContext);
         }
         if (source.highlighter() != null) {

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

@@ -114,7 +114,7 @@ public Aggregator createInternal(SearchContext searchContext,
             subSearchContext.docValuesContext(docValuesContext);
         }
         if (fetchFields != null) {
-            FetchFieldsContext fieldsContext = new FetchFieldsContext(fetchFields);
+            FetchFieldsContext fieldsContext = new FetchFieldsContext(fetchFields, true);
             subSearchContext.fetchFieldsContext(fieldsContext);
         }
         for (ScriptFieldsContext.ScriptField field : scriptFields) {

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

@@ -25,12 +25,18 @@
  */
 public class FetchFieldsContext {
     private final List<FieldAndFormat> fields;
+    private final boolean includeUnmapped;
 
-    public FetchFieldsContext(List<FieldAndFormat> fields) {
+    public FetchFieldsContext(List<FieldAndFormat> fields, boolean includeUnmapped) {
         this.fields = fields;
+        this.includeUnmapped = includeUnmapped;
     }
 
     public List<FieldAndFormat> fields() {
         return fields;
     }
+
+    public boolean includeUnmapped() {
+        return includeUnmapped;
+    }
 }

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

@@ -53,7 +53,13 @@ public FetchSubPhaseProcessor getProcessor(FetchContext fetchContext) {
                 "in the mappings for index [" + fetchContext.getIndexName() + "]");
         }
 
-        FieldFetcher fieldFetcher = FieldFetcher.create(fetchContext.getQueryShardContext(), searchLookup, fetchFieldsContext.fields());
+        FieldFetcher fieldFetcher = FieldFetcher.create(
+            fetchContext.getQueryShardContext(),
+            searchLookup,
+            fetchFieldsContext.fields(),
+            fetchFieldsContext.includeUnmapped()
+        );
+
         return new FetchSubPhaseProcessor() {
             @Override
             public void setNextReader(LeafReaderContext readerContext) {

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

@@ -21,6 +21,7 @@
 
 import org.apache.lucene.index.LeafReaderContext;
 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;
@@ -30,7 +31,9 @@
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -42,32 +45,48 @@
 public class FieldFetcher {
     public static FieldFetcher create(QueryShardContext context,
                                       SearchLookup searchLookup,
-                                      Collection<FieldAndFormat> fieldAndFormats) {
+                                      Collection<FieldAndFormat> fieldAndFormats,
+                                      boolean includeUnmapped) {
 
         List<FieldContext> fieldContexts = new ArrayList<>();
+        List<String> originalPattern = new ArrayList<>();
+        List<String> mappedFields = new ArrayList<>();
 
         for (FieldAndFormat fieldAndFormat : fieldAndFormats) {
             String fieldPattern = fieldAndFormat.field;
             String format = fieldAndFormat.format;
 
             Collection<String> concreteFields = context.simpleMatchToIndexNames(fieldPattern);
+            originalPattern.add(fieldAndFormat.field);
             for (String field : concreteFields) {
                 MappedFieldType ft = context.getFieldType(field);
                 if (ft == null || context.isMetadataField(field)) {
                     continue;
                 }
                 ValueFetcher valueFetcher = ft.valueFetcher(context, searchLookup, format);
+                mappedFields.add(field);
                 fieldContexts.add(new FieldContext(field, valueFetcher));
             }
         }
 
-        return new FieldFetcher(fieldContexts);
+        return new FieldFetcher(fieldContexts, originalPattern, mappedFields, includeUnmapped);
     }
 
     private final List<FieldContext> fieldContexts;
-
-    private FieldFetcher(List<FieldContext> fieldContexts) {
+    private final List<String> originalPattern;
+    private final List<String> mappedFields;
+    private final boolean includeUnmapped;
+
+    private FieldFetcher(
+        List<FieldContext> fieldContexts,
+        List<String> originalPattern,
+        List<String> mappedFields,
+        boolean includeUnmapped
+    ) {
         this.fieldContexts = fieldContexts;
+        this.originalPattern = originalPattern;
+        this.mappedFields = mappedFields;
+        this.includeUnmapped = includeUnmapped;
     }
 
     public Map<String, DocumentField> fetch(SourceLookup sourceLookup, Set<String> ignoredFields) throws IOException {
@@ -85,9 +104,60 @@ public Map<String, DocumentField> fetch(SourceLookup sourceLookup, Set<String> i
                 documentFields.put(field, new DocumentField(field, parsedValues));
             }
         }
+        if (includeUnmapped) {
+            // also look up unmapped fields from source
+            Set<String> allSourcePaths = extractAllLeafPaths(sourceLookup.loadSourceIfNeeded());
+            for (String fetchFieldPattern : originalPattern) {
+                if (Regex.isSimpleMatchPattern(fetchFieldPattern) == false) {
+                    // if pattern has no wildcard, simply look up field if not already present
+                    if (allSourcePaths.contains(fetchFieldPattern)) {
+                        addValueFromSource(sourceLookup, fetchFieldPattern, documentFields);
+                    }
+                } else {
+                    for (String singlePath : allSourcePaths) {
+                        if (Regex.simpleMatch(fetchFieldPattern, singlePath)) {
+                            addValueFromSource(sourceLookup, singlePath, documentFields);
+                        }
+                    }
+                }
+            }
+        }
         return documentFields;
     }
 
+    private void addValueFromSource(SourceLookup sourceLookup, String path, Map<String, DocumentField> documentFields) {
+        // checking mapped fields here again to avoid adding from _source where some value was already added or ignored on purpose before
+        if (mappedFields.contains(path) == false) {
+            Object object = sourceLookup.extractValue(path, null);
+            DocumentField f;
+            if (object instanceof List) {
+                f = new DocumentField(path, (List) object);
+            } else {
+                f = new DocumentField(path, Collections.singletonList(object));
+            }
+            if (f.getValue() != null) {
+                documentFields.put(path, f);
+            }
+        }
+    }
+
+    static Set<String> extractAllLeafPaths(Map<String, Object> map) {
+        Set<String> allPaths = new HashSet<>();
+        collectAllPaths(allPaths, "", map);
+        return allPaths;
+    }
+
+    private static void collectAllPaths(Set<String> allPaths, String prefix, Map<String, Object> source) {
+        for (String key : source.keySet()) {
+            Object value = source.get(key);
+            if (value instanceof Map) {
+                collectAllPaths(allPaths, prefix + key + ".", (Map<String, Object>) value);
+            } else {
+                allPaths.add(prefix + key);
+            }
+        }
+    }
+
     public void setNextReader(LeafReaderContext readerContext) {
         for (FieldContext field : fieldContexts) {
             field.valueFetcher.setNextReader(readerContext);