Bulk merge field-caps responses using mapping hash (#86323)

The most common usage of field-caps is retrieving the field-caps of 
group indices having the same index mappings. We can speed up the
merging process by performing bulk merges for index responses with the
same mapping hash.

This change reduces the response time by 10 times in the many_shards 
benchmark.

GET /auditbeat*/_field_caps?fields=* (single index mapping)
|  50th percentile latency |     field-caps |  4420.91  |  374.729  |   -4046.19  | ms |  -91.52% |
|  90th percentile latency |     field-caps |  5126.87  |  402.883  |   -4723.98  | ms |  -92.14% |
|  99th percentile latency |     field-caps |  5529.41  |  576.324  |   -4953.08  | ms |  -89.58% |
| 100th percentile latency |     field-caps |  6096.73  |  643.252  |   -5453.48  | ms |  -89.45% |

GET /*/_field_caps?fields=* * (i.e. multiple index mappings)
|  50th percentile latency | field-caps-all |  4475.04  |  395.844  |   -4079.2   | ms |  -91.15% |
|  90th percentile latency | field-caps-all |  5334.01  |  425.248  |   -4908.76  | ms |  -92.03% |
|  99th percentile latency | field-caps-all |  5628.16  |  606.959  |   -5021.2   | ms |  -89.22% |
| 100th percentile latency | field-caps-all |  6292.63  |  675.807  |   -5616.82  | ms |  -89.26% |

docs/changelog/86323.yaml

@@ -0,0 +1,5 @@
+pr: 86323
+summary: Bulk merge field-caps responses using mapping hash
+area: Search
+type: enhancement
+issues: []

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

@@ -58,8 +58,10 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Predicate;
+import java.util.stream.IntStream;
 
 import static java.util.Collections.singletonList;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
@@ -70,6 +72,8 @@ import static org.hamcrest.Matchers.containsInAnyOrder;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.hasKey;
 import static org.hamcrest.Matchers.hasSize;
+import static org.hamcrest.Matchers.not;
+import static org.hamcrest.Matchers.nullValue;
 
 public class FieldCapabilitiesIT extends ESIntegTestCase {
 
@@ -575,6 +579,67 @@ public class FieldCapabilitiesIT extends ESIntegTestCase {
         }
     }
 
+    public void testManyIndicesWithSameMapping() {
+        final String mapping = """
+             {
+                 "properties": {
+                   "message_field": { "type": "text" },
+                   "value_field": { "type": "long" },
+                   "multi_field" : { "type" : "ip", "fields" : { "keyword" : { "type" : "keyword" } } },
+                   "timestamp": {"type": "date"}
+                 }
+             }
+            """;
+        String[] indices = IntStream.range(0, between(1, 9)).mapToObj(n -> "test_many_index_" + n).toArray(String[]::new);
+        for (String index : indices) {
+            assertAcked(client().admin().indices().prepareCreate(index).setMapping(mapping).get());
+        }
+        FieldCapabilitiesRequest request = new FieldCapabilitiesRequest();
+        request.indices("test_many_index_*");
+        request.fields("*");
+        boolean excludeMultiField = randomBoolean();
+        if (excludeMultiField) {
+            request.filters("-multifield");
+        }
+        Consumer<FieldCapabilitiesResponse> verifyResponse = resp -> {
+            assertThat(resp.getIndices(), equalTo(indices));
+            assertThat(resp.getField("message_field"), hasKey("text"));
+            assertThat(resp.getField("message_field").get("text").indices(), nullValue());
+            assertTrue(resp.getField("message_field").get("text").isSearchable());
+            assertFalse(resp.getField("message_field").get("text").isAggregatable());
+
+            assertThat(resp.getField("value_field"), hasKey("long"));
+            assertThat(resp.getField("value_field").get("long").indices(), nullValue());
+            assertTrue(resp.getField("value_field").get("long").isSearchable());
+            assertTrue(resp.getField("value_field").get("long").isAggregatable());
+
+            assertThat(resp.getField("timestamp"), hasKey("date"));
+
+            assertThat(resp.getField("multi_field"), hasKey("ip"));
+            if (excludeMultiField) {
+                assertThat(resp.getField("multi_field.keyword"), not(hasKey("keyword")));
+            } else {
+                assertThat(resp.getField("multi_field.keyword"), hasKey("keyword"));
+            }
+        };
+        // Single mapping
+        verifyResponse.accept(client().execute(FieldCapabilitiesAction.INSTANCE, request).actionGet());
+
+        // add an extra field for some indices
+        String[] indicesWithExtraField = randomSubsetOf(between(1, indices.length), indices).stream().sorted().toArray(String[]::new);
+        ensureGreen(indices);
+        assertAcked(client().admin().indices().preparePutMapping(indicesWithExtraField).setSource("extra_field", "type=integer").get());
+        for (String index : indicesWithExtraField) {
+            client().prepareIndex(index).setSource("extra_field", randomIntBetween(1, 1000)).get();
+        }
+        FieldCapabilitiesResponse resp = client().execute(FieldCapabilitiesAction.INSTANCE, request).actionGet();
+        verifyResponse.accept(resp);
+        assertThat(resp.getField("extra_field"), hasKey("integer"));
+        assertThat(resp.getField("extra_field").get("integer").indices(), nullValue());
+        assertTrue(resp.getField("extra_field").get("integer").isSearchable());
+        assertTrue(resp.getField("extra_field").get("integer").isAggregatable());
+    }
+
     private void assertIndices(FieldCapabilitiesResponse response, String... indices) {
         assertNotNull(response.getIndices());
         Arrays.sort(indices);

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

@@ -486,23 +486,37 @@ public class FieldCapabilities implements Writeable, ToXContentObject {
         private int dimensionIndices = 0;
         private TimeSeriesParams.MetricType metricType;
         private boolean hasConflictMetricType;
-        private final List<IndexCaps> indiceList;
+        private final List<IndexCaps> indicesList;
         private final Map<String, Set<String>> meta;
+        private int totalIndices;
 
         Builder(String name, String type) {
             this.name = name;
             this.type = type;
             this.metricType = null;
             this.hasConflictMetricType = false;
-            this.indiceList = new ArrayList<>();
+            this.indicesList = new ArrayList<>();
             this.meta = new HashMap<>();
         }
 
+        private boolean assertIndicesSorted(String[] indices) {
+            for (int i = 1; i < indices.length; i++) {
+                assert indices[i - 1].compareTo(indices[i]) < 0 : "indices [" + Arrays.toString(indices) + "] aren't sorted";
+            }
+            if (indicesList.isEmpty() == false) {
+                final IndexCaps lastCaps = indicesList.get(indicesList.size() - 1);
+                final String lastIndex = lastCaps.indices[lastCaps.indices.length - 1];
+                assert lastIndex.compareTo(indices[0]) < 0
+                    : "indices aren't sorted; previous [" + lastIndex + "], current [" + indices[0] + "]";
+            }
+            return true;
+        }
+
         /**
          * Collect the field capabilities for an index.
          */
         void add(
-            String index,
+            String[] indices,
             boolean isMetadataField,
             boolean search,
             boolean agg,
@@ -510,82 +524,87 @@ public class FieldCapabilities implements Writeable, ToXContentObject {
             TimeSeriesParams.MetricType metricType,
             Map<String, String> meta
         ) {
-            assert indiceList.isEmpty() || indiceList.get(indiceList.size() - 1).name.compareTo(index) < 0
-                : "indices aren't sorted; previous [" + indiceList.get(indiceList.size() - 1).name + "], current [" + index + "]";
+            assert assertIndicesSorted(indices);
+            totalIndices += indices.length;
             if (search) {
-                searchableIndices++;
+                searchableIndices += indices.length;
             }
             if (agg) {
-                aggregatableIndices++;
+                aggregatableIndices += indices.length;
             }
             if (isDimension) {
-                dimensionIndices++;
+                dimensionIndices += indices.length;
             }
             this.isMetadataField |= isMetadataField;
             // If we have discrepancy in metric types or in some indices this field is not marked as a metric field - we will
             // treat is a non-metric field and report this discrepancy in metricConflictsIndices
-            if (indiceList.isEmpty()) {
+            if (indicesList.isEmpty()) {
                 this.metricType = metricType;
             } else if (this.metricType != metricType) {
                 hasConflictMetricType = true;
                 this.metricType = null;
             }
-            IndexCaps indexCaps = new IndexCaps(index, search, agg, isDimension, metricType);
-            indiceList.add(indexCaps);
+            indicesList.add(new IndexCaps(indices, search, agg, isDimension, metricType));
             for (Map.Entry<String, String> entry : meta.entrySet()) {
                 this.meta.computeIfAbsent(entry.getKey(), key -> new HashSet<>()).add(entry.getValue());
             }
         }
 
         Stream<String> getIndices() {
-            return indiceList.stream().map(c -> c.name);
+            return indicesList.stream().flatMap(c -> Arrays.stream(c.indices));
         }
 
-        private String[] getNonFeatureIndices(boolean featureInAll, int featureIndices, Predicate<IndexCaps> hasFeature) {
-            if (featureInAll || featureIndices == 0) {
-                return null;
-            }
-            String[] nonFeatureIndices = new String[indiceList.size() - featureIndices];
+        private String[] filterIndices(int length, Predicate<IndexCaps> pred) {
             int index = 0;
-            for (IndexCaps indexCaps : indiceList) {
-                if (hasFeature.test(indexCaps) == false) {
-                    nonFeatureIndices[index++] = indexCaps.name;
+            final String[] dst = new String[length];
+            for (IndexCaps indexCaps : indicesList) {
+                if (pred.test(indexCaps)) {
+                    System.arraycopy(indexCaps.indices, 0, dst, index, indexCaps.indices.length);
+                    index += indexCaps.indices.length;
                 }
             }
-            return nonFeatureIndices;
+            assert index == length : index + "!=" + length;
+            return dst;
         }
 
         FieldCapabilities build(boolean withIndices) {
-            final String[] indices;
-            if (withIndices) {
-                indices = indiceList.stream().map(caps -> caps.name).toArray(String[]::new);
-            } else {
-                indices = null;
-            }
+            final String[] indices = withIndices ? filterIndices(totalIndices, ic -> true) : null;
 
             // Iff this field is searchable in some indices AND non-searchable in others
             // we record the list of non-searchable indices
-            boolean isSearchable = searchableIndices == indiceList.size();
-            String[] nonSearchableIndices = getNonFeatureIndices(isSearchable, searchableIndices, IndexCaps::isSearchable);
+            final boolean isSearchable = searchableIndices == totalIndices;
+            final String[] nonSearchableIndices;
+            if (isSearchable || searchableIndices == 0) {
+                nonSearchableIndices = null;
+            } else {
+                nonSearchableIndices = filterIndices(totalIndices - searchableIndices, ic -> ic.isSearchable == false);
+            }
 
             // Iff this field is aggregatable in some indices AND non-aggregatable in others
             // we keep the list of non-aggregatable indices
-            boolean isAggregatable = aggregatableIndices == indiceList.size();
-            String[] nonAggregatableIndices = getNonFeatureIndices(isAggregatable, aggregatableIndices, IndexCaps::isAggregatable);
+            final boolean isAggregatable = aggregatableIndices == totalIndices;
+            final String[] nonAggregatableIndices;
+            if (isAggregatable || aggregatableIndices == 0) {
+                nonAggregatableIndices = null;
+            } else {
+                nonAggregatableIndices = filterIndices(totalIndices - aggregatableIndices, ic -> ic.isAggregatable == false);
+            }
 
             // Collect all indices that have dimension == false if this field is marked as a dimension in at least one index
-            boolean isDimension = dimensionIndices == indiceList.size();
-            String[] nonDimensionIndices = getNonFeatureIndices(isDimension, dimensionIndices, IndexCaps::isDimension);
+            final boolean isDimension = dimensionIndices == totalIndices;
+            final String[] nonDimensionIndices;
+            if (isDimension || dimensionIndices == 0) {
+                nonDimensionIndices = null;
+            } else {
+                nonDimensionIndices = filterIndices(totalIndices - dimensionIndices, ic -> ic.isDimension == false);
+            }
 
             final String[] metricConflictsIndices;
             if (hasConflictMetricType) {
                 // Collect all indices that have this field. If it is marked differently in different indices, we cannot really
                 // make a decisions which index is "right" and which index is "wrong" so collecting all indices where this field
                 // is present is probably the only sensible thing to do here
-                metricConflictsIndices = Objects.requireNonNullElseGet(
-                    indices,
-                    () -> indiceList.stream().map(caps -> caps.name).toArray(String[]::new)
-                );
+                metricConflictsIndices = Objects.requireNonNullElseGet(indices, () -> filterIndices(totalIndices, ic -> true));
             } else {
                 metricConflictsIndices = null;
             }
@@ -613,7 +632,7 @@ public class FieldCapabilities implements Writeable, ToXContentObject {
     }
 
     private record IndexCaps(
-        String name,
+        String[] indices,
         boolean isSearchable,
         boolean isAggregatable,
         boolean isDimension,

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

@@ -8,6 +8,7 @@
 
 package org.elasticsearch.action.fieldcaps;
 
+import org.apache.lucene.util.ArrayUtil;
 import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionRunnable;
@@ -36,14 +37,15 @@ import org.elasticsearch.transport.TransportRequestHandler;
 import org.elasticsearch.transport.TransportService;
 
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
+import java.util.Comparator;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
-import java.util.TreeMap;
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.stream.Collectors;
@@ -114,12 +116,12 @@ public class TransportFieldCapabilitiesAction extends HandledTransportAction<Fie
 
         final Map<String, FieldCapabilitiesIndexResponse> indexResponses = Collections.synchronizedMap(new HashMap<>());
         // This map is used to share the index response for indices which have the same index mapping hash to reduce the memory usage.
-        final Map<String, Map<String, IndexFieldCapabilities>> indexMappingHashToResponses = Collections.synchronizedMap(new HashMap<>());
+        final Map<String, FieldCapabilitiesIndexResponse> indexMappingHashToResponses = Collections.synchronizedMap(new HashMap<>());
         final Consumer<FieldCapabilitiesIndexResponse> handleIndexResponse = resp -> {
             if (resp.canMatch() && resp.getIndexMappingHash() != null) {
-                Map<String, IndexFieldCapabilities> curr = indexMappingHashToResponses.putIfAbsent(resp.getIndexMappingHash(), resp.get());
+                FieldCapabilitiesIndexResponse curr = indexMappingHashToResponses.putIfAbsent(resp.getIndexMappingHash(), resp);
                 if (curr != null) {
-                    resp = new FieldCapabilitiesIndexResponse(resp.getIndexName(), resp.getIndexMappingHash(), curr, true);
+                    resp = new FieldCapabilitiesIndexResponse(resp.getIndexName(), curr.getIndexMappingHash(), curr.get(), true);
                 }
             }
             indexResponses.putIfAbsent(resp.getIndexName(), resp);
@@ -228,15 +230,35 @@ public class TransportFieldCapabilitiesAction extends HandledTransportAction<Fie
         return remoteRequest;
     }
 
+    private static boolean hasSameMappingHash(FieldCapabilitiesIndexResponse r1, FieldCapabilitiesIndexResponse r2) {
+        return r1.getIndexMappingHash() != null
+            && r2.getIndexMappingHash() != null
+            && r1.getIndexMappingHash().equals(r2.getIndexMappingHash());
+    }
+
     private FieldCapabilitiesResponse merge(
         Map<String, FieldCapabilitiesIndexResponse> indexResponsesMap,
         FieldCapabilitiesRequest request,
         List<FieldCapabilitiesFailure> failures
     ) {
-        Map<String, FieldCapabilitiesIndexResponse> responses = new TreeMap<>(indexResponsesMap);
-        Map<String, Map<String, FieldCapabilities.Builder>> responseMapBuilder = new HashMap<>();
-        for (FieldCapabilitiesIndexResponse response : responses.values()) {
-            innerMerge(responseMapBuilder, request, response);
+        final FieldCapabilitiesIndexResponse[] indexResponses = indexResponsesMap.values()
+            .stream()
+            .sorted(Comparator.comparing(FieldCapabilitiesIndexResponse::getIndexName))
+            .toArray(FieldCapabilitiesIndexResponse[]::new);
+        final String[] indices = Arrays.stream(indexResponses).map(FieldCapabilitiesIndexResponse::getIndexName).toArray(String[]::new);
+        final Map<String, Map<String, FieldCapabilities.Builder>> responseMapBuilder = new HashMap<>();
+        int lastPendingIndex = 0;
+        for (int i = 1; i <= indexResponses.length; i++) {
+            if (i == indexResponses.length || hasSameMappingHash(indexResponses[lastPendingIndex], indexResponses[i]) == false) {
+                final String[] subIndices;
+                if (lastPendingIndex == 0 && i == indexResponses.length) {
+                    subIndices = indices;
+                } else {
+                    subIndices = ArrayUtil.copyOfSubArray(indices, lastPendingIndex, i);
+                }
+                innerMerge(subIndices, responseMapBuilder, request, indexResponses[lastPendingIndex]);
+                lastPendingIndex = i;
+            }
         }
 
         Map<String, Map<String, FieldCapabilities>> responseMap = new HashMap<>();
@@ -247,7 +269,7 @@ public class TransportFieldCapabilitiesAction extends HandledTransportAction<Fie
             if (request.includeUnmapped()) {
                 // do this directly, rather than using the builder, to save creating a whole lot of objects we don't need
                 unmapped = getUnmappedFields(
-                    responses.keySet(),
+                    indexResponsesMap.keySet(),
                     entry.getKey(),
                     typeMapBuilder.values().stream().flatMap(FieldCapabilities.Builder::getIndices).collect(Collectors.toSet())
                 );
@@ -266,7 +288,7 @@ public class TransportFieldCapabilitiesAction extends HandledTransportAction<Fie
                 )
             );
         }
-        return new FieldCapabilitiesResponse(responses.keySet().toArray(String[]::new), Collections.unmodifiableMap(responseMap), failures);
+        return new FieldCapabilitiesResponse(indices, Collections.unmodifiableMap(responseMap), failures);
     }
 
     private static Optional<Function<Boolean, FieldCapabilities>> getUnmappedFields(
@@ -287,6 +309,7 @@ public class TransportFieldCapabilitiesAction extends HandledTransportAction<Fie
     }
 
     private void innerMerge(
+        String[] indices,
         Map<String, Map<String, FieldCapabilities.Builder>> responseMapBuilder,
         FieldCapabilitiesRequest request,
         FieldCapabilitiesIndexResponse response
@@ -306,7 +329,7 @@ public class TransportFieldCapabilitiesAction extends HandledTransportAction<Fie
                 key -> new FieldCapabilities.Builder(field, key)
             );
             builder.add(
-                response.getIndexName(),
+                indices,
                 fieldCap.isMetadatafield(),
                 fieldCap.isSearchable(),
                 fieldCap.isAggregatable(),

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

@@ -8,19 +8,20 @@
 
 package org.elasticsearch.action.fieldcaps;
 
+import org.apache.lucene.util.ArrayUtil;
 import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.util.iterable.Iterables;
-import org.elasticsearch.common.util.set.Sets;
 import org.elasticsearch.index.mapper.TimeSeriesParams;
 import org.elasticsearch.test.AbstractXContentSerializingTestCase;
 import org.elasticsearch.xcontent.XContentParser;
 
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
-import java.util.HashSet;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Set;
 import java.util.stream.IntStream;
@@ -48,9 +49,9 @@ public class FieldCapabilitiesTests extends AbstractXContentSerializingTestCase<
 
     public void testBuilder() {
         FieldCapabilities.Builder builder = new FieldCapabilities.Builder("field", "type");
-        builder.add("index1", false, true, false, false, null, Collections.emptyMap());
-        builder.add("index2", false, true, false, false, null, Collections.emptyMap());
-        builder.add("index3", false, true, false, false, null, Collections.emptyMap());
+        builder.add(new String[] { "index1" }, false, true, false, false, null, Collections.emptyMap());
+        builder.add(new String[] { "index2" }, false, true, false, false, null, Collections.emptyMap());
+        builder.add(new String[] { "index3" }, false, true, false, false, null, Collections.emptyMap());
 
         {
             FieldCapabilities cap1 = builder.build(false);
@@ -78,9 +79,9 @@ public class FieldCapabilitiesTests extends AbstractXContentSerializingTestCase<
         }
 
         builder = new FieldCapabilities.Builder("field", "type");
-        builder.add("index1", false, false, true, true, null, Collections.emptyMap());
-        builder.add("index2", false, true, false, false, TimeSeriesParams.MetricType.counter, Collections.emptyMap());
-        builder.add("index3", false, false, false, false, null, Collections.emptyMap());
+        builder.add(new String[] { "index1" }, false, false, true, true, null, Collections.emptyMap());
+        builder.add(new String[] { "index2" }, false, true, false, false, TimeSeriesParams.MetricType.counter, Collections.emptyMap());
+        builder.add(new String[] { "index3" }, false, false, false, false, null, Collections.emptyMap());
         {
             FieldCapabilities cap1 = builder.build(false);
             assertThat(cap1.isSearchable(), equalTo(false));
@@ -107,9 +108,9 @@ public class FieldCapabilitiesTests extends AbstractXContentSerializingTestCase<
         }
 
         builder = new FieldCapabilities.Builder("field", "type");
-        builder.add("index1", false, true, true, true, TimeSeriesParams.MetricType.counter, Collections.emptyMap());
-        builder.add("index2", false, true, true, true, TimeSeriesParams.MetricType.counter, Map.of("foo", "bar"));
-        builder.add("index3", false, true, true, true, TimeSeriesParams.MetricType.counter, Map.of("foo", "quux"));
+        builder.add(new String[] { "index1" }, false, true, true, true, TimeSeriesParams.MetricType.counter, Collections.emptyMap());
+        builder.add(new String[] { "index2" }, false, true, true, true, TimeSeriesParams.MetricType.counter, Map.of("foo", "bar"));
+        builder.add(new String[] { "index3" }, false, true, true, true, TimeSeriesParams.MetricType.counter, Map.of("foo", "quux"));
         {
             FieldCapabilities cap1 = builder.build(false);
             assertThat(cap1.isSearchable(), equalTo(true));
@@ -136,9 +137,9 @@ public class FieldCapabilitiesTests extends AbstractXContentSerializingTestCase<
         }
 
         builder = new FieldCapabilities.Builder("field", "type");
-        builder.add("index1", false, true, true, true, TimeSeriesParams.MetricType.counter, Collections.emptyMap());
-        builder.add("index2", false, true, true, true, TimeSeriesParams.MetricType.gauge, Map.of("foo", "bar"));
-        builder.add("index3", false, true, true, true, TimeSeriesParams.MetricType.counter, Map.of("foo", "quux"));
+        builder.add(new String[] { "index1" }, false, true, true, true, TimeSeriesParams.MetricType.counter, Collections.emptyMap());
+        builder.add(new String[] { "index2" }, false, true, true, true, TimeSeriesParams.MetricType.gauge, Map.of("foo", "bar"));
+        builder.add(new String[] { "index3" }, false, true, true, true, TimeSeriesParams.MetricType.counter, Map.of("foo", "quux"));
         {
             FieldCapabilities cap1 = builder.build(false);
             assertThat(cap1.isSearchable(), equalTo(true));
@@ -166,64 +167,73 @@ public class FieldCapabilitiesTests extends AbstractXContentSerializingTestCase<
     }
 
     public void testRandomBuilder() {
-        List<String> indices = IntStream.range(0, randomIntBetween(1, 50)).mapToObj(n -> formatted("index_%2d", n)).toList();
-        Set<String> searchableIndices = new HashSet<>(randomSubsetOf(indices));
-        Set<String> aggregatableIndices = new HashSet<>(randomSubsetOf(indices));
-        Set<String> dimensionIndices = new HashSet<>(randomSubsetOf(indices));
+        String[] indices = IntStream.range(0, randomIntBetween(1, 50))
+            .mapToObj(n -> String.format(Locale.ROOT, "index_%2d", n))
+            .toArray(String[]::new);
+
+        List<String> nonSearchableIndices = new ArrayList<>();
+        List<String> nonAggregatableIndices = new ArrayList<>();
+        List<String> nonDimensionIndices = new ArrayList<>();
+
         FieldCapabilities.Builder builder = new FieldCapabilities.Builder("field", "type");
-        for (String index : indices) {
-            builder.add(
-                index,
-                randomBoolean(),
-                searchableIndices.contains(index),
-                aggregatableIndices.contains(index),
-                dimensionIndices.contains(index),
-                null,
-                Map.of()
-            );
+        for (int i = 0; i < indices.length;) {
+            int bulkSize = randomIntBetween(1, indices.length - i);
+            String[] groupIndices = ArrayUtil.copyOfSubArray(indices, i, i + bulkSize);
+            final boolean searchable = randomBoolean();
+            if (searchable == false) {
+                nonSearchableIndices.addAll(Arrays.asList(groupIndices));
+            }
+            final boolean aggregatable = randomBoolean();
+            if (aggregatable == false) {
+                nonAggregatableIndices.addAll(Arrays.asList(groupIndices));
+            }
+            final boolean isDimension = randomBoolean();
+            if (isDimension == false) {
+                nonDimensionIndices.addAll(Arrays.asList(groupIndices));
+            }
+            builder.add(groupIndices, false, searchable, aggregatable, isDimension, null, Map.of());
+            i += bulkSize;
+        }
+        boolean withIndices = randomBoolean();
+        FieldCapabilities fieldCaps = builder.build(withIndices);
+        if (withIndices) {
+            assertThat(fieldCaps.indices(), equalTo(indices));
         }
-        FieldCapabilities fieldCaps = builder.build(randomBoolean());
         // search
-        if (searchableIndices.isEmpty()) {
-            assertFalse(fieldCaps.isSearchable());
-            assertNull(fieldCaps.nonSearchableIndices());
-        } else if (searchableIndices.size() == indices.size()) {
+        if (nonSearchableIndices.isEmpty()) {
             assertTrue(fieldCaps.isSearchable());
             assertNull(fieldCaps.nonSearchableIndices());
         } else {
             assertFalse(fieldCaps.isSearchable());
-            assertThat(
-                Sets.newHashSet(fieldCaps.nonSearchableIndices()),
-                equalTo(Sets.difference(Sets.newHashSet(indices), searchableIndices))
-            );
+            if (nonSearchableIndices.size() == indices.length) {
+                assertThat(fieldCaps.nonSearchableIndices(), equalTo(null));
+            } else {
+                assertThat(fieldCaps.nonSearchableIndices(), equalTo(nonSearchableIndices.toArray(String[]::new)));
+            }
         }
         // aggregate
-        if (aggregatableIndices.isEmpty()) {
-            assertFalse(fieldCaps.isAggregatable());
-            assertNull(fieldCaps.nonAggregatableIndices());
-        } else if (aggregatableIndices.size() == indices.size()) {
+        if (nonAggregatableIndices.isEmpty()) {
             assertTrue(fieldCaps.isAggregatable());
             assertNull(fieldCaps.nonAggregatableIndices());
         } else {
             assertFalse(fieldCaps.isAggregatable());
-            assertThat(
-                Sets.newHashSet(fieldCaps.nonAggregatableIndices()),
-                equalTo(Sets.difference(Sets.newHashSet(indices), aggregatableIndices))
-            );
+            if (nonAggregatableIndices.size() == indices.length) {
+                assertThat(fieldCaps.nonAggregatableIndices(), equalTo(null));
+            } else {
+                assertThat(fieldCaps.nonAggregatableIndices(), equalTo(nonAggregatableIndices.toArray(String[]::new)));
+            }
         }
         // dimension
-        if (dimensionIndices.isEmpty()) {
-            assertFalse(fieldCaps.isDimension());
-            assertNull(fieldCaps.nonDimensionIndices());
-        } else if (dimensionIndices.size() == indices.size()) {
+        if (nonDimensionIndices.isEmpty()) {
             assertTrue(fieldCaps.isDimension());
             assertNull(fieldCaps.nonDimensionIndices());
         } else {
             assertFalse(fieldCaps.isDimension());
-            assertThat(
-                Sets.newHashSet(fieldCaps.nonDimensionIndices()),
-                equalTo(Sets.difference(Sets.newHashSet(indices), dimensionIndices))
-            );
+            if (nonDimensionIndices.size() == indices.length) {
+                assertThat(fieldCaps.nonDimensionIndices(), equalTo(null));
+            } else {
+                assertThat(fieldCaps.nonDimensionIndices(), equalTo(nonDimensionIndices.toArray(String[]::new)));
+            }
         }
     }
 
@@ -232,7 +242,7 @@ public class FieldCapabilitiesTests extends AbstractXContentSerializingTestCase<
         TimeSeriesParams.MetricType metric = randomBoolean() ? null : randomFrom(TimeSeriesParams.MetricType.values());
         FieldCapabilities.Builder builder = new FieldCapabilities.Builder("field", "type");
         for (String index : indices) {
-            builder.add(index, randomBoolean(), randomBoolean(), randomBoolean(), randomBoolean(), metric, Map.of());
+            builder.add(new String[] { index }, randomBoolean(), randomBoolean(), randomBoolean(), randomBoolean(), metric, Map.of());
         }
         FieldCapabilities fieldCaps = builder.build(randomBoolean());
         assertThat(fieldCaps.getMetricType(), equalTo(metric));
@@ -249,7 +259,15 @@ public class FieldCapabilitiesTests extends AbstractXContentSerializingTestCase<
         }
         FieldCapabilities.Builder builder = new FieldCapabilities.Builder("field", "type");
         for (String index : indices) {
-            builder.add(index, randomBoolean(), randomBoolean(), randomBoolean(), randomBoolean(), metricTypes.get(index), Map.of());
+            builder.add(
+                new String[] { index },
+                randomBoolean(),
+                randomBoolean(),
+                randomBoolean(),
+                randomBoolean(),
+                metricTypes.get(index),
+                Map.of()
+            );
         }
         FieldCapabilities fieldCaps = builder.build(randomBoolean());
         if (metricTypes.isEmpty()) {
@@ -269,7 +287,7 @@ public class FieldCapabilitiesTests extends AbstractXContentSerializingTestCase<
         int numIndex = randomIntBetween(1, 5);
         for (int i = 1; i <= numIndex; i++) {
             builder.add(
-                "index-" + i,
+                new String[] { "index-" + i },
                 randomBoolean(),
                 randomBoolean(),
                 randomBoolean(),
@@ -281,7 +299,7 @@ public class FieldCapabilitiesTests extends AbstractXContentSerializingTestCase<
         final String outOfOrderIndex = randomBoolean() ? "abc" : "index-" + randomIntBetween(1, numIndex);
         AssertionError error = expectThrows(AssertionError.class, () -> {
             builder.add(
-                outOfOrderIndex,
+                new String[] { outOfOrderIndex },
                 randomBoolean(),
                 randomBoolean(),
                 randomBoolean(),