Group field producers once when performing shard level downsampling (#96795) (#96845)

Group field producers once when creating RollupBucketBuilder,
instead of everytime a rollup document needs to be built.

Author: martijnvg

x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/downsample/RollupShardIndexer.java

@@ -384,9 +384,28 @@ private class RollupBucketBuilder {
         private long timestamp;
         private int docCount;
         private final List<AbstractDownsampleFieldProducer> rollupFieldProducers;
+        private final List<DownsampleFieldSerializer> groupedProducers;
 
         RollupBucketBuilder(List<AbstractDownsampleFieldProducer> rollupFieldProducers) {
             this.rollupFieldProducers = rollupFieldProducers;
+            /*
+             * The rollup field producers for aggregate_metric_double all share the same name (this is
+             * the name they will be serialized in the target index). We group all field producers by
+             * name. If grouping yields multiple rollup field producers, we delegate serialization to
+             * the AggregateMetricFieldSerializer class.
+             */
+            groupedProducers = rollupFieldProducers.stream()
+                .collect(groupingBy(AbstractDownsampleFieldProducer::name))
+                .entrySet()
+                .stream()
+                .map(e -> {
+                    if (e.getValue().size() == 1) {
+                        return e.getValue().get(0);
+                    } else {
+                        return new AggregateMetricFieldSerializer(e.getKey(), e.getValue());
+                    }
+                })
+                .toList();
         }
 
         /**
@@ -435,25 +454,6 @@ public XContentBuilder buildRollupDocument() throws IOException {
                 builder.field((String) e.getKey(), e.getValue());
             }
 
-            /*
-             * The rollup field producers for aggregate_metric_double all share the same name (this is
-             * the name they will be serialized in the target index). We group all field producers by
-             * name. If grouping yields multiple rollup field producers, we delegate serialization to
-             * the AggregateMetricFieldSerializer class.
-             */
-            List<DownsampleFieldSerializer> groupedProducers = rollupFieldProducers.stream()
-                .collect(groupingBy(AbstractDownsampleFieldProducer::name))
-                .entrySet()
-                .stream()
-                .map(e -> {
-                    if (e.getValue().size() == 1) {
-                        return e.getValue().get(0);
-                    } else {
-                        return new AggregateMetricFieldSerializer(e.getKey(), e.getValue());
-                    }
-                })
-                .toList();
-
             // Serialize fields
             for (DownsampleFieldSerializer fieldProducer : groupedProducers) {
                 fieldProducer.write(builder);