Add support for querying JSON fields based on key. (#34621)

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

@@ -1334,3 +1334,59 @@ setup:
                 field: text
 
   - match: {hits.total: 3}
+
+---
+"Test exists query on JSON field":
+  - skip:
+      version: " - 6.99.99"
+      reason: "JSON fields are currently only implemented in 7.0."
+
+  - do:
+      indices.create:
+        index:  json_test
+        body:
+          mappings:
+            _doc:
+              dynamic: false
+              properties:
+                json:
+                  type: json
+  - do:
+      index:
+        index:  json_test
+        type:   _doc
+        id:     1
+        body:
+          json:
+            key: some_value
+        refresh: true
+
+  - do:
+      search:
+        index: json_test
+        body:
+          query:
+            exists:
+              field: json
+
+  - match: { hits.total: 1 }
+
+  - do:
+      search:
+        index: json_test
+        body:
+          query:
+            exists:
+              field: json.key
+
+  - match: { hits.total: 1 }
+
+  - do:
+      search:
+        index: json_test
+        body:
+          query:
+            exists:
+              field: json.nonexistent_key
+
+  - match: { hits.total: 0 }

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

@@ -61,3 +61,65 @@
         lenient: true
 
   - match: {hits.total: 0}
+
+---
+"search on JSON field":
+  - skip:
+      version: " - 6.99.99"
+      reason: "JSON fields are currently only implemented in 7.0."
+
+  - do:
+      indices.create:
+        index:  test
+        body:
+          mappings:
+            _doc:
+              properties:
+                headers:
+                  type: json
+
+  - do:
+      index:
+        index:  test
+        type:   _doc
+        id:     1
+        body:
+          headers:
+            content-type: application/javascript
+            origin: elastic.co
+        refresh: true
+
+  - do:
+      index:
+        index:  test
+        type:   _doc
+        id:     2
+        body:
+          headers:
+            content-type: text/plain
+            origin: elastic.co
+        refresh: true
+
+  - do:
+      search:
+        index: test
+        body:
+          query:
+            query_string:
+              query: "headers:text\\/plain"
+
+  - match: { hits.total: 1 }
+  - length: { hits.hits: 1 }
+  - match: { hits.hits.0._id: "2" }
+
+  - do:
+      search:
+        index: test
+        body:
+          query:
+            query_string:
+              query: "application\\/javascript AND headers.origin:elastic.co"
+
+  - match: { hits.total: 1 }
+  - length: { hits.hits: 1 }
+  - match: { hits.hits.0._id: "1" }

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

@@ -35,16 +35,20 @@ class FieldTypeLookup implements Iterable<MappedFieldType> {
 
     final CopyOnWriteHashMap<String, MappedFieldType> fullNameToFieldType;
     private final CopyOnWriteHashMap<String, String> aliasToConcreteName;
+    private final CopyOnWriteHashMap<String, JsonFieldMapper> fullNameToJsonMapper;
 
     FieldTypeLookup() {
         fullNameToFieldType = new CopyOnWriteHashMap<>();
         aliasToConcreteName = new CopyOnWriteHashMap<>();
+        fullNameToJsonMapper = new CopyOnWriteHashMap<>();
     }
 
     private FieldTypeLookup(CopyOnWriteHashMap<String, MappedFieldType> fullNameToFieldType,
-                            CopyOnWriteHashMap<String, String> aliasToConcreteName) {
+                            CopyOnWriteHashMap<String, String> aliasToConcreteName,
+                            CopyOnWriteHashMap<String, JsonFieldMapper> fullNameToJsonMapper) {
         this.fullNameToFieldType = fullNameToFieldType;
         this.aliasToConcreteName = aliasToConcreteName;
+        this.fullNameToJsonMapper = fullNameToJsonMapper;
     }
 
     /**
@@ -63,6 +67,7 @@ public FieldTypeLookup copyAndAddAll(String type,
 
         CopyOnWriteHashMap<String, MappedFieldType> fullName = this.fullNameToFieldType;
         CopyOnWriteHashMap<String, String> aliases = this.aliasToConcreteName;
+        CopyOnWriteHashMap<String, JsonFieldMapper> jsonMappers = this.fullNameToJsonMapper;
 
         for (FieldMapper fieldMapper : fieldMappers) {
             MappedFieldType fieldType = fieldMapper.fieldType();
@@ -71,6 +76,10 @@ public FieldTypeLookup copyAndAddAll(String type,
             if (!Objects.equals(fieldType, fullNameFieldType)) {
                 fullName = fullName.copyAndPut(fieldType.name(), fieldType);
             }
+
+            if (fieldMapper instanceof JsonFieldMapper) {
+                jsonMappers = fullNameToJsonMapper.copyAndPut(fieldType.name(), (JsonFieldMapper) fieldMapper);
+            }
         }
 
         for (FieldAliasMapper fieldAliasMapper : fieldAliasMappers) {
@@ -79,14 +88,42 @@ public FieldTypeLookup copyAndAddAll(String type,
             aliases = aliases.copyAndPut(aliasName, path);
         }
 
-        return new FieldTypeLookup(fullName, aliases);
+        return new FieldTypeLookup(fullName, aliases, jsonMappers);
     }
 
 
-    /** Returns the field for the given field */
+    /**
+     * Returns the mapped field type for the given field name.
+     */
     public MappedFieldType get(String field) {
         String concreteField = aliasToConcreteName.getOrDefault(field, field);
-        return fullNameToFieldType.get(concreteField);
+        MappedFieldType fieldType = fullNameToFieldType.get(concreteField);
+        if (fieldType != null) {
+            return fieldType;
+        }
+
+        // If the mapping contains JSON fields, check if this could correspond
+        // to a keyed JSON field of the form 'path_to_json_field.path_to_key'.
+        return !fullNameToJsonMapper.isEmpty() ? getKeyedJsonField(field) : null;
+    }
+
+    private MappedFieldType getKeyedJsonField(String field) {
+        int dotIndex = -1;
+        while (true) {
+            dotIndex = field.indexOf('.', dotIndex + 1);
+            if (dotIndex < 0) {
+                return null;
+            }
+
+            String parentField = field.substring(0, dotIndex);
+            String concreteField = aliasToConcreteName.getOrDefault(parentField, parentField);
+            JsonFieldMapper mapper = fullNameToJsonMapper.get(concreteField);
+
+            if (mapper != null) {
+                String key = field.substring(dotIndex + 1);
+                return mapper.keyedFieldType(key);
+            }
+        }
     }
 
     /**