Support fetching flattened subfields

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

Author: Christoph Büscher

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

@@ -132,6 +132,10 @@ Set<String> sourcePaths(String field) {
         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

@@ -163,17 +163,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,
+        public KeyedFlattenedFieldType(String name, String rootName, boolean indexed, boolean hasDocValues, String key,
                                        boolean splitQueriesOnWhitespace, Map<String, String> meta) {
             super(name, 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 name, String rootName, String key, RootFlattenedFieldType ref) {
+            this(name, rootName, ref.isSearchable(), ref.hasDocValues(), key, ref.splitQueriesOnWhitespace, ref.meta());
         }
 
         @Override
@@ -259,8 +261,7 @@ 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 -> List.of();
+            return SourceValueFetcher.identity(rootName + "." + key, context, format);
         }
     }
 
@@ -441,7 +442,7 @@ public RootFlattenedFieldType fieldType() {
 
     @Override
     public KeyedFlattenedFieldType keyedFieldType(String key) {
-        return new KeyedFlattenedFieldType(keyedFieldName(), key, fieldType());
+        return new KeyedFlattenedFieldType(keyedFieldName(), name(), key, fieldType());
     }
 
     public String keyedFieldName() {

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

@@ -31,7 +31,7 @@
 public class KeyedFlattenedFieldTypeTests extends FieldTypeTestCase {
 
     private static KeyedFlattenedFieldType createFieldType() {
-        return new KeyedFlattenedFieldType("field", true, true, "key", false, Collections.emptyMap());
+        return new KeyedFlattenedFieldType("field", "field", true, true, "key", false, Collections.emptyMap());
     }
 
     public void testIndexedValueForSearch() {
@@ -56,7 +56,7 @@ public void testTermQuery() {
         expected = AutomatonQueries.caseInsensitiveTermQuery(new Term("field", "key\0value"));
         assertEquals(expected, ft.termQueryCaseInsensitive("value", null));
 
-        KeyedFlattenedFieldType unsearchable = new KeyedFlattenedFieldType("field", false, true, "key",
+        KeyedFlattenedFieldType unsearchable = new KeyedFlattenedFieldType("field", "field", false, true, "key",
             false, Collections.emptyMap());
         IllegalArgumentException e = expectThrows(IllegalArgumentException.class,
             () -> unsearchable.termQuery("field", null));

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

@@ -669,6 +669,64 @@ public void testNestedFields() throws IOException {
         assertEquals("value4b", eval("inner_nested.0.f4.0", obj1));
     }
 
+    @SuppressWarnings("unchecked")
+    public void testFlattenedField() throws IOException {
+        XContentBuilder mapping = XContentFactory.jsonBuilder().startObject()
+        .startObject("_doc")
+            .startObject("properties")
+              .startObject("flat")
+                .field("type", "flattened")
+              .endObject()
+            .endObject()
+        .endObject()
+        .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> flattendedValue = (Map<String, Object>) fields.get("flat").getValue();
+        assertThat(flattendedValue.keySet(), containsInAnyOrder("f1", "f2"));
+        assertEquals("value1", flattendedValue.get("f1"));
+        assertEquals(1, flattendedValue.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"));
+        flattendedValue = (Map<String, Object>) fields.get("flat").getValue();
+        assertThat(flattendedValue.keySet(), containsInAnyOrder("f1", "f2"));
+        assertEquals("value1", flattendedValue.get("f1"));
+        assertEquals(1, flattendedValue.get("f2"));
+        assertThat(fields.get("flat.f1").getValue(), equalTo("value1"));
+
+        // retrieval of subfield with widlcard 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

@@ -28,6 +28,7 @@ public class LeafDocLookupTests extends ESTestCase {
     private ScriptDocValues<?> docValues;
     private LeafDocLookup docLookup;
 
+    @Override
     @Before
     public void setUp() throws Exception {
         super.setUp();
@@ -61,9 +62,9 @@ public void testFlattenedField() {
         IndexFieldData<?> fieldData2 = createFieldData(docValues2);
 
         FlattenedFieldMapper.KeyedFlattenedFieldType fieldType1
-            = new FlattenedFieldMapper.KeyedFlattenedFieldType("field", true, true, "key1", false, Collections.emptyMap());
+            = new FlattenedFieldMapper.KeyedFlattenedFieldType("field", "field", true, true, "key1", false, Collections.emptyMap());
         FlattenedFieldMapper.KeyedFlattenedFieldType fieldType2
-            = new FlattenedFieldMapper.KeyedFlattenedFieldType( "field", true, true, "key2", false, Collections.emptyMap());
+            = new FlattenedFieldMapper.KeyedFlattenedFieldType( "field", "field", true, true, "key2", false, Collections.emptyMap());
 
         Function<MappedFieldType, IndexFieldData<?>> fieldDataSupplier = fieldType -> {
             FlattenedFieldMapper.KeyedFlattenedFieldType keyedFieldType = (FlattenedFieldMapper.KeyedFlattenedFieldType) fieldType;