diff --git a/x-pack/plugin/mapper-aggregate-metric/build.gradle b/x-pack/plugin/mapper-aggregate-metric/build.gradle
index 6220ee78c62a9..35fc7d60b2afe 100644
--- a/x-pack/plugin/mapper-aggregate-metric/build.gradle
+++ b/x-pack/plugin/mapper-aggregate-metric/build.gradle
@@ -13,15 +13,13 @@ apply plugin: 'elasticsearch.internal-es-plugin'
 
 esplugin {
   name 'x-pack-aggregate-metric'
-  description 'Module for the aggregate_metric field type, which allows pre-aggregated fields to be stored a single field.'
+  description 'Module for the aggregate_metric_double field type, which allows pre-aggregated fields to be stored as a single field'
   classname 'org.elasticsearch.xpack.aggregatemetric.AggregateMetricMapperPlugin'
   extendedPlugins = ['x-pack-core']
 }
 archivesBaseName = 'x-pack-aggregate-metric'
 
 dependencies {
-  compileOnly project(":server")
-
   compileOnly project(path: xpackModule('core'))
   testImplementation(testArtifact(project(xpackModule('core'))))
 }
diff --git a/x-pack/plugin/mapper-aggregate-metric/src/main/java/org/elasticsearch/xpack/aggregatemetric/AggregateMetricMapperPlugin.java b/x-pack/plugin/mapper-aggregate-metric/src/main/java/org/elasticsearch/xpack/aggregatemetric/AggregateMetricMapperPlugin.java
index e44b65ee2212b..fea55e793d638 100644
--- a/x-pack/plugin/mapper-aggregate-metric/src/main/java/org/elasticsearch/xpack/aggregatemetric/AggregateMetricMapperPlugin.java
+++ b/x-pack/plugin/mapper-aggregate-metric/src/main/java/org/elasticsearch/xpack/aggregatemetric/AggregateMetricMapperPlugin.java
@@ -11,6 +11,7 @@
 import org.elasticsearch.action.ActionResponse;
 import org.elasticsearch.index.mapper.Mapper;
 import org.elasticsearch.plugins.ActionPlugin;
+import org.elasticsearch.plugins.ExtensiblePlugin;
 import org.elasticsearch.plugins.MapperPlugin;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.plugins.SearchPlugin;
@@ -27,7 +28,7 @@
 
 import static java.util.Collections.singletonMap;
 
-public class AggregateMetricMapperPlugin extends Plugin implements MapperPlugin, ActionPlugin, SearchPlugin {
+public class AggregateMetricMapperPlugin extends Plugin implements MapperPlugin, ActionPlugin, SearchPlugin, ExtensiblePlugin {
 
     @Override
     public Map<String, Mapper.TypeParser> getMappers() {
diff --git a/x-pack/plugin/mapper-aggregate-metric/src/main/java/org/elasticsearch/xpack/aggregatemetric/mapper/AggregateDoubleMetricFieldMapper.java b/x-pack/plugin/mapper-aggregate-metric/src/main/java/org/elasticsearch/xpack/aggregatemetric/mapper/AggregateDoubleMetricFieldMapper.java
index 02c31e668e7ae..d80da1772197e 100644
--- a/x-pack/plugin/mapper-aggregate-metric/src/main/java/org/elasticsearch/xpack/aggregatemetric/mapper/AggregateDoubleMetricFieldMapper.java
+++ b/x-pack/plugin/mapper-aggregate-metric/src/main/java/org/elasticsearch/xpack/aggregatemetric/mapper/AggregateDoubleMetricFieldMapper.java
@@ -307,6 +307,10 @@ private void setMetricFields(EnumMap<Metric, NumberFieldMapper.NumberFieldType>
             this.metricFields = metricFields;
         }
 
+        public Map<Metric, NumberFieldMapper.NumberFieldType> getMetricFields() {
+            return Collections.unmodifiableMap(metricFields);
+        }
+
         public void addMetricField(Metric m, NumberFieldMapper.NumberFieldType subfield) {
             if (metricFields == null) {
                 metricFields = new EnumMap<>(AggregateDoubleMetricFieldMapper.Metric.class);
diff --git a/x-pack/plugin/rollup/build.gradle b/x-pack/plugin/rollup/build.gradle
index 401e178fafca7..e2da0ba0f866e 100644
--- a/x-pack/plugin/rollup/build.gradle
+++ b/x-pack/plugin/rollup/build.gradle
@@ -1,22 +1,18 @@
-import org.elasticsearch.gradle.internal.info.BuildParams
-
 apply plugin: 'elasticsearch.internal-es-plugin'
 esplugin {
   name 'x-pack-rollup'
   description 'Elasticsearch Expanded Pack Plugin - Rollup'
   classname 'org.elasticsearch.xpack.rollup.Rollup'
-  extendedPlugins = ['x-pack-core']
+  extendedPlugins = ['x-pack-aggregate-metric']
 }
 
 archivesBaseName = 'x-pack-rollup'
 
 dependencies {
-  compileOnly project(":server")
   compileOnly project(path: xpackModule('core'))
-  compileOnly project(path: xpackModule('analytics'))
-  compileOnly project(path: xpackModule('mapper-aggregate-metric'))
-  compileOnly project(path: xpackModule('ilm'))
   compileOnly project(':modules:data-streams')
+  compileOnly project(path: xpackModule('ilm'))
+  compileOnly project(path: xpackModule('mapper-aggregate-metric'))
   testImplementation(testArtifact(project(xpackModule('core'))))
 }
 
diff --git a/x-pack/plugin/rollup/qa/rest/build.gradle b/x-pack/plugin/rollup/qa/rest/build.gradle
index 0c794c07b713b..6e69f1e929d8a 100644
--- a/x-pack/plugin/rollup/qa/rest/build.gradle
+++ b/x-pack/plugin/rollup/qa/rest/build.gradle
@@ -17,7 +17,7 @@ dependencies {
 
 restResources {
   restApi {
-    include '_common', 'bulk', 'cluster', 'indices', 'search', 'downsample'
+    include '_common', 'bulk', 'cluster', 'indices', 'search'
   }
 }
 
diff --git a/x-pack/plugin/rollup/qa/rest/src/yamlRestTest/resources/rest-api-spec/test/rollup/10_basic.yml b/x-pack/plugin/rollup/qa/rest/src/yamlRestTest/resources/rest-api-spec/test/rollup/10_basic.yml
index 75d80ec1b28da..8a17dccbf1441 100644
--- a/x-pack/plugin/rollup/qa/rest/src/yamlRestTest/resources/rest-api-spec/test/rollup/10_basic.yml
+++ b/x-pack/plugin/rollup/qa/rest/src/yamlRestTest/resources/rest-api-spec/test/rollup/10_basic.yml
@@ -87,7 +87,7 @@ setup:
           index.blocks.write: true
 
 ---
-"Rollup index":
+"Downsample index":
   - skip:
       version: " - 8.4.99"
       reason: "rollup renamed to downsample in 8.5.0"
@@ -141,6 +141,24 @@ setup:
   - match: { rollup-test.settings.index.number_of_shards: "1" }
   - match: { rollup-test.settings.index.number_of_replicas: "0" }
 
+  # Assert rollup index mapping
+  - do:
+      indices.get_mapping:
+        index: rollup-test
+
+  - match: { rollup-test.mappings.properties.@timestamp.type: date }
+  - match: { rollup-test.mappings.properties.@timestamp.meta.fixed_interval: 1h }
+  - match: { rollup-test.mappings.properties.@timestamp.meta.time_zone: UTC }
+  - match: { rollup-test.mappings.properties.k8s.properties.pod.properties.multi-gauge.type: aggregate_metric_double }
+  - match: { rollup-test.mappings.properties.k8s.properties.pod.properties.multi-gauge.metrics: [ "min", "max", "sum", "value_count" ] }
+  - match: { rollup-test.mappings.properties.k8s.properties.pod.properties.multi-gauge.default_metric: max }
+  - match: { rollup-test.mappings.properties.k8s.properties.pod.properties.multi-gauge.time_series_metric: gauge }
+  - match: { rollup-test.mappings.properties.k8s.properties.pod.properties.multi-counter.type: long }
+  - match: { rollup-test.mappings.properties.k8s.properties.pod.properties.multi-counter.time_series_metric: counter }
+  - match: { rollup-test.mappings.properties.k8s.properties.pod.properties.uid.type: keyword }
+  - match: { rollup-test.mappings.properties.k8s.properties.pod.properties.uid.time_series_dimension: true }
+
+
   # Assert source index has not been deleted
   - do:
       indices.get:
@@ -156,7 +174,7 @@ setup:
   - match:   { indices.rollup-test.shards.0.0.num_search_segments: 1}
 
 ---
-"Rollup non-existing index":
+"Downsample non-existing index":
   - skip:
       version: " - 8.4.99"
       reason: "rollup renamed to downsample in 8.5.0"
@@ -172,7 +190,7 @@ setup:
           }
 
 ---
-"Rollup to existing rollup index":
+"Downsample to existing index":
   - skip:
       version: " - 8.4.99"
       reason: "rollup renamed to downsample in 8.5.0"
@@ -192,7 +210,7 @@ setup:
           }
 
 ---
-"Rollup not time_series index":
+"Downsample not time_series index":
   - skip:
       version: " - 8.4.99"
       reason: "rollup renamed to downsample in 8.5.0"
@@ -213,7 +231,7 @@ setup:
 
 
 ---
-"Rollup no metric index":
+"Downsample no metric index":
   - skip:
       version: " - 8.4.99"
       reason: "rollup renamed to downsample in 8.5.0"
@@ -254,3 +272,142 @@ setup:
           {
             "fixed_interval": "1h"
           }
+
+---
+"Downsample a downsampled index":
+  - skip:
+      version: " - 8.4.99"
+      reason: "Rollup of rollups introduced in 8.5.0"
+
+  - do:
+      indices.downsample:
+        index: test
+        target_index: rollup-test
+        body: >
+          {
+            "fixed_interval": "1h"
+          }
+  - is_true: acknowledged
+
+  - do:
+      indices.downsample:
+        index: rollup-test
+        target_index: rollup-test-2
+        body: >
+          {
+            "fixed_interval": "2h"
+          }
+  - is_true: acknowledged
+
+
+  # Assert rollup index mapping
+  - do:
+      indices.get_mapping:
+        index: rollup-test-2
+
+  - match: { rollup-test-2.mappings.properties.@timestamp.type: date }
+  - match: { rollup-test-2.mappings.properties.@timestamp.meta.fixed_interval: 2h }
+  - match: { rollup-test-2.mappings.properties.@timestamp.meta.time_zone: UTC }
+  - match: { rollup-test-2.mappings.properties.k8s.properties.pod.properties.multi-gauge.type: aggregate_metric_double }
+  - match: { rollup-test-2.mappings.properties.k8s.properties.pod.properties.multi-gauge.metrics: [ "min", "max", "sum", "value_count" ] }
+  - match: { rollup-test-2.mappings.properties.k8s.properties.pod.properties.multi-gauge.default_metric: max }
+  - match: { rollup-test-2.mappings.properties.k8s.properties.pod.properties.multi-gauge.time_series_metric: gauge }
+  - match: { rollup-test-2.mappings.properties.k8s.properties.pod.properties.multi-counter.type: long }
+  - match: { rollup-test-2.mappings.properties.k8s.properties.pod.properties.multi-counter.time_series_metric: counter }
+  - match: { rollup-test-2.mappings.properties.k8s.properties.pod.properties.uid.type: keyword }
+  - match: { rollup-test-2.mappings.properties.k8s.properties.pod.properties.uid.time_series_dimension: true }
+  - match: { rollup-test-2.mappings.properties.k8s.properties.pod.properties.network.properties.tx.type: aggregate_metric_double }
+  - match: { rollup-test-2.mappings.properties.k8s.properties.pod.properties.network.properties.tx.metrics: [ "min", "max", "sum", "value_count" ] }
+  - match: { rollup-test-2.mappings.properties.k8s.properties.pod.properties.network.properties.tx.default_metric: max }
+  - match: { rollup-test-2.mappings.properties.k8s.properties.pod.properties.network.properties.tx.time_series_metric: gauge }
+
+  - do:
+      search:
+        index: rollup-test-2
+        body:
+          sort: [ "_tsid", "@timestamp" ]
+
+  - length: { hits.hits: 3 }
+  - match: { hits.hits.0._source._doc_count: 2 }
+  - match: { hits.hits.0._source.k8s\.pod\.uid: 947e4ced-1786-4e53-9e0c-5c447e959507 }
+  - match: { hits.hits.0._source.metricset: pod }
+  - match: { hits.hits.0._source.@timestamp: 2021-04-28T18:00:00.000Z }
+  - match: { hits.hits.0._source.k8s\.pod\.multi-counter: 21 }
+  - match: { hits.hits.0._source.k8s\.pod\.multi-gauge.min: 90 }
+  - match: { hits.hits.0._source.k8s\.pod\.multi-gauge.max: 200 }
+  - match: { hits.hits.0._source.k8s\.pod\.multi-gauge.sum: 726 }
+  - match: { hits.hits.0._source.k8s\.pod\.multi-gauge.value_count: 6 }
+  - match: { hits.hits.0._source.k8s\.pod\.network\.tx.min: 2001818691 }
+  - match: { hits.hits.0._source.k8s\.pod\.network\.tx.max: 2005177954 }
+  - match: { hits.hits.0._source.k8s\.pod\.network\.tx.value_count: 2 }
+  - match: { hits.hits.0._source.k8s\.pod\.ip: "10.10.55.26" }
+  - match: { hits.hits.0._source.k8s\.pod\.created_at: "2021-04-28T19:35:00.000Z" }
+  - match: { hits.hits.0._source.k8s\.pod\.number_of_containers: 2 }
+  - match: { hits.hits.0._source.k8s\.pod\.tags: [ "backend", "prod", "us-west1" ] }
+  - match: { hits.hits.0._source.k8s\.pod\.values: [ 1, 1, 3 ] }
+
+  - match: { hits.hits.1._source.k8s\.pod\.uid: 947e4ced-1786-4e53-9e0c-5c447e959507 }
+  - match: { hits.hits.1._source.metricset: pod }
+  - match: { hits.hits.1._source.@timestamp: 2021-04-28T20:00:00.000Z }
+  - match: { hits.hits.1._source._doc_count: 2 }
+
+  - match: { hits.hits.2._source.k8s\.pod\.uid: df3145b3-0563-4d3b-a0f7-897eb2876ea9 }
+  - match: { hits.hits.2._source.metricset: pod }
+  - match: { hits.hits.2._source.@timestamp: 2021-04-28T18:00:00.000Z }
+  - match: { hits.hits.2._source._doc_count: 4 }
+
+  - do:
+      indices.downsample:
+        index: rollup-test
+        target_index: rollup-test-3
+        body: >
+          {
+            "fixed_interval": "180m"
+          }
+  - is_true: acknowledged
+
+---
+"Downsample a downsampled index with wrong intervals":
+  - skip:
+      version: " - 8.4.99"
+      reason: "Rollup of rollups introduced in 8.5.0"
+
+  - do:
+      indices.downsample:
+        index: test
+        target_index: rollup-test
+        body: >
+          {
+            "fixed_interval": "1h"
+          }
+  - is_true: acknowledged
+
+  - do:
+      catch: /Downsampling interval \[1h\] must be greater than the the source index interval \[1h\]/
+      indices.downsample:
+        index: rollup-test
+        target_index: rollup-test-2
+        body: >
+          {
+            "fixed_interval": "1h"
+          }
+
+  - do:
+      catch: /Downsampling interval \[30m\] must be greater than the the source index interval \[1h\]/
+      indices.downsample:
+        index: rollup-test
+        target_index: rollup-test-2
+        body: >
+          {
+            "fixed_interval": "30m"
+          }
+
+  - do:
+      catch: /Downsampling interval \[90m\] must be a multiple of the source index interval \[1h\]/
+      indices.downsample:
+        index: rollup-test
+        target_index: rollup-test-2
+        body: >
+          {
+            "fixed_interval": "90m"
+          }
diff --git a/x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/downsample/AbstractRollupFieldProducer.java b/x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/downsample/AbstractRollupFieldProducer.java
index 21afa1859a190..44bf433ada13c 100644
--- a/x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/downsample/AbstractRollupFieldProducer.java
+++ b/x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/downsample/AbstractRollupFieldProducer.java
@@ -7,6 +7,10 @@
 
 package org.elasticsearch.xpack.downsample;
 
+import org.elasticsearch.xcontent.XContentBuilder;
+
+import java.io.IOException;
+
 /**
  * Base class for classes that read metric and label fields.
  */
@@ -22,9 +26,11 @@ abstract class AbstractRollupFieldProducer<T> {
 
     /**
      * Collect a value for the field applying the specific subclass collection strategy.
+     *
+     * @param field the name of the field to collect
      * @param value the value to collect.
      */
-    public abstract void collect(T value);
+    public abstract void collect(String field, T value);
 
     /**
      * @return the name of the field.
@@ -34,14 +40,14 @@ public String name() {
     }
 
     /**
-     * @return the value of the field.
+     * Resets the producer to an empty value.
      */
-    public abstract Object value();
+    public abstract void reset();
 
     /**
-     * Resets the collected value to the specific subclass reset value.
+     * Serialize the downsampled value of the field.
      */
-    public abstract void reset();
+    public abstract void write(XContentBuilder builder) throws IOException;
 
     /**
      * @return true if the field has not collected any value.
diff --git a/x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/downsample/FieldValueFetcher.java b/x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/downsample/FieldValueFetcher.java
index 367c80691ab55..5c99d6135b3e0 100644
--- a/x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/downsample/FieldValueFetcher.java
+++ b/x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/downsample/FieldValueFetcher.java
@@ -8,16 +8,17 @@
 package org.elasticsearch.xpack.downsample;
 
 import org.apache.lucene.index.LeafReaderContext;
-import org.apache.lucene.util.BytesRef;
 import org.elasticsearch.index.fielddata.FormattedDocValues;
 import org.elasticsearch.index.fielddata.IndexFieldData;
 import org.elasticsearch.index.mapper.MappedFieldType;
+import org.elasticsearch.index.mapper.NumberFieldMapper;
 import org.elasticsearch.index.query.SearchExecutionContext;
 import org.elasticsearch.search.DocValueFormat;
+import org.elasticsearch.xpack.aggregatemetric.mapper.AggregateDoubleMetricFieldMapper;
 
-import java.util.ArrayList;
 import java.util.Collections;
-import java.util.List;
+import java.util.LinkedHashMap;
+import java.util.Map;
 
 /**
  * Utility class used for fetching field values by reading field data
@@ -55,42 +56,39 @@ FormattedDocValues getLeaf(LeafReaderContext context) {
         return fieldData.load(context).getFormattedValues(format);
     }
 
-    Object format(Object value) {
-        if (value instanceof Long l) {
-            return format.format(l);
-        } else if (value instanceof Double d) {
-            return format.format(d);
-        } else if (value instanceof BytesRef b) {
-            return format.format(b);
-        } else if (value instanceof String s) {
-            return s;
-        } else {
-            throw new IllegalArgumentException("Invalid type: [" + value.getClass() + "]");
-        }
-    }
-
     /**
-     * Retrieve field fetchers for a list of fields.
+     * Retrieve field value fetchers for a list of fields.
      */
-    private static List<FieldValueFetcher> build(SearchExecutionContext context, String[] fields) {
-        List<FieldValueFetcher> fetchers = new ArrayList<>(fields.length);
+    static Map<String, FieldValueFetcher> create(SearchExecutionContext context, String[] fields) {
+        Map<String, FieldValueFetcher> fetchers = new LinkedHashMap<>();
         for (String field : fields) {
             MappedFieldType fieldType = context.getFieldType(field);
-            if (fieldType == null) {
-                throw new IllegalArgumentException("Unknown field: [" + field + "]");
+            assert fieldType != null : "Unknown field type for field: [" + field + "]";
+
+            if (fieldType instanceof AggregateDoubleMetricFieldMapper.AggregateDoubleMetricFieldType aggMetricFieldType) {
+                // If the field is an aggregate_metric_double field, we should load all its subfields
+                // This is a rollup-of-rollup case
+                for (NumberFieldMapper.NumberFieldType metricSubField : aggMetricFieldType.getMetricFields().values()) {
+                    if (context.fieldExistsInIndex(metricSubField.name())) {
+                        IndexFieldData<?> fieldData = context.getForField(metricSubField, MappedFieldType.FielddataOperation.SEARCH);
+                        fetchers.put(metricSubField.name(), new FieldValueFetcher(metricSubField.name(), fieldType, fieldData));
+                    }
+                }
+            } else {
+                if (context.fieldExistsInIndex(field)) {
+                    IndexFieldData<?> fieldData = context.getForField(fieldType, MappedFieldType.FielddataOperation.SEARCH);
+                    fetchers.put(field, new FieldValueFetcher(field, fieldType, fieldData));
+                }
             }
-            IndexFieldData<?> fieldData = context.getForField(fieldType, MappedFieldType.FielddataOperation.SEARCH);
-            fetchers.add(new FieldValueFetcher(field, fieldType, fieldData));
         }
-        return Collections.unmodifiableList(fetchers);
+        return Collections.unmodifiableMap(fetchers);
     }
 
-    static List<FieldValueFetcher> forMetrics(SearchExecutionContext context, String[] metricFields) {
-        return build(context, metricFields);
-    }
-
-    static List<FieldValueFetcher> forLabels(SearchExecutionContext context, String[] labelFields) {
-        return build(context, labelFields);
+    static Map<String, FormattedDocValues> docValuesFetchers(LeafReaderContext ctx, Map<String, FieldValueFetcher> fieldValueFetchers) {
+        final Map<String, FormattedDocValues> docValuesFetchers = new LinkedHashMap<>(fieldValueFetchers.size());
+        for (FieldValueFetcher fetcher : fieldValueFetchers.values()) {
+            docValuesFetchers.put(fetcher.name(), fetcher.getLeaf(ctx));
+        }
+        return Collections.unmodifiableMap(docValuesFetchers);
     }
-
 }
diff --git a/x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/downsample/LabelFieldProducer.java b/x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/downsample/LabelFieldProducer.java
index c10eeaec5dc91..cfaeeae7ca9af 100644
--- a/x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/downsample/LabelFieldProducer.java
+++ b/x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/downsample/LabelFieldProducer.java
@@ -7,8 +7,14 @@
 
 package org.elasticsearch.xpack.downsample;
 
+import org.elasticsearch.index.mapper.MappedFieldType;
+import org.elasticsearch.index.mapper.NumberFieldMapper;
 import org.elasticsearch.index.query.SearchExecutionContext;
+import org.elasticsearch.xcontent.XContentBuilder;
+import org.elasticsearch.xpack.aggregatemetric.mapper.AggregateDoubleMetricFieldMapper;
 
+import java.io.IOException;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.LinkedHashMap;
 import java.util.Map;
@@ -31,7 +37,7 @@ public String name() {
 
     /** Collect the value of a raw field  */
     @Override
-    public void collect(Object value) {
+    public void collect(String field, Object value) {
         label.collect(value);
         isEmpty = false;
     }
@@ -55,13 +61,17 @@ abstract static class Label {
         final String name;
 
         /**
-         * Abstract class that defines the how a label is computed.
-         * @param name
+         * Abstract class that defines how a label is downsampled.
+         * @param name the name of the field as it will be stored in the downsampled document
          */
         protected Label(String name) {
             this.name = name;
         }
 
+        public String name() {
+            return name;
+        }
+
         abstract void collect(Object value);
 
         abstract Object get();
@@ -79,6 +89,10 @@ protected Label(String name) {
     static class LastValueLabel extends Label {
         private Object lastValue;
 
+        LastValueLabel(String name) {
+            super(name);
+        }
+
         LastValueLabel() {
             super("last_value");
         }
@@ -114,16 +128,85 @@ static class LabelLastValueFieldProducer extends LabelFieldProducer {
         public Object value() {
             return label().get();
         }
+
+        @Override
+        public void write(XContentBuilder builder) throws IOException {
+            if (isEmpty() == false) {
+                builder.field(name(), value());
+            }
+        }
+    }
+
+    static class AggregateMetricFieldProducer extends LabelFieldProducer {
+
+        private Map<String, Label> labelsByField = new LinkedHashMap<>();
+
+        AggregateMetricFieldProducer(String name) {
+            super(name, null);
+        }
+
+        public void addLabel(String field, Label label) {
+            labelsByField.put(field, label);
+        }
+
+        @Override
+        public void collect(String field, Object value) {
+            labelsByField.get(field).collect(value);
+            isEmpty = false;
+        }
+
+        @Override
+        public void write(XContentBuilder builder) throws IOException {
+            if (isEmpty() == false) {
+                builder.startObject(name());
+                for (Label label : labels()) {
+                    if (label.get() != null) {
+                        builder.field(label.name(), label.get());
+                    }
+                }
+                builder.endObject();
+            }
+        }
+
+        public Collection<Label> labels() {
+            return labelsByField.values();
+        }
+
+        @Override
+        public Object value() {
+            return labelsByField;
+        }
+
+        @Override
+        public void reset() {
+            labels().forEach(Label::reset);
+            isEmpty = true;
+        }
     }
 
     /**
-     * Produce a collection of label field producers.
+     * Create a collection of label field producers.
      */
-    static Map<String, LabelFieldProducer> buildLabelFieldProducers(SearchExecutionContext context, String[] labelFields) {
+    static Map<String, LabelFieldProducer> createLabelFieldProducers(SearchExecutionContext context, String[] labelFields) {
         final Map<String, LabelFieldProducer> fields = new LinkedHashMap<>();
         for (String field : labelFields) {
-            LabelFieldProducer producer = new LabelLastValueFieldProducer(field);
-            fields.put(field, producer);
+            MappedFieldType fieldType = context.getFieldType(field);
+            assert fieldType != null : "Unknown field type for field: [" + field + "]";
+
+            if (fieldType instanceof AggregateDoubleMetricFieldMapper.AggregateDoubleMetricFieldType aggMetricFieldType) {
+                // If the field is an aggregate_metric_double field, we should use the correct subfields
+                // for each aggregation. This is a rollup-of-rollup case
+                AggregateMetricFieldProducer producer = new AggregateMetricFieldProducer.AggregateMetricFieldProducer(field);
+                for (var e : aggMetricFieldType.getMetricFields().entrySet()) {
+                    AggregateDoubleMetricFieldMapper.Metric metric = e.getKey();
+                    NumberFieldMapper.NumberFieldType metricSubField = e.getValue();
+                    producer.addLabel(metricSubField.name(), new LastValueLabel(metric.name()));
+                    fields.put(metricSubField.name(), producer);
+                }
+            } else {
+                LabelFieldProducer producer = new LabelLastValueFieldProducer(field);
+                fields.put(field, producer);
+            }
         }
         return Collections.unmodifiableMap(fields);
     }
diff --git a/x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/downsample/MetricFieldProducer.java b/x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/downsample/MetricFieldProducer.java
index 5164e6d8518df..a849ac915eeb6 100644
--- a/x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/downsample/MetricFieldProducer.java
+++ b/x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/downsample/MetricFieldProducer.java
@@ -8,11 +8,15 @@
 package org.elasticsearch.xpack.downsample;
 
 import org.elasticsearch.index.mapper.MappedFieldType;
+import org.elasticsearch.index.mapper.NumberFieldMapper;
 import org.elasticsearch.index.query.SearchExecutionContext;
 import org.elasticsearch.search.aggregations.metrics.CompensatedSum;
+import org.elasticsearch.xcontent.XContentBuilder;
+import org.elasticsearch.xpack.aggregatemetric.mapper.AggregateDoubleMetricFieldMapper;
 
+import java.io.IOException;
+import java.util.Collection;
 import java.util.Collections;
-import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
@@ -37,7 +41,7 @@ abstract class MetricFieldProducer extends AbstractRollupFieldProducer<Number> {
      * Reset all values collected for the field
      */
     public void reset() {
-        for (Metric metric : metrics) {
+        for (Metric metric : metrics()) {
             metric.reset();
         }
         isEmpty = true;
@@ -48,36 +52,34 @@ public String name() {
     }
 
     /** return the list of metrics that are computed for the field */
-    public List<Metric> metrics() {
+    public Collection<Metric> metrics() {
         return metrics;
     }
 
     /** Collect the value of a raw field and compute all downsampled metrics */
     @Override
-    public void collect(Number value) {
-        for (MetricFieldProducer.Metric metric : metrics) {
+    public void collect(String field, Number value) {
+        for (MetricFieldProducer.Metric metric : metrics()) {
             metric.collect(value);
         }
         isEmpty = false;
     }
 
-    /**
-     * Return the downsampled value as computed after collecting all raw values.
-     * @return
-     */
-    public abstract Object value();
-
     abstract static class Metric {
         final String name;
 
         /**
-         * Abstract class that defines the how a metric is computed.
-         * @param name
+         * Abstract class that defines how a metric is computed.
+         * @param name the name of the metric as it will be output in the downsampled document
          */
         protected Metric(String name) {
             this.name = name;
         }
 
+        public String name() {
+            return name;
+        }
+
         abstract void collect(Number number);
 
         abstract Number get();
@@ -147,6 +149,10 @@ static class Sum extends Metric {
             super("sum");
         }
 
+        Sum(String name) {
+            super(name);
+        }
+
         @Override
         void collect(Number value) {
             kahanSummation.add(value.doubleValue());
@@ -227,13 +233,19 @@ void reset() {
     static class CounterMetricFieldProducer extends MetricFieldProducer {
 
         CounterMetricFieldProducer(String name) {
-            super(name, List.of(new LastValue()));
+            super(name, Collections.singletonList(new LastValue()));
         }
 
-        @Override
         public Object value() {
-            assert metrics().size() == 1 : "Counters have only one metric";
-            return metrics().get(0).get();
+            assert metrics().size() == 1 : "Single value producers must have only one metric";
+            return metrics().iterator().next().get();
+        }
+
+        @Override
+        public void write(XContentBuilder builder) throws IOException {
+            if (isEmpty() == false) {
+                builder.field(name(), value());
+            }
         }
     }
 
@@ -243,38 +255,99 @@ public Object value() {
     static class GaugeMetricFieldProducer extends MetricFieldProducer {
 
         GaugeMetricFieldProducer(String name) {
-            super(name, List.of(new Min(), new Max(), new Sum(), new ValueCount()));
+            this(name, List.of(new Min(), new Max(), new Sum(), new ValueCount()));
+        }
+
+        GaugeMetricFieldProducer(String name, List<Metric> metrics) {
+            super(name, metrics);
         }
 
         @Override
-        public Object value() {
-            Map<String, Object> metricValues = new HashMap<>();
-            for (MetricFieldProducer.Metric metric : metrics()) {
-                if (metric.get() != null) {
-                    metricValues.put(metric.name, metric.get());
+        public void write(XContentBuilder builder) throws IOException {
+            if (isEmpty() == false) {
+                builder.startObject(name());
+                for (MetricFieldProducer.Metric metric : metrics()) {
+                    if (metric.get() != null) {
+                        builder.field(metric.name(), metric.get());
+                    }
                 }
+                builder.endObject();
             }
-            return Collections.unmodifiableMap(metricValues);
+        }
+    }
+
+    static class AggregateMetricFieldProducer extends MetricFieldProducer {
+
+        private Map<String, Metric> metricsByField = new LinkedHashMap<>();
+
+        AggregateMetricFieldProducer(String name) {
+            super(name, Collections.emptyList());
+        }
+
+        public void addMetric(String field, Metric metric) {
+            metricsByField.put(field, metric);
+        }
+
+        @Override
+        public void collect(String field, Number value) {
+            metricsByField.get(field).collect(value);
+            isEmpty = false;
+        }
+
+        @Override
+        public void write(XContentBuilder builder) throws IOException {
+            if (isEmpty() == false) {
+                builder.startObject(name());
+                for (MetricFieldProducer.Metric metric : metrics()) {
+                    if (metric.get() != null) {
+                        builder.field(metric.name(), metric.get());
+                    }
+                }
+                builder.endObject();
+            }
+        }
+
+        @Override
+        public Collection<Metric> metrics() {
+            return metricsByField.values();
         }
     }
 
     /**
-     * Produce a collection of metric field producers based on the metric_type mapping parameter in the field
+     * Create a collection of metric field producers based on the metric_type mapping parameter in the field
      * mapping.
      */
-    static Map<String, MetricFieldProducer> buildMetricFieldProducers(SearchExecutionContext context, String[] metricFields) {
+    static Map<String, MetricFieldProducer> createMetricFieldProducers(SearchExecutionContext context, String[] metricFields) {
         final Map<String, MetricFieldProducer> fields = new LinkedHashMap<>();
         for (String field : metricFields) {
             MappedFieldType fieldType = context.getFieldType(field);
-            assert fieldType.getMetricType() != null;
-
-            MetricFieldProducer producer = switch (fieldType.getMetricType()) {
-                case gauge -> new GaugeMetricFieldProducer(field);
-                case counter -> new CounterMetricFieldProducer(field);
-                default -> throw new IllegalArgumentException("Unsupported metric type [" + fieldType.getMetricType() + "]");
-            };
-
-            fields.put(field, producer);
+            assert fieldType != null : "Unknown field type for field: [" + field + "]";
+            assert fieldType.getMetricType() != null : "Unknown metric type for metric field: [" + field + "]";
+
+            if (fieldType instanceof AggregateDoubleMetricFieldMapper.AggregateDoubleMetricFieldType aggMetricFieldType) {
+                // If the field is an aggregate_metric_double field, we should use the correct subfields
+                // for each aggregation. This is a rollup-of-rollup case
+                AggregateMetricFieldProducer producer = new AggregateMetricFieldProducer(field);
+                for (var e : aggMetricFieldType.getMetricFields().entrySet()) {
+                    AggregateDoubleMetricFieldMapper.Metric metric = e.getKey();
+                    NumberFieldMapper.NumberFieldType metricSubField = e.getValue();
+                    Metric metricOperation = switch (metric) {
+                        case max -> new Max();
+                        case min -> new Min();
+                        case sum -> new Sum();
+                        // To compute value_count summary, we must sum all field values
+                        case value_count -> new Sum(AggregateDoubleMetricFieldMapper.Metric.value_count.name());
+                    };
+                    producer.addMetric(metricSubField.name(), metricOperation);
+                    fields.put(metricSubField.name(), producer);
+                }
+            } else {
+                MetricFieldProducer producer = switch (fieldType.getMetricType()) {
+                    case gauge -> new GaugeMetricFieldProducer(field);
+                    case counter -> new CounterMetricFieldProducer(field);
+                };
+                fields.put(field, producer);
+            }
         }
         return Collections.unmodifiableMap(fields);
     }
diff --git a/x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/downsample/RollupShardIndexer.java b/x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/downsample/RollupShardIndexer.java
index 758b46dd45124..856fcb8d53c3c 100644
--- a/x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/downsample/RollupShardIndexer.java
+++ b/x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/downsample/RollupShardIndexer.java
@@ -24,7 +24,7 @@
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.unit.ByteSizeUnit;
 import org.elasticsearch.common.unit.ByteSizeValue;
-import org.elasticsearch.common.util.set.Sets;
+import org.elasticsearch.common.util.ArrayUtils;
 import org.elasticsearch.core.IOUtils;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.index.IndexService;
@@ -53,21 +53,19 @@
 import java.io.IOException;
 import java.util.Arrays;
 import java.util.Collections;
-import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
-import java.util.Set;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Function;
 import java.util.stream.Collectors;
-import java.util.stream.Stream;
 
 import static org.elasticsearch.core.Strings.format;
 
 /**
- * An indexer for rollups that iterates documents collected by {@link TimeSeriesIndexSearcher},
- * computes the rollup buckets and stores the buckets in the rollup index.
+ * An indexer for downsampling that iterates documents collected by {@link TimeSeriesIndexSearcher},
+ * computes the rollup buckets and stores the buckets in the downsampled index.
  *
  * The documents collected by the {@link TimeSeriesIndexSearcher} are expected to be sorted
  * by _tsid in ascending order and @timestamp in descending order.
@@ -91,9 +89,7 @@ class RollupShardIndexer {
     private final String[] dimensionFields;
     private final String[] metricFields;
     private final String[] labelFields;
-    private final List<FieldValueFetcher> metricFieldFetchers;
-    private final List<FieldValueFetcher> labelFieldFetchers;
-
+    private final Map<String, FieldValueFetcher> fieldValueFetchers;
     private final AtomicLong numSent = new AtomicLong();
     private final AtomicLong numIndexed = new AtomicLong();
     private final AtomicLong numFailed = new AtomicLong();
@@ -116,7 +112,7 @@ class RollupShardIndexer {
         this.metricFields = metricFields;
         this.labelFields = labelFields;
 
-        this.searcher = indexShard.acquireSearcher("rollup");
+        this.searcher = indexShard.acquireSearcher("downsampling");
         Closeable toClose = searcher;
         try {
             this.searchExecutionContext = indexService.newSearchExecutionContext(
@@ -130,8 +126,7 @@ class RollupShardIndexer {
             this.timestampField = searchExecutionContext.getFieldType(DataStreamTimestampFieldMapper.DEFAULT_PATH);
             this.timestampFormat = timestampField.docValueFormat(null, null);
             this.rounding = config.createRounding();
-            this.metricFieldFetchers = FieldValueFetcher.forMetrics(searchExecutionContext, metricFields);
-            this.labelFieldFetchers = FieldValueFetcher.forLabels(searchExecutionContext, labelFields);
+            this.fieldValueFetchers = FieldValueFetcher.create(searchExecutionContext, ArrayUtils.concat(metricFields, labelFields));
             toClose = null;
         } finally {
             IOUtils.closeWhileHandlingException(toClose);
@@ -234,20 +229,7 @@ public LeafBucketCollector getLeafCollector(final AggregationExecutionContext ag
             final LeafReaderContext ctx = aggCtx.getLeafReaderContext();
             final DocCountProvider docCountProvider = new DocCountProvider();
             docCountProvider.setLeafReaderContext(ctx);
-            final Map<String, FormattedDocValues> metricsFieldLeaves = new HashMap<>();
-            for (FieldValueFetcher fetcher : metricFieldFetchers) {
-                metricsFieldLeaves.put(fetcher.name(), fetcher.getLeaf(ctx));
-            }
-
-            final Map<String, FormattedDocValues> labelFieldLeaves = new HashMap<>();
-            for (FieldValueFetcher fetcher : labelFieldFetchers) {
-                labelFieldLeaves.put(fetcher.name(), fetcher.getLeaf(ctx));
-            }
-
-            Set<Map.Entry<String, FormattedDocValues>> fieldFetchers = Sets.union(
-                metricsFieldLeaves.entrySet(),
-                labelFieldLeaves.entrySet()
-            );
+            final Map<String, FormattedDocValues> docValuesFetchers = FieldValueFetcher.docValuesFetchers(ctx, fieldValueFetchers);
 
             return new LeafBucketCollector() {
                 @Override
@@ -314,7 +296,7 @@ public void collect(int docId, long owningBucketOrd) throws IOException {
 
                     final int docCount = docCountProvider.getDocCount(docId);
                     rollupBucketBuilder.collectDocCount(docCount);
-                    for (Map.Entry<String, FormattedDocValues> e : fieldFetchers) {
+                    for (Map.Entry<String, FormattedDocValues> e : docValuesFetchers.entrySet()) {
                         final String fieldName = e.getKey();
                         final FormattedDocValues leafField = e.getValue();
 
@@ -327,7 +309,6 @@ public void collect(int docId, long owningBucketOrd) throws IOException {
                                     } catch (IOException ex) {
                                         throw new ElasticsearchException("Failed to read values for field [" + fieldName + "]");
                                     }
-
                                 }
                                 return values;
                             });
@@ -378,8 +359,8 @@ private class RollupBucketBuilder {
         private final Map<String, LabelFieldProducer> labelFieldProducers;
 
         RollupBucketBuilder() {
-            this.metricFieldProducers = MetricFieldProducer.buildMetricFieldProducers(searchExecutionContext, metricFields);
-            this.labelFieldProducers = LabelFieldProducer.buildLabelFieldProducers(searchExecutionContext, labelFields);
+            this.metricFieldProducers = MetricFieldProducer.createMetricFieldProducers(searchExecutionContext, metricFields);
+            this.labelFieldProducers = LabelFieldProducer.createLabelFieldProducers(searchExecutionContext, labelFields);
         }
 
         /**
@@ -434,13 +415,13 @@ public void collect(final String field, int docValueCount, final Function<Intege
         }
 
         private void collectLabel(final String field, final Object value) {
-            labelFieldProducers.get(field).collect(value);
+            labelFieldProducers.get(field).collect(field, value);
         }
 
         private void collectMetric(final String field, final Object[] values) {
-            for (var value : values) {
+            for (Object value : values) {
                 if (value instanceof Number number) {
-                    metricFieldProducers.get(field).collect(number);
+                    metricFieldProducers.get(field).collect(field, number);
                 } else {
                     throw new IllegalArgumentException(
                         "Expected numeric value for field '" + field + "' but got non numeric value: '" + value + "'"
@@ -471,17 +452,14 @@ public XContentBuilder buildRollupDocument() throws IOException {
                 builder.field(e.getKey(), e.getValue());
             }
 
-            for (AbstractRollupFieldProducer<?> fieldProducer : Stream.concat(
-                metricFieldProducers.values().stream(),
-                labelFieldProducers.values().stream()
-            ).toList()) {
-                if (fieldProducer.isEmpty() == false) {
-                    String field = fieldProducer.name();
-                    Object value = fieldProducer.value();
-                    if (value != null) {
-                        builder.field(field, value);
-                    }
-                }
+            // Serialize all metric fields
+            for (var producer : new HashSet<>(metricFieldProducers.values())) {
+                producer.write(builder);
+            }
+
+            // Serialize all label fields
+            for (var producer : new HashSet<>(labelFieldProducers.values())) {
+                producer.write(builder);
             }
 
             builder.endObject();
diff --git a/x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/downsample/TimeseriesFieldTypeHelper.java b/x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/downsample/TimeseriesFieldTypeHelper.java
index 21cd5a27aa2b7..d3f0874f5abd1 100644
--- a/x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/downsample/TimeseriesFieldTypeHelper.java
+++ b/x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/downsample/TimeseriesFieldTypeHelper.java
@@ -7,15 +7,12 @@
 
 package org.elasticsearch.xpack.downsample;
 
-import org.elasticsearch.cluster.metadata.IndexMetadata;
-import org.elasticsearch.common.compress.CompressedXContent;
 import org.elasticsearch.index.mapper.MappedFieldType;
 import org.elasticsearch.index.mapper.MapperService;
 import org.elasticsearch.index.mapper.TimeSeriesParams;
-import org.elasticsearch.indices.IndicesService;
 
 import java.io.IOException;
-import java.util.Arrays;
+import java.util.List;
 import java.util.Map;
 
 import static org.elasticsearch.index.mapper.TimeSeriesParams.TIME_SERIES_DIMENSION_PARAM;
@@ -43,7 +40,7 @@ public boolean isTimeSeriesLabel(final String field, final Map<String, ?> unused
     public boolean isTimeSeriesMetric(final String unused, final Map<String, ?> fieldMapping) {
         final String metricType = (String) fieldMapping.get(TIME_SERIES_METRIC_PARAM);
         return metricType != null
-            && Arrays.asList(TimeSeriesParams.MetricType.values()).contains(TimeSeriesParams.MetricType.valueOf(metricType));
+            && List.of(TimeSeriesParams.MetricType.values()).contains(TimeSeriesParams.MetricType.valueOf(metricType));
     }
 
     public boolean isTimeSeriesDimension(final String unused, final Map<String, ?> fieldMapping) {
@@ -51,20 +48,13 @@ public boolean isTimeSeriesDimension(final String unused, final Map<String, ?> f
     }
 
     static class Builder {
-        private final IndicesService indicesService;
-        private final Map<String, Object> indexMapping;
-        private final IndexMetadata indexMetadata;
+        private final MapperService mapperService;
 
-        Builder(final IndicesService indicesService, final Map<String, Object> indexMapping, final IndexMetadata indexMetadata) {
-            this.indicesService = indicesService;
-            this.indexMapping = indexMapping;
-            this.indexMetadata = indexMetadata;
+        Builder(final MapperService mapperService) {
+            this.mapperService = mapperService;
         }
 
         public TimeseriesFieldTypeHelper build(final String timestampField) throws IOException {
-            final MapperService mapperService = indicesService.createIndexMapperServiceForValidation(indexMetadata);
-            final CompressedXContent sourceIndexCompressedXContent = new CompressedXContent(indexMapping);
-            mapperService.merge(MapperService.SINGLE_MAPPING_NAME, sourceIndexCompressedXContent, MapperService.MergeReason.INDEX_TEMPLATE);
             return new TimeseriesFieldTypeHelper(mapperService, timestampField);
         }
     }
diff --git a/x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/downsample/TransportRollupAction.java b/x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/downsample/TransportRollupAction.java
index b15942eff8146..458f4f386b7c1 100644
--- a/x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/downsample/TransportRollupAction.java
+++ b/x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/downsample/TransportRollupAction.java
@@ -48,9 +48,11 @@
 import org.elasticsearch.index.IndexNotFoundException;
 import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.index.mapper.DateFieldMapper;
+import org.elasticsearch.index.mapper.MappedFieldType;
 import org.elasticsearch.index.mapper.MapperService;
 import org.elasticsearch.index.mapper.TimeSeriesParams;
 import org.elasticsearch.indices.IndicesService;
+import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramInterval;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.tasks.TaskId;
 import org.elasticsearch.threadpool.ThreadPool;
@@ -206,14 +208,19 @@ protected void masterOperation(
                 .orElseThrow(() -> new IllegalArgumentException("No mapping found for rollup source index [" + sourceIndexName + "]"));
 
             // 2. Extract rollup config from index mappings
+            final MapperService mapperService = indicesService.createIndexMapperServiceForValidation(sourceIndexMetadata);
+            final CompressedXContent sourceIndexCompressedXContent = new CompressedXContent(sourceIndexMappings);
+            mapperService.merge(MapperService.SINGLE_MAPPING_NAME, sourceIndexCompressedXContent, MapperService.MergeReason.INDEX_TEMPLATE);
+
+            // Validate downsampling interval
+            validateDownsamplingInterval(mapperService, request.getDownsampleConfig());
+
             final List<String> dimensionFields = new ArrayList<>();
             final List<String> metricFields = new ArrayList<>();
             final List<String> labelFields = new ArrayList<>();
-            final TimeseriesFieldTypeHelper helper = new TimeseriesFieldTypeHelper.Builder(
-                indicesService,
-                sourceIndexMappings,
-                sourceIndexMetadata
-            ).build(request.getDownsampleConfig().getTimestampField());
+            final TimeseriesFieldTypeHelper helper = new TimeseriesFieldTypeHelper.Builder(mapperService).build(
+                request.getDownsampleConfig().getTimestampField()
+            );
             MappingVisitor.visitMapping(sourceIndexMappings, (field, mapping) -> {
                 if (helper.isTimeSeriesDimension(field, mapping)) {
                     dimensionFields.add(field);
@@ -237,10 +244,6 @@ protected void masterOperation(
                 return;
             }
 
-            final MapperService mapperService = indicesService.createIndexMapperServiceForValidation(sourceIndexMetadata);
-            final CompressedXContent sourceIndexCompressedXContent = new CompressedXContent(sourceIndexMappings);
-            mapperService.merge(MapperService.SINGLE_MAPPING_NAME, sourceIndexCompressedXContent, MapperService.MergeReason.INDEX_TEMPLATE);
-
             final String mapping;
             try {
                 mapping = createRollupIndexMapping(helper, request.getDownsampleConfig(), mapperService, sourceIndexMappings);
@@ -474,6 +477,59 @@ private static void addMetricFieldMapping(final XContentBuilder builder, final S
         }
     }
 
+    private static void validateDownsamplingInterval(MapperService mapperService, DownsampleConfig config) {
+        MappedFieldType timestampFieldType = mapperService.fieldType(config.getTimestampField());
+        assert timestampFieldType != null : "Cannot find timestamp field [" + config.getTimestampField() + "] in the mapping";
+        RollupActionRequestValidationException e = new RollupActionRequestValidationException();
+
+        Map<String, String> meta = timestampFieldType.meta();
+        if (meta.isEmpty() == false) {
+            String interval = meta.get(config.getIntervalType());
+            if (interval != null) {
+                DateHistogramInterval sourceIndexInterval = new DateHistogramInterval(interval);
+                DateHistogramInterval targetIndexInterval = config.getInterval();
+                long sourceMillis = sourceIndexInterval.estimateMillis();
+                long targetMillis = targetIndexInterval.estimateMillis();
+                if (sourceMillis >= targetMillis) {
+                    // Downsampling interval must be greater than source interval
+                    e.addValidationError(
+                        "Source index is a downsampled index. Downsampling interval ["
+                            + targetIndexInterval
+                            + "] must be greater than the the source index interval ["
+                            + sourceIndexInterval
+                            + "]"
+                    );
+                } else if (targetMillis % sourceMillis != 0) {
+                    // Downsampling interval must be a multiple of the source interval
+                    e.addValidationError(
+                        "Source index is a downsampled index. Downsampling interval ["
+                            + targetIndexInterval
+                            + "] must be a multiple of the source index interval ["
+                            + sourceIndexInterval
+                            + "]"
+                    );
+                }
+            }
+
+            // Validate that timezones match
+            String sourceTimezone = meta.get(DownsampleConfig.TIME_ZONE);
+            if (sourceTimezone != null && sourceTimezone.equals(config.getTimeZone()) == false) {
+                e.addValidationError(
+                    "Source index is a downsampled index. Downsampling timezone ["
+                        + config.getTimeZone()
+                        + "] cannot be different than the source index timezone ["
+                        + sourceTimezone
+                        + "]"
+                );
+            }
+
+            if (e.validationErrors().isEmpty() == false) {
+                throw e;
+            }
+
+        }
+    }
+
     /**
      * Copy index settings from the source index to the rollup index. Settings that
      * have already been set in the rollup index will not be overridden.
diff --git a/x-pack/plugin/rollup/src/test/java/org/elasticsearch/xpack/downsample/DownsampleActionSingleNodeTests.java b/x-pack/plugin/rollup/src/test/java/org/elasticsearch/xpack/downsample/DownsampleActionSingleNodeTests.java
index 948458f798902..f48ba9bf27bea 100644
--- a/x-pack/plugin/rollup/src/test/java/org/elasticsearch/xpack/downsample/DownsampleActionSingleNodeTests.java
+++ b/x-pack/plugin/rollup/src/test/java/org/elasticsearch/xpack/downsample/DownsampleActionSingleNodeTests.java
@@ -69,7 +69,6 @@
 import org.elasticsearch.xcontent.XContentBuilder;
 import org.elasticsearch.xcontent.XContentFactory;
 import org.elasticsearch.xpack.aggregatemetric.AggregateMetricMapperPlugin;
-import org.elasticsearch.xpack.analytics.AnalyticsPlugin;
 import org.elasticsearch.xpack.core.LocalStateCompositeXPackPlugin;
 import org.elasticsearch.xpack.core.downsample.DownsampleAction;
 import org.elasticsearch.xpack.core.downsample.DownsampleConfig;
@@ -101,6 +100,7 @@
 import static org.elasticsearch.index.mapper.TimeSeriesParams.TIME_SERIES_METRIC_PARAM;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertHitCount;
+import static org.elasticsearch.xcontent.XContentFactory.jsonBuilder;
 import static org.hamcrest.Matchers.containsString;
 
 public class DownsampleActionSingleNodeTests extends ESSingleNodeTestCase {
@@ -111,6 +111,7 @@ public class DownsampleActionSingleNodeTests extends ESSingleNodeTestCase {
     public static final String FIELD_DIMENSION_2 = "dimension_long";
     public static final String FIELD_NUMERIC_1 = "numeric_1";
     public static final String FIELD_NUMERIC_2 = "numeric_2";
+    public static final String FIELD_AGG_METRIC = "agg_metric_1";
     public static final String FIELD_METRIC_LABEL_DOUBLE = "metric_label_double";
     public static final String FIELD_LABEL_DOUBLE = "label_double";
     public static final String FIELD_LABEL_INTEGER = "label_integer";
@@ -123,6 +124,7 @@ public class DownsampleActionSingleNodeTests extends ESSingleNodeTestCase {
     public static final String FIELD_LABEL_UNMAPPED = "label_unmapped";
     public static final String FIELD_LABEL_KEYWORD_ARRAY = "label_keyword_array";
     public static final String FIELD_LABEL_DOUBLE_ARRAY = "label_double_array";
+    public static final String FIELD_LABEL_AGG_METRIC = "label_agg_metric";
 
     private static final int MAX_DIM_VALUES = 5;
     private static final long MAX_NUM_BUCKETS = 10;
@@ -137,7 +139,6 @@ protected Collection<Class<? extends Plugin>> getPlugins() {
         return List.of(
             LocalStateCompositeXPackPlugin.class,
             Rollup.class,
-            AnalyticsPlugin.class,
             AggregateMetricMapperPlugin.class,
             DataStreamsPlugin.class,
             IndexLifecycle.class
@@ -145,7 +146,7 @@ protected Collection<Class<? extends Plugin>> getPlugins() {
     }
 
     @Before
-    public void setup() {
+    public void setup() throws IOException {
         sourceIndex = randomAlphaOfLength(14).toLowerCase(Locale.ROOT);
         rollupIndex = "rollup-" + sourceIndex;
         startTime = randomLongBetween(946769284000L, 1607470084000L); // random date between 2000-2020
@@ -180,47 +181,47 @@ public void setup() {
         if (randomBoolean()) {
             settings.put(IndexMetadata.SETTING_INDEX_HIDDEN, randomBoolean());
         }
-        assertAcked(
-            client().admin()
-                .indices()
-                .prepareCreate(sourceIndex)
-                .setSettings(settings.build())
-                .setMapping(
-                    FIELD_TIMESTAMP,
-                    "type=date",
-                    FIELD_DIMENSION_1,
-                    "type=keyword,time_series_dimension=true",
-                    FIELD_DIMENSION_2,
-                    "type=long,time_series_dimension=true",
-                    FIELD_NUMERIC_1,
-                    "type=long,time_series_metric=gauge",
-                    FIELD_NUMERIC_2,
-                    "type=double,time_series_metric=counter",
-                    FIELD_LABEL_DOUBLE,
-                    "type=double",
-                    FIELD_LABEL_INTEGER,
-                    "type=integer",
-                    FIELD_LABEL_KEYWORD,
-                    "type=keyword",
-                    FIELD_LABEL_TEXT,
-                    "type=text",
-                    FIELD_LABEL_BOOLEAN,
-                    "type=boolean",
-                    FIELD_METRIC_LABEL_DOUBLE, /* numeric label indexed as a metric */
-                    "type=double,time_series_metric=counter",
-                    FIELD_LABEL_IPv4_ADDRESS,
-                    "type=ip",
-                    FIELD_LABEL_IPv6_ADDRESS,
-                    "type=ip",
-                    FIELD_LABEL_DATE,
-                    "type=date,format=date_optional_time",
-                    FIELD_LABEL_KEYWORD_ARRAY,
-                    "type=keyword",
-                    FIELD_LABEL_DOUBLE_ARRAY,
-                    "type=double"
-                )
-                .get()
-        );
+
+        XContentBuilder mapping = jsonBuilder().startObject().startObject("_doc").startObject("properties");
+        mapping.startObject(FIELD_TIMESTAMP).field("type", "date").endObject();
+
+        // Dimensions
+        mapping.startObject(FIELD_DIMENSION_1).field("type", "keyword").field("time_series_dimension", true).endObject();
+        mapping.startObject(FIELD_DIMENSION_2).field("type", "long").field("time_series_dimension", true).endObject();
+
+        // Metrics
+        mapping.startObject(FIELD_NUMERIC_1).field("type", "long").field("time_series_metric", "gauge").endObject();
+        mapping.startObject(FIELD_NUMERIC_2).field("type", "double").field("time_series_metric", "counter").endObject();
+        mapping.startObject(FIELD_AGG_METRIC)
+            .field("type", "aggregate_metric_double")
+            .field("time_series_metric", "gauge")
+            .array("metrics", new String[] { "min", "max", "sum", "value_count" })
+            .field("default_metric", "value_count")
+            .endObject();
+        mapping.startObject(FIELD_METRIC_LABEL_DOUBLE)
+            .field("type", "double") /* numeric label indexed as a metric */
+            .field("time_series_metric", "counter")
+            .endObject();
+
+        // Labels
+        mapping.startObject(FIELD_LABEL_DOUBLE).field("type", "double").endObject();
+        mapping.startObject(FIELD_LABEL_INTEGER).field("type", "integer").endObject();
+        mapping.startObject(FIELD_LABEL_KEYWORD).field("type", "keyword").endObject();
+        mapping.startObject(FIELD_LABEL_TEXT).field("type", "text").endObject();
+        mapping.startObject(FIELD_LABEL_BOOLEAN).field("type", "boolean").endObject();
+        mapping.startObject(FIELD_LABEL_IPv4_ADDRESS).field("type", "ip").endObject();
+        mapping.startObject(FIELD_LABEL_IPv6_ADDRESS).field("type", "ip").endObject();
+        mapping.startObject(FIELD_LABEL_DATE).field("type", "date").field("format", "date_optional_time").endObject();
+        mapping.startObject(FIELD_LABEL_KEYWORD_ARRAY).field("type", "keyword").endObject();
+        mapping.startObject(FIELD_LABEL_DOUBLE_ARRAY).field("type", "double").endObject();
+        mapping.startObject(FIELD_LABEL_AGG_METRIC)
+            .field("type", "aggregate_metric_double")
+            .array("metrics", new String[] { "min", "max", "sum", "value_count" })
+            .field("default_metric", "value_count")
+            .endObject();
+
+        mapping.endObject().endObject().endObject();
+        assertAcked(client().admin().indices().prepareCreate(sourceIndex).setSettings(settings.build()).setMapping(mapping).get());
     }
 
     public void testRollupIndex() throws IOException {
@@ -250,6 +251,12 @@ public void testRollupIndex() throws IOException {
                 .field(FIELD_DIMENSION_2, randomIntBetween(1, 10))
                 .field(FIELD_NUMERIC_1, randomInt())
                 .field(FIELD_NUMERIC_2, DATE_FORMATTER.parseMillis(ts))
+                .startObject(FIELD_AGG_METRIC)
+                .field("min", randomDoubleBetween(-1000, 1000, true))
+                .field("max", randomDoubleBetween(-1000, 1000, true))
+                .field("sum", randomIntBetween(100, 10000))
+                .field("value_count", randomIntBetween(100, 1000))
+                .endObject()
                 .field(FIELD_LABEL_DOUBLE, labelDoubleValue)
                 .field(FIELD_METRIC_LABEL_DOUBLE, labelDoubleValue)
                 .field(FIELD_LABEL_INTEGER, labelIntegerValue)
@@ -262,12 +269,61 @@ public void testRollupIndex() throws IOException {
                 .field(FIELD_LABEL_DATE, labelDateValue)
                 .field(FIELD_LABEL_KEYWORD_ARRAY, keywordArray)
                 .field(FIELD_LABEL_DOUBLE_ARRAY, doubleArray)
+                .startObject(FIELD_LABEL_AGG_METRIC)
+                .field("min", randomDoubleBetween(-1000, 1000, true))
+                .field("max", randomDoubleBetween(-1000, 1000, true))
+                .field("sum", Double.valueOf(randomIntBetween(100, 10000)))
+                .field("value_count", randomIntBetween(100, 1000))
+                .endObject()
+                .endObject();
+        };
+        bulkIndex(sourceSupplier);
+        prepareSourceIndex(sourceIndex);
+        rollup(sourceIndex, rollupIndex, config);
+        assertRollupIndex(sourceIndex, rollupIndex, config);
+    }
+
+    public void testRollupOfRollups() throws IOException {
+        int intervalMinutes = randomIntBetween(10, 120);
+        DownsampleConfig config = new DownsampleConfig(DateHistogramInterval.minutes(intervalMinutes));
+        SourceSupplier sourceSupplier = () -> {
+            String ts = randomDateForInterval(config.getInterval());
+            double labelDoubleValue = DATE_FORMATTER.parseMillis(ts);
+
+            return XContentFactory.jsonBuilder()
+                .startObject()
+                .field(FIELD_TIMESTAMP, ts)
+                .field(FIELD_DIMENSION_1, randomFrom(dimensionValues))
+                .field(FIELD_NUMERIC_1, randomInt())
+                .field(FIELD_NUMERIC_2, DATE_FORMATTER.parseMillis(ts))
+                .startObject(FIELD_AGG_METRIC)
+                .field("min", randomDoubleBetween(-1000, 1000, true))
+                .field("max", randomDoubleBetween(-1000, 1000, true))
+                .field("sum", randomIntBetween(100, 10000))
+                .field("value_count", randomIntBetween(100, 1000))
+                .endObject()
+                .field(FIELD_LABEL_DOUBLE, labelDoubleValue)
+                .field(FIELD_METRIC_LABEL_DOUBLE, labelDoubleValue)
+                .startObject(FIELD_LABEL_AGG_METRIC)
+                .field("min", randomDoubleBetween(-1000, 1000, true))
+                .field("max", randomDoubleBetween(-1000, 1000, true))
+                .field("sum", Double.valueOf(randomIntBetween(100, 10000)))
+                .field("value_count", randomIntBetween(100, 1000))
+                .endObject()
                 .endObject();
         };
         bulkIndex(sourceSupplier);
+
+        // Downsample the source index
         prepareSourceIndex(sourceIndex);
         rollup(sourceIndex, rollupIndex, config);
         assertRollupIndex(sourceIndex, rollupIndex, config);
+
+        // Downsample the rollup index. The downsampling interval is a multiple of the previous downsampling interval.
+        String rollupIndex2 = rollupIndex + "-2";
+        DownsampleConfig config2 = new DownsampleConfig(DateHistogramInterval.minutes(intervalMinutes * randomIntBetween(2, 50)));
+        rollup(rollupIndex, rollupIndex2, config2);
+        assertRollupIndex(sourceIndex, rollupIndex2, config2);
     }
 
     private Date randomDate() {
@@ -593,12 +649,13 @@ private void assertRollupIndex(String sourceIndex, String rollupIndex, Downsampl
             .map(mappingMetadata -> mappingMetadata.getValue().sourceAsMap())
             .orElseThrow(() -> new IllegalArgumentException("No mapping found for rollup source index [" + sourceIndex + "]"));
 
-        IndexMetadata indexMetadata = client().admin().cluster().prepareState().get().getState().getMetadata().index(sourceIndex);
-        TimeseriesFieldTypeHelper helper = new TimeseriesFieldTypeHelper.Builder(
-            getInstanceFromNode(IndicesService.class),
-            sourceIndexMappings,
-            indexMetadata
-        ).build(config.getTimestampField());
+        final IndexMetadata indexMetadata = client().admin().cluster().prepareState().get().getState().getMetadata().index(sourceIndex);
+        final IndicesService indicesService = getInstanceFromNode(IndicesService.class);
+        final MapperService mapperService = indicesService.createIndexMapperServiceForValidation(indexMetadata);
+        final CompressedXContent sourceIndexCompressedXContent = new CompressedXContent(sourceIndexMappings);
+        mapperService.merge(MapperService.SINGLE_MAPPING_NAME, sourceIndexCompressedXContent, MapperService.MergeReason.INDEX_TEMPLATE);
+        TimeseriesFieldTypeHelper helper = new TimeseriesFieldTypeHelper.Builder(mapperService).build(config.getTimestampField());
+
         Map<String, TimeSeriesParams.MetricType> metricFields = new HashMap<>();
         Map<String, String> labelFields = new HashMap<>();
         MappingVisitor.visitMapping(sourceIndexMappings, (field, fieldMapping) -> {
@@ -724,8 +781,18 @@ private void assertRollupIndexAggregations(
                             // `counter`. This way we can actually check that the label value stored in the rollup index
                             // is the last value (which is what we store for a metric of type counter) by comparing the metric
                             // field value to the label field value.
-                            originalFieldsList.forEach(field -> assertTrue(rollupFieldsList.contains(field)));
-                            rollupFieldsList.forEach(field -> assertTrue(originalFieldsList.contains(field)));
+                            originalFieldsList.forEach(
+                                field -> assertTrue(
+                                    "Field [" + field + "] is not included in the rollup fields: " + rollupFieldsList,
+                                    rollupFieldsList.contains(field)
+                                )
+                            );
+                            rollupFieldsList.forEach(
+                                field -> assertTrue(
+                                    "Field [" + field + "] is not included in the source fields: " + originalFieldsList,
+                                    originalFieldsList.contains(field)
+                                )
+                            );
                             Object originalLabelValue = originalHit.getDocumentFields().values().stream().toList().get(0).getValue();
                             Object rollupLabelValue = rollupHit.getDocumentFields().values().stream().toList().get(0).getValue();
                             Optional<Aggregation> labelAsMetric = nonTopHitsOriginalAggregations.stream()
diff --git a/x-pack/plugin/rollup/src/test/java/org/elasticsearch/xpack/downsample/LabelFieldProducerTests.java b/x-pack/plugin/rollup/src/test/java/org/elasticsearch/xpack/downsample/LabelFieldProducerTests.java
index ee88dc83bbb18..641cbe228e02a 100644
--- a/x-pack/plugin/rollup/src/test/java/org/elasticsearch/xpack/downsample/LabelFieldProducerTests.java
+++ b/x-pack/plugin/rollup/src/test/java/org/elasticsearch/xpack/downsample/LabelFieldProducerTests.java
@@ -66,7 +66,7 @@ public void testLabelFieldProducer() {
         assertTrue(producer.isEmpty());
         assertEquals("dummy", producer.name());
         assertEquals("last_value", producer.label().name);
-        producer.collect("aaaa");
+        producer.collect("dummy", "aaaa");
         assertFalse(producer.isEmpty());
         assertEquals("aaaa", producer.value());
         producer.reset();
diff --git a/x-pack/plugin/rollup/src/test/java/org/elasticsearch/xpack/downsample/MetricFieldProducerTests.java b/x-pack/plugin/rollup/src/test/java/org/elasticsearch/xpack/downsample/MetricFieldProducerTests.java
index d27008729ae2c..0c93f010ec99e 100644
--- a/x-pack/plugin/rollup/src/test/java/org/elasticsearch/xpack/downsample/MetricFieldProducerTests.java
+++ b/x-pack/plugin/rollup/src/test/java/org/elasticsearch/xpack/downsample/MetricFieldProducerTests.java
@@ -7,18 +7,20 @@
 
 package org.elasticsearch.xpack.downsample;
 
+import org.elasticsearch.common.Strings;
 import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.index.mapper.MappedFieldType;
 import org.elasticsearch.index.mapper.NumberFieldMapper;
 import org.elasticsearch.index.mapper.TimeSeriesParams;
 import org.elasticsearch.index.query.SearchExecutionContext;
 import org.elasticsearch.search.aggregations.AggregatorTestCase;
+import org.elasticsearch.xcontent.XContentBuilder;
+import org.elasticsearch.xcontent.json.JsonXContent;
 
+import java.io.IOException;
 import java.util.Map;
 
 import static java.util.Collections.emptyMap;
-import static org.elasticsearch.test.MapMatcher.assertMap;
-import static org.elasticsearch.test.MapMatcher.matchesMap;
 
 public class MetricFieldProducerTests extends AggregatorTestCase {
 
@@ -119,36 +121,41 @@ public void testLastValueMetric() {
         assertNull(metric.get());
     }
 
-    public void testCounterMetricFieldProducer() {
-        MetricFieldProducer producer = new MetricFieldProducer.CounterMetricFieldProducer("field");
+    public void testCounterMetricFieldProducer() throws IOException {
+        final String field = "field";
+        var producer = new MetricFieldProducer.CounterMetricFieldProducer(field);
         assertTrue(producer.isEmpty());
-        producer.collect(55.0);
-        producer.collect(12.2);
-        producer.collect(5.5);
+        producer.collect(field, 55.0);
+        producer.collect(field, 12.2);
+        producer.collect(field, 5.5);
 
         assertFalse(producer.isEmpty());
         Object o = producer.value();
         assertEquals(55.0, o);
         assertEquals("field", producer.name());
+
+        XContentBuilder builder = JsonXContent.contentBuilder().startObject();
+        producer.write(builder);
+        builder.endObject();
+        assertEquals("{\"field\":55.0}", Strings.toString(builder));
     }
 
-    public void testGaugeMetricFieldProducer() {
-        MetricFieldProducer producer = new MetricFieldProducer.GaugeMetricFieldProducer("field");
+    public void testGaugeMetricFieldProducer() throws IOException {
+        final String field = "field";
+        MetricFieldProducer producer = new MetricFieldProducer.GaugeMetricFieldProducer(field);
         assertTrue(producer.isEmpty());
-        producer.collect(55.0);
-        producer.collect(12.2);
-        producer.collect(5.5);
+        producer.collect(field, 55.0);
+        producer.collect(field, 12.2);
+        producer.collect(field, 5.5);
 
         assertFalse(producer.isEmpty());
-        Object o = producer.value();
-        if (o instanceof Map) {
-            Map<?, ?> m = (Map<?, ?>) o;
-            assertMap(m, matchesMap().entry("min", 5.5).entry("max", 55.0).entry("value_count", 3L).entry("sum", 72.7));
-            assertEquals(4, m.size());
-        } else {
-            fail("Value is not a Map");
-        }
-        assertEquals("field", producer.name());
+
+        XContentBuilder builder = JsonXContent.contentBuilder().startObject();
+        producer.write(builder);
+        builder.endObject();
+        assertEquals("{\"field\":{\"min\":5.5,\"max\":55.0,\"sum\":72.7,\"value_count\":3}}", Strings.toString(builder));
+
+        assertEquals(field, producer.name());
     }
 
     public void testBuildMetricProducers() {
@@ -211,7 +218,7 @@ public MappedFieldType getFieldType(String name) {
             }
         };
 
-        Map<String, MetricFieldProducer> producers = MetricFieldProducer.buildMetricFieldProducers(
+        Map<String, MetricFieldProducer> producers = MetricFieldProducer.createMetricFieldProducers(
             searchExecutionContext,
             new String[] { "gauge_field", "counter_field" }
         );