Support fetching flattened subfields (elastic#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 elastic#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

@@ -138,6 +138,10 @@ Set<String> sourcePaths(String field) {
         if (fullNameToFieldType.isEmpty()) {
             return org.elasticsearch.common.collect.Set.of();
         }
+        if (dynamicKeyLookup.get(field) != null) {
+            return Collections.singleton(field);
+        }
+
         String resolvedField = field;
         int lastDotIndex = field.lastIndexOf('.');
         if (lastDotIndex > 0) {

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

@@ -55,7 +55,6 @@
 
 import java.io.IOException;
 import java.util.Arrays;
-import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.function.Supplier;
@@ -167,17 +166,19 @@ public FlattenedFieldMapper build(ContentPath contentPath) {
      */
     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
@@ -267,8 +268,11 @@ public IndexFieldData.Builder fielddataBuilder(String fullyQualifiedIndexName, S
 
         @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 -> Collections.emptyList();
+            if (format != null) {
+                throw new IllegalArgumentException("Field [" + rootName + "." + key + "] of type [" + typeName() +
+                    "] doesn't support formats.");
+            }
+            return SourceValueFetcher.identity(rootName + "." + key, context, format);
         }
     }
 
@@ -432,7 +436,7 @@ private FlattenedFieldMapper(String simpleName,
                                  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());
     }
 
@@ -456,11 +460,7 @@ public RootFlattenedFieldType fieldType() {
 
     @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

@@ -198,7 +198,7 @@ public void testFlattenedLookup() {
         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;
@@ -219,7 +219,7 @@ public void testFlattenedLookupWithAlias() {
         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 void testGlobalFieldDataCaching() throws IOException {
             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 void onCache(ShardId shardId, String fieldName, Accountable ramUsage) {
             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,14 +20,20 @@
 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 static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
 public class KeyedFlattenedFieldTypeTests extends FieldTypeTestCase {
 
     private static KeyedFlattenedFieldType createFieldType() {
@@ -50,23 +56,22 @@ public void testIndexedValueForSearch() {
     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 +86,17 @@ public void testTermsQuery() {
     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 +116,12 @@ public void testFuzzyQuery() {
     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 +165,23 @@ public void testFetchIsEmpty() throws IOException {
         assertEquals(Collections.emptyList(), fetchSourceValue(ft, sourceValue));
         assertEquals(Collections.emptyList(), fetchSourceValue(ft, null));
     }
+
+    public void testFetchSourceValue() throws IOException {
+        KeyedFlattenedFieldType ft = createFieldType();
+        Map<String, Object> sourceValue = Collections.singletonMap("key", "value");
+
+        SearchExecutionContext searchExecutionContext = mock(SearchExecutionContext.class);
+        when(searchExecutionContext.sourcePath("field.key")).thenReturn(Collections.singleton("field.key"));
+
+        ValueFetcher fetcher = ft.valueFetcher(searchExecutionContext, null);
+        SourceLookup lookup = new SourceLookup();
+        lookup.setSource(Collections.singletonMap("field", sourceValue));
+
+        assertEquals(Collections.singletonList("value"), fetcher.fetchValues(lookup));
+        lookup.setSource(Collections.singletonMap("field", null));
+        assertEquals(Collections.emptyList(), 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());
+    }
 }