Prevent slow field lookups when JSON fields are present. (elastic#39872)

We now track the maximum depth of any JSON field, which allows the JSON field
lookup to be short-circuited as soon as that depth is reached. This helps
prevent slow lookups when the user is searching over a very deep field that is
not in the mappings.

rest-api-spec/src/main/resources/rest-api-spec/test/search/160_exists_query.yml

@@ -1338,8 +1338,8 @@ setup:
 ---
 "Test exists query on JSON field":
   - skip:
-      version: " - 6.99.99"
-      reason: "JSON fields are currently only implemented in 7.0."
+      version: " - 7.99.99"
+      reason: "JSON fields are currently only implemented in 8.0."
 
   - do:
       indices.create:

rest-api-spec/src/main/resources/rest-api-spec/test/search/60_query_string.yml

@@ -65,8 +65,8 @@
 ---
 "search on JSON field":
   - skip:
-      version: " - 6.99.99"
-      reason: "JSON fields are currently only implemented in 7.0."
+      version: " - 7.99.99"
+      reason: "JSON fields are currently only implemented in 8.0."
 
   - do:
       indices.create:

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

@@ -25,6 +25,7 @@
 import java.util.Collection;
 import java.util.HashSet;
 import java.util.Iterator;
+import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
 
@@ -35,20 +36,25 @@ class FieldTypeLookup implements Iterable<MappedFieldType> {
 
     final CopyOnWriteHashMap<String, MappedFieldType> fullNameToFieldType;
     private final CopyOnWriteHashMap<String, String> aliasToConcreteName;
+
     private final CopyOnWriteHashMap<String, JsonFieldMapper> fullNameToJsonMapper;
+    private final int maxJsonFieldDepth;
 
     FieldTypeLookup() {
         fullNameToFieldType = new CopyOnWriteHashMap<>();
         aliasToConcreteName = new CopyOnWriteHashMap<>();
         fullNameToJsonMapper = new CopyOnWriteHashMap<>();
+        maxJsonFieldDepth = 0;
     }
 
     private FieldTypeLookup(CopyOnWriteHashMap<String, MappedFieldType> fullNameToFieldType,
                             CopyOnWriteHashMap<String, String> aliasToConcreteName,
-                            CopyOnWriteHashMap<String, JsonFieldMapper> fullNameToJsonMapper) {
+                            CopyOnWriteHashMap<String, JsonFieldMapper> fullNameToJsonMapper,
+                            int maxJsonFieldDepth) {
         this.fullNameToFieldType = fullNameToFieldType;
         this.aliasToConcreteName = aliasToConcreteName;
         this.fullNameToJsonMapper = fullNameToJsonMapper;
+        this.maxJsonFieldDepth = maxJsonFieldDepth;
     }
 
     /**
@@ -70,6 +76,7 @@ public FieldTypeLookup copyAndAddAll(String type,
         CopyOnWriteHashMap<String, JsonFieldMapper> jsonMappers = this.fullNameToJsonMapper;
 
         for (FieldMapper fieldMapper : fieldMappers) {
+            String fieldName = fieldMapper.name();
             MappedFieldType fieldType = fieldMapper.fieldType();
             MappedFieldType fullNameFieldType = fullName.get(fieldType.name());
 
@@ -78,7 +85,7 @@ public FieldTypeLookup copyAndAddAll(String type,
             }
 
             if (fieldMapper instanceof JsonFieldMapper) {
-                jsonMappers = fullNameToJsonMapper.copyAndPut(fieldType.name(), (JsonFieldMapper) fieldMapper);
+                jsonMappers = fullNameToJsonMapper.copyAndPut(fieldName, (JsonFieldMapper) fieldMapper);
             }
         }
 
@@ -92,7 +99,43 @@ public FieldTypeLookup copyAndAddAll(String type,
             }
         }
 
-        return new FieldTypeLookup(fullName, aliases, jsonMappers);
+        int maxFieldDepth = getMaxJsonFieldDepth(aliases, jsonMappers);
+
+        return new FieldTypeLookup(fullName, aliases, jsonMappers, maxFieldDepth);
+    }
+
+    private static int getMaxJsonFieldDepth(CopyOnWriteHashMap<String, String> aliases,
+                                            CopyOnWriteHashMap<String, JsonFieldMapper> jsonMappers) {
+        int maxFieldDepth = 0;
+        for (Map.Entry<String, String> entry : aliases.entrySet()) {
+            String aliasName = entry.getKey();
+            String path = entry.getValue();
+            if (jsonMappers.containsKey(path)) {
+                maxFieldDepth = Math.max(maxFieldDepth, fieldDepth(aliasName));
+            }
+        }
+
+        for (String fieldName : jsonMappers.keySet()) {
+            if (jsonMappers.containsKey(fieldName)) {
+                maxFieldDepth = Math.max(maxFieldDepth, fieldDepth(fieldName));
+            }
+        }
+
+        return maxFieldDepth;
+    }
+
+    /**
+     * Computes the total depth of this field by counting the number of parent fields
+     * in its path. As an example, the field 'parent1.parent2.field' has depth 3.
+     */
+    private static int fieldDepth(String field) {
+        int numDots = 0;
+        for (int i = 0; i < field.length(); ++i) {
+            if (field.charAt(i) == '.') {
+                numDots++;
+            }
+        }
+        return numDots + 1;
     }
 
 
@@ -111,9 +154,20 @@ public MappedFieldType get(String field) {
         return !fullNameToJsonMapper.isEmpty() ? getKeyedJsonField(field) : null;
     }
 
+    /**
+     * Check if the given field corresponds to a keyed JSON field of the form
+     * 'path_to_json_field.path_to_key'. If so, returns a field type that can
+     * be used to perform searches on this field.
+     */
     private MappedFieldType getKeyedJsonField(String field) {
         int dotIndex = -1;
+        int fieldDepth = 0;
+
         while (true) {
+            if (++fieldDepth > maxJsonFieldDepth) {
+                return null;
+            }
+
             dotIndex = field.indexOf('.', dotIndex + 1);
             if (dotIndex < 0) {
                 return null;
@@ -152,4 +206,9 @@ public Collection<String> simpleMatchToFullName(String pattern) {
     public Iterator<MappedFieldType> iterator() {
         return fullNameToFieldType.values().iterator();
     }
+
+    // Visible for testing.
+    int maxJsonFieldDepth() {
+        return maxJsonFieldDepth;
+    }
 }

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

@@ -186,6 +186,36 @@ public void testJsonFieldTypeWithAlias() {
         assertEquals(objectKey, keyedFieldType.key());
     }
 
+    public void testMaxJsonFieldDepth() {
+        FieldTypeLookup lookup = new FieldTypeLookup();
+        assertEquals(0, lookup.maxJsonFieldDepth());
+
+        // Add a JSON field.
+        String jsonFieldName = "object1.object2.field";
+        JsonFieldMapper jsonField = createJsonMapper(jsonFieldName);
+        lookup = lookup.copyAndAddAll("type", newList(jsonField), emptyList());
+        assertEquals(3, lookup.maxJsonFieldDepth());
+
+        // Add a short alias to that field.
+        String aliasName = "alias";
+        FieldAliasMapper alias = new FieldAliasMapper(aliasName, aliasName, jsonFieldName);
+        lookup = lookup.copyAndAddAll("type", emptyList(), newList(alias));
+        assertEquals(3, lookup.maxJsonFieldDepth());
+
+        // Add a longer alias to that field.
+        String longAliasName = "object1.object2.object3.alias";
+        FieldAliasMapper longAlias = new FieldAliasMapper(longAliasName, longAliasName, jsonFieldName);
+        lookup = lookup.copyAndAddAll("type", emptyList(), newList(longAlias));
+        assertEquals(4, lookup.maxJsonFieldDepth());
+
+        // Update the long alias to refer to a non-JSON field.
+        String fieldName = "field";
+        MockFieldMapper field = new MockFieldMapper(fieldName);
+        longAlias = new FieldAliasMapper(longAliasName, longAliasName, fieldName);
+        lookup = lookup.copyAndAddAll("type", newList(field), newList(longAlias));
+        assertEquals(3, lookup.maxJsonFieldDepth());
+    }
+
     private JsonFieldMapper createJsonMapper(String fieldName) {
         Settings settings = Settings.builder()
             .put("index.version.created", Version.CURRENT)