Flag in _field_caps to return only fields with values in index (#103651)

We are adding a query parameter to the field_caps api in order to filter out 
fields with no values. The parameter is called `include_empty_fields`  and 
defaults to true, and if set to false it will filter out from the field_caps 
response all the fields that has no value in the index.
We keep track of FieldInfos during refresh in order to know which field has 
value in an index. We added also a system property 
`es.field_caps_empty_fields_filter` in order to disable this feature if needed.

---------

Co-authored-by: Matthias Wilhelm <ankertal@gmail.com>

docs/changelog/103651.yaml

@@ -0,0 +1,5 @@
+pr: 103651
+summary: Flag in `_field_caps` to return only fields with values in index
+area: Search
+type: enhancement
+issues: []

docs/reference/search/field-caps.asciidoc

@@ -77,6 +77,10 @@ include::{es-repo-dir}/rest-api/common-parms.asciidoc[tag=index-ignore-unavailab
   (Optional, Boolean) If `true`, unmapped fields that are mapped in one index but not in another are included in the response. Fields that don't have any mapping are never included.
   Defaults to `false`.
 
+`include_empty_fields`::
+  (Optional, Boolean) If `false`, fields that never had a value in any shards are not included in the response. Fields that are not empty are always included. This flag does not consider deletions and updates.  If a field was non-empty and all the documents containing that field were deleted or the field was removed by updates,  it will still be returned even if the flag is `false`.
+  Defaults to `true`.
+
 `filters`::
 (Optional, string) Comma-separated list of filters to apply to the response.
 +

modules/mapper-extras/src/main/java/org/elasticsearch/index/mapper/extras/RankFeatureFieldMapper.java

@@ -9,6 +9,8 @@
 package org.elasticsearch.index.mapper.extras;
 
 import org.apache.lucene.document.FeatureField;
+import org.apache.lucene.index.FieldInfo;
+import org.apache.lucene.index.FieldInfos;
 import org.apache.lucene.index.Term;
 import org.apache.lucene.search.Query;
 import org.apache.lucene.search.TermQuery;
@@ -38,6 +40,7 @@ import java.util.Set;
  */
 public class RankFeatureFieldMapper extends FieldMapper {
 
+    public static final String NAME = "_feature";
     public static final String CONTENT_TYPE = "rank_feature";
 
     private static RankFeatureFieldType ft(FieldMapper in) {
@@ -128,7 +131,17 @@ public class RankFeatureFieldMapper extends FieldMapper {
 
         @Override
         public Query existsQuery(SearchExecutionContext context) {
-            return new TermQuery(new Term("_feature", name()));
+            return new TermQuery(new Term(NAME, name()));
+        }
+
+        @Override
+        public boolean fieldHasValue(FieldInfos fieldInfos) {
+            for (FieldInfo fieldInfo : fieldInfos) {
+                if (fieldInfo.getName().equals(NAME)) {
+                    return true;
+                }
+            }
+            return false;
         }
 
         @Override
@@ -208,7 +221,7 @@ public class RankFeatureFieldMapper extends FieldMapper {
             value = 1 / value;
         }
 
-        context.doc().addWithKey(name(), new FeatureField("_feature", name(), value));
+        context.doc().addWithKey(name(), new FeatureField(NAME, name(), value));
     }
 
     private static Float objectToFloat(Object value) {

modules/mapper-extras/src/main/java/org/elasticsearch/index/mapper/extras/RankFeatureMetaFieldMapper.java

@@ -8,6 +8,8 @@
 
 package org.elasticsearch.index.mapper.extras;
 
+import org.apache.lucene.index.FieldInfo;
+import org.apache.lucene.index.FieldInfos;
 import org.apache.lucene.search.Query;
 import org.elasticsearch.index.mapper.MappedFieldType;
 import org.elasticsearch.index.mapper.MetadataFieldMapper;
@@ -35,7 +37,8 @@ public class RankFeatureMetaFieldMapper extends MetadataFieldMapper {
 
         public static final RankFeatureMetaFieldType INSTANCE = new RankFeatureMetaFieldType();
 
-        private RankFeatureMetaFieldType() {
+        // made visible for tests
+        RankFeatureMetaFieldType() {
             super(NAME, false, false, false, TextSearchInfo.NONE, Collections.emptyMap());
         }
 
@@ -54,6 +57,16 @@ public class RankFeatureMetaFieldMapper extends MetadataFieldMapper {
             throw new UnsupportedOperationException("Cannot run exists query on [_feature]");
         }
 
+        @Override
+        public boolean fieldHasValue(FieldInfos fieldInfos) {
+            for (FieldInfo fieldInfo : fieldInfos) {
+                if (fieldInfo.getName().equals(NAME)) {
+                    return true;
+                }
+            }
+            return false;
+        }
+
         @Override
         public Query termQuery(Object value, SearchExecutionContext context) {
             throw new UnsupportedOperationException("The [_feature] field may not be queried directly");

modules/mapper-extras/src/test/java/org/elasticsearch/index/mapper/extras/FieldCapsRankFeatureTests.java

@@ -0,0 +1,114 @@
+/*
+ * 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.extras;
+
+import org.elasticsearch.action.fieldcaps.FieldCapabilities;
+import org.elasticsearch.action.fieldcaps.FieldCapabilitiesResponse;
+import org.elasticsearch.action.support.ActiveShardCount;
+import org.elasticsearch.plugins.Plugin;
+import org.elasticsearch.test.ESIntegTestCase;
+import org.hamcrest.Matchers;
+import org.junit.Before;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Map;
+
+import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
+
+@ESIntegTestCase.ClusterScope(scope = ESIntegTestCase.Scope.TEST)
+public class FieldCapsRankFeatureTests extends ESIntegTestCase {
+    private final String INDEX = "index-1";
+
+    @Override
+    protected Collection<Class<? extends Plugin>> nodePlugins() {
+        var plugins = new ArrayList<>(super.nodePlugins());
+        plugins.add(MapperExtrasPlugin.class);
+        return plugins;
+    }
+
+    @Before
+    public void setUpIndices() {
+        assertAcked(
+            prepareCreate(INDEX).setWaitForActiveShards(ActiveShardCount.ALL)
+                .setSettings(indexSettings())
+                .setMapping("fooRank", "type=rank_feature", "barRank", "type=rank_feature")
+        );
+    }
+
+    public void testRankFeatureInIndex() {
+        FieldCapabilitiesResponse response = client().prepareFieldCaps(INDEX).setFields("*").setincludeEmptyFields(false).get();
+        assertFalse(response.get().containsKey("fooRank"));
+        assertFalse(response.get().containsKey("barRank"));
+        prepareIndex(INDEX).setSource("fooRank", 8).setSource("barRank", 8).get();
+        refresh(INDEX);
+
+        response = client().prepareFieldCaps(INDEX).setFields("*").setincludeEmptyFields(false).get();
+        assertEquals(1, response.getIndices().length);
+        assertEquals(response.getIndices()[0], INDEX);
+        assertThat(response.get(), Matchers.hasKey("fooRank"));
+        // Check the capabilities for the 'fooRank' field.
+        Map<String, FieldCapabilities> fooRankField = response.getField("fooRank");
+        assertEquals(1, fooRankField.size());
+        assertThat(fooRankField, Matchers.hasKey("rank_feature"));
+        assertEquals(
+            new FieldCapabilities("fooRank", "rank_feature", false, true, false, null, null, null, Collections.emptyMap()),
+            fooRankField.get("rank_feature")
+        );
+    }
+
+    public void testRankFeatureInIndexAfterRestart() throws Exception {
+        prepareIndex(INDEX).setSource("fooRank", 8).get();
+        internalCluster().fullRestart();
+        ensureGreen(INDEX);
+
+        FieldCapabilitiesResponse response = client().prepareFieldCaps(INDEX).setFields("*").setincludeEmptyFields(false).get();
+
+        assertEquals(1, response.getIndices().length);
+        assertEquals(response.getIndices()[0], INDEX);
+        assertThat(response.get(), Matchers.hasKey("fooRank"));
+        // Check the capabilities for the 'fooRank' field.
+        Map<String, FieldCapabilities> fooRankField = response.getField("fooRank");
+        assertEquals(1, fooRankField.size());
+        assertThat(fooRankField, Matchers.hasKey("rank_feature"));
+        assertEquals(
+            new FieldCapabilities("fooRank", "rank_feature", false, true, false, null, null, null, Collections.emptyMap()),
+            fooRankField.get("rank_feature")
+        );
+    }
+
+    public void testAllRankFeatureReturnedIfOneIsPresent() {
+        prepareIndex(INDEX).setSource("fooRank", 8).get();
+        refresh(INDEX);
+
+        FieldCapabilitiesResponse response = client().prepareFieldCaps(INDEX).setFields("*").setincludeEmptyFields(false).get();
+
+        assertEquals(1, response.getIndices().length);
+        assertEquals(response.getIndices()[0], INDEX);
+        assertThat(response.get(), Matchers.hasKey("fooRank"));
+        // Check the capabilities for the 'fooRank' field.
+        Map<String, FieldCapabilities> fooRankField = response.getField("fooRank");
+        assertEquals(1, fooRankField.size());
+        assertThat(fooRankField, Matchers.hasKey("rank_feature"));
+        assertEquals(
+            new FieldCapabilities("fooRank", "rank_feature", false, true, false, null, null, null, Collections.emptyMap()),
+            fooRankField.get("rank_feature")
+        );
+        assertThat(response.get(), Matchers.hasKey("barRank"));
+        // Check the capabilities for the 'barRank' field.
+        Map<String, FieldCapabilities> barRankField = response.getField("barRank");
+        assertEquals(1, barRankField.size());
+        assertThat(barRankField, Matchers.hasKey("rank_feature"));
+        assertEquals(
+            new FieldCapabilities("barRank", "rank_feature", false, true, false, null, null, null, Collections.emptyMap()),
+            barRankField.get("rank_feature")
+        );
+    }
+}

modules/mapper-extras/src/test/java/org/elasticsearch/index/mapper/extras/RankFeatureFieldTypeTests.java

@@ -8,6 +8,8 @@
 
 package org.elasticsearch.index.mapper.extras;
 
+import org.apache.lucene.index.FieldInfo;
+import org.apache.lucene.index.FieldInfos;
 import org.elasticsearch.index.mapper.FieldTypeTestCase;
 import org.elasticsearch.index.mapper.MappedFieldType;
 import org.elasticsearch.index.mapper.MapperBuilderContext;
@@ -19,7 +21,7 @@ import java.util.List;
 public class RankFeatureFieldTypeTests extends FieldTypeTestCase {
 
     public void testIsNotAggregatable() {
-        MappedFieldType fieldType = new RankFeatureFieldMapper.RankFeatureFieldType("field", Collections.emptyMap(), true, null);
+        MappedFieldType fieldType = getMappedFieldType();
         assertFalse(fieldType.isAggregatable());
     }
 
@@ -32,4 +34,28 @@ public class RankFeatureFieldTypeTests extends FieldTypeTestCase {
         assertEquals(List.of(42.9f), fetchSourceValue(mapper, "42.9"));
         assertEquals(List.of(2.0f), fetchSourceValue(mapper, null));
     }
+
+    @Override
+    public void testFieldHasValue() {
+        MappedFieldType fieldType = getMappedFieldType();
+        FieldInfos fieldInfos = new FieldInfos(new FieldInfo[] { getFieldInfoWithName("_feature") });
+        assertTrue(fieldType.fieldHasValue(fieldInfos));
+    }
+
+    @Override
+    public void testFieldHasValueWithEmptyFieldInfos() {
+        MappedFieldType fieldType = getMappedFieldType();
+        assertFalse(fieldType.fieldHasValue(FieldInfos.EMPTY));
+    }
+
+    public void testFieldEmptyIfNameIsPresentInFieldInfos() {
+        MappedFieldType fieldType = getMappedFieldType();
+        FieldInfos fieldInfos = new FieldInfos(new FieldInfo[] { getFieldInfoWithName("field") });
+        assertFalse(fieldType.fieldHasValue(fieldInfos));
+    }
+
+    @Override
+    public MappedFieldType getMappedFieldType() {
+        return new RankFeatureFieldMapper.RankFeatureFieldType("field", Collections.emptyMap(), true, null);
+    }
 }

modules/mapper-extras/src/test/java/org/elasticsearch/index/mapper/extras/RankFeatureMetaFieldMapperTests.java

@@ -8,11 +8,14 @@
 
 package org.elasticsearch.index.mapper.extras;
 
+import org.apache.lucene.index.FieldInfo;
+import org.apache.lucene.index.FieldInfos;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.compress.CompressedXContent;
 import org.elasticsearch.index.mapper.DocumentMapper;
 import org.elasticsearch.index.mapper.DocumentParsingException;
+import org.elasticsearch.index.mapper.MappedFieldType;
 import org.elasticsearch.index.mapper.MapperService;
 import org.elasticsearch.index.mapper.MapperServiceTestCase;
 import org.elasticsearch.index.mapper.Mapping;
@@ -73,4 +76,19 @@ public class RankFeatureMetaFieldMapperTests extends MapperServiceTestCase {
             CoreMatchers.containsString("Field [" + rfMetaField + "] is a metadata field and cannot be added inside a document.")
         );
     }
+
+    @Override
+    public void testFieldHasValue() {
+        assertTrue(getMappedFieldType().fieldHasValue(new FieldInfos(new FieldInfo[] { getFieldInfoWithName("_feature") })));
+    }
+
+    @Override
+    public void testFieldHasValueWithEmptyFieldInfos() {
+        assertFalse(getMappedFieldType().fieldHasValue(FieldInfos.EMPTY));
+    }
+
+    @Override
+    public MappedFieldType getMappedFieldType() {
+        return new RankFeatureMetaFieldMapper.RankFeatureMetaFieldType();
+    }
 }

qa/multi-cluster-search/src/test/resources/rest-api-spec/test/multi_cluster/30_field_caps.yml

@@ -65,7 +65,6 @@
       field_caps:
         index: 'my_remote_cluster:some_index_that_doesnt_exist'
         fields: [number]
-        
   - match: { error.type: "index_not_found_exception" }
   - match: { error.reason: "no such index [some_index_that_doesnt_exist]" }
 
@@ -156,3 +155,38 @@
   - length: {fields.number:                             1}
   - match: {fields.number.long.searchable:              true}
   - match: {fields.number.long.aggregatable:            true}
+
+---
+"Field caps with with include_empty_fields false":
+  - skip:
+      version: " - 8.12.99"
+      reason: include_empty_fields has been added in 8.13.0
+  - do:
+        indices.create:
+          index: field_caps_index_5
+          body:
+            mappings:
+              properties:
+                number:
+                  type: double
+                empty-baz:
+                  type: text
+
+  - do:
+      index:
+        index:  field_caps_index_5
+        body:   { number: "42", unmapped-bar: "bar" }
+
+  - do:
+      indices.refresh:
+        index: [field_caps_index_5]
+  - do:
+      field_caps:
+        include_empty_fields: false
+        index: 'field_caps_index_5,my_remote_cluster:field_*'
+        fields: '*'
+
+  - is_true: fields.number
+  - is_false: fields.empty-baz
+  - is_true: fields.unmapped-bar
+  - is_true: fields._index

qa/multi-cluster-search/src/test/resources/rest-api-spec/test/remote_cluster/10_basic.yml

@@ -82,6 +82,10 @@
                       nested2:
                         type: keyword
                         doc_values: false
+  - do:
+        index:
+          index:  field_caps_index_1
+          body:   { number: "42", unmapped-bar: "bar" }
   - do:
       indices.create:
         index: test_index

rest-api-spec/src/main/resources/rest-api-spec/api/field_caps.json

@@ -71,6 +71,11 @@
       "types": {
         "type": "list",
         "description":"Only return results for fields that have one of the types in the list"
+      },
+      "include_empty_fields": {
+        "type":"boolean",
+        "default": true,
+        "description":"Include empty fields in result"
       }
     },
     "body":{

rest-api-spec/src/yamlRestTest/resources/rest-api-spec/test/field_caps/60_non_empty_fields.yml

@@ -0,0 +1,39 @@
+---
+setup:
+  - do:
+        indices.create:
+          index: test
+          body:
+            mappings:
+              properties:
+                foo:
+                  type: long
+                empty-baz:
+                  type: text
+
+  - do:
+      index:
+        index:  test
+        body:   { foo: "42", unmapped-bar: "bar" }
+
+  - do:
+      indices.refresh:
+        index: [test]
+
+---
+"Field caps with with include_empty_fields false":
+  - skip:
+      version: " - 8.12.99"
+      reason: include_empty_fields has been added in 8.13.0
+
+  - do:
+      field_caps:
+        include_empty_fields: false
+        index: test
+        fields: "*"
+
+  - match: {indices: ["test"]}
+  - is_true: fields.foo
+  - is_false: fields.empty-baz
+  - is_true: fields.unmapped-bar
+  - is_true: fields._index

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

@@ -0,0 +1,433 @@
+/*
+ * 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.search.fieldcaps;
+
+import org.elasticsearch.action.DocWriteResponse;
+import org.elasticsearch.action.fieldcaps.FieldCapabilities;
+import org.elasticsearch.action.fieldcaps.FieldCapabilitiesResponse;
+import org.elasticsearch.action.index.IndexRequestBuilder;
+import org.elasticsearch.action.support.ActiveShardCount;
+import org.elasticsearch.index.query.QueryBuilders;
+import org.elasticsearch.test.ESIntegTestCase;
+import org.hamcrest.Matchers;
+import org.junit.Before;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
+
+@ESIntegTestCase.ClusterScope(scope = ESIntegTestCase.Scope.TEST)
+public class FieldCapsHasValueTests extends ESIntegTestCase {
+    private final String INDEX1 = "index-1";
+    private final String ALIAS1 = "alias-1";
+    private final String INDEX2 = "index-2";
+    private final String INDEX3 = "index-3";
+
+    @Before
+    public void setUpIndices() {
+        assertAcked(
+            prepareCreate(INDEX1).setWaitForActiveShards(ActiveShardCount.ALL)
+                .setSettings(indexSettings())
+                .setMapping("foo", "type=text", "bar", "type=keyword", "bar-alias", "type=alias,path=bar", "timestamp", "type=date")
+        );
+        assertAcked(
+            prepareCreate(INDEX2).setWaitForActiveShards(ActiveShardCount.ALL)
+                .setSettings(indexSettings())
+                .setMapping("bar", "type=date", "timestamp", "type=date")
+        );
+        assertAcked(
+            prepareCreate(INDEX3).setWaitForActiveShards(ActiveShardCount.ALL)
+                .setSettings(indexSettings())
+                .setMapping("nested_type", "type=nested", "object.sub_field", "type=keyword,store=true")
+        );
+        assertAcked(indicesAdmin().prepareAliases().addAlias(INDEX1, ALIAS1));
+    }
+
+    public void testNoFieldsInEmptyIndex() {
+        FieldCapabilitiesResponse response = client().prepareFieldCaps().setFields("*").setincludeEmptyFields(false).get();
+
+        assertIndices(response, INDEX1, INDEX2, INDEX3);
+        // Ensure the response has no mapped fields.
+        assertFalse(response.get().containsKey("foo"));
+        assertFalse(response.get().containsKey("bar"));
+        assertFalse(response.get().containsKey("bar-alias"));
+    }
+
+    public void testOnlyFieldsWithValueInIndex() {
+        prepareIndex(INDEX1).setSource("foo", "foo-text").get();
+        refresh(INDEX1);
+
+        FieldCapabilitiesResponse response = client().prepareFieldCaps().setFields("*").setincludeEmptyFields(false).get();
+
+        assertIndices(response, INDEX1, INDEX2, INDEX3);
+        assertThat(response.get(), Matchers.hasKey("foo"));
+        // Check the capabilities for the 'foo' field.
+        Map<String, FieldCapabilities> fooField = response.getField("foo");
+        assertEquals(1, fooField.size());
+        assertThat(fooField, Matchers.hasKey("text"));
+        assertEquals(
+            new FieldCapabilities("foo", "text", false, true, false, null, null, null, Collections.emptyMap()),
+            fooField.get("text")
+        );
+    }
+
+    public void testOnlyFieldsWithValueInAlias() {
+        prepareIndex(ALIAS1).setSource("foo", "foo-text").get();
+        refresh(ALIAS1);
+
+        FieldCapabilitiesResponse response = client().prepareFieldCaps().setFields("*").setincludeEmptyFields(false).get();
+
+        assertIndices(response, INDEX1, INDEX2, INDEX3);
+        assertThat(response.get(), Matchers.hasKey("foo"));
+        // Check the capabilities for the 'foo' field.
+        Map<String, FieldCapabilities> fooField = response.getField("foo");
+        assertEquals(1, fooField.size());
+        assertThat(fooField, Matchers.hasKey("text"));
+        assertEquals(
+            new FieldCapabilities("foo", "text", false, true, false, null, null, null, Collections.emptyMap()),
+            fooField.get("text")
+        );
+    }
+
+    public void testOnlyFieldsWithValueInSpecifiedIndex() {
+        prepareIndex(INDEX1).setSource("foo", "foo-text").get();
+        refresh(INDEX1);
+
+        FieldCapabilitiesResponse response = client().prepareFieldCaps(INDEX1).setFields("*").setincludeEmptyFields(false).get();
+
+        assertIndices(response, INDEX1);
+        assertThat(response.get(), Matchers.hasKey("foo"));
+        // Check the capabilities for the 'foo' field.
+        Map<String, FieldCapabilities> fooField = response.getField("foo");
+        assertEquals(1, fooField.size());
+        assertThat(fooField, Matchers.hasKey("text"));
+        assertEquals(
+            new FieldCapabilities("foo", "text", false, true, false, null, null, null, Collections.emptyMap()),
+            fooField.get("text")
+        );
+    }
+
+    public void testOnlyFieldsWithValueInSpecifiedAlias() {
+        prepareIndex(ALIAS1).setSource("foo", "foo-text").get();
+        refresh(ALIAS1);
+
+        FieldCapabilitiesResponse response = client().prepareFieldCaps(ALIAS1).setFields("*").setincludeEmptyFields(false).get();
+
+        assertIndices(response, INDEX1);
+        assertThat(response.get(), Matchers.hasKey("foo"));
+        // Check the capabilities for the 'foo' field.
+        Map<String, FieldCapabilities> fooField = response.getField("foo");
+        assertEquals(1, fooField.size());
+        assertThat(fooField, Matchers.hasKey("text"));
+        assertEquals(
+            new FieldCapabilities("foo", "text", false, true, false, null, null, null, Collections.emptyMap()),
+            fooField.get("text")
+        );
+    }
+
+    public void testFieldsWithValueAfterUpdate() {
+        DocWriteResponse doc = prepareIndex(INDEX1).setSource("foo", "foo-text").get();
+        prepareIndex(INDEX1).setId(doc.getId()).setSource("bar", "bar-keyword").get();
+        refresh(INDEX1);
+
+        FieldCapabilitiesResponse response = client().prepareFieldCaps(INDEX1).setFields("*").setincludeEmptyFields(false).get();
+
+        assertIndices(response, INDEX1);
+        assertThat(response.get(), Matchers.hasKey("foo"));
+        assertThat(response.get(), Matchers.hasKey("bar"));
+        // Check the capabilities for the 'foo' field.
+        Map<String, FieldCapabilities> fooField = response.getField("foo");
+        assertEquals(1, fooField.size());
+        assertThat(fooField, Matchers.hasKey("text"));
+        assertEquals(
+            new FieldCapabilities("foo", "text", false, true, false, null, null, null, Collections.emptyMap()),
+            fooField.get("text")
+        );
+        // Check the capabilities for the 'bar' field.
+        Map<String, FieldCapabilities> barField = response.getField("bar");
+        assertEquals(1, barField.size());
+        assertThat(barField, Matchers.hasKey("keyword"));
+        assertEquals(
+            new FieldCapabilities("bar", "keyword", false, true, true, null, null, null, Collections.emptyMap()),
+            barField.get("keyword")
+        );
+    }
+
+    public void testOnlyFieldsWithValueAfterNodesRestart() throws Exception {
+        prepareIndex(INDEX1).setSource("foo", "foo-text").get();
+        internalCluster().fullRestart();
+        ensureGreen(INDEX1);
+
+        FieldCapabilitiesResponse response = client().prepareFieldCaps(INDEX1).setFields("*").setincludeEmptyFields(false).get();
+
+        assertIndices(response, INDEX1);
+        assertThat(response.get(), Matchers.hasKey("foo"));
+        // Check the capabilities for the 'foo' field.
+        Map<String, FieldCapabilities> fooField = response.getField("foo");
+        assertEquals(1, fooField.size());
+        assertThat(fooField, Matchers.hasKey("text"));
+        assertEquals(
+            new FieldCapabilities("foo", "text", false, true, false, null, null, null, Collections.emptyMap()),
+            fooField.get("text")
+        );
+    }
+
+    public void testFieldsAndAliasWithValue() {
+        prepareIndex(INDEX1).setSource("foo", "foo-text").get();
+        prepareIndex(INDEX1).setSource("bar", "bar-keyword").get();
+        refresh(INDEX1);
+
+        FieldCapabilitiesResponse response = client().prepareFieldCaps().setFields("*").setincludeEmptyFields(false).get();
+
+        assertIndices(response, INDEX1, INDEX2, INDEX3);
+        assertThat(response.get(), Matchers.hasKey("foo"));
+        assertThat(response.get(), Matchers.hasKey("bar"));
+        assertThat(response.get(), Matchers.hasKey("bar-alias"));
+        // Check the capabilities for the 'foo' field.
+        Map<String, FieldCapabilities> fooField = response.getField("foo");
+        assertEquals(1, fooField.size());
+        assertThat(fooField, Matchers.hasKey("text"));
+        assertEquals(
+            new FieldCapabilities("foo", "text", false, true, false, null, null, null, Collections.emptyMap()),
+            fooField.get("text")
+        );
+        // Check the capabilities for the 'bar' field.
+        Map<String, FieldCapabilities> barField = response.getField("bar");
+        assertEquals(1, barField.size());
+        assertThat(barField, Matchers.hasKey("keyword"));
+        assertEquals(
+            new FieldCapabilities("bar", "keyword", false, true, true, null, null, null, Collections.emptyMap()),
+            barField.get("keyword")
+        );
+        // Check the capabilities for the 'bar-alias' field.
+        Map<String, FieldCapabilities> barAlias = response.getField("bar-alias");
+        assertEquals(1, barAlias.size());
+        assertThat(barAlias, Matchers.hasKey("keyword"));
+        assertEquals(
+            new FieldCapabilities("bar-alias", "keyword", false, true, true, null, null, null, Collections.emptyMap()),
+            barAlias.get("keyword")
+        );
+    }
+
+    public void testUnmappedFieldsWithValueAfterRestart() throws Exception {
+        prepareIndex(INDEX1).setSource("unmapped", "unmapped-text").get();
+        internalCluster().fullRestart();
+        ensureGreen(INDEX1);
+
+        FieldCapabilitiesResponse response = client().prepareFieldCaps()
+            .setFields("*")
+            .setIncludeUnmapped(true)
+            .setincludeEmptyFields(false)
+            .get();
+
+        assertIndices(response, INDEX1, INDEX2, INDEX3);
+        assertThat(response.get(), Matchers.hasKey("unmapped"));
+        // Check the capabilities for the 'unmapped' field.
+        Map<String, FieldCapabilities> unmappedField = response.getField("unmapped");
+        assertEquals(2, unmappedField.size());
+        assertThat(unmappedField, Matchers.hasKey("text"));
+        assertEquals(
+            new FieldCapabilities("unmapped", "text", false, true, false, new String[] { INDEX1 }, null, null, Collections.emptyMap()),
+            unmappedField.get("text")
+        );
+    }
+
+    public void testTwoFieldsNameTwoIndices() {
+        prepareIndex(INDEX1).setSource("foo", "foo-text").get();
+        prepareIndex(INDEX2).setSource("bar", 1704293160000L).get();
+        refresh(INDEX1, INDEX2);
+
+        FieldCapabilitiesResponse response = client().prepareFieldCaps().setFields("*").setincludeEmptyFields(false).get();
+
+        assertIndices(response, INDEX1, INDEX2, INDEX3);
+        assertThat(response.get(), Matchers.hasKey("foo"));
+        assertThat(response.get(), Matchers.hasKey("bar"));
+        // Check the capabilities for the 'foo' field.
+        Map<String, FieldCapabilities> fooField = response.getField("foo");
+        assertEquals(1, fooField.size());
+        assertThat(fooField, Matchers.hasKey("text"));
+        assertEquals(
+            new FieldCapabilities("foo", "text", false, true, false, null, null, null, Collections.emptyMap()),
+            fooField.get("text")
+        );
+        // Check the capabilities for the 'bar' field.
+        Map<String, FieldCapabilities> barField = response.getField("bar");
+        assertEquals(1, barField.size());
+        assertThat(barField, Matchers.hasKey("date"));
+        assertEquals(
+            new FieldCapabilities("bar", "date", false, true, true, null, null, null, Collections.emptyMap()),
+            barField.get("date")
+        );
+    }
+
+    public void testSameFieldNameTwoIndices() {
+        prepareIndex(INDEX1).setSource("bar", "bar-text").get();
+        prepareIndex(INDEX2).setSource("bar", 1704293160000L).get();
+        refresh(INDEX1, INDEX2);
+
+        FieldCapabilitiesResponse response = client().prepareFieldCaps().setFields("*").setincludeEmptyFields(false).get();
+
+        assertIndices(response, INDEX1, INDEX2, INDEX3);
+        assertThat(response.get(), Matchers.hasKey("bar"));
+        // Check the capabilities for the 'bar' field.
+        Map<String, FieldCapabilities> barField = response.getField("bar");
+        assertEquals(2, barField.size());
+        assertThat(barField, Matchers.hasKey("keyword"));
+        assertEquals(
+            new FieldCapabilities("bar", "keyword", false, true, true, new String[] { INDEX1 }, null, null, Collections.emptyMap()),
+            barField.get("keyword")
+        );
+        assertThat(barField, Matchers.hasKey("date"));
+        assertEquals(
+            new FieldCapabilities("bar", "date", false, true, true, new String[] { INDEX2 }, null, null, Collections.emptyMap()),
+            barField.get("date")
+        );
+    }
+
+    public void testDeletedDocsReturned() {
+        // In this current implementation we do not handle deleted documents (without a restart).
+        // This test should fail if in a future implementation we handle deletes.
+        DocWriteResponse foo = prepareIndex(INDEX1).setSource("foo", "foo-text").get();
+        client().prepareDelete().setIndex(INDEX1).setId(foo.getId()).get();
+        client().admin().indices().prepareForceMerge(INDEX1).setFlush(true).setMaxNumSegments(1).get();
+        refresh(INDEX1);
+
+        FieldCapabilitiesResponse response = client().prepareFieldCaps().setFields("*").setincludeEmptyFields(false).get();
+
+        assertIndices(response, INDEX1, INDEX2, INDEX3);
+        assertThat(response.get(), Matchers.hasKey("foo"));
+        // Check the capabilities for the 'foo' field.
+        Map<String, FieldCapabilities> fooField = response.getField("foo");
+        assertEquals(1, fooField.size());
+        assertThat(fooField, Matchers.hasKey("text"));
+        assertEquals(
+            new FieldCapabilities("foo", "text", false, true, false, null, null, null, Collections.emptyMap()),
+            fooField.get("text")
+        );
+    }
+
+    public void testNoNestedFieldsInEmptyIndex() {
+        FieldCapabilitiesResponse response = client().prepareFieldCaps(INDEX3).setFields("*").setincludeEmptyFields(false).get();
+
+        assertIndices(response, INDEX3);
+        assertFalse(response.get().containsKey("nested_type"));
+        assertFalse(response.get().containsKey("nested_type.nested_field"));
+    }
+
+    public void testNestedFields() {
+        prepareIndex(INDEX3).setSource("nested_type", Collections.singletonMap("nested_field", "value")).get();
+        refresh(INDEX3);
+
+        FieldCapabilitiesResponse response = client().prepareFieldCaps(INDEX3).setFields("*").setincludeEmptyFields(false).get();
+
+        assertIndices(response, INDEX3);
+        assertThat(response.get(), Matchers.hasKey("nested_type"));
+        assertThat(response.get(), Matchers.hasKey("nested_type.nested_field"));
+        // Check the capabilities for the 'nested_type' field.
+        Map<String, FieldCapabilities> nestedTypeField = response.getField("nested_type");
+        assertEquals(1, nestedTypeField.size());
+        assertThat(nestedTypeField, Matchers.hasKey("nested"));
+        assertEquals(
+            new FieldCapabilities("nested_type", "nested", false, false, false, null, null, null, Collections.emptyMap()),
+            nestedTypeField.get("nested")
+        );
+        // Check the capabilities for the 'nested_type.nested_field' field.
+        Map<String, FieldCapabilities> nestedTypeNestedField = response.getField("nested_type.nested_field");
+        assertEquals(1, nestedTypeNestedField.size());
+        assertThat(nestedTypeNestedField, Matchers.hasKey("text"));
+        assertEquals(
+            new FieldCapabilities("nested_type.nested_field", "text", false, true, false, null, null, null, Collections.emptyMap()),
+            nestedTypeNestedField.get("text")
+        );
+    }
+
+    public void testNoObjectFieldsInEmptyIndex() {
+        FieldCapabilitiesResponse response = client().prepareFieldCaps(INDEX3).setFields("*").setincludeEmptyFields(false).get();
+
+        assertIndices(response, INDEX3);
+        assertFalse(response.get().containsKey("object"));
+        assertFalse(response.get().containsKey("object.sub_field"));
+    }
+
+    public void testObjectFields() {
+        prepareIndex(INDEX3).setSource("object.sub_field", "value").get();
+        refresh(INDEX3);
+
+        FieldCapabilitiesResponse response = client().prepareFieldCaps(INDEX3).setFields("*").setincludeEmptyFields(false).get();
+
+        assertIndices(response, INDEX3);
+        assertThat(response.get(), Matchers.hasKey("object"));
+        assertThat(response.get(), Matchers.hasKey("object.sub_field"));
+        // Check the capabilities for the 'object' field.
+        Map<String, FieldCapabilities> objectTypeField = response.getField("object");
+        assertEquals(1, objectTypeField.size());
+        assertThat(objectTypeField, Matchers.hasKey("object"));
+        assertEquals(
+            new FieldCapabilities("object", "object", false, false, false, null, null, null, Collections.emptyMap()),
+            objectTypeField.get("object")
+        );
+        // Check the capabilities for the 'object.sub_field' field.
+        Map<String, FieldCapabilities> objectSubfield = response.getField("object.sub_field");
+        assertEquals(1, objectSubfield.size());
+        assertThat(objectSubfield, Matchers.hasKey("keyword"));
+        assertEquals(
+            new FieldCapabilities("object.sub_field", "keyword", false, true, true, null, null, null, Collections.emptyMap()),
+            objectSubfield.get("keyword")
+        );
+    }
+
+    public void testWithIndexFilter() throws InterruptedException {
+
+        List<IndexRequestBuilder> reqs = new ArrayList<>();
+        reqs.add(prepareIndex(INDEX1).setSource("timestamp", "2015-07-08"));
+        reqs.add(prepareIndex(INDEX1).setSource("timestamp", "2018-07-08"));
+        reqs.add(prepareIndex(INDEX2).setSource("timestamp", "2019-10-12"));
+        reqs.add(prepareIndex(INDEX2).setSource("timestamp", "2020-07-08"));
+        indexRandom(true, reqs);
+
+        FieldCapabilitiesResponse response = client().prepareFieldCaps("index-*")
+            .setFields("*")
+            .setIndexFilter(QueryBuilders.rangeQuery("timestamp").gte("2019-11-01"))
+            .setincludeEmptyFields(false)
+            .get();
+        assertIndices(response, INDEX2);
+        // Check the capabilities for the 'timestamp' field.
+        Map<String, FieldCapabilities> timestampField = response.getField("timestamp");
+        assertEquals(1, timestampField.size());
+        assertThat(timestampField, Matchers.hasKey("date"));
+        assertNull(response.getField("foo"));
+        assertNull(response.getField("bar"));
+        assertNull(response.getField("bar"));
+
+        response = client().prepareFieldCaps("index-*")
+            .setFields("*")
+            .setIndexFilter(QueryBuilders.rangeQuery("timestamp").lte("2017-01-01"))
+            .setincludeEmptyFields(false)
+            .get();
+        assertIndices(response, INDEX1);
+        // Check the capabilities for the 'timestamp' field.
+        timestampField = response.getField("timestamp");
+        assertEquals(1, timestampField.size());
+        assertThat(timestampField, Matchers.hasKey("date"));
+        assertNull(response.getField("foo"));
+        assertNull(response.getField("bar"));
+    }
+
+    private void assertIndices(FieldCapabilitiesResponse response, String... indices) {
+        assertNotNull(response.getIndices());
+        Arrays.sort(indices);
+        Arrays.sort(response.getIndices());
+        assertArrayEquals(indices, response.getIndices());
+    }
+
+}

server/src/main/java/org/elasticsearch/TransportVersions.java

@@ -127,6 +127,8 @@ public class TransportVersions {
     public static final TransportVersion ML_TEXT_EMBEDDING_INFERENCE_SERVICE_ADDED = def(8_587_00_0);
     public static final TransportVersion HEALTH_INFO_ENRICHED_WITH_REPOS = def(8_588_00_0);
     public static final TransportVersion RESOLVE_CLUSTER_ENDPOINT_ADDED = def(8_589_00_0);
+    public static final TransportVersion FIELD_CAPS_FIELD_HAS_VALUE = def(8_590_00_0);
+
     /*
      * STOP! READ THIS FIRST! No, really,
      *        ____ _____ ___  ____  _        ____  _____    _    ____    _____ _   _ ___ ____    _____ ___ ____  ____ _____ _

server/src/main/java/org/elasticsearch/action/fieldcaps/FieldCapabilitiesFetcher.java

@@ -9,6 +9,7 @@
 package org.elasticsearch.action.fieldcaps;
 
 import org.elasticsearch.cluster.metadata.MappingMetadata;
+import org.elasticsearch.core.Booleans;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.index.IndexService;
 import org.elasticsearch.index.engine.Engine;
@@ -39,10 +40,15 @@ import java.util.function.Predicate;
  */
 class FieldCapabilitiesFetcher {
     private final IndicesService indicesService;
+    private final boolean includeEmptyFields;
     private final Map<String, Map<String, IndexFieldCapabilities>> indexMappingHashToResponses = new HashMap<>();
+    private static final boolean enableFieldHasValue = Booleans.parseBoolean(
+        System.getProperty("es.field_caps_empty_fields_filter", Boolean.TRUE.toString())
+    );
 
-    FieldCapabilitiesFetcher(IndicesService indicesService) {
+    FieldCapabilitiesFetcher(IndicesService indicesService, boolean includeEmptyFields) {
         this.indicesService = indicesService;
+        this.includeEmptyFields = includeEmptyFields;
     }
 
     FieldCapabilitiesIndexResponse fetch(
@@ -107,7 +113,19 @@ class FieldCapabilitiesFetcher {
         }
 
         final MappingMetadata mapping = indexService.getMetadata().mapping();
-        final String indexMappingHash = mapping != null ? mapping.getSha256() : null;
+        String indexMappingHash;
+        if (includeEmptyFields || enableFieldHasValue == false) {
+            indexMappingHash = mapping != null ? mapping.getSha256() : null;
+        } else {
+            // even if the mapping is the same if we return only fields with values we need
+            // to make sure that we consider all the shard-mappings pair, that is why we
+            // calculate a different hash for this particular case.
+            StringBuilder sb = new StringBuilder(indexService.getShard(shardId.getId()).getShardUuid());
+            if (mapping != null) {
+                sb.append(mapping.getSha256());
+            }
+            indexMappingHash = sb.toString();
+        }
         if (indexMappingHash != null) {
             final Map<String, IndexFieldCapabilities> existing = indexMappingHashToResponses.get(indexMappingHash);
             if (existing != null) {
@@ -121,7 +139,9 @@ class FieldCapabilitiesFetcher {
             fieldNameFilter,
             filters,
             fieldTypes,
-            fieldPredicate
+            fieldPredicate,
+            indicesService.getShardOrNull(shardId),
+            includeEmptyFields
         );
         if (indexMappingHash != null) {
             indexMappingHashToResponses.put(indexMappingHash, responseMap);
@@ -134,7 +154,9 @@ class FieldCapabilitiesFetcher {
         Predicate<String> fieldNameFilter,
         String[] filters,
         String[] types,
-        Predicate<String> indexFieldfilter
+        Predicate<String> indexFieldfilter,
+        IndexShard indexShard,
+        boolean includeEmptyFields
     ) {
         boolean includeParentObjects = checkIncludeParents(filters);
 
@@ -146,7 +168,8 @@ class FieldCapabilitiesFetcher {
                 continue;
             }
             MappedFieldType ft = context.getFieldType(field);
-            if (filter.test(ft)) {
+            boolean includeField = includeEmptyFields || enableFieldHasValue == false || ft.fieldHasValue(indexShard.getFieldInfos());
+            if (includeField && filter.test(ft)) {
                 IndexFieldCapabilities fieldCap = new IndexFieldCapabilities(
                     field,
                     ft.familyTypeName(),

server/src/main/java/org/elasticsearch/action/fieldcaps/FieldCapabilitiesNodeRequest.java

@@ -39,6 +39,7 @@ class FieldCapabilitiesNodeRequest extends ActionRequest implements IndicesReque
     private final QueryBuilder indexFilter;
     private final long nowInMillis;
     private final Map<String, Object> runtimeFields;
+    private final boolean includeEmptyFields;
 
     FieldCapabilitiesNodeRequest(StreamInput in) throws IOException {
         super(in);
@@ -55,6 +56,11 @@ class FieldCapabilitiesNodeRequest extends ActionRequest implements IndicesReque
         indexFilter = in.readOptionalNamedWriteable(QueryBuilder.class);
         nowInMillis = in.readLong();
         runtimeFields = in.readGenericMap();
+        if (in.getTransportVersion().onOrAfter(TransportVersions.FIELD_CAPS_FIELD_HAS_VALUE)) {
+            includeEmptyFields = in.readBoolean();
+        } else {
+            includeEmptyFields = true;
+        }
     }
 
     FieldCapabilitiesNodeRequest(
@@ -65,7 +71,8 @@ class FieldCapabilitiesNodeRequest extends ActionRequest implements IndicesReque
         OriginalIndices originalIndices,
         QueryBuilder indexFilter,
         long nowInMillis,
-        Map<String, Object> runtimeFields
+        Map<String, Object> runtimeFields,
+        boolean includeEmptyFields
     ) {
         this.shardIds = Objects.requireNonNull(shardIds);
         this.fields = fields;
@@ -75,6 +82,7 @@ class FieldCapabilitiesNodeRequest extends ActionRequest implements IndicesReque
         this.indexFilter = indexFilter;
         this.nowInMillis = nowInMillis;
         this.runtimeFields = runtimeFields;
+        this.includeEmptyFields = includeEmptyFields;
     }
 
     public String[] fields() {
@@ -119,6 +127,10 @@ class FieldCapabilitiesNodeRequest extends ActionRequest implements IndicesReque
         return nowInMillis;
     }
 
+    public boolean includeEmptyFields() {
+        return includeEmptyFields;
+    }
+
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
@@ -132,6 +144,9 @@ class FieldCapabilitiesNodeRequest extends ActionRequest implements IndicesReque
         out.writeOptionalNamedWriteable(indexFilter);
         out.writeLong(nowInMillis);
         out.writeGenericMap(runtimeFields);
+        if (out.getTransportVersion().onOrAfter(TransportVersions.FIELD_CAPS_FIELD_HAS_VALUE)) {
+            out.writeBoolean(includeEmptyFields);
+        }
     }
 
     @Override
@@ -143,16 +158,24 @@ class FieldCapabilitiesNodeRequest extends ActionRequest implements IndicesReque
     public String getDescription() {
         final StringBuilder stringBuilder = new StringBuilder("shards[");
         Strings.collectionToDelimitedStringWithLimit(shardIds, ",", "", "", 1024, stringBuilder);
-        return completeDescription(stringBuilder, fields, filters, allowedTypes);
+        return completeDescription(stringBuilder, fields, filters, allowedTypes, includeEmptyFields);
     }
 
-    static String completeDescription(StringBuilder stringBuilder, String[] fields, String[] filters, String[] allowedTypes) {
+    static String completeDescription(
+        StringBuilder stringBuilder,
+        String[] fields,
+        String[] filters,
+        String[] allowedTypes,
+        boolean includeEmptyFields
+    ) {
         stringBuilder.append("], fields[");
         Strings.collectionToDelimitedStringWithLimit(Arrays.asList(fields), ",", "", "", 1024, stringBuilder);
         stringBuilder.append("], filters[");
         Strings.collectionToDelimitedString(Arrays.asList(filters), ",", "", "", stringBuilder);
         stringBuilder.append("], types[");
         Strings.collectionToDelimitedString(Arrays.asList(allowedTypes), ",", "", "", stringBuilder);
+        stringBuilder.append("], includeEmptyFields[");
+        stringBuilder.append(includeEmptyFields);
         stringBuilder.append("]");
         return stringBuilder.toString();
     }
@@ -179,12 +202,13 @@ class FieldCapabilitiesNodeRequest extends ActionRequest implements IndicesReque
             && Arrays.equals(allowedTypes, that.allowedTypes)
             && Objects.equals(originalIndices, that.originalIndices)
             && Objects.equals(indexFilter, that.indexFilter)
-            && Objects.equals(runtimeFields, that.runtimeFields);
+            && Objects.equals(runtimeFields, that.runtimeFields)
+            && includeEmptyFields == that.includeEmptyFields;
     }
 
     @Override
     public int hashCode() {
-        int result = Objects.hash(originalIndices, indexFilter, nowInMillis, runtimeFields);
+        int result = Objects.hash(originalIndices, indexFilter, nowInMillis, runtimeFields, includeEmptyFields);
         result = 31 * result + shardIds.hashCode();
         result = 31 * result + Arrays.hashCode(fields);
         result = 31 * result + Arrays.hashCode(filters);

server/src/main/java/org/elasticsearch/action/fieldcaps/FieldCapabilitiesRequest.java

@@ -42,6 +42,7 @@ public final class FieldCapabilitiesRequest extends ActionRequest implements Ind
     private String[] filters = Strings.EMPTY_ARRAY;
     private String[] types = Strings.EMPTY_ARRAY;
     private boolean includeUnmapped = false;
+    private boolean includeEmptyFields = true;
     // pkg private API mainly for cross cluster search to signal that we do multiple reductions ie. the results should not be merged
     private boolean mergeResults = true;
     private QueryBuilder indexFilter;
@@ -62,6 +63,9 @@ public final class FieldCapabilitiesRequest extends ActionRequest implements Ind
             filters = in.readStringArray();
             types = in.readStringArray();
         }
+        if (in.getTransportVersion().onOrAfter(TransportVersions.FIELD_CAPS_FIELD_HAS_VALUE)) {
+            includeEmptyFields = in.readBoolean();
+        }
     }
 
     public FieldCapabilitiesRequest() {}
@@ -100,6 +104,9 @@ public final class FieldCapabilitiesRequest extends ActionRequest implements Ind
             out.writeStringArray(filters);
             out.writeStringArray(types);
         }
+        if (out.getTransportVersion().onOrAfter(TransportVersions.FIELD_CAPS_FIELD_HAS_VALUE)) {
+            out.writeBoolean(includeEmptyFields);
+        }
     }
 
     @Override
@@ -168,6 +175,11 @@ public final class FieldCapabilitiesRequest extends ActionRequest implements Ind
         return this;
     }
 
+    public FieldCapabilitiesRequest includeEmptyFields(boolean includeEmptyFields) {
+        this.includeEmptyFields = includeEmptyFields;
+        return this;
+    }
+
     @Override
     public String[] indices() {
         return indices;
@@ -192,6 +204,10 @@ public final class FieldCapabilitiesRequest extends ActionRequest implements Ind
         return includeUnmapped;
     }
 
+    public boolean includeEmptyFields() {
+        return includeEmptyFields;
+    }
+
     /**
      * Allows to filter indices if the provided {@link QueryBuilder} rewrites to `match_none` on every shard.
      */
@@ -247,12 +263,21 @@ public final class FieldCapabilitiesRequest extends ActionRequest implements Ind
             && Objects.equals(nowInMillis, that.nowInMillis)
             && Arrays.equals(filters, that.filters)
             && Arrays.equals(types, that.types)
-            && Objects.equals(runtimeFields, that.runtimeFields);
+            && Objects.equals(runtimeFields, that.runtimeFields)
+            && includeEmptyFields == that.includeEmptyFields;
     }
 
     @Override
     public int hashCode() {
-        int result = Objects.hash(indicesOptions, includeUnmapped, mergeResults, indexFilter, nowInMillis, runtimeFields);
+        int result = Objects.hash(
+            indicesOptions,
+            includeUnmapped,
+            mergeResults,
+            indexFilter,
+            nowInMillis,
+            runtimeFields,
+            includeEmptyFields
+        );
         result = 31 * result + Arrays.hashCode(indices);
         result = 31 * result + Arrays.hashCode(fields);
         result = 31 * result + Arrays.hashCode(filters);
@@ -264,7 +289,7 @@ public final class FieldCapabilitiesRequest extends ActionRequest implements Ind
     public String getDescription() {
         final StringBuilder stringBuilder = new StringBuilder("indices[");
         Strings.collectionToDelimitedStringWithLimit(Arrays.asList(indices), ",", "", "", 1024, stringBuilder);
-        return FieldCapabilitiesNodeRequest.completeDescription(stringBuilder, fields, filters, types);
+        return FieldCapabilitiesNodeRequest.completeDescription(stringBuilder, fields, filters, types, includeEmptyFields);
     }
 
     @Override

server/src/main/java/org/elasticsearch/action/fieldcaps/FieldCapabilitiesRequestBuilder.java

@@ -32,6 +32,11 @@ public class FieldCapabilitiesRequestBuilder extends ActionRequestBuilder<FieldC
         return this;
     }
 
+    public FieldCapabilitiesRequestBuilder setincludeEmptyFields(boolean includeEmptyFields) {
+        request().includeEmptyFields(includeEmptyFields);
+        return this;
+    }
+
     public FieldCapabilitiesRequestBuilder setIndexFilter(QueryBuilder indexFilter) {
         request().indexFilter(indexFilter);
         return this;

server/src/main/java/org/elasticsearch/action/fieldcaps/RequestDispatcher.java

@@ -179,7 +179,8 @@ final class RequestDispatcher {
             originalIndices,
             fieldCapsRequest.indexFilter(),
             nowInMillis,
-            fieldCapsRequest.runtimeFields()
+            fieldCapsRequest.runtimeFields(),
+            fieldCapsRequest.includeEmptyFields()
         );
         transportService.sendChildRequest(
             node,
@@ -203,7 +204,10 @@ final class RequestDispatcher {
     private void onRequestResponse(List<ShardId> shardIds, FieldCapabilitiesNodeResponse nodeResponse) {
         for (FieldCapabilitiesIndexResponse indexResponse : nodeResponse.getIndexResponses()) {
             if (indexResponse.canMatch()) {
-                if (indexSelectors.remove(indexResponse.getIndexName()) != null) {
+                if (fieldCapsRequest.includeEmptyFields() == false) {
+                    // we accept all the responses because they may vary from node to node if we exclude empty fields
+                    onIndexResponse.accept(indexResponse);
+                } else if (indexSelectors.remove(indexResponse.getIndexName()) != null) {
                     onIndexResponse.accept(indexResponse);
                 }
             }

server/src/main/java/org/elasticsearch/action/fieldcaps/TransportFieldCapabilitiesAction.java

@@ -167,7 +167,18 @@ public class TransportFieldCapabilitiesAction extends HandledTransportAction<Fie
                     resp = new FieldCapabilitiesIndexResponse(resp.getIndexName(), curr.getIndexMappingHash(), curr.get(), true);
                 }
             }
-            indexResponses.putIfAbsent(resp.getIndexName(), resp);
+            if (request.includeEmptyFields()) {
+                indexResponses.putIfAbsent(resp.getIndexName(), resp);
+            } else {
+                indexResponses.merge(resp.getIndexName(), resp, (a, b) -> {
+                    if (a.get().equals(b.get())) {
+                        return a;
+                    }
+                    Map<String, IndexFieldCapabilities> mergedCaps = new HashMap<>(a.get());
+                    mergedCaps.putAll(b.get());
+                    return new FieldCapabilitiesIndexResponse(a.getIndexName(), a.getIndexMappingHash(), mergedCaps, true);
+                });
+            }
             if (fieldCapTask.isCancelled()) {
                 releaseResourcesOnCancel.run();
             }
@@ -294,6 +305,7 @@ public class TransportFieldCapabilitiesAction extends HandledTransportAction<Fie
         remoteRequest.runtimeFields(request.runtimeFields());
         remoteRequest.indexFilter(request.indexFilter());
         remoteRequest.nowInMillis(nowInMillis);
+        remoteRequest.includeEmptyFields(request.includeEmptyFields());
         return remoteRequest;
     }
 
@@ -519,7 +531,7 @@ public class TransportFieldCapabilitiesAction extends HandledTransportAction<Fie
                 final Map<String, List<ShardId>> groupedShardIds = request.shardIds()
                     .stream()
                     .collect(Collectors.groupingBy(ShardId::getIndexName));
-                final FieldCapabilitiesFetcher fetcher = new FieldCapabilitiesFetcher(indicesService);
+                final FieldCapabilitiesFetcher fetcher = new FieldCapabilitiesFetcher(indicesService, request.includeEmptyFields());
                 final Predicate<String> fieldNameFilter = Regex.simpleMatcher(request.fields());
                 for (List<ShardId> shardIds : groupedShardIds.values()) {
                     final Map<ShardId, Exception> failures = new HashMap<>();
@@ -537,10 +549,12 @@ public class TransportFieldCapabilitiesAction extends HandledTransportAction<Fie
                                 request.runtimeFields()
                             );
                             if (response.canMatch()) {
-                                unmatched.clear();
-                                failures.clear();
                                 allResponses.add(response);
-                                break;
+                                if (request.includeEmptyFields()) {
+                                    unmatched.clear();
+                                    failures.clear();
+                                    break;
+                                }
                             } else {
                                 unmatched.add(shardId);
                             }

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

@@ -9,6 +9,7 @@
 package org.elasticsearch.index.mapper;
 
 import org.apache.lucene.analysis.TokenStream;
+import org.apache.lucene.index.FieldInfos;
 import org.apache.lucene.queries.spans.SpanMultiTermQueryWrapper;
 import org.apache.lucene.queries.spans.SpanQuery;
 import org.apache.lucene.search.MultiTermQuery;
@@ -217,6 +218,14 @@ public abstract class AbstractScriptFieldType<LeafFactory> extends MappedFieldTy
         );
     }
 
+    @Override
+    public final boolean fieldHasValue(FieldInfos fieldInfos) {
+        // To know whether script field types have value we would need to run the script,
+        // this because script fields do not have footprint in Lucene. Since running the
+        // script would be too expensive for _field_caps we consider them as always non-empty.
+        return true;
+    }
+
     // Placeholder Script for source-only fields
     // TODO rework things so that we don't need this
     protected static final Script DEFAULT_SCRIPT = new Script("");

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

@@ -8,6 +8,7 @@
 
 package org.elasticsearch.index.mapper;
 
+import org.apache.lucene.index.FieldInfos;
 import org.apache.lucene.search.MatchAllDocsQuery;
 import org.apache.lucene.search.MatchNoDocsQuery;
 import org.apache.lucene.search.MultiTermQuery;
@@ -132,4 +133,10 @@ public abstract class ConstantFieldType extends MappedFieldType {
             return new MatchNoDocsQuery();
         }
     }
+
+    @Override
+    public final boolean fieldHasValue(FieldInfos fieldInfos) {
+        // We consider constant field types to always have value.
+        return true;
+    }
 }

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

@@ -9,6 +9,8 @@
 package org.elasticsearch.index.mapper;
 
 import org.apache.lucene.analysis.TokenStream;
+import org.apache.lucene.index.FieldInfo;
+import org.apache.lucene.index.FieldInfos;
 import org.apache.lucene.index.IndexReader;
 import org.apache.lucene.index.PrefixCodedTerms;
 import org.apache.lucene.index.PrefixCodedTerms.TermIterator;
@@ -632,6 +634,24 @@ public abstract class MappedFieldType {
         );
     }
 
+    /**
+     * This method is used to support _field_caps when include_empty_fields is set to
+     * {@code false}. In that case we return only fields with value in an index. This method
+     * gets as input FieldInfos and returns if the field is non-empty. This method needs to
+     * be overwritten where fields don't have footprint in Lucene or their name differs from
+     * {@link MappedFieldType#name()}
+     * @param fieldInfos field information
+     * @return {@code true} if field is present in fieldInfos {@code false} otherwise
+     */
+    public boolean fieldHasValue(FieldInfos fieldInfos) {
+        for (FieldInfo fieldInfo : fieldInfos) {
+            if (fieldInfo.getName().equals(name())) {
+                return true;
+            }
+        }
+        return false;
+    }
+
     /**
      * Returns a loader for ESQL or {@code null} if the field doesn't support
      * ESQL.

server/src/main/java/org/elasticsearch/index/shard/IndexShard.java

@@ -13,6 +13,7 @@ import org.apache.lucene.analysis.Analyzer;
 import org.apache.lucene.analysis.DelegatingAnalyzerWrapper;
 import org.apache.lucene.index.CheckIndex;
 import org.apache.lucene.index.DirectoryReader;
+import org.apache.lucene.index.FieldInfos;
 import org.apache.lucene.index.FilterDirectoryReader;
 import org.apache.lucene.index.IndexCommit;
 import org.apache.lucene.index.LeafReader;
@@ -223,6 +224,12 @@ public class IndexShard extends AbstractIndexShardComponent implements IndicesCl
     private final ReplicationTracker replicationTracker;
     private final IndexStorePlugin.SnapshotCommitSupplier snapshotCommitSupplier;
     private final Engine.IndexCommitListener indexCommitListener;
+    private FieldInfos fieldInfos;
+    // sys prop to disable the field has value feature, defaults to true (enabled) if set to false (disabled) the
+    // field caps always returns empty fields ignoring the value of the query param `field_caps_empty_fields_filter`.
+    private final boolean enableFieldHasValue = Booleans.parseBoolean(
+        System.getProperty("es.field_caps_empty_fields_filter", Boolean.TRUE.toString())
+    );
 
     protected volatile ShardRouting shardRouting;
     protected volatile IndexShardState state;
@@ -281,6 +288,7 @@ public class IndexShard extends AbstractIndexShardComponent implements IndicesCl
     private final AtomicLong lastSearcherAccess = new AtomicLong();
     private final AtomicReference<Translog.Location> pendingRefreshLocation = new AtomicReference<>();
     private final RefreshPendingLocationListener refreshPendingLocationListener;
+    private final RefreshFieldHasValueListener refreshFieldHasValueListener;
     private volatile boolean useRetentionLeasesInPeerRecovery;
     private final LongSupplier relativeTimeInNanosSupplier;
     private volatile long startedRelativeTimeInNanos;
@@ -396,8 +404,10 @@ public class IndexShard extends AbstractIndexShardComponent implements IndicesCl
         persistMetadata(path, indexSettings, shardRouting, null, logger);
         this.useRetentionLeasesInPeerRecovery = replicationTracker.hasAllPeerRecoveryRetentionLeases();
         this.refreshPendingLocationListener = new RefreshPendingLocationListener();
+        this.refreshFieldHasValueListener = new RefreshFieldHasValueListener();
         this.relativeTimeInNanosSupplier = relativeTimeInNanosSupplier;
         this.indexCommitListener = indexCommitListener;
+        this.fieldInfos = FieldInfos.EMPTY;
     }
 
     public ThreadPool getThreadPool() {
@@ -983,10 +993,17 @@ public class IndexShard extends AbstractIndexShardComponent implements IndicesCl
             verifyNotClosed(e);
             return new Engine.IndexResult(e, version, opPrimaryTerm, seqNo, sourceToParse.id());
         }
-
         return index(engine, operation);
     }
 
+    public void setFieldInfos(FieldInfos fieldInfos) {
+        this.fieldInfos = fieldInfos;
+    }
+
+    public FieldInfos getFieldInfos() {
+        return fieldInfos;
+    }
+
     public static Engine.Index prepareIndex(
         MapperService mapperService,
         SourceToParse source,
@@ -3433,7 +3450,7 @@ public class IndexShard extends AbstractIndexShardComponent implements IndicesCl
             cachingPolicy,
             translogConfig,
             IndexingMemoryController.SHARD_INACTIVE_TIME_SETTING.get(indexSettings.getSettings()),
-            List.of(refreshListeners, refreshPendingLocationListener),
+            List.of(refreshListeners, refreshPendingLocationListener, refreshFieldHasValueListener),
             Collections.singletonList(new RefreshMetricUpdater(refreshMetric)),
             indexSort,
             circuitBreakerService,
@@ -3987,6 +4004,20 @@ public class IndexShard extends AbstractIndexShardComponent implements IndicesCl
         }
     }
 
+    private class RefreshFieldHasValueListener implements ReferenceManager.RefreshListener {
+        @Override
+        public void beforeRefresh() {}
+
+        @Override
+        public void afterRefresh(boolean didRefresh) {
+            if (enableFieldHasValue) {
+                try (Engine.Searcher hasValueSearcher = getEngine().acquireSearcher("field_has_value")) {
+                    setFieldInfos(FieldInfos.getMergedFieldInfos(hasValueSearcher.getIndexReader()));
+                }
+            }
+        }
+    }
+
     /**
      * Ensures this shard is search active before invoking the provided listener.
      * <p>

server/src/main/java/org/elasticsearch/indices/IndicesService.java

@@ -577,7 +577,7 @@ public class IndicesService extends AbstractLifecycleComponent
      * Creates a new {@link IndexService} for the given metadata.
      *
      * @param indexMetadata          the index metadata to create the index for
-     * @param builtInListeners       a list of built-in lifecycle {@link IndexEventListener} that should should be used along side with the
+     * @param builtInListeners       a list of built-in lifecycle {@link IndexEventListener} that should be used alongside with the
      *                               per-index listeners
      * @throws ResourceAlreadyExistsException if the index already exists.
      */

server/src/main/java/org/elasticsearch/rest/action/RestFieldCapabilitiesAction.java

@@ -53,6 +53,7 @@ public class RestFieldCapabilitiesAction extends BaseRestHandler {
 
         fieldRequest.indicesOptions(IndicesOptions.fromRequest(request, fieldRequest.indicesOptions()));
         fieldRequest.includeUnmapped(request.paramAsBoolean("include_unmapped", false));
+        fieldRequest.includeEmptyFields(request.paramAsBoolean("include_empty_fields", true));
         fieldRequest.filters(request.paramAsStringArray("filters", Strings.EMPTY_ARRAY));
         fieldRequest.types(request.paramAsStringArray("types", Strings.EMPTY_ARRAY));
         request.withContentOrSourceParamParserOrNull(parser -> {

server/src/test/java/org/elasticsearch/action/fieldcaps/FieldCapabilitiesFilterTests.java

@@ -8,15 +8,20 @@
 
 package org.elasticsearch.action.fieldcaps;
 
+import org.apache.lucene.index.FieldInfos;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.index.mapper.MapperService;
 import org.elasticsearch.index.mapper.MapperServiceTestCase;
 import org.elasticsearch.index.query.SearchExecutionContext;
+import org.elasticsearch.index.shard.IndexShard;
 
 import java.io.IOException;
 import java.util.Map;
 import java.util.function.Predicate;
 
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
 public class FieldCapabilitiesFilterTests extends MapperServiceTestCase {
 
     public void testExcludeNestedFields() throws IOException {
@@ -41,7 +46,9 @@ public class FieldCapabilitiesFilterTests extends MapperServiceTestCase {
             s -> true,
             new String[] { "-nested" },
             Strings.EMPTY_ARRAY,
-            f -> true
+            f -> true,
+            getMockIndexShard(),
+            true
         );
 
         assertNotNull(response.get("field1"));
@@ -67,7 +74,9 @@ public class FieldCapabilitiesFilterTests extends MapperServiceTestCase {
                 s -> true,
                 new String[] { "+metadata" },
                 Strings.EMPTY_ARRAY,
-                f -> true
+                f -> true,
+                getMockIndexShard(),
+                true
             );
             assertNotNull(response.get("_index"));
             assertNull(response.get("field1"));
@@ -78,7 +87,9 @@ public class FieldCapabilitiesFilterTests extends MapperServiceTestCase {
                 s -> true,
                 new String[] { "-metadata" },
                 Strings.EMPTY_ARRAY,
-                f -> true
+                f -> true,
+                getMockIndexShard(),
+                true
             );
             assertNull(response.get("_index"));
             assertNotNull(response.get("field1"));
@@ -109,7 +120,9 @@ public class FieldCapabilitiesFilterTests extends MapperServiceTestCase {
             s -> true,
             new String[] { "-multifield" },
             Strings.EMPTY_ARRAY,
-            f -> true
+            f -> true,
+            getMockIndexShard(),
+            true
         );
         assertNotNull(response.get("field1"));
         assertNull(response.get("field1.keyword"));
@@ -138,7 +151,9 @@ public class FieldCapabilitiesFilterTests extends MapperServiceTestCase {
             s -> true,
             new String[] { "-parent" },
             Strings.EMPTY_ARRAY,
-            f -> true
+            f -> true,
+            getMockIndexShard(),
+            true
         );
         assertNotNull(response.get("parent.field1"));
         assertNotNull(response.get("parent.field2"));
@@ -164,7 +179,9 @@ public class FieldCapabilitiesFilterTests extends MapperServiceTestCase {
                 s -> true,
                 Strings.EMPTY_ARRAY,
                 Strings.EMPTY_ARRAY,
-                securityFilter
+                securityFilter,
+                getMockIndexShard(),
+                true
             );
 
             assertNotNull(response.get("permitted1"));
@@ -178,7 +195,9 @@ public class FieldCapabilitiesFilterTests extends MapperServiceTestCase {
                 s -> true,
                 new String[] { "-metadata" },
                 Strings.EMPTY_ARRAY,
-                securityFilter
+                securityFilter,
+                getMockIndexShard(),
+                true
             );
 
             assertNotNull(response.get("permitted1"));
@@ -204,11 +223,20 @@ public class FieldCapabilitiesFilterTests extends MapperServiceTestCase {
             s -> true,
             Strings.EMPTY_ARRAY,
             new String[] { "text", "keyword" },
-            f -> true
+            f -> true,
+            getMockIndexShard(),
+            true
         );
         assertNotNull(response.get("field1"));
         assertNull(response.get("field2"));
         assertNotNull(response.get("field3"));
         assertNull(response.get("_index"));
     }
+
+    private IndexShard getMockIndexShard() {
+        IndexShard indexShard = mock(IndexShard.class);
+        when(indexShard.getFieldInfos()).thenReturn(FieldInfos.EMPTY);
+        return indexShard;
+    }
+
 }

server/src/test/java/org/elasticsearch/action/fieldcaps/FieldCapabilitiesNodeRequestTests.java

@@ -52,7 +52,8 @@ public class FieldCapabilitiesNodeRequestTests extends AbstractWireSerializingTe
             originalIndices,
             indexFilter,
             nowInMillis,
-            runtimeFields
+            runtimeFields,
+            true
         );
     }
 
@@ -94,7 +95,7 @@ public class FieldCapabilitiesNodeRequestTests extends AbstractWireSerializingTe
 
     @Override
     protected FieldCapabilitiesNodeRequest mutateInstance(FieldCapabilitiesNodeRequest instance) {
-        switch (random().nextInt(7)) {
+        switch (random().nextInt(8)) {
             case 0 -> {
                 List<ShardId> shardIds = randomShardIds(instance.shardIds().size() + 1);
                 return new FieldCapabilitiesNodeRequest(
@@ -105,7 +106,8 @@ public class FieldCapabilitiesNodeRequestTests extends AbstractWireSerializingTe
                     instance.originalIndices(),
                     instance.indexFilter(),
                     instance.nowInMillis(),
-                    instance.runtimeFields()
+                    instance.runtimeFields(),
+                    true
                 );
             }
             case 1 -> {
@@ -118,7 +120,8 @@ public class FieldCapabilitiesNodeRequestTests extends AbstractWireSerializingTe
                     instance.originalIndices(),
                     instance.indexFilter(),
                     instance.nowInMillis(),
-                    instance.runtimeFields()
+                    instance.runtimeFields(),
+                    true
                 );
             }
             case 2 -> {
@@ -131,7 +134,8 @@ public class FieldCapabilitiesNodeRequestTests extends AbstractWireSerializingTe
                     originalIndices,
                     instance.indexFilter(),
                     instance.nowInMillis(),
-                    instance.runtimeFields()
+                    instance.runtimeFields(),
+                    true
                 );
             }
             case 3 -> {
@@ -144,7 +148,8 @@ public class FieldCapabilitiesNodeRequestTests extends AbstractWireSerializingTe
                     instance.originalIndices(),
                     indexFilter,
                     instance.nowInMillis(),
-                    instance.runtimeFields()
+                    instance.runtimeFields(),
+                    true
                 );
             }
             case 4 -> {
@@ -157,7 +162,8 @@ public class FieldCapabilitiesNodeRequestTests extends AbstractWireSerializingTe
                     instance.originalIndices(),
                     instance.indexFilter(),
                     nowInMillis,
-                    instance.runtimeFields()
+                    instance.runtimeFields(),
+                    true
                 );
             }
             case 5 -> {
@@ -172,7 +178,8 @@ public class FieldCapabilitiesNodeRequestTests extends AbstractWireSerializingTe
                     instance.originalIndices(),
                     instance.indexFilter(),
                     instance.nowInMillis(),
-                    runtimeFields
+                    runtimeFields,
+                    true
                 );
             }
             case 6 -> {
@@ -185,7 +192,8 @@ public class FieldCapabilitiesNodeRequestTests extends AbstractWireSerializingTe
                     instance.originalIndices(),
                     instance.indexFilter(),
                     instance.nowInMillis(),
-                    instance.runtimeFields()
+                    instance.runtimeFields(),
+                    true
                 );
             }
             case 7 -> {
@@ -198,10 +206,24 @@ public class FieldCapabilitiesNodeRequestTests extends AbstractWireSerializingTe
                     instance.originalIndices(),
                     instance.indexFilter(),
                     instance.nowInMillis(),
-                    instance.runtimeFields()
+                    instance.runtimeFields(),
+                    true
                 );
             }
-            default -> throw new IllegalStateException("The test should only allow 7 parameters mutated");
+            case 8 -> {
+                return new FieldCapabilitiesNodeRequest(
+                    instance.shardIds(),
+                    instance.fields(),
+                    instance.filters(),
+                    instance.allowedTypes(),
+                    instance.originalIndices(),
+                    instance.indexFilter(),
+                    instance.nowInMillis(),
+                    instance.runtimeFields(),
+                    false
+                );
+            }
+            default -> throw new IllegalStateException("The test should only allow 8 parameters mutated");
         }
     }
 
@@ -214,9 +236,13 @@ public class FieldCapabilitiesNodeRequestTests extends AbstractWireSerializingTe
             randomOriginalIndices(1),
             null,
             randomNonNegativeLong(),
-            Map.of()
+            Map.of(),
+            true
+        );
+        assertThat(
+            r1.getDescription(),
+            equalTo("shards[[index-1][0],[index-2][3]], fields[field-1,field-2], filters[], types[], includeEmptyFields[true]")
         );
-        assertThat(r1.getDescription(), equalTo("shards[[index-1][0],[index-2][3]], fields[field-1,field-2], filters[], types[]"));
 
         FieldCapabilitiesNodeRequest r2 = new FieldCapabilitiesNodeRequest(
             List.of(new ShardId("index-1", "n/a", 0)),
@@ -226,8 +252,12 @@ public class FieldCapabilitiesNodeRequestTests extends AbstractWireSerializingTe
             randomOriginalIndices(1),
             null,
             randomNonNegativeLong(),
-            Map.of()
+            Map.of(),
+            false
+        );
+        assertThat(
+            r2.getDescription(),
+            equalTo("shards[[index-1][0]], fields[*], filters[-nested,-metadata], types[], includeEmptyFields[false]")
         );
-        assertThat(r2.getDescription(), equalTo("shards[[index-1][0]], fields[*], filters[-nested,-metadata], types[]"));
     }
 }

server/src/test/java/org/elasticsearch/action/fieldcaps/FieldCapabilitiesRequestTests.java

@@ -171,20 +171,23 @@ public class FieldCapabilitiesRequestTests extends AbstractWireSerializingTestCa
 
     public void testGetDescription() {
         final FieldCapabilitiesRequest request = new FieldCapabilitiesRequest();
-        assertThat(request.getDescription(), equalTo("indices[], fields[], filters[], types[]"));
+        assertThat(request.getDescription(), equalTo("indices[], fields[], filters[], types[], includeEmptyFields[true]"));
 
         request.fields("a", "b");
         assertThat(
             request.getDescription(),
-            anyOf(equalTo("indices[], fields[a,b], filters[], types[]"), equalTo("indices[], fields[b,a], filters[], types[]"))
+            anyOf(
+                equalTo("indices[], fields[a,b], filters[], types[], includeEmptyFields[true]"),
+                equalTo("indices[], fields[b,a], filters[], types[], includeEmptyFields[true]")
+            )
         );
 
         request.indices("x", "y", "z");
         request.fields("a");
-        assertThat(request.getDescription(), equalTo("indices[x,y,z], fields[a], filters[], types[]"));
+        assertThat(request.getDescription(), equalTo("indices[x,y,z], fields[a], filters[], types[], includeEmptyFields[true]"));
 
         request.filters("-metadata", "-multifields");
-        assertThat(request.getDescription(), endsWith("filters[-metadata,-multifields], types[]"));
+        assertThat(request.getDescription(), endsWith("filters[-metadata,-multifields], types[], includeEmptyFields[true]"));
 
         final String[] lots = new String[between(1024, 2048)];
         for (int i = 0; i < lots.length; i++) {
@@ -205,7 +208,10 @@ public class FieldCapabilitiesRequestTests extends AbstractWireSerializingTestCa
         );
         assertThat(
             request.getDescription().length(),
-            lessThanOrEqualTo(1024 + ("indices[x,y,z], fields[" + "s9999,... (9999 in total, 9999 omitted)], filters[], types[]").length())
+            lessThanOrEqualTo(
+                1024 + ("indices[x,y,z], fields["
+                    + "s9999,... (9999 in total, 9999 omitted)], filters[], types[], includeEmptyFields[true]").length()
+            )
         );
 
         request.fields("a");
@@ -217,12 +223,15 @@ public class FieldCapabilitiesRequestTests extends AbstractWireSerializingTestCa
                 containsString("..."),
                 containsString(lots.length + " in total"),
                 containsString("omitted"),
-                endsWith("], fields[a], filters[], types[]")
+                endsWith("], fields[a], filters[], types[], includeEmptyFields[true]")
             )
         );
         assertThat(
             request.getDescription().length(),
-            lessThanOrEqualTo(1024 + ("indices[" + "s9999,... (9999 in total, 9999 omitted)], fields[a], filters[], types[]").length())
+            lessThanOrEqualTo(
+                1024 + ("indices[" + "s9999,... (9999 in total, 9999 omitted)], fields[a], filters[], types[], includeEmptyFields[true]")
+                    .length()
+            )
         );
 
         final FieldCapabilitiesRequest randomRequest = createTestInstance();

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

@@ -16,14 +16,13 @@ import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.index.IndexVersion;
 import org.elasticsearch.index.query.QueryShardException;
 import org.elasticsearch.index.query.SearchExecutionContext;
-import org.elasticsearch.test.ESTestCase;
 
 import java.util.Collections;
 import java.util.function.Predicate;
 
 import static org.hamcrest.Matchers.containsString;
 
-public class IndexFieldTypeTests extends ESTestCase {
+public class IndexFieldTypeTests extends ConstantFieldTypeTestCase {
 
     public void testPrefixQuery() {
         MappedFieldType ft = IndexFieldMapper.IndexFieldType.INSTANCE;
@@ -51,6 +50,11 @@ public class IndexFieldTypeTests extends ESTestCase {
         assertThat(e.getMessage(), containsString("Can only use regexp queries on keyword and text fields"));
     }
 
+    @Override
+    public MappedFieldType getMappedFieldType() {
+        return IndexFieldMapper.IndexFieldType.INSTANCE;
+    }
+
     private SearchExecutionContext createContext() {
         IndexMetadata indexMetadata = IndexMetadata.builder("index")
             .settings(Settings.builder().put(IndexMetadata.SETTING_VERSION_CREATED, IndexVersion.current()))

test/framework/src/main/java/org/elasticsearch/index/mapper/AbstractScriptFieldTypeTestCase.java

@@ -10,6 +10,8 @@ package org.elasticsearch.index.mapper;
 
 import org.apache.lucene.document.StoredField;
 import org.apache.lucene.index.DirectoryReader;
+import org.apache.lucene.index.FieldInfo;
+import org.apache.lucene.index.FieldInfos;
 import org.apache.lucene.index.IndexReader;
 import org.apache.lucene.index.LeafReaderContext;
 import org.apache.lucene.search.IndexSearcher;
@@ -198,6 +200,21 @@ public abstract class AbstractScriptFieldTypeTestCase extends MapperServiceTestC
         }
     }
 
+    @Override
+    public void testFieldHasValue() {
+        assertTrue(getMappedFieldType().fieldHasValue(new FieldInfos(new FieldInfo[] { getFieldInfoWithName(randomAlphaOfLength(5)) })));
+    }
+
+    @Override
+    public void testFieldHasValueWithEmptyFieldInfos() {
+        assertTrue(getMappedFieldType().fieldHasValue(FieldInfos.EMPTY));
+    }
+
+    @Override
+    public MappedFieldType getMappedFieldType() {
+        return simpleMappedFieldType();
+    }
+
     protected abstract AbstractScriptFieldType<?> build(String error, Map<String, Object> emptyMap, OnScriptError onScriptError);
 
     @SuppressWarnings("unused")

test/framework/src/main/java/org/elasticsearch/index/mapper/ConstantFieldTypeTestCase.java

@@ -0,0 +1,24 @@
+/*
+ * 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;
+
+import org.apache.lucene.index.FieldInfo;
+import org.apache.lucene.index.FieldInfos;
+
+public class ConstantFieldTypeTestCase extends FieldTypeTestCase {
+    @Override
+    public void testFieldHasValue() {
+        assertTrue(getMappedFieldType().fieldHasValue(new FieldInfos(new FieldInfo[] { getFieldInfoWithName(randomAlphaOfLength(5)) })));
+    }
+
+    @Override
+    public void testFieldHasValueWithEmptyFieldInfos() {
+        assertTrue(getMappedFieldType().fieldHasValue(FieldInfos.EMPTY));
+    }
+}

test/framework/src/main/java/org/elasticsearch/index/mapper/FieldTypeTestCase.java

@@ -7,6 +7,13 @@
  */
 package org.elasticsearch.index.mapper;
 
+import org.apache.lucene.index.DocValuesType;
+import org.apache.lucene.index.FieldInfo;
+import org.apache.lucene.index.FieldInfos;
+import org.apache.lucene.index.IndexOptions;
+import org.apache.lucene.index.VectorEncoding;
+import org.apache.lucene.index.VectorSimilarityFunction;
+import org.apache.lucene.search.Query;
 import org.elasticsearch.index.IndexVersion;
 import org.elasticsearch.index.query.SearchExecutionContext;
 import org.elasticsearch.search.lookup.FieldLookup;
@@ -20,6 +27,7 @@ import org.elasticsearch.xcontent.XContentType;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Set;
 
@@ -89,4 +97,56 @@ public abstract class FieldTypeTestCase extends ESTestCase {
         fetcher.setNextReader(null);
         return fetcher.fetchValues(null, -1, new ArrayList<>());
     }
+
+    public void testFieldHasValue() {
+        MappedFieldType fieldType = getMappedFieldType();
+        FieldInfos fieldInfos = new FieldInfos(new FieldInfo[] { getFieldInfoWithName("field") });
+        assertTrue(fieldType.fieldHasValue(fieldInfos));
+    }
+
+    public void testFieldHasValueWithEmptyFieldInfos() {
+        MappedFieldType fieldType = getMappedFieldType();
+        assertFalse(fieldType.fieldHasValue(FieldInfos.EMPTY));
+    }
+
+    public MappedFieldType getMappedFieldType() {
+        return new MappedFieldType("field", false, false, false, TextSearchInfo.NONE, Collections.emptyMap()) {
+
+            @Override
+            public ValueFetcher valueFetcher(SearchExecutionContext context, String format) {
+                return null;
+            }
+
+            @Override
+            public String typeName() {
+                return null;
+            }
+
+            @Override
+            public Query termQuery(Object value, SearchExecutionContext context) {
+                return null;
+            }
+        };
+    }
+
+    public FieldInfo getFieldInfoWithName(String name) {
+        return new FieldInfo(
+            name,
+            1,
+            randomBoolean(),
+            randomBoolean(),
+            randomBoolean(),
+            IndexOptions.NONE,
+            DocValuesType.NONE,
+            -1,
+            new HashMap<>(),
+            1,
+            1,
+            1,
+            1,
+            VectorEncoding.BYTE,
+            VectorSimilarityFunction.COSINE,
+            randomBoolean()
+        );
+    }
 }

test/framework/src/main/java/org/elasticsearch/index/mapper/MapperServiceTestCase.java

@@ -72,7 +72,6 @@ import org.elasticsearch.search.sort.BucketedSort;
 import org.elasticsearch.search.sort.BucketedSort.ExtraData;
 import org.elasticsearch.search.sort.SortAndFormats;
 import org.elasticsearch.search.sort.SortBuilder;
-import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.test.FieldMaskingReader;
 import org.elasticsearch.xcontent.ToXContent;
 import org.elasticsearch.xcontent.XContentBuilder;
@@ -101,7 +100,7 @@ import static org.mockito.Mockito.mock;
  * mapping. Useful when you don't need to spin up an entire index but do
  * need most of the trapping of the mapping.
  */
-public abstract class MapperServiceTestCase extends ESTestCase {
+public abstract class MapperServiceTestCase extends FieldTypeTestCase {
 
     protected static final Settings SETTINGS = Settings.builder()
         .put(IndexMetadata.SETTING_VERSION_CREATED, IndexVersion.current())

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/cluster/routing/allocation/mapper/DataTierFieldTypeTests.java

@@ -7,6 +7,8 @@
 
 package org.elasticsearch.xpack.cluster.routing.allocation.mapper;
 
+import org.apache.lucene.index.FieldInfo;
+import org.apache.lucene.index.FieldInfos;
 import org.apache.lucene.search.MatchAllDocsQuery;
 import org.apache.lucene.search.MatchNoDocsQuery;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
@@ -125,4 +127,19 @@ public class DataTierFieldTypeTests extends MapperServiceTestCase {
         IndexSettings indexSettings = new IndexSettings(indexMetadata, Settings.EMPTY);
         return SearchExecutionContextHelper.createSimple(indexSettings, parserConfig(), writableRegistry());
     }
+
+    @Override
+    public void testFieldHasValue() {
+        assertTrue(getMappedFieldType().fieldHasValue(new FieldInfos(new FieldInfo[] { getFieldInfoWithName(randomAlphaOfLength(5)) })));
+    }
+
+    @Override
+    public void testFieldHasValueWithEmptyFieldInfos() {
+        assertTrue(getMappedFieldType().fieldHasValue(FieldInfos.EMPTY));
+    }
+
+    @Override
+    public MappedFieldType getMappedFieldType() {
+        return DataTierFieldMapper.DataTierFieldType.INSTANCE;
+    }
 }

x-pack/plugin/mapper-constant-keyword/src/test/java/org/elasticsearch/xpack/constantkeyword/mapper/ConstantKeywordFieldTypeTests.java

@@ -11,7 +11,7 @@ import org.apache.lucene.search.MatchAllDocsQuery;
 import org.apache.lucene.search.MatchNoDocsQuery;
 import org.apache.lucene.util.automaton.RegExp;
 import org.elasticsearch.common.unit.Fuzziness;
-import org.elasticsearch.index.mapper.FieldTypeTestCase;
+import org.elasticsearch.index.mapper.ConstantFieldTypeTestCase;
 import org.elasticsearch.index.mapper.MappedFieldType;
 import org.elasticsearch.index.mapper.ValueFetcher;
 import org.elasticsearch.search.lookup.Source;
@@ -24,7 +24,7 @@ import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 
-public class ConstantKeywordFieldTypeTests extends FieldTypeTestCase {
+public class ConstantKeywordFieldTypeTests extends ConstantFieldTypeTestCase {
 
     public void testTermQuery() {
         ConstantKeywordFieldType ft = new ConstantKeywordFieldType("f", "foo");
@@ -126,4 +126,9 @@ public class ConstantKeywordFieldTypeTests extends FieldTypeTestCase {
         assertEquals(List.of("foo"), fetcher.fetchValues(sourceWithNoFieldValue, -1, ignoredValues));
         assertEquals(List.of("foo"), fetcher.fetchValues(sourceWithNullFieldValue, -1, ignoredValues));
     }
+
+    @Override
+    public MappedFieldType getMappedFieldType() {
+        return new ConstantKeywordFieldMapper.ConstantKeywordFieldType(randomAlphaOfLength(5), randomAlphaOfLength(5));
+    }
 }