Merge dynamic field type lookup into FieldTypeLookup

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

@@ -0,0 +1,34 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.index.mapper;
+
+/**
+ * Defines a MappedFieldType that exposes dynamic child field types
+ *
+ * If the field is named 'my_field', then a user is able to search on
+ * the field in both of the following ways:
+ *  - Using the field name 'my_field', which will delegate to the field type
+ *    as usual.
+ *  - Using any sub-key, for example 'my_field.some_key'. In this case, the
+ *    search is delegated to {@link #getChildFieldType(String)}, with 'some_key'
+ *    passed as the argument. The field may create a new field type dynamically
+ *    in order to handle the search.
+ *
+ *  To prevent conflicts between these dynamic sub-keys and multi-fields, any
+ *  field mappers generating field types that implement this interface should
+ *  explicitly disallow multi-fields.
+ */
+public interface DynamicFieldType {
+
+    /**
+     * Returns a dynamic MappedFieldType for the given path
+     */
+    MappedFieldType getChildFieldType(String path);
+
+}

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

@@ -15,13 +15,15 @@
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Set;
 
 /**
  * An immutable container for looking up {@link MappedFieldType}s by their name.
  */
 final class FieldTypeLookup {
     private final Map<String, MappedFieldType> fullNameToFieldType = new HashMap<>();
+    private final Map<String, DynamicFieldType> dynamicFieldTypes = new HashMap<>();
 
     /**
      * A map from field name to all fields whose content has been copied into it
@@ -31,23 +33,22 @@ 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<>();
 
         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);
+            if (fieldType instanceof DynamicFieldType) {
+                dynamicFieldTypes.put(fieldType.name(), (DynamicFieldType) fieldType);
             }
-
             for (String targetField : fieldMapper.copyTo().copyToFields()) {
                 Set<String> sourcePath = fieldToCopiedFields.get(targetField);
                 if (sourcePath == null) {
@@ -59,21 +60,51 @@ final class FieldTypeLookup {
             }
         }
 
-        final Map<String, String> aliasToConcreteName = new HashMap<>();
+        int maxParentPathDots = 0;
+        for (String dynamicRoot : dynamicFieldTypes.keySet()) {
+            maxParentPathDots = Math.max(maxParentPathDots, dotCount(dynamicRoot));
+        }
+        this.maxParentPathDots = maxParentPathDots;
+
         for (FieldAliasMapper fieldAliasMapper : fieldAliasMappers) {
             String aliasName = fieldAliasMapper.name();
             String path = fieldAliasMapper.path();
-            aliasToConcreteName.put(aliasName, path);
-            fullNameToFieldType.put(aliasName, fullNameToFieldType.get(path));
+            MappedFieldType fieldType = fullNameToFieldType.get(path);
+            fullNameToFieldType.put(aliasName, fieldType);
+            if (fieldType instanceof DynamicFieldType) {
+                dynamicFieldTypes.put(aliasName, (DynamicFieldType) fieldType);
+            }
         }
 
         for (RuntimeField runtimeField : runtimeFields) {
             MappedFieldType runtimeFieldType = runtimeField.asMappedFieldType();
             //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;
     }
 
     /**
@@ -84,10 +115,37 @@ MappedFieldType get(String field) {
         if (fieldType != null) {
             return fieldType;
         }
+        return getDynamicField(field);
+    }
+
+    // Check if the given field corresponds to a dynamic key mapper of the
+    // form 'path_to_field.path_to_key'. If so, returns a field type that
+    // can be used to perform searches on this field. Otherwise returns null.
+    private MappedFieldType getDynamicField(String field) {
+        if (dynamicFieldTypes.isEmpty()) {
+            // no parent fields defined
+            return null;
+        }
+        int dotIndex = -1;
+        int fieldDepth = -1;
+
+        while (true) {
+            if (++fieldDepth > maxParentPathDots) {
+                return null;
+            }
 
-        // 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);
+            dotIndex = field.indexOf('.', dotIndex + 1);
+            if (dotIndex < 0) {
+                return null;
+            }
+
+            String parentField = field.substring(0, dotIndex);
+            DynamicFieldType dft = dynamicFieldTypes.get(parentField);
+            if (dft != null && Objects.equals(field, parentField) == false) {
+                String key = field.substring(dotIndex + 1);
+                return dft.getChildFieldType(key);
+            }
+        }
     }
 
     /**
@@ -130,7 +188,10 @@ Set<String> sourcePaths(String field) {
         if (fullNameToFieldType.isEmpty()) {
             return Set.of();
         }
-        if (dynamicKeyLookup.get(field) != null) {
+
+        // If the field is dynamically generated then return its full path
+        MappedFieldType fieldType = getDynamicField(field);
+        if (fieldType != null) {
             return Set.of(field);
         }