Support fetching flattened subfields (#70916)

Currently the `fields` API fetches the root flattened field and returns it in a
structured way in the response. In addition this change makes it possible to
directly query subfields. However, requesting flattened subfields via wildcard
patterns is not possible.

Closes #70605

docs/reference/mapping/types/flattened.asciidoc

@@ -121,6 +121,76 @@ lexicographically.
 Flattened object fields currently cannot be stored. It is not possible to
 specify the <<mapping-store, `store`>> parameter in the mapping.
 
+[[search-fields-flattened]]
+==== Retrieving flattened fields
+
+Field values and concrete subfields can be retrieved using the 
+<<search-fields-param,fields parameter>>. content. Since the `flattened` field maps an 
+entire object with potentially many subfields as a single field, the response contains
+the unaltered structure from `_source`.
+
+Single subfields, however, can be fetched by specifying them explicitly in the request.
+This only works for concrete paths, but not using wildcards:
+
+[source,console]
+--------------------------------------------------
+PUT my-index-000001
+{
+  "mappings": {
+    "properties": {
+      "flattened_field": {
+        "type": "flattened"
+      }
+    }
+  }
+}
+
+PUT my-index-000001/_doc/1?refresh=true
+{
+  "flattened_field" : {
+    "subfield" : "value"
+  }
+}
+
+POST my-index-000001/_search
+{
+  "fields": ["flattened_field.subfield"],
+  "_source": false
+}
+--------------------------------------------------
+
+[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": {
+        "flattened_field.subfield" : [ "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/]
+
 [[flattened-params]]
 ==== Parameters for flattened object fields
 

rest-api-spec/src/yamlRestTest/resources/rest-api-spec/test/search/340_flattened.yml

@@ -180,3 +180,60 @@ setup:
   - length: { hits.hits: 1 }
   - length: { hits.hits.0.fields: 1 }
   - match:  { hits.hits.0.fields.flattened: [ { "some_field": "some_value" } ] }
+
+---
+"Test fetching flattened subfields via fields option":
+  - do:
+      indices.create:
+        index:  test
+        body:
+          mappings:
+            properties:
+              flattened:
+                type: flattened
+
+  - do:
+      index:
+        index:  test
+        id:     1
+        body:
+          flattened:
+            some_field: some_value
+            some_fields:
+               - value1
+               - value2
+        refresh: true
+
+  - do:
+      search:
+        index: test
+        body:
+          fields: [ { "field" : "flattened.some_field" } ]
+
+  - length: { hits.hits.0.fields: 1 }
+  - match:  { hits.hits.0.fields.flattened\.some_field: [ "some_value" ] }
+
+  - do:
+      search:
+        index: test
+        body:
+          fields: [ { "field" : "flattened.some_fields" } ]
+
+  - length: { hits.hits.0.fields: 1 }
+  - match:  { hits.hits.0.fields.flattened\.some_fields: [ "value1", "value2" ] }
+
+  - do:
+      search:
+        index: test
+        body:
+          fields: [ { "field" : "flattened.some*" } ]
+
+  - is_false: hits.hits.0.fields
+
+  - do:
+      search:
+        index: test
+        body:
+          fields: [ { "field" : "flattened.non_existing_field" } ]
+
+  - is_false: hits.hits.0.fields

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

@@ -130,6 +130,10 @@ final class FieldTypeLookup {
         if (fullNameToFieldType.isEmpty()) {
             return Set.of();
         }
+        if (dynamicKeyLookup.get(field) != null) {
+            return Set.of(field);
+        }
+
         String resolvedField = field;
         int lastDotIndex = field.lastIndexOf('.');
         if (lastDotIndex > 0) {

server/src/main/java/org/elasticsearch/index/mapper/flattened/FlattenedFieldMapper.java

@@ -165,17 +165,19 @@ public final class FlattenedFieldMapper extends DynamicKeyFieldMapper {
      */
     public static final class KeyedFlattenedFieldType extends StringFieldType {
         private final String key;
+        private final String rootName;
 
-        public KeyedFlattenedFieldType(String name, boolean indexed, boolean hasDocValues, String key,
+        KeyedFlattenedFieldType(String rootName, boolean indexed, boolean hasDocValues, String key,
                                        boolean splitQueriesOnWhitespace, Map<String, String> meta) {
-            super(name, indexed, false, hasDocValues,
+            super(rootName + KEYED_FIELD_SUFFIX, indexed, false, hasDocValues,
                 splitQueriesOnWhitespace ? TextSearchInfo.WHITESPACE_MATCH_ONLY : TextSearchInfo.SIMPLE_MATCH_ONLY,
                 meta);
             this.key = key;
+            this.rootName = rootName;
         }
 
-        private KeyedFlattenedFieldType(String name, String key, RootFlattenedFieldType ref) {
-            this(name, ref.isSearchable(), ref.hasDocValues(), key, ref.splitQueriesOnWhitespace, ref.meta());
+        private KeyedFlattenedFieldType(String rootName, String key, RootFlattenedFieldType ref) {
+            this(rootName, ref.isSearchable(), ref.hasDocValues(), key, ref.splitQueriesOnWhitespace, ref.meta());
         }
 
         @Override
@@ -261,8 +263,11 @@ public final class FlattenedFieldMapper extends DynamicKeyFieldMapper {
 
         @Override
         public ValueFetcher valueFetcher(SearchExecutionContext context, String format) {
-            // This is an internal field but it can match a field pattern so we return an empty list.
-            return lookup -> List.of();
+            if (format != null) {
+                throw new IllegalArgumentException("Field [" + rootName + "." + key + "] of type [" + typeName() +
+                    "] doesn't support formats.");
+            }
+            return SourceValueFetcher.identity(rootName + "." + key, context, format);
         }
     }
 
@@ -426,7 +431,7 @@ public final class FlattenedFieldMapper extends DynamicKeyFieldMapper {
                                  Builder builder) {
         super(simpleName, mappedFieldType, Lucene.KEYWORD_ANALYZER, CopyTo.empty());
         this.builder = builder;
-        this.fieldParser = new FlattenedFieldParser(mappedFieldType.name(), keyedFieldName(),
+        this.fieldParser = new FlattenedFieldParser(mappedFieldType.name(), mappedFieldType.name() + KEYED_FIELD_SUFFIX,
             mappedFieldType, builder.depthLimit.get(), builder.ignoreAbove.get(), builder.nullValue.get());
     }
 
@@ -450,11 +455,7 @@ public final class FlattenedFieldMapper extends DynamicKeyFieldMapper {
 
     @Override
     public KeyedFlattenedFieldType keyedFieldType(String key) {
-        return new KeyedFlattenedFieldType(keyedFieldName(), key, fieldType());
-    }
-
-    public String keyedFieldName() {
-        return mappedFieldType.name() + KEYED_FIELD_SUFFIX;
+        return new KeyedFlattenedFieldType(name(), key, fieldType());
     }
 
     @Override

server/src/test/java/org/elasticsearch/index/mapper/FieldTypeLookupTests.java

@@ -189,7 +189,7 @@ public class FieldTypeLookupTests extends ESTestCase {
         String searchFieldName = fieldName + "." + objectKey;
 
         MappedFieldType searchFieldType = lookup.get(searchFieldName);
-        assertEquals(mapper.keyedFieldName(), searchFieldType.name());
+        assertEquals(mapper.keyedFieldType(objectKey).name(), searchFieldType.name());
         assertThat(searchFieldType, Matchers.instanceOf(FlattenedFieldMapper.KeyedFlattenedFieldType.class));
 
         FlattenedFieldMapper.KeyedFlattenedFieldType keyedFieldType = (FlattenedFieldMapper.KeyedFlattenedFieldType) searchFieldType;
@@ -210,7 +210,7 @@ public class FieldTypeLookupTests extends ESTestCase {
         String searchFieldName = aliasName + "." + objectKey;
 
         MappedFieldType searchFieldType = lookup.get(searchFieldName);
-        assertEquals(mapper.keyedFieldName(), searchFieldType.name());
+        assertEquals(mapper.keyedFieldType(objectKey).name(), searchFieldType.name());
         assertThat(searchFieldType, Matchers.instanceOf(FlattenedFieldMapper.KeyedFlattenedFieldType.class));
 
         FlattenedFieldMapper.KeyedFlattenedFieldType keyedFieldType = (FlattenedFieldMapper.KeyedFlattenedFieldType) searchFieldType;

server/src/test/java/org/elasticsearch/index/mapper/flattened/FlattenedIndexFieldDataTests.java

@@ -46,12 +46,13 @@ public class FlattenedIndexFieldDataTests extends ESSingleNodeTestCase  {
             indexService.mapperService());
 
         FlattenedFieldMapper fieldMapper = new FlattenedFieldMapper.Builder("json").build(new ContentPath(1));
+        KeyedFlattenedFieldType fieldType1 = fieldMapper.keyedFieldType("key");
 
         AtomicInteger onCacheCalled = new AtomicInteger();
         ifdService.setListener(new IndexFieldDataCache.Listener() {
             @Override
             public void onCache(ShardId shardId, String fieldName, Accountable ramUsage) {
-                assertEquals(fieldMapper.keyedFieldName(), fieldName);
+                assertEquals(fieldType1.name(), fieldName);
                 onCacheCalled.incrementAndGet();
             }
         });
@@ -71,7 +72,6 @@ public class FlattenedIndexFieldDataTests extends ESSingleNodeTestCase  {
             new ShardId("test", "_na_", 1));
 
         // Load global field data for subfield 'key'.
-        KeyedFlattenedFieldType fieldType1 = fieldMapper.keyedFieldType("key");
         IndexFieldData<?> ifd1 = ifdService.getForField(fieldType1, "test", () -> {
             throw new UnsupportedOperationException("search lookup not available");
         });

server/src/test/java/org/elasticsearch/index/mapper/flattened/KeyedFlattenedFieldTypeTests.java

@@ -20,13 +20,20 @@ import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.common.lucene.search.AutomatonQueries;
 import org.elasticsearch.common.unit.Fuzziness;
 import org.elasticsearch.index.mapper.FieldTypeTestCase;
+import org.elasticsearch.index.mapper.ValueFetcher;
 import org.elasticsearch.index.mapper.flattened.FlattenedFieldMapper.KeyedFlattenedFieldType;
+import org.elasticsearch.index.query.SearchExecutionContext;
+import org.elasticsearch.search.lookup.SourceLookup;
 
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
 
 public class KeyedFlattenedFieldTypeTests extends FieldTypeTestCase {
 
@@ -50,23 +57,22 @@ public class KeyedFlattenedFieldTypeTests extends FieldTypeTestCase {
     public void testTermQuery() {
         KeyedFlattenedFieldType ft = createFieldType();
 
-        Query expected = new TermQuery(new Term("field", "key\0value"));
+        Query expected = new TermQuery(new Term(ft.name(), "key\0value"));
         assertEquals(expected, ft.termQuery("value", null));
 
-        expected = AutomatonQueries.caseInsensitiveTermQuery(new Term("field", "key\0value"));
+        expected = AutomatonQueries.caseInsensitiveTermQuery(new Term(ft.name(), "key\0value"));
         assertEquals(expected, ft.termQueryCaseInsensitive("value", null));
 
-        KeyedFlattenedFieldType unsearchable = new KeyedFlattenedFieldType("field", false, true, "key",
-            false, Collections.emptyMap());
+        KeyedFlattenedFieldType unsearchable = new KeyedFlattenedFieldType("field", false, true, "key", false, Collections.emptyMap());
         IllegalArgumentException e = expectThrows(IllegalArgumentException.class,
             () -> unsearchable.termQuery("field", null));
-        assertEquals("Cannot search on field [field] since it is not indexed.", e.getMessage());
+        assertEquals("Cannot search on field [" + ft.name() + "] since it is not indexed.", e.getMessage());
     }
 
     public void testTermsQuery() {
         KeyedFlattenedFieldType ft = createFieldType();
 
-        Query expected = new TermInSetQuery("field",
+        Query expected = new TermInSetQuery(ft.name(),
             new BytesRef("key\0value1"),
             new BytesRef("key\0value2"));
 
@@ -81,17 +87,17 @@ public class KeyedFlattenedFieldTypeTests extends FieldTypeTestCase {
     public void testExistsQuery() {
         KeyedFlattenedFieldType ft = createFieldType();
 
-        Query expected = new PrefixQuery(new Term("field", "key\0"));
+        Query expected = new PrefixQuery(new Term(ft.name(), "key\0"));
         assertEquals(expected, ft.existsQuery(null));
     }
 
     public void testPrefixQuery() {
         KeyedFlattenedFieldType ft = createFieldType();
 
-        Query expected = new PrefixQuery(new Term("field", "key\0val"));
+        Query expected = new PrefixQuery(new Term(ft.name(), "key\0val"));
         assertEquals(expected, ft.prefixQuery("val", MultiTermQuery.CONSTANT_SCORE_REWRITE, false, MOCK_CONTEXT));
 
-        expected = AutomatonQueries.caseInsensitivePrefixQuery(new Term("field", "key\0vAl"));
+        expected = AutomatonQueries.caseInsensitivePrefixQuery(new Term(ft.name(), "key\0vAl"));
         assertEquals(expected, ft.prefixQuery("vAl", MultiTermQuery.CONSTANT_SCORE_REWRITE, true, MOCK_CONTEXT));
 
         ElasticsearchException ee = expectThrows(ElasticsearchException.class,
@@ -111,12 +117,12 @@ public class KeyedFlattenedFieldTypeTests extends FieldTypeTestCase {
     public void testRangeQuery() {
         KeyedFlattenedFieldType ft = createFieldType();
 
-        TermRangeQuery expected = new TermRangeQuery("field",
+        TermRangeQuery expected = new TermRangeQuery(ft.name(),
             new BytesRef("key\0lower"),
             new BytesRef("key\0upper"), false, false);
         assertEquals(expected, ft.rangeQuery("lower", "upper", false, false, MOCK_CONTEXT));
 
-        expected = new TermRangeQuery("field",
+        expected = new TermRangeQuery(ft.name(),
             new BytesRef("key\0lower"),
             new BytesRef("key\0upper"), true, true);
         assertEquals(expected, ft.rangeQuery("lower", "upper", true, true, MOCK_CONTEXT));
@@ -160,4 +166,23 @@ public class KeyedFlattenedFieldTypeTests extends FieldTypeTestCase {
         assertEquals(List.of(), fetchSourceValue(ft, sourceValue));
         assertEquals(List.of(), fetchSourceValue(ft, null));
     }
+
+    public void testFetchSourceValue() throws IOException {
+        KeyedFlattenedFieldType ft = createFieldType();
+        Map<String, Object> sourceValue = Map.of("key", "value");
+
+        SearchExecutionContext searchExecutionContext = mock(SearchExecutionContext.class);
+        when(searchExecutionContext.sourcePath("field.key")).thenReturn(Set.of("field.key"));
+
+        ValueFetcher fetcher = ft.valueFetcher(searchExecutionContext, null);
+        SourceLookup lookup = new SourceLookup();
+        lookup.setSource(Collections.singletonMap("field", sourceValue));
+
+        assertEquals(List.of("value"), fetcher.fetchValues(lookup));
+        lookup.setSource(Collections.singletonMap("field", null));
+        assertEquals(List.of(), fetcher.fetchValues(lookup));
+
+        IllegalArgumentException e = expectThrows(IllegalArgumentException.class, () -> ft.valueFetcher(searchExecutionContext, "format"));
+        assertEquals("Field [field.key] of type [flattened] doesn't support formats.", e.getMessage());
+    }
 }

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

@@ -676,6 +676,55 @@ public class FieldFetcherTests extends MapperServiceTestCase {
         assertEquals("value4b", eval("inner_nested.0.f4.0", obj1));
     }
 
+    @SuppressWarnings("unchecked")
+    public void testFlattenedField() throws IOException {
+        XContentBuilder mapping = mapping(b -> b.startObject("flat").field("type", "flattened").endObject());
+        MapperService mapperService = createMapperService(mapping);
+
+        XContentBuilder source = XContentFactory.jsonBuilder().startObject()
+            .startObject("flat")
+              .field("f1", "value1")
+              .field("f2", 1)
+            .endObject()
+          .endObject();
+
+        // requesting via wildcard should retrieve the root field as a structured map
+        Map<String, DocumentField> fields = fetchFields(mapperService, source, fieldAndFormatList("*", null, false));
+        assertEquals(1, fields.size());
+        assertThat(fields.keySet(), containsInAnyOrder("flat"));
+        Map<String, Object> flattenedValue = (Map<String, Object>) fields.get("flat").getValue();
+        assertThat(flattenedValue.keySet(), containsInAnyOrder("f1", "f2"));
+        assertEquals("value1", flattenedValue.get("f1"));
+        assertEquals(1, flattenedValue.get("f2"));
+
+        // direct retrieval of subfield is possible
+        List<FieldAndFormat> fieldAndFormatList = new ArrayList<>();
+        fieldAndFormatList.add(new FieldAndFormat("flat.f1", null));
+        fields = fetchFields(mapperService, source, fieldAndFormatList);
+        assertEquals(1, fields.size());
+        assertThat(fields.keySet(), containsInAnyOrder("flat.f1"));
+        assertThat(fields.get("flat.f1").getValue(), equalTo("value1"));
+
+        // direct retrieval of root field and subfield is possible
+        fieldAndFormatList.add(new FieldAndFormat("*", null));
+        fields = fetchFields(mapperService, source, fieldAndFormatList);
+        assertEquals(2, fields.size());
+        assertThat(fields.keySet(), containsInAnyOrder("flat", "flat.f1"));
+        flattenedValue = (Map<String, Object>) fields.get("flat").getValue();
+        assertThat(flattenedValue.keySet(), containsInAnyOrder("f1", "f2"));
+        assertEquals("value1", flattenedValue.get("f1"));
+        assertEquals(1, flattenedValue.get("f2"));
+        assertThat(fields.get("flat.f1").getValue(), equalTo("value1"));
+
+        // retrieval of subfield with wildcard is not possible
+        fields = fetchFields(mapperService, source, fieldAndFormatList("flat.f*", null, false));
+        assertEquals(0, fields.size());
+
+        // retrieval of non-existing subfield returns empty result
+        fields = fetchFields(mapperService, source, fieldAndFormatList("flat.baz", null, false));
+        assertEquals(0, fields.size());
+    }
+
     public void testUnmappedFieldsInsideObject() throws IOException {
         XContentBuilder mapping = XContentFactory.jsonBuilder().startObject()
             .startObject("_doc")

server/src/test/java/org/elasticsearch/search/lookup/LeafDocLookupTests.java

@@ -10,12 +10,12 @@ package org.elasticsearch.search.lookup;
 import org.elasticsearch.index.fielddata.IndexFieldData;
 import org.elasticsearch.index.fielddata.LeafFieldData;
 import org.elasticsearch.index.fielddata.ScriptDocValues;
+import org.elasticsearch.index.mapper.ContentPath;
 import org.elasticsearch.index.mapper.MappedFieldType;
 import org.elasticsearch.index.mapper.flattened.FlattenedFieldMapper;
 import org.elasticsearch.test.ESTestCase;
 import org.junit.Before;
 
-import java.util.Collections;
 import java.util.function.Function;
 
 import static org.mockito.AdditionalAnswers.returnsFirstArg;
@@ -28,6 +28,7 @@ public class LeafDocLookupTests extends ESTestCase {
     private ScriptDocValues<?> docValues;
     private LeafDocLookup docLookup;
 
+    @Override
     @Before
     public void setUp() throws Exception {
         super.setUp();
@@ -60,10 +61,9 @@ public class LeafDocLookupTests extends ESTestCase {
         ScriptDocValues<?> docValues2 = mock(ScriptDocValues.class);
         IndexFieldData<?> fieldData2 = createFieldData(docValues2);
 
-        FlattenedFieldMapper.KeyedFlattenedFieldType fieldType1
-            = new FlattenedFieldMapper.KeyedFlattenedFieldType("field", true, true, "key1", false, Collections.emptyMap());
-        FlattenedFieldMapper.KeyedFlattenedFieldType fieldType2
-            = new FlattenedFieldMapper.KeyedFlattenedFieldType( "field", true, true, "key2", false, Collections.emptyMap());
+        FlattenedFieldMapper fieldMapper = new FlattenedFieldMapper.Builder("field").build(new ContentPath(1));
+        FlattenedFieldMapper.KeyedFlattenedFieldType fieldType1 = fieldMapper.keyedFieldType("key1");
+        FlattenedFieldMapper.KeyedFlattenedFieldType fieldType2 = fieldMapper.keyedFieldType("key2");
 
         Function<MappedFieldType, IndexFieldData<?>> fieldDataSupplier = fieldType -> {
             FlattenedFieldMapper.KeyedFlattenedFieldType keyedFieldType = (FlattenedFieldMapper.KeyedFlattenedFieldType) fieldType;