Merge dynamic field type lookup into FieldTypeLookup

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

@@ -9,6 +9,7 @@
 package org.elasticsearch.index.mapper;
 
 import org.elasticsearch.common.regex.Regex;
+import org.elasticsearch.index.mapper.flattened.FlattenedFieldMapper;
 
 import java.util.Collection;
 import java.util.Collections;
@@ -31,23 +32,21 @@ final class FieldTypeLookup {
      * For convenience, the set of copied fields includes the field itself.
      */
     private final Map<String, Set<String>> fieldToCopiedFields = new HashMap<>();
-    private final DynamicKeyFieldTypeLookup dynamicKeyLookup;
+
+    private final int maxParentPathDots;
 
     FieldTypeLookup(
         Collection<FieldMapper> fieldMappers,
         Collection<FieldAliasMapper> fieldAliasMappers,
         Collection<RuntimeField> runtimeFields
     ) {
-        Map<String, DynamicKeyFieldMapper> dynamicKeyMappers = new HashMap<>();
 
+        int maxParentPathDots = 0;
         for (FieldMapper fieldMapper : fieldMappers) {
             String fieldName = fieldMapper.name();
             MappedFieldType fieldType = fieldMapper.fieldType();
             fullNameToFieldType.put(fieldType.name(), fieldType);
-            if (fieldMapper instanceof DynamicKeyFieldMapper) {
-                dynamicKeyMappers.put(fieldName, (DynamicKeyFieldMapper) fieldMapper);
-            }
-
+            maxParentPathDots = Math.max(maxParentPathDots, dotCount(fieldType.name()));
             for (String targetField : fieldMapper.copyTo().copyToFields()) {
                 Set<String> sourcePath = fieldToCopiedFields.get(targetField);
                 if (sourcePath == null) {
@@ -58,12 +57,11 @@ final class FieldTypeLookup {
                 fieldToCopiedFields.get(targetField).add(fieldName);
             }
         }
+        this.maxParentPathDots = maxParentPathDots;
 
-        final Map<String, String> aliasToConcreteName = new HashMap<>();
         for (FieldAliasMapper fieldAliasMapper : fieldAliasMappers) {
             String aliasName = fieldAliasMapper.name();
             String path = fieldAliasMapper.path();
-            aliasToConcreteName.put(aliasName, path);
             fullNameToFieldType.put(aliasName, fullNameToFieldType.get(path));
         }
 
@@ -72,8 +70,30 @@ final class FieldTypeLookup {
             //this will override concrete fields with runtime fields that have the same name
             fullNameToFieldType.put(runtimeFieldType.name(), runtimeFieldType);
         }
+    }
+
+    private static int dotCount(String path) {
+        int dotCount = 0;
+        for (int i = 0; i < path.length(); i++) {
+            if (path.charAt(i) == '.') {
+                dotCount++;
+            }
+        }
+        return dotCount;
+    }
 
-        this.dynamicKeyLookup = new DynamicKeyFieldTypeLookup(dynamicKeyMappers, aliasToConcreteName);
+    // for testing
+    String longestPossibleParent(String path) {
+        int dotCount = 0;
+        for (int i = 0; i < path.length(); i++) {
+            if (path.charAt(i) == '.') {
+                dotCount++;
+                if (dotCount > maxParentPathDots) {
+                    return path.substring(0, i);
+                }
+            }
+        }
+        return path;
     }
 
     /**
@@ -85,9 +105,20 @@ MappedFieldType get(String field) {
             return fieldType;
         }
 
-        // If the mapping contains fields that support dynamic sub-key lookup, check
-        // if this could correspond to a keyed field of the form 'path_to_field.path_to_key'.
-        return dynamicKeyLookup.get(field);
+        // Try parent fields instead!
+        String parentField = longestPossibleParent(field);
+        while (true) {
+            fieldType = fullNameToFieldType.get(parentField);
+            if (fieldType != null) {
+                return fieldType.childFieldType(field.substring(parentField.length() + 1));
+            }
+            if (parentField.contains(".") == false) {
+                break;
+            }
+            parentField = parentField.substring(0, parentField.lastIndexOf("."));
+        }
+
+        return null;
     }
 
     /**
@@ -130,7 +161,10 @@ Set<String> sourcePaths(String field) {
         if (fullNameToFieldType.isEmpty()) {
             return Set.of();
         }
-        if (dynamicKeyLookup.get(field) != null) {
+
+        // TODO there must be a nicer way of doing this...
+        MappedFieldType fieldType = get(field);
+        if (fieldType instanceof FlattenedFieldMapper.KeyedFlattenedFieldType) {
             return Set.of(field);
         }
 

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

@@ -422,4 +422,13 @@ public enum CollapseType {
         KEYWORD,
         NUMERIC
     }
+
+    /**
+     * Returns a MappedFieldType for a given child of this field type
+     * @param childPath the child field, which may contain dots
+     * @return a child MappedFieldType, or {@code null} if no such child exists
+     */
+    public MappedFieldType childFieldType(String childPath) {
+        return null;
+    }
 }

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

@@ -31,7 +31,6 @@
 import org.elasticsearch.index.fielddata.plain.AbstractLeafOrdinalsFieldData;
 import org.elasticsearch.index.fielddata.plain.SortedSetOrdinalsIndexFieldData;
 import org.elasticsearch.index.mapper.ContentPath;
-import org.elasticsearch.index.mapper.DynamicKeyFieldMapper;
 import org.elasticsearch.index.mapper.FieldMapper;
 import org.elasticsearch.index.mapper.MappedFieldType;
 import org.elasticsearch.index.mapper.Mapper;
@@ -83,7 +82,7 @@
  * "key\0some value" and "key2.key3\0true". Note that \0 is used as a reserved separator
  *  character (see {@link FlattenedFieldParser#SEPARATOR}).
  */
-public final class FlattenedFieldMapper extends DynamicKeyFieldMapper {
+public final class FlattenedFieldMapper extends FieldMapper {
 
     public static final String CONTENT_TYPE = "flattened";
     private static final String KEYED_FIELD_SUFFIX = "._keyed";
@@ -421,6 +420,11 @@ public IndexFieldData.Builder fielddataBuilder(String fullyQualifiedIndexName, S
         public ValueFetcher valueFetcher(SearchExecutionContext context, String format) {
             return SourceValueFetcher.identity(name(), context, format);
         }
+
+        @Override
+        public MappedFieldType childFieldType(String childPath) {
+            return new KeyedFlattenedFieldType(name(), childPath, this);
+        }
     }
 
     private final FlattenedFieldParser fieldParser;
@@ -429,7 +433,7 @@ public ValueFetcher valueFetcher(SearchExecutionContext context, String format)
     private FlattenedFieldMapper(String simpleName,
                                  MappedFieldType mappedFieldType,
                                  Builder builder) {
-        super(simpleName, mappedFieldType, Lucene.KEYWORD_ANALYZER, CopyTo.empty());
+        super(simpleName, mappedFieldType, Lucene.KEYWORD_ANALYZER, MultiFields.empty(), CopyTo.empty());
         this.builder = builder;
         this.fieldParser = new FlattenedFieldParser(mappedFieldType.name(), mappedFieldType.name() + KEYED_FIELD_SUFFIX,
             mappedFieldType, builder.depthLimit.get(), builder.ignoreAbove.get(), builder.nullValue.get());
@@ -453,11 +457,6 @@ public RootFlattenedFieldType fieldType() {
         return (RootFlattenedFieldType) super.fieldType();
     }
 
-    @Override
-    public KeyedFlattenedFieldType keyedFieldType(String key) {
-        return new KeyedFlattenedFieldType(name(), key, fieldType());
-    }
-
     @Override
     protected void parseCreateField(ParseContext context) throws IOException {
         if (context.parser().currentToken() == XContentParser.Token.VALUE_NULL) {