Expose if a field is a metadata field in the field capabilities respo…

…nse (elastic#69977)

This change exposes for each field in the _field_caps response if the field is a metadata field.
This is needed for consumers of this API that want to filter these fields. Currently ML keeps a static list
and QL checks that the family type starts with `_`. In order to ease the addition of new metadata fields, this
change reworks the strategy in this solution and now only checks for the new flag.
Note that the new flag is also applied at the coordinator level in a best-effort to apply the logic on older nodes
in a mixed-version cluster.

client/rest-high-level/src/test/java/org/elasticsearch/client/SearchIT.java

@@ -1247,11 +1247,11 @@ public void testFieldCaps() throws IOException {
         assertEquals(2, ratingResponse.size());
 
         FieldCapabilities expectedKeywordCapabilities = new FieldCapabilities(
-            "rating", "keyword", true, true, new String[]{"index2"}, null, null, Collections.emptyMap());
+            "rating", "keyword", false, true, true, new String[]{"index2"}, null, null, Collections.emptyMap());
         assertEquals(expectedKeywordCapabilities, ratingResponse.get("keyword"));
 
         FieldCapabilities expectedLongCapabilities = new FieldCapabilities(
-            "rating", "long", true, true, new String[]{"index1"}, null, null, Collections.emptyMap());
+            "rating", "long", false, true, true, new String[]{"index1"}, null, null, Collections.emptyMap());
         assertEquals(expectedLongCapabilities, ratingResponse.get("long"));
 
         // Check the capabilities for the 'field' field.
@@ -1260,7 +1260,7 @@ public void testFieldCaps() throws IOException {
         assertEquals(1, fieldResponse.size());
 
         FieldCapabilities expectedTextCapabilities = new FieldCapabilities(
-            "field", "text", true, false, null, null, null, Collections.emptyMap());
+            "field", "text", false, true, false, null, null, null, Collections.emptyMap());
         assertEquals(expectedTextCapabilities, fieldResponse.get("text"));
     }
 

docs/reference/search/field-caps.asciidoc

@@ -106,6 +106,8 @@ described using a type family. For example, `keyword`, `constant_keyword` and `w
 field types are all described as the `keyword` type family.
 
 
+`metadata_field`::
+  Whether this field is registered as a <<mapping-fields,metadata field>>.
 
 `searchable`::
   Whether this field is indexed for search on all indices.
@@ -162,12 +164,14 @@ The API returns the following response:
   "fields": {
     "rating": {                                   <1>
       "long": {
+        "metadata_field": false,
         "searchable": true,
         "aggregatable": false,
         "indices": [ "index1", "index2" ],
         "non_aggregatable_indices": [ "index1" ]  <2>
       },
       "keyword": {
+        "metadata_field": false,
         "searchable": false,
         "aggregatable": true,
         "indices": [ "index3", "index4" ],
@@ -176,6 +180,7 @@ The API returns the following response:
     },
     "title": {                                    <4>
       "text": {
+        "metadata_field": false,
         "searchable": true,
         "aggregatable": false
 
@@ -211,30 +216,35 @@ in some indices but not all:
   "fields": {
     "rating": {
       "long": {
+        "metadata_field": false,
         "searchable": true,
         "aggregatable": false,
         "indices": [ "index1", "index2" ],
         "non_aggregatable_indices": [ "index1" ]
       },
       "keyword": {
+        "metadata_field": false,
         "searchable": false,
         "aggregatable": true,
         "indices": [ "index3", "index4" ],
         "non_searchable_indices": [ "index4" ]
       },
       "unmapped": {                               <1>
+        "metadata_field": false,
         "indices": [ "index5" ],
         "searchable": false,
         "aggregatable": false
       }
     },
     "title": {
       "text": {
+        "metadata_field": false,
         "indices": [ "index1", "index2", "index3", "index4" ],
         "searchable": true,
         "aggregatable": false
       },
       "unmapped": {                               <2>
+        "metadata_field": false,
         "indices": [ "index5" ],
         "searchable": false,
         "aggregatable": false

server/src/internalClusterTest/java/org/elasticsearch/search/fieldcaps/FieldCapabilitiesIT.java

@@ -13,12 +13,16 @@
 import org.elasticsearch.action.index.IndexRequestBuilder;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentFactory;
+import org.elasticsearch.index.mapper.KeywordFieldMapper;
+import org.elasticsearch.index.mapper.MetadataFieldMapper;
+import org.elasticsearch.index.mapper.ParseContext;
 import org.elasticsearch.index.query.QueryBuilders;
 import org.elasticsearch.plugins.MapperPlugin;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.test.ESIntegTestCase;
 import org.junit.Before;
 
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -86,16 +90,9 @@ public void setUp() throws Exception {
         assertAcked(client().admin().indices().prepareAliases().addAlias("new_index", "current"));
     }
 
-    public static class FieldFilterPlugin extends Plugin implements MapperPlugin {
-        @Override
-        public Function<String, Predicate<String>> getFieldFilter() {
-            return index -> field -> field.equals("playlist") == false;
-        }
-    }
-
     @Override
     protected Collection<Class<? extends Plugin>> nodePlugins() {
-        return Collections.singleton(FieldFilterPlugin.class);
+        return List.of(TestMapperPlugin.class);
     }
 
     public void testFieldAlias() {
@@ -112,13 +109,13 @@ public void testFieldAlias() {
 
         assertTrue(distance.containsKey("double"));
         assertEquals(
-            new FieldCapabilities("distance", "double", true, true, new String[] {"old_index"}, null, null,
+            new FieldCapabilities("distance", "double", false, true, true, new String[] {"old_index"}, null, null,
                     Collections.emptyMap()),
             distance.get("double"));
 
         assertTrue(distance.containsKey("text"));
         assertEquals(
-            new FieldCapabilities("distance", "text", true, false, new String[] {"new_index"}, null, null,
+            new FieldCapabilities("distance", "text", false, true, false, new String[] {"new_index"}, null, null,
                     Collections.emptyMap()),
             distance.get("text"));
 
@@ -128,7 +125,7 @@ public void testFieldAlias() {
 
         assertTrue(routeLength.containsKey("double"));
         assertEquals(
-            new FieldCapabilities("route_length_miles", "double", true, true, null, null, null, Collections.emptyMap()),
+            new FieldCapabilities("route_length_miles", "double", false, true, true, null, null, null, Collections.emptyMap()),
             routeLength.get("double"));
     }
 
@@ -169,13 +166,13 @@ public void testWithUnmapped() {
 
         assertTrue(oldField.containsKey("long"));
         assertEquals(
-            new FieldCapabilities("old_field", "long", true, true, new String[] {"old_index"}, null, null,
+            new FieldCapabilities("old_field", "long", false, true, true, new String[] {"old_index"}, null, null,
                     Collections.emptyMap()),
             oldField.get("long"));
 
         assertTrue(oldField.containsKey("unmapped"));
         assertEquals(
-            new FieldCapabilities("old_field", "unmapped", false, false, new String[] {"new_index"}, null, null,
+            new FieldCapabilities("old_field", "unmapped", false, false, false, new String[] {"new_index"}, null, null,
                     Collections.emptyMap()),
             oldField.get("unmapped"));
 
@@ -184,7 +181,7 @@ public void testWithUnmapped() {
 
         assertTrue(newField.containsKey("long"));
         assertEquals(
-            new FieldCapabilities("new_field", "long", true, true, null, null, null, Collections.emptyMap()),
+            new FieldCapabilities("new_field", "long", false, true, true, null, null, null, Collections.emptyMap()),
             newField.get("long"));
     }
 
@@ -235,10 +232,67 @@ public void testWithIndexFilter() throws InterruptedException {
         assertTrue(newField.containsKey("keyword"));
     }
 
+    public void testMetadataFields() {
+        for (int i = 0; i < 2; i++) {
+            String[] fields = i == 0 ? new String[] { "*" } : new String[] { "_id", "_test" };
+            FieldCapabilitiesResponse response = client()
+                .prepareFieldCaps()
+                .setFields(fields)
+                .get();
+
+            Map<String, FieldCapabilities> idField = response.getField("_id");
+            assertEquals(1, idField.size());
+
+            assertTrue(idField.containsKey("_id"));
+            assertEquals(
+                new FieldCapabilities("_id", "_id", true, true, false, null, null, null, Collections.emptyMap()),
+                idField.get("_id"));
+
+            Map<String, FieldCapabilities> testField = response.getField("_test");
+            assertEquals(1, testField.size());
+
+            assertTrue(testField.containsKey("keyword"));
+            assertEquals(
+                new FieldCapabilities("_test", "keyword", true, true, true, null, null, null, Collections.emptyMap()),
+                testField.get("keyword"));
+        }
+    }
+
     private void assertIndices(FieldCapabilitiesResponse response, String... indices) {
         assertNotNull(response.getIndices());
         Arrays.sort(indices);
         Arrays.sort(response.getIndices());
         assertArrayEquals(indices, response.getIndices());
     }
+
+    public static final class TestMapperPlugin extends Plugin implements MapperPlugin {
+        @Override
+        public Map<String, MetadataFieldMapper.TypeParser> getMetadataMappers() {
+            return Collections.singletonMap(TestMetadataMapper.CONTENT_TYPE, TestMetadataMapper.PARSER);
+        }
+
+        @Override
+        public Function<String, Predicate<String>> getFieldFilter() {
+            return index -> field -> field.equals("playlist") == false;
+        }
+    }
+
+    private static final class TestMetadataMapper extends MetadataFieldMapper {
+        private static final String CONTENT_TYPE = "_test";
+        private static final String FIELD_NAME = "_test";
+
+        protected TestMetadataMapper() {
+            super(new KeywordFieldMapper.KeywordFieldType(FIELD_NAME));
+        }
+
+        @Override
+        protected void parseCreateField(ParseContext context) throws IOException {}
+
+        @Override
+        protected String contentType() {
+            return CONTENT_TYPE;
+        }
+
+        private static final TypeParser PARSER = new FixedTypeParser(c -> new TestMetadataMapper());
+    }
 }