Implement aggregations on aggregate metric fields (#56745)

In the process of developing a new implementation for the Elasticsearch Rollups functionality we came up with the concept of the aggregate metric field type.

The aggregate_metric_double field type can store the results of aggregations (currently min, max, sum, value_count and avg are supported - more to come).

This field allows us to run (min, max, sum, value_count, avg) aggregations on the container field and the field will return the correct metric depending on the aggregation that is computed.

docs/reference/rest-api/info.asciidoc

@@ -67,6 +67,10 @@ Example response:
         "available" : true,
         "enabled" : true
       },
+     "aggregate_metric" : {
+          "available" : true,
+          "enabled" : true
+      },
       "analytics" : {
           "available" : true,
           "enabled" : true

docs/reference/rest-api/usage.asciidoc

@@ -337,6 +337,10 @@ GET /_xpack/usage
       "primary_shard_size_median_bytes" : 0,
       "primary_shard_size_mad_bytes" : 0
     }
+  },
+  "aggregate_metric" : {
+    "available" : true,
+    "enabled" : true
   }
 }
 ------------------------------------------------------------

server/src/main/java/org/elasticsearch/search/aggregations/metrics/InternalSum.java

@@ -32,7 +32,7 @@ import java.util.Objects;
 public class InternalSum extends InternalNumericMetricsAggregation.SingleValue implements Sum {
     private final double sum;
 
-    public InternalSum(String name, double sum, DocValueFormat formatter, Map<String, Object> metadata) {
+    public  InternalSum(String name, double sum, DocValueFormat formatter, Map<String, Object> metadata) {
         super(name, metadata);
         this.sum = sum;
         this.format = formatter;

server/src/main/java/org/elasticsearch/search/aggregations/metrics/MinAggregator.java

@@ -42,7 +42,7 @@ import java.io.IOException;
 import java.util.Map;
 import java.util.function.Function;
 
-class MinAggregator extends NumericMetricsAggregator.SingleValue {
+public class MinAggregator extends NumericMetricsAggregator.SingleValue {
     private static final int MAX_BKD_LOOKUPS = 1024;
 
     final ValuesSource.Numeric valuesSource;

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/XPackClientPlugin.java

@@ -33,6 +33,7 @@ import org.elasticsearch.tasks.Task;
 import org.elasticsearch.xpack.ccr.CCRInfoTransportAction;
 import org.elasticsearch.xpack.core.action.XPackInfoAction;
 import org.elasticsearch.xpack.core.action.XPackUsageAction;
+import org.elasticsearch.xpack.core.aggregatemetric.AggregateMetricFeatureSetUsage;
 import org.elasticsearch.xpack.core.analytics.AnalyticsFeatureSetUsage;
 import org.elasticsearch.xpack.core.async.DeleteAsyncResultAction;
 import org.elasticsearch.xpack.core.ccr.AutoFollowMetadata;
@@ -500,6 +501,8 @@ public class XPackClientPlugin extends Plugin implements ActionPlugin, NetworkPl
             new NamedWriteableRegistry.Entry(XPackFeatureSet.Usage.class, XPackField.SPATIAL, SpatialFeatureSetUsage::new),
             // Analytics
             new NamedWriteableRegistry.Entry(XPackFeatureSet.Usage.class, XPackField.ANALYTICS, AnalyticsFeatureSetUsage::new),
+            // Aggregate metric field type
+            new NamedWriteableRegistry.Entry(XPackFeatureSet.Usage.class, XPackField.AGGREGATE_METRIC, AggregateMetricFeatureSetUsage::new),
             // Enrich
             new NamedWriteableRegistry.Entry(XPackFeatureSet.Usage.class, XPackField.ENRICH, EnrichFeatureSetUsage::new),
             new NamedWriteableRegistry.Entry(Task.Status.class, ExecuteEnrichPolicyStatus.NAME, ExecuteEnrichPolicyStatus::new),

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/XPackField.java

@@ -65,6 +65,8 @@ public final class XPackField {
     public static final String DATA_STREAMS = "data_streams";
     /** Name constant for the data tiers feature. */
     public static final String DATA_TIERS = "data_tiers";
+    /** Name constant for the aggregate_metric plugin. */
+    public static final String AGGREGATE_METRIC = "aggregate_metric";
 
     private XPackField() {}
 

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/action/XPackInfoFeatureAction.java

@@ -46,13 +46,15 @@ public class XPackInfoFeatureAction extends ActionType<XPackInfoFeatureResponse>
     public static final XPackInfoFeatureAction SEARCHABLE_SNAPSHOTS = new XPackInfoFeatureAction(XPackField.SEARCHABLE_SNAPSHOTS);
     public static final XPackInfoFeatureAction DATA_STREAMS = new XPackInfoFeatureAction(XPackField.DATA_STREAMS);
     public static final XPackInfoFeatureAction DATA_TIERS = new XPackInfoFeatureAction(XPackField.DATA_TIERS);
+    public static final XPackInfoFeatureAction AGGREGATE_METRIC = new XPackInfoFeatureAction(XPackField.AGGREGATE_METRIC);
 
     public static final List<XPackInfoFeatureAction> ALL;
     static {
         final List<XPackInfoFeatureAction> actions = new ArrayList<>();
         actions.addAll(Arrays.asList(
             SECURITY, MONITORING, WATCHER, GRAPH, MACHINE_LEARNING, LOGSTASH, EQL, SQL, ROLLUP, INDEX_LIFECYCLE, SNAPSHOT_LIFECYCLE, CCR,
-            TRANSFORM, VECTORS, VOTING_ONLY, FROZEN_INDICES, SPATIAL, ANALYTICS, ENRICH, DATA_STREAMS, SEARCHABLE_SNAPSHOTS, DATA_TIERS
+            TRANSFORM, VECTORS, VOTING_ONLY, FROZEN_INDICES, SPATIAL, ANALYTICS, ENRICH, DATA_STREAMS, SEARCHABLE_SNAPSHOTS, DATA_TIERS,
+            AGGREGATE_METRIC
         ));
         ALL = Collections.unmodifiableList(actions);
     }

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/action/XPackUsageFeatureAction.java

@@ -46,13 +46,15 @@ public class XPackUsageFeatureAction extends ActionType<XPackUsageFeatureRespons
     public static final XPackUsageFeatureAction SEARCHABLE_SNAPSHOTS = new XPackUsageFeatureAction(XPackField.SEARCHABLE_SNAPSHOTS);
     public static final XPackUsageFeatureAction DATA_STREAMS = new XPackUsageFeatureAction(XPackField.DATA_STREAMS);
     public static final XPackUsageFeatureAction DATA_TIERS = new XPackUsageFeatureAction(XPackField.DATA_TIERS);
+    public static final XPackUsageFeatureAction AGGREGATE_METRIC = new XPackUsageFeatureAction(XPackField.AGGREGATE_METRIC);
 
     public static final List<XPackUsageFeatureAction> ALL;
     static {
         final List<XPackUsageFeatureAction> actions = new ArrayList<>();
         actions.addAll(Arrays.asList(
             SECURITY, MONITORING, WATCHER, GRAPH, MACHINE_LEARNING, LOGSTASH, EQL, SQL, ROLLUP, INDEX_LIFECYCLE, SNAPSHOT_LIFECYCLE, CCR,
-            TRANSFORM, VECTORS, VOTING_ONLY, FROZEN_INDICES, SPATIAL, ANALYTICS, DATA_STREAMS, SEARCHABLE_SNAPSHOTS, DATA_TIERS
+            TRANSFORM, VECTORS, VOTING_ONLY, FROZEN_INDICES, SPATIAL, ANALYTICS, DATA_STREAMS, SEARCHABLE_SNAPSHOTS, DATA_TIERS,
+            AGGREGATE_METRIC
         ));
         ALL = Collections.unmodifiableList(actions);
     }

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/aggregatemetric/AggregateMetricFeatureSetUsage.java

@@ -0,0 +1,48 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.xpack.core.aggregatemetric;
+
+import org.elasticsearch.Version;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.xpack.core.XPackFeatureSet;
+import org.elasticsearch.xpack.core.XPackField;
+
+import java.io.IOException;
+import java.util.Objects;
+
+public class AggregateMetricFeatureSetUsage extends XPackFeatureSet.Usage {
+
+    public AggregateMetricFeatureSetUsage(StreamInput input) throws IOException {
+        super(input);
+    }
+
+    public AggregateMetricFeatureSetUsage(boolean available, boolean enabled) {
+        super(XPackField.AGGREGATE_METRIC, available, enabled);
+    }
+
+    @Override public Version getMinimalSupportedVersion() {
+        return Version.V_8_0_0;
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (obj == null) {
+            return false;
+        }
+        if (getClass() != obj.getClass()) {
+            return false;
+        }
+        AggregateMetricFeatureSetUsage other = (AggregateMetricFeatureSetUsage) obj;
+        return Objects.equals(available, other.available) &&
+            Objects.equals(enabled, other.enabled);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(available, enabled);
+    }
+}

x-pack/plugin/mapper-aggregate-metric/build.gradle

@@ -0,0 +1,24 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+evaluationDependsOn(xpackModule('core'))
+
+apply plugin: 'elasticsearch.esplugin'
+
+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.'
+  classname 'org.elasticsearch.xpack.aggregatemetric.AggregateMetricMapperPlugin'
+  extendedPlugins = ['x-pack-core']
+}
+archivesBaseName = 'x-pack-aggregate-metric'
+
+dependencies {
+  compileOnly project(":server")
+
+  compileOnly project(path: xpackModule('core'), configuration: 'default')
+  testImplementation project(path: xpackModule('core'), configuration: 'testArtifacts')
+}

x-pack/plugin/mapper-aggregate-metric/src/main/java/org/elasticsearch/xpack/aggregatemetric/AggregateMetricInfoTransportAction.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.xpack.aggregatemetric;
+
+import org.elasticsearch.action.support.ActionFilters;
+import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.license.XPackLicenseState;
+import org.elasticsearch.transport.TransportService;
+import org.elasticsearch.xpack.core.XPackField;
+import org.elasticsearch.xpack.core.action.XPackInfoFeatureAction;
+import org.elasticsearch.xpack.core.action.XPackInfoFeatureTransportAction;
+
+public class AggregateMetricInfoTransportAction extends XPackInfoFeatureTransportAction {
+
+    @Inject
+    public AggregateMetricInfoTransportAction(
+        TransportService transportService,
+        ActionFilters actionFilters,
+        Settings settings,
+        XPackLicenseState licenseState
+    ) {
+        super(XPackInfoFeatureAction.AGGREGATE_METRIC.name(), transportService, actionFilters);
+    }
+
+    @Override
+    public String name() {
+        return XPackField.AGGREGATE_METRIC;
+    }
+
+    @Override
+    public boolean available() {
+        return true;
+    }
+
+    @Override
+    public boolean enabled() {
+        return true;
+    }
+
+}

x-pack/plugin/mapper-aggregate-metric/src/main/java/org/elasticsearch/xpack/aggregatemetric/AggregateMetricMapperPlugin.java

@@ -0,0 +1,54 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.xpack.aggregatemetric;
+
+import org.elasticsearch.action.ActionRequest;
+import org.elasticsearch.action.ActionResponse;
+import org.elasticsearch.index.mapper.Mapper;
+import org.elasticsearch.plugins.ActionPlugin;
+import org.elasticsearch.plugins.MapperPlugin;
+import org.elasticsearch.plugins.Plugin;
+import org.elasticsearch.plugins.SearchPlugin;
+import org.elasticsearch.search.aggregations.support.ValuesSourceRegistry;
+import org.elasticsearch.xpack.aggregatemetric.aggregations.metrics.AggregateMetricsAggregatorsRegistrar;
+import org.elasticsearch.xpack.aggregatemetric.mapper.AggregateDoubleMetricFieldMapper;
+import org.elasticsearch.xpack.core.action.XPackInfoFeatureAction;
+import org.elasticsearch.xpack.core.action.XPackUsageFeatureAction;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Consumer;
+
+import static java.util.Collections.singletonMap;
+
+public class AggregateMetricMapperPlugin extends Plugin implements MapperPlugin, ActionPlugin, SearchPlugin {
+
+    @Override
+    public Map<String, Mapper.TypeParser> getMappers() {
+        return singletonMap(AggregateDoubleMetricFieldMapper.CONTENT_TYPE, AggregateDoubleMetricFieldMapper.PARSER);
+    }
+
+    @Override
+    public List<ActionHandler<? extends ActionRequest, ? extends ActionResponse>> getActions() {
+        return Arrays.asList(
+            new ActionHandler<>(XPackUsageFeatureAction.AGGREGATE_METRIC, AggregateMetricUsageTransportAction.class),
+            new ActionHandler<>(XPackInfoFeatureAction.AGGREGATE_METRIC, AggregateMetricInfoTransportAction.class)
+        );
+    }
+
+    @Override
+    public List<Consumer<ValuesSourceRegistry.Builder>> getAggregationExtentions() {
+        return List.of(
+            AggregateMetricsAggregatorsRegistrar::registerSumAggregator,
+            AggregateMetricsAggregatorsRegistrar::registerAvgAggregator,
+            AggregateMetricsAggregatorsRegistrar::registerMinAggregator,
+            AggregateMetricsAggregatorsRegistrar::registerMaxAggregator,
+            AggregateMetricsAggregatorsRegistrar::registerValueCountAggregator
+        );
+    }
+}

x-pack/plugin/mapper-aggregate-metric/src/main/java/org/elasticsearch/xpack/aggregatemetric/AggregateMetricUsageTransportAction.java

@@ -0,0 +1,58 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.xpack.aggregatemetric;
+
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.support.ActionFilters;
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
+import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.license.XPackLicenseState;
+import org.elasticsearch.protocol.xpack.XPackUsageRequest;
+import org.elasticsearch.tasks.Task;
+import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.transport.TransportService;
+import org.elasticsearch.xpack.core.action.XPackUsageFeatureAction;
+import org.elasticsearch.xpack.core.action.XPackUsageFeatureResponse;
+import org.elasticsearch.xpack.core.action.XPackUsageFeatureTransportAction;
+import org.elasticsearch.xpack.core.aggregatemetric.AggregateMetricFeatureSetUsage;
+
+public class AggregateMetricUsageTransportAction extends XPackUsageFeatureTransportAction {
+
+    @Inject
+    public AggregateMetricUsageTransportAction(
+        TransportService transportService,
+        ClusterService clusterService,
+        ThreadPool threadPool,
+        ActionFilters actionFilters,
+        IndexNameExpressionResolver indexNameExpressionResolver,
+        Settings settings,
+        XPackLicenseState licenseState
+    ) {
+        super(
+            XPackUsageFeatureAction.AGGREGATE_METRIC.name(),
+            transportService,
+            clusterService,
+            threadPool,
+            actionFilters,
+            indexNameExpressionResolver
+        );
+    }
+
+    @Override
+    protected void masterOperation(
+        Task task,
+        XPackUsageRequest request,
+        ClusterState state,
+        ActionListener<XPackUsageFeatureResponse> listener
+    ) {
+        AggregateMetricFeatureSetUsage usage = new AggregateMetricFeatureSetUsage(true, true);
+        listener.onResponse(new XPackUsageFeatureResponse(usage));
+    }
+}

x-pack/plugin/mapper-aggregate-metric/src/main/java/org/elasticsearch/xpack/aggregatemetric/aggregations/metrics/AggregateMetricBackedAvgAggregator.java

@@ -0,0 +1,138 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.xpack.aggregatemetric.aggregations.metrics;
+
+import org.apache.lucene.index.LeafReaderContext;
+import org.apache.lucene.search.ScoreMode;
+import org.elasticsearch.common.lease.Releasables;
+import org.elasticsearch.common.util.BigArrays;
+import org.elasticsearch.common.util.DoubleArray;
+import org.elasticsearch.common.util.LongArray;
+import org.elasticsearch.index.fielddata.SortedNumericDoubleValues;
+import org.elasticsearch.search.DocValueFormat;
+import org.elasticsearch.search.aggregations.Aggregator;
+import org.elasticsearch.search.aggregations.InternalAggregation;
+import org.elasticsearch.search.aggregations.LeafBucketCollector;
+import org.elasticsearch.search.aggregations.LeafBucketCollectorBase;
+import org.elasticsearch.search.aggregations.metrics.CompensatedSum;
+import org.elasticsearch.search.aggregations.metrics.InternalAvg;
+import org.elasticsearch.search.aggregations.metrics.NumericMetricsAggregator;
+import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
+import org.elasticsearch.search.internal.SearchContext;
+import org.elasticsearch.xpack.aggregatemetric.aggregations.support.AggregateMetricsValuesSource;
+import org.elasticsearch.xpack.aggregatemetric.mapper.AggregateDoubleMetricFieldMapper.Metric;
+
+import java.io.IOException;
+import java.util.Map;
+
+class AggregateMetricBackedAvgAggregator extends NumericMetricsAggregator.SingleValue {
+
+    final AggregateMetricsValuesSource.AggregateDoubleMetric valuesSource;
+
+    LongArray counts;
+    DoubleArray sums;
+    DoubleArray compensations;
+    DocValueFormat format;
+
+    AggregateMetricBackedAvgAggregator(
+        String name,
+        ValuesSourceConfig valuesSourceConfig,
+        SearchContext context,
+        Aggregator parent,
+        Map<String, Object> metadata
+    ) throws IOException {
+        super(name, context, parent, metadata);
+        // TODO: stop expecting nulls here
+        this.valuesSource = valuesSourceConfig.hasValues()
+            ? (AggregateMetricsValuesSource.AggregateDoubleMetric) valuesSourceConfig.getValuesSource()
+            : null;
+        if (valuesSource != null) {
+            final BigArrays bigArrays = context.bigArrays();
+            counts = bigArrays.newLongArray(1, true);
+            sums = bigArrays.newDoubleArray(1, true);
+            compensations = bigArrays.newDoubleArray(1, true);
+        }
+        this.format = valuesSourceConfig.format();
+    }
+
+    @Override
+    public ScoreMode scoreMode() {
+        return valuesSource != null && valuesSource.needsScores() ? ScoreMode.COMPLETE : ScoreMode.COMPLETE_NO_SCORES;
+    }
+
+    @Override
+    public LeafBucketCollector getLeafCollector(LeafReaderContext ctx, final LeafBucketCollector sub) throws IOException {
+        if (valuesSource == null) {
+            return LeafBucketCollector.NO_OP_COLLECTOR;
+        }
+        final BigArrays bigArrays = bigArrays();
+        // Retrieve aggregate values for metrics sum and value_count
+        final SortedNumericDoubleValues aggregateSums = valuesSource.getAggregateMetricValues(ctx, Metric.sum);
+        final SortedNumericDoubleValues aggregateValueCounts = valuesSource.getAggregateMetricValues(ctx, Metric.value_count);
+        final CompensatedSum kahanSummation = new CompensatedSum(0, 0);
+        return new LeafBucketCollectorBase(sub, sums) {
+            @Override
+            public void collect(int doc, long bucket) throws IOException {
+                sums = bigArrays.grow(sums, bucket + 1);
+                compensations = bigArrays.grow(compensations, bucket + 1);
+
+                // Read aggregate values for sums
+                if (aggregateSums.advanceExact(doc)) {
+                    // Compute the sum of double values with Kahan summation algorithm which is more
+                    // accurate than naive summation.
+                    double sum = sums.get(bucket);
+                    double compensation = compensations.get(bucket);
+
+                    kahanSummation.reset(sum, compensation);
+                    for (int i = 0; i < aggregateSums.docValueCount(); i++) {
+                        double value = aggregateSums.nextValue();
+                        kahanSummation.add(value);
+                    }
+
+                    sums.set(bucket, kahanSummation.value());
+                    compensations.set(bucket, kahanSummation.delta());
+                }
+
+                counts = bigArrays.grow(counts, bucket + 1);
+                // Read aggregate values for value_count
+                if (aggregateValueCounts.advanceExact(doc)) {
+                    for (int i = 0; i < aggregateValueCounts.docValueCount(); i++) {
+                        double d = aggregateValueCounts.nextValue();
+                        long value = Double.valueOf(d).longValue();
+                        counts.increment(bucket, value);
+                    }
+                }
+            }
+        };
+    }
+
+    @Override
+    public double metric(long owningBucketOrd) {
+        if (valuesSource == null || owningBucketOrd >= sums.size()) {
+            return Double.NaN;
+        }
+        return sums.get(owningBucketOrd) / counts.get(owningBucketOrd);
+    }
+
+    @Override
+    public InternalAggregation buildAggregation(long bucket) {
+        if (valuesSource == null || bucket >= sums.size()) {
+            return buildEmptyAggregation();
+        }
+        return new InternalAvg(name, sums.get(bucket), counts.get(bucket), format, metadata());
+    }
+
+    @Override
+    public InternalAggregation buildEmptyAggregation() {
+        return new InternalAvg(name, 0.0, 0L, format, metadata());
+    }
+
+    @Override
+    public void doClose() {
+        Releasables.close(counts, sums, compensations);
+    }
+
+}

x-pack/plugin/mapper-aggregate-metric/src/main/java/org/elasticsearch/xpack/aggregatemetric/aggregations/metrics/AggregateMetricBackedMaxAggregator.java

@@ -0,0 +1,117 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.xpack.aggregatemetric.aggregations.metrics;
+
+import org.apache.lucene.index.LeafReaderContext;
+import org.apache.lucene.search.CollectionTerminatedException;
+import org.apache.lucene.search.ScoreMode;
+import org.elasticsearch.common.lease.Releasables;
+import org.elasticsearch.common.util.BigArrays;
+import org.elasticsearch.common.util.DoubleArray;
+import org.elasticsearch.index.fielddata.NumericDoubleValues;
+import org.elasticsearch.index.fielddata.SortedNumericDoubleValues;
+import org.elasticsearch.search.DocValueFormat;
+import org.elasticsearch.search.MultiValueMode;
+import org.elasticsearch.search.aggregations.Aggregator;
+import org.elasticsearch.search.aggregations.InternalAggregation;
+import org.elasticsearch.search.aggregations.LeafBucketCollector;
+import org.elasticsearch.search.aggregations.LeafBucketCollectorBase;
+import org.elasticsearch.search.aggregations.metrics.InternalMax;
+import org.elasticsearch.search.aggregations.metrics.NumericMetricsAggregator;
+import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
+import org.elasticsearch.search.internal.SearchContext;
+import org.elasticsearch.xpack.aggregatemetric.aggregations.support.AggregateMetricsValuesSource;
+import org.elasticsearch.xpack.aggregatemetric.mapper.AggregateDoubleMetricFieldMapper.Metric;
+
+import java.io.IOException;
+import java.util.Map;
+
+class AggregateMetricBackedMaxAggregator extends NumericMetricsAggregator.SingleValue {
+
+    private final AggregateMetricsValuesSource.AggregateDoubleMetric valuesSource;
+    final DocValueFormat formatter;
+    DoubleArray maxes;
+
+    AggregateMetricBackedMaxAggregator(
+        String name,
+        ValuesSourceConfig config,
+        SearchContext context,
+        Aggregator parent,
+        Map<String, Object> metadata
+    ) throws IOException {
+        super(name, context, parent, metadata);
+        this.valuesSource = config.hasValues() ? (AggregateMetricsValuesSource.AggregateDoubleMetric) config.getValuesSource() : null;
+        if (valuesSource != null) {
+            maxes = context.bigArrays().newDoubleArray(1, false);
+            maxes.fill(0, maxes.size(), Double.NEGATIVE_INFINITY);
+        }
+        this.formatter = config.format();
+    }
+
+    @Override
+    public ScoreMode scoreMode() {
+        return valuesSource != null && valuesSource.needsScores() ? ScoreMode.COMPLETE : ScoreMode.COMPLETE_NO_SCORES;
+    }
+
+    @Override
+    public LeafBucketCollector getLeafCollector(LeafReaderContext ctx, final LeafBucketCollector sub) throws IOException {
+        if (valuesSource == null) {
+            if (parent != null) {
+                return LeafBucketCollector.NO_OP_COLLECTOR;
+            } else {
+                // we have no parent and the values source is empty so we can skip collecting hits.
+                throw new CollectionTerminatedException();
+            }
+        }
+
+        final BigArrays bigArrays = bigArrays();
+        final SortedNumericDoubleValues allValues = valuesSource.getAggregateMetricValues(ctx, Metric.max);
+        final NumericDoubleValues values = MultiValueMode.MAX.select(allValues);
+        return new LeafBucketCollectorBase(sub, allValues) {
+
+            @Override
+            public void collect(int doc, long bucket) throws IOException {
+                if (bucket >= maxes.size()) {
+                    long from = maxes.size();
+                    maxes = bigArrays.grow(maxes, bucket + 1);
+                    maxes.fill(from, maxes.size(), Double.NEGATIVE_INFINITY);
+                }
+                if (values.advanceExact(doc)) {
+                    final double value = values.doubleValue();
+                    double max = maxes.get(bucket);
+                    max = Math.max(max, value);
+                    maxes.set(bucket, max);
+                }
+            }
+        };
+    }
+
+    @Override
+    public double metric(long owningBucketOrd) {
+        if (valuesSource == null || owningBucketOrd >= maxes.size()) {
+            return Double.NEGATIVE_INFINITY;
+        }
+        return maxes.get(owningBucketOrd);
+    }
+
+    @Override
+    public InternalAggregation buildAggregation(long bucket) {
+        if (valuesSource == null || bucket >= maxes.size()) {
+            return buildEmptyAggregation();
+        }
+        return new InternalMax(name, maxes.get(bucket), formatter, metadata());
+    }
+
+    @Override
+    public InternalAggregation buildEmptyAggregation() {
+        return new InternalMax(name, Double.NEGATIVE_INFINITY, formatter, metadata());
+    }
+
+    @Override
+    public void doClose() {
+        Releasables.close(maxes);
+    }
+}

x-pack/plugin/mapper-aggregate-metric/src/main/java/org/elasticsearch/xpack/aggregatemetric/aggregations/metrics/AggregateMetricBackedMinAggregator.java

@@ -0,0 +1,117 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.xpack.aggregatemetric.aggregations.metrics;
+
+import org.apache.lucene.index.LeafReaderContext;
+import org.apache.lucene.search.CollectionTerminatedException;
+import org.apache.lucene.search.ScoreMode;
+import org.elasticsearch.common.lease.Releasables;
+import org.elasticsearch.common.util.BigArrays;
+import org.elasticsearch.common.util.DoubleArray;
+import org.elasticsearch.index.fielddata.NumericDoubleValues;
+import org.elasticsearch.index.fielddata.SortedNumericDoubleValues;
+import org.elasticsearch.search.DocValueFormat;
+import org.elasticsearch.search.MultiValueMode;
+import org.elasticsearch.search.aggregations.Aggregator;
+import org.elasticsearch.search.aggregations.InternalAggregation;
+import org.elasticsearch.search.aggregations.LeafBucketCollector;
+import org.elasticsearch.search.aggregations.LeafBucketCollectorBase;
+import org.elasticsearch.search.aggregations.metrics.InternalMin;
+import org.elasticsearch.search.aggregations.metrics.NumericMetricsAggregator;
+import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
+import org.elasticsearch.search.internal.SearchContext;
+import org.elasticsearch.xpack.aggregatemetric.aggregations.support.AggregateMetricsValuesSource;
+import org.elasticsearch.xpack.aggregatemetric.mapper.AggregateDoubleMetricFieldMapper.Metric;
+
+import java.io.IOException;
+import java.util.Map;
+
+class AggregateMetricBackedMinAggregator extends NumericMetricsAggregator.SingleValue {
+
+    private final AggregateMetricsValuesSource.AggregateDoubleMetric valuesSource;
+    final DocValueFormat format;
+    DoubleArray mins;
+
+    AggregateMetricBackedMinAggregator(
+        String name,
+        ValuesSourceConfig config,
+        SearchContext context,
+        Aggregator parent,
+        Map<String, Object> metadata
+    ) throws IOException {
+        super(name, context, parent, metadata);
+        this.valuesSource = config.hasValues() ? (AggregateMetricsValuesSource.AggregateDoubleMetric) config.getValuesSource() : null;
+        if (valuesSource != null) {
+            mins = context.bigArrays().newDoubleArray(1, false);
+            mins.fill(0, mins.size(), Double.POSITIVE_INFINITY);
+        }
+        this.format = config.format();
+    }
+
+    @Override
+    public ScoreMode scoreMode() {
+        return valuesSource != null && valuesSource.needsScores() ? ScoreMode.COMPLETE : ScoreMode.COMPLETE_NO_SCORES;
+    }
+
+    @Override
+    public LeafBucketCollector getLeafCollector(LeafReaderContext ctx, final LeafBucketCollector sub) throws IOException {
+        if (valuesSource == null) {
+            if (parent == null) {
+                return LeafBucketCollector.NO_OP_COLLECTOR;
+            } else {
+                // we have no parent and the values source is empty so we can skip collecting hits.
+                throw new CollectionTerminatedException();
+            }
+        }
+
+        final BigArrays bigArrays = bigArrays();
+        final SortedNumericDoubleValues allValues = valuesSource.getAggregateMetricValues(ctx, Metric.min);
+        final NumericDoubleValues values = MultiValueMode.MIN.select(allValues);
+        return new LeafBucketCollectorBase(sub, allValues) {
+
+            @Override
+            public void collect(int doc, long bucket) throws IOException {
+                if (bucket >= mins.size()) {
+                    long from = mins.size();
+                    mins = bigArrays.grow(mins, bucket + 1);
+                    mins.fill(from, mins.size(), Double.POSITIVE_INFINITY);
+                }
+                if (values.advanceExact(doc)) {
+                    final double value = values.doubleValue();
+                    double min = mins.get(bucket);
+                    min = Math.min(min, value);
+                    mins.set(bucket, min);
+                }
+            }
+        };
+    }
+
+    @Override
+    public double metric(long owningBucketOrd) {
+        if (valuesSource == null || owningBucketOrd >= mins.size()) {
+            return Double.POSITIVE_INFINITY;
+        }
+        return mins.get(owningBucketOrd);
+    }
+
+    @Override
+    public InternalAggregation buildAggregation(long bucket) {
+        if (valuesSource == null || bucket >= mins.size()) {
+            return buildEmptyAggregation();
+        }
+        return new InternalMin(name, mins.get(bucket), format, metadata());
+    }
+
+    @Override
+    public InternalAggregation buildEmptyAggregation() {
+        return new InternalMin(name, Double.POSITIVE_INFINITY, format, metadata());
+    }
+
+    @Override
+    public void doClose() {
+        Releasables.close(mins);
+    }
+}

x-pack/plugin/mapper-aggregate-metric/src/main/java/org/elasticsearch/xpack/aggregatemetric/aggregations/metrics/AggregateMetricBackedSumAggregator.java

@@ -0,0 +1,120 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.xpack.aggregatemetric.aggregations.metrics;
+
+import org.apache.lucene.index.LeafReaderContext;
+import org.apache.lucene.search.ScoreMode;
+import org.elasticsearch.common.lease.Releasables;
+import org.elasticsearch.common.util.BigArrays;
+import org.elasticsearch.common.util.DoubleArray;
+import org.elasticsearch.index.fielddata.SortedNumericDoubleValues;
+import org.elasticsearch.search.DocValueFormat;
+import org.elasticsearch.search.aggregations.Aggregator;
+import org.elasticsearch.search.aggregations.InternalAggregation;
+import org.elasticsearch.search.aggregations.LeafBucketCollector;
+import org.elasticsearch.search.aggregations.LeafBucketCollectorBase;
+import org.elasticsearch.search.aggregations.metrics.CompensatedSum;
+import org.elasticsearch.search.aggregations.metrics.InternalSum;
+import org.elasticsearch.search.aggregations.metrics.NumericMetricsAggregator;
+import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
+import org.elasticsearch.search.internal.SearchContext;
+import org.elasticsearch.xpack.aggregatemetric.aggregations.support.AggregateMetricsValuesSource;
+import org.elasticsearch.xpack.aggregatemetric.mapper.AggregateDoubleMetricFieldMapper.Metric;
+
+import java.io.IOException;
+import java.util.Map;
+
+class AggregateMetricBackedSumAggregator extends NumericMetricsAggregator.SingleValue {
+
+    private final AggregateMetricsValuesSource.AggregateDoubleMetric valuesSource;
+    private final DocValueFormat format;
+
+    private DoubleArray sums;
+    private DoubleArray compensations;
+
+    AggregateMetricBackedSumAggregator(
+        String name,
+        ValuesSourceConfig valuesSourceConfig,
+        SearchContext context,
+        Aggregator parent,
+        Map<String, Object> metadata
+    ) throws IOException {
+        super(name, context, parent, metadata);
+        // TODO: stop expecting nulls here
+        this.valuesSource = valuesSourceConfig.hasValues()
+            ? (AggregateMetricsValuesSource.AggregateDoubleMetric) valuesSourceConfig.getValuesSource()
+            : null;
+        if (valuesSource != null) {
+            sums = context.bigArrays().newDoubleArray(1, true);
+            compensations = context.bigArrays().newDoubleArray(1, true);
+        }
+        this.format = valuesSourceConfig.format();
+    }
+
+    @Override
+    public ScoreMode scoreMode() {
+        return valuesSource != null && valuesSource.needsScores() ? ScoreMode.COMPLETE : ScoreMode.COMPLETE_NO_SCORES;
+    }
+
+    @Override
+    public LeafBucketCollector getLeafCollector(LeafReaderContext ctx, final LeafBucketCollector sub) throws IOException {
+        if (valuesSource == null) {
+            return LeafBucketCollector.NO_OP_COLLECTOR;
+        }
+        final BigArrays bigArrays = bigArrays();
+        final SortedNumericDoubleValues values = valuesSource.getAggregateMetricValues(ctx, Metric.sum);
+        final CompensatedSum kahanSummation = new CompensatedSum(0, 0);
+        return new LeafBucketCollectorBase(sub, values) {
+            @Override
+            public void collect(int doc, long bucket) throws IOException {
+                sums = bigArrays.grow(sums, bucket + 1);
+                compensations = bigArrays.grow(compensations, bucket + 1);
+
+                if (values.advanceExact(doc)) {
+                    final int valuesCount = values.docValueCount(); // For aggregate metric this should always equal to 1
+                    // Compute the sum of double values with Kahan summation algorithm which is more
+                    // accurate than naive summation.
+                    double sum = sums.get(bucket);
+                    double compensation = compensations.get(bucket);
+                    kahanSummation.reset(sum, compensation);
+
+                    for (int i = 0; i < valuesCount; i++) {
+                        double value = values.nextValue();
+                        kahanSummation.add(value);
+                    }
+                    compensations.set(bucket, kahanSummation.delta());
+                    sums.set(bucket, kahanSummation.value());
+                }
+            }
+        };
+    }
+
+    @Override
+    public double metric(long owningBucketOrd) {
+        if (valuesSource == null || owningBucketOrd >= sums.size()) {
+            return 0.0;
+        }
+        return sums.get(owningBucketOrd);
+    }
+
+    @Override
+    public InternalAggregation buildAggregation(long bucket) {
+        if (valuesSource == null || bucket >= sums.size()) {
+            return buildEmptyAggregation();
+        }
+        return new InternalSum(name, sums.get(bucket), format, metadata());
+    }
+
+    @Override
+    public InternalAggregation buildEmptyAggregation() {
+        return new InternalSum(name, 0.0, format, metadata());
+    }
+
+    @Override
+    public void doClose() {
+        Releasables.close(sums, compensations);
+    }
+}

x-pack/plugin/mapper-aggregate-metric/src/main/java/org/elasticsearch/xpack/aggregatemetric/aggregations/metrics/AggregateMetricBackedValueCountAggregator.java

@@ -0,0 +1,104 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.xpack.aggregatemetric.aggregations.metrics;
+
+import org.apache.lucene.index.LeafReaderContext;
+import org.elasticsearch.common.lease.Releasables;
+import org.elasticsearch.common.util.BigArrays;
+import org.elasticsearch.common.util.LongArray;
+import org.elasticsearch.index.fielddata.SortedNumericDoubleValues;
+import org.elasticsearch.search.aggregations.Aggregator;
+import org.elasticsearch.search.aggregations.InternalAggregation;
+import org.elasticsearch.search.aggregations.LeafBucketCollector;
+import org.elasticsearch.search.aggregations.LeafBucketCollectorBase;
+import org.elasticsearch.search.aggregations.metrics.InternalValueCount;
+import org.elasticsearch.search.aggregations.metrics.NumericMetricsAggregator;
+import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
+import org.elasticsearch.search.internal.SearchContext;
+import org.elasticsearch.xpack.aggregatemetric.aggregations.support.AggregateMetricsValuesSource;
+import org.elasticsearch.xpack.aggregatemetric.mapper.AggregateDoubleMetricFieldMapper;
+
+import java.io.IOException;
+import java.util.Map;
+
+/**
+ * A field data based aggregator that adds all values in the value_count metric sub-field from an aggregate_metric field.
+ * This aggregator works in a multi-bucket mode, that is, when serves as a sub-aggregator, a single aggregator instance
+ * aggregates the counts for all buckets owned by the parent aggregator)
+ */
+class AggregateMetricBackedValueCountAggregator extends NumericMetricsAggregator.SingleValue {
+
+    private final AggregateMetricsValuesSource.AggregateDoubleMetric valuesSource;
+
+    // a count per bucket
+    LongArray counts;
+
+    AggregateMetricBackedValueCountAggregator(
+        String name,
+        ValuesSourceConfig valuesSourceConfig,
+        SearchContext aggregationContext,
+        Aggregator parent,
+        Map<String, Object> metadata
+    ) throws IOException {
+        super(name, aggregationContext, parent, metadata);
+        // TODO: stop expecting nulls here
+        this.valuesSource = valuesSourceConfig.hasValues()
+            ? (AggregateMetricsValuesSource.AggregateDoubleMetric) valuesSourceConfig.getValuesSource()
+            : null;
+        if (valuesSource != null) {
+            counts = bigArrays().newLongArray(1, true);
+        }
+    }
+
+    @Override
+    public LeafBucketCollector getLeafCollector(LeafReaderContext ctx, final LeafBucketCollector sub) throws IOException {
+        if (valuesSource == null) {
+            return LeafBucketCollector.NO_OP_COLLECTOR;
+        }
+        final BigArrays bigArrays = bigArrays();
+        final SortedNumericDoubleValues values = valuesSource.getAggregateMetricValues(
+            ctx,
+            AggregateDoubleMetricFieldMapper.Metric.value_count
+        );
+
+        return new LeafBucketCollectorBase(sub, values) {
+            @Override
+            public void collect(int doc, long bucket) throws IOException {
+                counts = bigArrays.grow(counts, bucket + 1);
+                if (values.advanceExact(doc)) {
+                    for (int i = 0; i < values.docValueCount(); i++) { // For aggregate metric this should always equal to 1
+                        long value = Double.valueOf(values.nextValue()).longValue();
+                        counts.increment(bucket, value);
+                    }
+                }
+            }
+        };
+    }
+
+    @Override
+    public double metric(long owningBucketOrd) {
+        return (valuesSource == null || owningBucketOrd >= counts.size()) ? 0 : counts.get(owningBucketOrd);
+    }
+
+    @Override
+    public InternalAggregation buildAggregation(long bucket) {
+        if (valuesSource == null || bucket >= counts.size()) {
+            return buildEmptyAggregation();
+        }
+        return new InternalValueCount(name, counts.get(bucket), metadata());
+    }
+
+    @Override
+    public InternalAggregation buildEmptyAggregation() {
+        return new InternalValueCount(name, 0L, metadata());
+    }
+
+    @Override
+    public void doClose() {
+        Releasables.close(counts);
+    }
+
+}

x-pack/plugin/mapper-aggregate-metric/src/main/java/org/elasticsearch/xpack/aggregatemetric/aggregations/metrics/AggregateMetricsAggregatorsRegistrar.java

@@ -0,0 +1,65 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.xpack.aggregatemetric.aggregations.metrics;
+
+import org.elasticsearch.search.aggregations.metrics.AvgAggregationBuilder;
+import org.elasticsearch.search.aggregations.metrics.MaxAggregationBuilder;
+import org.elasticsearch.search.aggregations.metrics.MinAggregationBuilder;
+import org.elasticsearch.search.aggregations.metrics.SumAggregationBuilder;
+import org.elasticsearch.search.aggregations.metrics.ValueCountAggregationBuilder;
+import org.elasticsearch.search.aggregations.support.ValuesSourceRegistry;
+import org.elasticsearch.xpack.aggregatemetric.aggregations.support.AggregateMetricsValuesSourceType;
+
+/**
+ * Utility class providing static methods to register aggregators for the aggregate_metric values source
+ */
+public class AggregateMetricsAggregatorsRegistrar {
+
+    public static void registerSumAggregator(ValuesSourceRegistry.Builder builder) {
+        builder.register(
+            SumAggregationBuilder.REGISTRY_KEY,
+            AggregateMetricsValuesSourceType.AGGREGATE_METRIC,
+            AggregateMetricBackedSumAggregator::new,
+            true
+        );
+    }
+
+    public static void registerAvgAggregator(ValuesSourceRegistry.Builder builder) {
+        builder.register(
+            AvgAggregationBuilder.REGISTRY_KEY,
+            AggregateMetricsValuesSourceType.AGGREGATE_METRIC,
+            AggregateMetricBackedAvgAggregator::new,
+            true
+        );
+    }
+
+    public static void registerMinAggregator(ValuesSourceRegistry.Builder builder) {
+        builder.register(
+            MinAggregationBuilder.REGISTRY_KEY,
+            AggregateMetricsValuesSourceType.AGGREGATE_METRIC,
+            AggregateMetricBackedMinAggregator::new,
+            true
+        );
+    }
+
+    public static void registerMaxAggregator(ValuesSourceRegistry.Builder builder) {
+        builder.register(
+            MaxAggregationBuilder.REGISTRY_KEY,
+            AggregateMetricsValuesSourceType.AGGREGATE_METRIC,
+            AggregateMetricBackedMaxAggregator::new,
+            true
+        );
+    }
+
+    public static void registerValueCountAggregator(ValuesSourceRegistry.Builder builder) {
+        builder.register(
+            ValueCountAggregationBuilder.REGISTRY_KEY,
+            AggregateMetricsValuesSourceType.AGGREGATE_METRIC,
+            AggregateMetricBackedValueCountAggregator::new,
+            true
+        );
+    }
+}

x-pack/plugin/mapper-aggregate-metric/src/main/java/org/elasticsearch/xpack/aggregatemetric/aggregations/support/AggregateMetricsValuesSource.java

@@ -0,0 +1,60 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.xpack.aggregatemetric.aggregations.support;
+
+import org.apache.lucene.index.LeafReaderContext;
+import org.elasticsearch.common.Rounding;
+import org.elasticsearch.index.fielddata.DocValueBits;
+import org.elasticsearch.index.fielddata.SortedBinaryDocValues;
+import org.elasticsearch.index.fielddata.SortedNumericDoubleValues;
+import org.elasticsearch.search.aggregations.AggregationExecutionException;
+import org.elasticsearch.xpack.aggregatemetric.fielddata.IndexAggregateDoubleMetricFieldData;
+import org.elasticsearch.xpack.aggregatemetric.mapper.AggregateDoubleMetricFieldMapper;
+import org.elasticsearch.xpack.aggregatemetric.mapper.AggregateDoubleMetricFieldMapper.Metric;
+
+import java.io.IOException;
+import java.util.function.Function;
+
+public class AggregateMetricsValuesSource {
+    public abstract static class AggregateDoubleMetric extends org.elasticsearch.search.aggregations.support.ValuesSource {
+
+        public abstract SortedNumericDoubleValues getAggregateMetricValues(LeafReaderContext context, Metric metric) throws IOException;
+
+        public static class Fielddata extends AggregateDoubleMetric {
+
+            protected final IndexAggregateDoubleMetricFieldData indexFieldData;
+
+            public Fielddata(IndexAggregateDoubleMetricFieldData indexFieldData) {
+                this.indexFieldData = indexFieldData;
+            }
+
+            @Override
+            public SortedBinaryDocValues bytesValues(LeafReaderContext context) {
+                return indexFieldData.load(context).getBytesValues();
+            }
+
+            @Override
+            public DocValueBits docsWithValue(LeafReaderContext context) throws IOException {
+                SortedNumericDoubleValues values = getAggregateMetricValues(context, null);
+                return new DocValueBits() {
+                    @Override
+                    public boolean advanceExact(int doc) throws IOException {
+                        return values.advanceExact(doc);
+                    }
+                };
+            }
+
+            @Override
+            protected Function<Rounding, Rounding.Prepared> roundingPreparer() throws IOException {
+                throw new AggregationExecutionException("Can't round an [" + AggregateDoubleMetricFieldMapper.CONTENT_TYPE + "]");
+            }
+
+            public SortedNumericDoubleValues getAggregateMetricValues(LeafReaderContext context, Metric metric) throws IOException {
+                return indexFieldData.load(context).getAggregateMetricValues(metric);
+            }
+        }
+    }
+}

x-pack/plugin/mapper-aggregate-metric/src/main/java/org/elasticsearch/xpack/aggregatemetric/aggregations/support/AggregateMetricsValuesSourceType.java

@@ -0,0 +1,73 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.xpack.aggregatemetric.aggregations.support;
+
+import org.elasticsearch.index.fielddata.IndexFieldData;
+import org.elasticsearch.script.AggregationScript;
+import org.elasticsearch.search.DocValueFormat;
+import org.elasticsearch.search.aggregations.AggregationExecutionException;
+import org.elasticsearch.search.aggregations.support.AggregationContext;
+import org.elasticsearch.search.aggregations.support.FieldContext;
+import org.elasticsearch.search.aggregations.support.ValueType;
+import org.elasticsearch.search.aggregations.support.ValuesSource;
+import org.elasticsearch.search.aggregations.support.ValuesSourceType;
+import org.elasticsearch.xpack.aggregatemetric.fielddata.IndexAggregateDoubleMetricFieldData;
+
+import java.util.Locale;
+
+public enum AggregateMetricsValuesSourceType implements ValuesSourceType {
+
+    AGGREGATE_METRIC() {
+        @Override
+        public ValuesSource getEmpty() {
+            throw new IllegalArgumentException("Can't deal with unmapped AggregateMetricsValuesSource type " + this.value());
+        }
+
+        @Override
+        public ValuesSource getScript(AggregationScript.LeafFactory script, ValueType scriptValueType) {
+            throw new AggregationExecutionException("Value source of type [" + this.value() + "] is not supported by scripts");
+        }
+
+        @Override
+        public ValuesSource getField(FieldContext fieldContext, AggregationScript.LeafFactory script, AggregationContext context) {
+            final IndexFieldData<?> indexFieldData = fieldContext.indexFieldData();
+
+            if ((indexFieldData instanceof IndexAggregateDoubleMetricFieldData) == false) {
+                throw new IllegalArgumentException(
+                    "Expected aggregate_metric_double type on field ["
+                        + fieldContext.field()
+                        + "], but got ["
+                        + fieldContext.fieldType().typeName()
+                        + "]"
+                );
+            }
+            return new AggregateMetricsValuesSource.AggregateDoubleMetric.Fielddata((IndexAggregateDoubleMetricFieldData) indexFieldData);
+        }
+
+        @Override
+        public ValuesSource replaceMissing(
+            ValuesSource valuesSource,
+            Object rawMissing,
+            DocValueFormat docValueFormat,
+            AggregationContext context
+        ) {
+            throw new IllegalArgumentException("Can't apply missing values on a " + valuesSource.getClass());
+        }
+    };
+
+    @Override
+    public String typeName() {
+        return value();
+    }
+
+    public static ValuesSourceType fromString(String name) {
+        return valueOf(name.trim().toUpperCase(Locale.ROOT));
+    }
+
+    public String value() {
+        return name().toLowerCase(Locale.ROOT);
+    }
+}

x-pack/plugin/mapper-aggregate-metric/src/main/java/org/elasticsearch/xpack/aggregatemetric/fielddata/IndexAggregateDoubleMetricFieldData.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.xpack.aggregatemetric.fielddata;
+
+import org.elasticsearch.index.fielddata.IndexFieldData;
+import org.elasticsearch.search.aggregations.support.ValuesSourceType;
+
+/**
+ * Specialization of {@link IndexFieldData} for aggregate_metric.
+ */
+public abstract class IndexAggregateDoubleMetricFieldData implements IndexFieldData<LeafAggregateDoubleMetricFieldData> {
+
+    protected final String fieldName;
+    protected final ValuesSourceType valuesSourceType;
+
+    public IndexAggregateDoubleMetricFieldData(String fieldName, ValuesSourceType valuesSourceType) {
+        this.fieldName = fieldName;
+        this.valuesSourceType = valuesSourceType;
+    }
+
+    @Override
+    public final String getFieldName() {
+        return fieldName;
+    }
+
+    @Override
+    public ValuesSourceType getValuesSourceType() {
+        return valuesSourceType;
+    }
+}

x-pack/plugin/mapper-aggregate-metric/src/main/java/org/elasticsearch/xpack/aggregatemetric/fielddata/LeafAggregateDoubleMetricFieldData.java

@@ -0,0 +1,22 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.xpack.aggregatemetric.fielddata;
+
+import org.elasticsearch.index.fielddata.LeafFieldData;
+import org.elasticsearch.index.fielddata.SortedNumericDoubleValues;
+import org.elasticsearch.xpack.aggregatemetric.mapper.AggregateDoubleMetricFieldMapper.Metric;
+
+/**
+ * {@link LeafFieldData} specialization for aggregate_double_metric data.
+ */
+public interface LeafAggregateDoubleMetricFieldData extends LeafFieldData {
+
+    /**
+     * Return aggregate_metric of double values for a given metric
+     */
+    SortedNumericDoubleValues getAggregateMetricValues(Metric metric);
+
+}

x-pack/plugin/mapper-aggregate-metric/src/main/java/org/elasticsearch/xpack/aggregatemetric/mapper/AggregateDoubleMetricFieldMapper.java

@@ -0,0 +1,597 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.xpack.aggregatemetric.mapper;
+
+import org.apache.lucene.index.DocValues;
+import org.apache.lucene.index.IndexReader;
+import org.apache.lucene.index.IndexableField;
+import org.apache.lucene.index.LeafReaderContext;
+import org.apache.lucene.index.SortedNumericDocValues;
+import org.apache.lucene.search.Query;
+import org.apache.lucene.search.SortField;
+import org.apache.lucene.search.SortedNumericSortField;
+import org.apache.lucene.util.NumericUtils;
+import org.elasticsearch.common.time.DateMathParser;
+import org.elasticsearch.common.util.BigArrays;
+import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.common.xcontent.XContentSubParser;
+import org.elasticsearch.index.fielddata.IndexFieldData;
+import org.elasticsearch.index.fielddata.ScriptDocValues;
+import org.elasticsearch.index.fielddata.SortedBinaryDocValues;
+import org.elasticsearch.index.fielddata.SortedNumericDoubleValues;
+import org.elasticsearch.index.mapper.ContentPath;
+import org.elasticsearch.index.mapper.FieldMapper;
+import org.elasticsearch.index.mapper.MappedFieldType;
+import org.elasticsearch.index.mapper.Mapper;
+import org.elasticsearch.index.mapper.NumberFieldMapper;
+import org.elasticsearch.index.mapper.ParseContext;
+import org.elasticsearch.index.mapper.SimpleMappedFieldType;
+import org.elasticsearch.index.mapper.SourceValueFetcher;
+import org.elasticsearch.index.mapper.TextSearchInfo;
+import org.elasticsearch.index.mapper.ValueFetcher;
+import org.elasticsearch.index.query.QueryRewriteContext;
+import org.elasticsearch.index.query.QueryShardContext;
+import org.elasticsearch.search.DocValueFormat;
+import org.elasticsearch.search.MultiValueMode;
+import org.elasticsearch.search.lookup.SearchLookup;
+import org.elasticsearch.search.sort.BucketedSort;
+import org.elasticsearch.search.sort.SortOrder;
+import org.elasticsearch.xpack.aggregatemetric.aggregations.support.AggregateMetricsValuesSourceType;
+import org.elasticsearch.xpack.aggregatemetric.fielddata.IndexAggregateDoubleMetricFieldData;
+import org.elasticsearch.xpack.aggregatemetric.fielddata.LeafAggregateDoubleMetricFieldData;
+
+import java.io.IOException;
+import java.time.ZoneId;
+import java.util.Collections;
+import java.util.EnumMap;
+import java.util.EnumSet;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+
+import static org.elasticsearch.common.xcontent.XContentParserUtils.ensureExpectedToken;
+
+/** A {@link FieldMapper} for a field containing aggregate metrics such as min/max/value_count etc. */
+public class AggregateDoubleMetricFieldMapper extends FieldMapper {
+
+    public static final String CONTENT_TYPE = "aggregate_metric_double";
+    public static final String SUBFIELD_SEPARATOR = ".";
+
+    private static AggregateDoubleMetricFieldMapper toType(FieldMapper in) {
+        return (AggregateDoubleMetricFieldMapper) in;
+    }
+
+    /**
+     * Return the name of a subfield of an aggregate metric field
+     *
+     * @param fieldName the name of the aggregate metric field
+     * @param metric    the metric type the subfield corresponds to
+     * @return the name of the subfield
+     */
+    public static String subfieldName(String fieldName, Metric metric) {
+        return fieldName + AggregateDoubleMetricFieldMapper.SUBFIELD_SEPARATOR + metric.name();
+    }
+
+    /**
+     * Mapping field names
+     */
+    public static class Names {
+        public static final String IGNORE_MALFORMED = "ignore_malformed";
+        public static final String METRICS = "metrics";
+        public static final String DEFAULT_METRIC = "default_metric";
+    }
+
+    /**
+     * Enum of aggregate metrics supported by this field mapper
+     */
+    public enum Metric {
+        min,
+        max,
+        sum,
+        value_count
+    }
+
+    public static class Defaults {
+        public static final Set<Metric> METRICS = Collections.emptySet();
+    }
+
+    public static class Builder extends FieldMapper.Builder {
+
+        private final Parameter<Map<String, String>> meta = Parameter.metaParam();
+
+        private final Parameter<Boolean> ignoreMalformed;
+
+        private final Parameter<Set<Metric>> metrics = new Parameter<>(Names.METRICS, false, () -> Defaults.METRICS, (n, c, o) -> {
+            @SuppressWarnings("unchecked")
+            List<String> metricsList = (List<String>) o;
+            EnumSet<Metric> parsedMetrics = EnumSet.noneOf(Metric.class);
+            for (String s : metricsList) {
+                try {
+                    Metric m = Metric.valueOf(s);
+                    parsedMetrics.add(m);
+                } catch (IllegalArgumentException e) {
+                    throw new IllegalArgumentException("Metric [" + s + "] is not supported.", e);
+                }
+            }
+            return parsedMetrics;
+        }, m -> toType(m).metrics).setValidator(v -> {
+            if (v == null || v.isEmpty()) {
+                throw new IllegalArgumentException("Property [" + Names.METRICS + "] is required for field [" + name() + "].");
+            }
+        });
+
+        /**
+         * Set the default metric so that query operations are delegated to it.
+         */
+        private final Parameter<Metric> defaultMetric = new Parameter<>(Names.DEFAULT_METRIC, false, () -> null, (n, c, o) -> {
+            try {
+                return Metric.valueOf(o.toString());
+            } catch (IllegalArgumentException e) {
+                throw new IllegalArgumentException("Metric [" + o.toString() + "] is not supported.", e);
+            }
+        }, m -> toType(m).defaultMetric);
+
+        public Builder(String name, Boolean ignoreMalformedByDefault) {
+            super(name);
+            this.ignoreMalformed = Parameter.boolParam(
+                Names.IGNORE_MALFORMED,
+                true,
+                m -> toType(m).ignoreMalformed,
+                ignoreMalformedByDefault
+            );
+        }
+
+        @Override
+        protected List<Parameter<?>> getParameters() {
+            return List.of(ignoreMalformed, metrics, defaultMetric, meta);
+        }
+
+        @Override
+        public AggregateDoubleMetricFieldMapper build(ContentPath context) {
+            if (defaultMetric.isConfigured() == false) {
+                // If a single metric is contained, this should be the default
+                if (metrics.getValue().size() == 1) {
+                    Metric m = metrics.getValue().iterator().next();
+                    defaultMetric.setValue(m);
+                }
+
+                if (metrics.getValue().contains(defaultMetric.getValue()) == false) {
+                    throw new IllegalArgumentException("Property [" + Names.DEFAULT_METRIC + "] is required for field [" + name() + "].");
+                }
+            }
+
+            if (metrics.getValue().contains(defaultMetric.getValue()) == false) {
+                // The default_metric is not defined in the "metrics" field
+                throw new IllegalArgumentException(
+                    "Default metric [" + defaultMetric.getValue() + "] is not defined in the metrics of field [" + name() + "]."
+                );
+            }
+
+            EnumMap<Metric, NumberFieldMapper> metricMappers = new EnumMap<>(Metric.class);
+            // Instantiate one NumberFieldMapper instance for each metric
+            for (Metric m : this.metrics.getValue()) {
+                String fieldName = subfieldName(name, m);
+                NumberFieldMapper.Builder builder;
+
+                if (m == Metric.value_count) {
+                    // value_count metric can only be an integer and not a double
+                    builder = new NumberFieldMapper.Builder(fieldName, NumberFieldMapper.NumberType.INTEGER, false, false);
+                } else {
+                    builder = new NumberFieldMapper.Builder(fieldName, NumberFieldMapper.NumberType.DOUBLE, false, true);
+                }
+                NumberFieldMapper fieldMapper = builder.build(context);
+                metricMappers.put(m, fieldMapper);
+            }
+
+            EnumMap<Metric, NumberFieldMapper.NumberFieldType> metricFields = metricMappers.entrySet()
+                .stream()
+                .collect(
+                    Collectors.toMap(
+                        Map.Entry::getKey,
+                        e -> e.getValue().fieldType(),
+                        (l, r) -> { throw new IllegalArgumentException("Duplicate keys " + l + "and " + r + "."); },
+                        () -> new EnumMap<>(Metric.class)
+                    )
+                );
+
+            AggregateDoubleMetricFieldType metricFieldType = new AggregateDoubleMetricFieldType(buildFullName(context), meta.getValue());
+            metricFieldType.setMetricFields(metricFields);
+            metricFieldType.setDefaultMetric(defaultMetric.getValue());
+
+            return new AggregateDoubleMetricFieldMapper(name, metricFieldType, metricMappers, this);
+        }
+    }
+
+    public static final FieldMapper.TypeParser PARSER = new TypeParser(
+        (n, c) -> new Builder(n, IGNORE_MALFORMED_SETTING.get(c.getSettings()))
+    );
+
+    public static final class AggregateDoubleMetricFieldType extends SimpleMappedFieldType {
+
+        private EnumMap<Metric, NumberFieldMapper.NumberFieldType> metricFields;
+
+        private Metric defaultMetric;
+
+        public AggregateDoubleMetricFieldType(String name) {
+            this(name, Collections.emptyMap());
+        }
+
+        public AggregateDoubleMetricFieldType(String name, Map<String, String> meta) {
+            super(name, true, false, false, TextSearchInfo.SIMPLE_MATCH_WITHOUT_TERMS, meta);
+        }
+
+        /**
+         * Return a delegate field type for a given metric sub-field
+         * @return a field type
+         */
+        private NumberFieldMapper.NumberFieldType delegateFieldType(Metric metric) {
+            return metricFields.get(metric);
+        }
+
+        /**
+         * Return a delegate field type for the default metric sub-field
+         * @return a field type
+         */
+        private NumberFieldMapper.NumberFieldType delegateFieldType() {
+            return delegateFieldType(defaultMetric);
+        }
+
+        @Override
+        public String familyTypeName() {
+            return NumberFieldMapper.NumberType.DOUBLE.typeName();
+        }
+
+        @Override
+        public String typeName() {
+            return CONTENT_TYPE;
+        }
+
+        private void setMetricFields(EnumMap<Metric, NumberFieldMapper.NumberFieldType> metricFields) {
+            this.metricFields = metricFields;
+        }
+
+        public void addMetricField(Metric m, NumberFieldMapper.NumberFieldType subfield) {
+            if (metricFields == null) {
+                metricFields = new EnumMap<>(AggregateDoubleMetricFieldMapper.Metric.class);
+            }
+
+            if (name() == null) {
+                throw new IllegalArgumentException("Field of type [" + typeName() + "] must have a name before adding a subfield");
+            }
+            metricFields.put(m, subfield);
+        }
+
+        public void setDefaultMetric(Metric defaultMetric) {
+            this.defaultMetric = defaultMetric;
+        }
+
+        Metric getDefaultMetric() {
+            return defaultMetric;
+        }
+
+        @Override
+        public Query existsQuery(QueryShardContext context) {
+            return delegateFieldType().existsQuery(context);
+        }
+
+        @Override
+        public Query termQuery(Object value, QueryShardContext context) {
+            if (value == null) {
+                throw new IllegalArgumentException("Cannot search for null.");
+            }
+            return delegateFieldType().termQuery(value, context);
+        }
+
+        @Override
+        public Query termsQuery(List<?> values, QueryShardContext context) {
+            return delegateFieldType().termsQuery(values, context);
+        }
+
+        @Override
+        public Query rangeQuery(Object lowerTerm, Object upperTerm, boolean includeLower, boolean includeUpper, QueryShardContext context) {
+            return delegateFieldType().rangeQuery(lowerTerm, upperTerm, includeLower, includeUpper, context);
+        }
+
+        @Override
+        public Object valueForDisplay(Object value) {
+            return delegateFieldType().valueForDisplay(value);
+        }
+
+        @Override
+        public DocValueFormat docValueFormat(String format, ZoneId timeZone) {
+            return delegateFieldType().docValueFormat(format, timeZone);
+        }
+
+        @Override
+        public Relation isFieldWithinQuery(
+            IndexReader reader,
+            Object from,
+            Object to,
+            boolean includeLower,
+            boolean includeUpper,
+            ZoneId timeZone,
+            DateMathParser dateMathParser,
+            QueryRewriteContext context
+        ) throws IOException {
+            return delegateFieldType().isFieldWithinQuery(reader, from, to, includeLower, includeUpper, timeZone, dateMathParser, context);
+        }
+
+        @Override
+        public IndexFieldData.Builder fielddataBuilder(String fullyQualifiedIndexName, Supplier<SearchLookup> searchLookup) {
+            return (cache, breakerService) -> new IndexAggregateDoubleMetricFieldData(
+                name(),
+                AggregateMetricsValuesSourceType.AGGREGATE_METRIC
+            ) {
+                @Override
+                public LeafAggregateDoubleMetricFieldData load(LeafReaderContext context) {
+                    return new LeafAggregateDoubleMetricFieldData() {
+                        @Override
+                        public SortedNumericDoubleValues getAggregateMetricValues(final Metric metric) {
+                            try {
+                                final SortedNumericDocValues values = DocValues.getSortedNumeric(
+                                    context.reader(),
+                                    subfieldName(getFieldName(), metric)
+                                );
+
+                                return new SortedNumericDoubleValues() {
+                                    @Override
+                                    public int docValueCount() {
+                                        return values.docValueCount();
+                                    }
+
+                                    @Override
+                                    public boolean advanceExact(int doc) throws IOException {
+                                        return values.advanceExact(doc);
+                                    }
+
+                                    @Override
+                                    public double nextValue() throws IOException {
+                                        long v = values.nextValue();
+                                        if (metric == Metric.value_count) {
+                                            // Only value_count metrics are encoded as integers
+                                            return v;
+                                        } else {
+                                            // All other metrics are encoded as doubles
+                                            return NumericUtils.sortableLongToDouble(v);
+                                        }
+                                    }
+                                };
+                            } catch (IOException e) {
+                                throw new IllegalStateException("Cannot load doc values", e);
+                            }
+                        }
+
+                        @Override
+                        public ScriptDocValues<?> getScriptValues() {
+                            // getAggregateMetricValues returns all metric as doubles, including `value_count`
+                            return new ScriptDocValues.Doubles(getAggregateMetricValues(defaultMetric));
+                        }
+
+                        @Override
+                        public SortedBinaryDocValues getBytesValues() {
+                            throw new UnsupportedOperationException(
+                                "String representation of doc values " + "for [" + CONTENT_TYPE + "] fields is not supported"
+                            );
+                        }
+
+                        @Override
+                        public long ramBytesUsed() {
+                            return 0; // Unknown
+                        }
+
+                        @Override
+                        public void close() {}
+                    };
+                }
+
+                @Override
+                public LeafAggregateDoubleMetricFieldData loadDirect(LeafReaderContext context) {
+                    return load(context);
+                }
+
+                @Override
+                public SortField sortField(
+                    Object missingValue,
+                    MultiValueMode sortMode,
+                    XFieldComparatorSource.Nested nested,
+                    boolean reverse
+                ) {
+                    return new SortedNumericSortField(delegateFieldType().name(), SortField.Type.DOUBLE, reverse);
+                }
+
+                @Override
+                public BucketedSort newBucketedSort(
+                    BigArrays bigArrays,
+                    Object missingValue,
+                    MultiValueMode sortMode,
+                    XFieldComparatorSource.Nested nested,
+                    SortOrder sortOrder,
+                    DocValueFormat format,
+                    int bucketSize,
+                    BucketedSort.ExtraData extra
+                ) {
+                    throw new IllegalArgumentException("Can't sort on the [" + CONTENT_TYPE + "] field");
+                }
+            };
+        }
+
+        @Override
+        public ValueFetcher valueFetcher(QueryShardContext context, SearchLookup searchLookup, String format) {
+            if (format != null) {
+                throw new IllegalArgumentException("Field [" + name() + "] of type [" + typeName() + "] doesn't support formats.");
+            }
+
+            return new SourceValueFetcher(name(), context) {
+                @Override
+                @SuppressWarnings("unchecked")
+                protected Object parseSourceValue(Object value) {
+                    Map<String, Double> metrics = (Map<String, Double>) value;
+                    return metrics.get(defaultMetric.name());
+                }
+            };
+        }
+    }
+
+    private final EnumMap<Metric, NumberFieldMapper> metricFieldMappers;
+
+    private final boolean ignoreMalformed;
+
+    private final boolean ignoreMalformedByDefault;
+
+    /** A set of metrics supported */
+    private final Set<Metric> metrics;
+
+    /** The default metric to be when querying this field type */
+    protected Metric defaultMetric;
+
+    private AggregateDoubleMetricFieldMapper(
+        String simpleName,
+        MappedFieldType mappedFieldType,
+        EnumMap<Metric, NumberFieldMapper> metricFieldMappers,
+        Builder builder
+    ) {
+        super(simpleName, mappedFieldType, MultiFields.empty(), CopyTo.empty());
+        this.ignoreMalformed = builder.ignoreMalformed.getValue();
+        this.ignoreMalformedByDefault = builder.ignoreMalformed.getDefaultValue();
+        this.metrics = builder.metrics.getValue();
+        this.defaultMetric = builder.defaultMetric.getValue();
+        this.metricFieldMappers = metricFieldMappers;
+    }
+
+    boolean ignoreMalformed() {
+        return ignoreMalformed;
+    }
+
+    Metric defaultMetric() {
+        return defaultMetric;
+    }
+
+    @Override
+    public AggregateDoubleMetricFieldType fieldType() {
+        return (AggregateDoubleMetricFieldType) super.fieldType();
+    }
+
+    @Override
+    protected String contentType() {
+        return CONTENT_TYPE;
+    }
+
+    @Override
+    public Iterator<Mapper> iterator() {
+        return Collections.emptyIterator();
+    }
+
+    @Override
+    protected void parseCreateField(ParseContext context) throws IOException {
+        if (context.externalValueSet()) {
+            throw new IllegalArgumentException("Field [" + name() + "] of type [" + typeName() + "] can't be used in multi-fields");
+        }
+
+        context.path().add(simpleName());
+        XContentParser.Token token;
+        XContentSubParser subParser = null;
+        EnumSet<Metric> metricsParsed = EnumSet.noneOf(Metric.class);
+        try {
+            token = context.parser().currentToken();
+            if (token == XContentParser.Token.VALUE_NULL) {
+                context.path().remove();
+                return;
+            }
+            ensureExpectedToken(XContentParser.Token.START_OBJECT, token, context.parser());
+            subParser = new XContentSubParser(context.parser());
+            token = subParser.nextToken();
+            while (token != XContentParser.Token.END_OBJECT) {
+                // should be an object sub-field with name a metric name
+                ensureExpectedToken(XContentParser.Token.FIELD_NAME, token, subParser);
+                String fieldName = subParser.currentName();
+                Metric metric = Metric.valueOf(fieldName);
+
+                if (metrics.contains(metric) == false) {
+                    throw new IllegalArgumentException(
+                        "Aggregate metric [" + metric + "] does not exist in the mapping of field [" + mappedFieldType.name() + "]"
+                    );
+                }
+
+                token = subParser.nextToken();
+                // Make sure that the value is a number. Probably this will change when
+                // new aggregate metric types are added (histogram, cardinality etc)
+                ensureExpectedToken(XContentParser.Token.VALUE_NUMBER, token, subParser);
+                NumberFieldMapper delegateFieldMapper = metricFieldMappers.get(metric);
+                // We don't accept arrays of metrics
+                if (context.doc().getField(delegateFieldMapper.fieldType().name()) != null) {
+                    throw new IllegalArgumentException(
+                        "Field ["
+                            + name()
+                            + "] of type ["
+                            + typeName()
+                            + "] does not support indexing multiple values for the same field in the same document"
+                    );
+                }
+                // Delegate parsing the field to a numeric field mapper
+                delegateFieldMapper.parse(context);
+
+                // Ensure a value_count metric does not have a negative value
+                if (Metric.value_count == metric) {
+                    // context.doc().getField() method iterates over all fields in the document.
+                    // Making the following call slow down. Maybe we can think something smarter.
+                    Number n = context.doc().getField(delegateFieldMapper.name()).numericValue();
+                    if (n.intValue() < 0) {
+                        throw new IllegalArgumentException(
+                            "Aggregate metric [" + metric.name() + "] of field [" + mappedFieldType.name() + "] cannot be a negative number"
+                        );
+                    }
+                }
+                metricsParsed.add(metric);
+                token = subParser.nextToken();
+            }
+
+            // Check if all required metrics have been parsed.
+            if (metricsParsed.containsAll(metrics) == false) {
+                throw new IllegalArgumentException(
+                    "Aggregate metric field [" + mappedFieldType.name() + "] must contain all metrics " + metrics.toString()
+                );
+            }
+        } catch (Exception e) {
+            if (ignoreMalformed) {
+                if (subParser != null) {
+                    // close the subParser so we advance to the end of the object
+                    subParser.close();
+                }
+                // If ignoreMalformed == true, clear all parsed fields
+                Set<String> ignoreFieldNames = new HashSet<>(metricFieldMappers.size());
+                for (NumberFieldMapper m : metricFieldMappers.values()) {
+                    context.addIgnoredField(m.fieldType().name());
+                    ignoreFieldNames.add(m.fieldType().name());
+                }
+                // Parsing a metric sub-field is delegated to the delegate field mapper by calling method
+                // delegateFieldMapper.parse(context). Unfortunately, this method adds the parsed sub-field
+                // to the document automatically. So, at this point we must undo this by removing all metric
+                // sub-fields from the document. To do so, we iterate over the document fields and remove
+                // the ones whose names match.
+                for (Iterator<IndexableField> iter = context.doc().getFields().iterator(); iter.hasNext();) {
+                    IndexableField field = iter.next();
+                    if (ignoreFieldNames.contains(field.name())) {
+                        iter.remove();
+                    }
+                }
+            } else {
+                // Rethrow exception as is. It is going to be caught and nested in a MapperParsingException
+                // by its FieldMapper.MappedFieldType#parse()
+                throw e;
+            }
+        }
+        context.path().remove();
+    }
+
+    @Override
+    public FieldMapper.Builder getMergeBuilder() {
+        return new Builder(simpleName(), ignoreMalformedByDefault).init(this);
+    }
+}

x-pack/plugin/mapper-aggregate-metric/src/test/java/org/elasticsearch/xpack/aggregatemetric/aggregations/metrics/AggregateMetricBackedAvgAggregatorTests.java

@@ -0,0 +1,160 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.xpack.aggregatemetric.aggregations.metrics;
+
+import org.apache.lucene.document.Field;
+import org.apache.lucene.document.NumericDocValuesField;
+import org.apache.lucene.document.StringField;
+import org.apache.lucene.index.RandomIndexWriter;
+import org.apache.lucene.index.Term;
+import org.apache.lucene.search.MatchAllDocsQuery;
+import org.apache.lucene.search.Query;
+import org.apache.lucene.search.TermQuery;
+import org.elasticsearch.common.CheckedConsumer;
+import org.elasticsearch.index.mapper.MappedFieldType;
+import org.elasticsearch.index.mapper.NumberFieldMapper;
+import org.elasticsearch.plugins.SearchPlugin;
+import org.elasticsearch.search.aggregations.AggregationBuilder;
+import org.elasticsearch.search.aggregations.AggregatorTestCase;
+import org.elasticsearch.search.aggregations.metrics.AvgAggregationBuilder;
+import org.elasticsearch.search.aggregations.metrics.InternalAvg;
+import org.elasticsearch.search.aggregations.support.AggregationInspectionHelper;
+import org.elasticsearch.search.aggregations.support.CoreValuesSourceType;
+import org.elasticsearch.search.aggregations.support.ValuesSourceType;
+import org.elasticsearch.xpack.aggregatemetric.AggregateMetricMapperPlugin;
+import org.elasticsearch.xpack.aggregatemetric.aggregations.support.AggregateMetricsValuesSourceType;
+import org.elasticsearch.xpack.aggregatemetric.mapper.AggregateDoubleMetricFieldMapper.AggregateDoubleMetricFieldType;
+import org.elasticsearch.xpack.aggregatemetric.mapper.AggregateDoubleMetricFieldMapper.Metric;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.function.Consumer;
+
+import static java.util.Collections.singleton;
+import static org.elasticsearch.xpack.aggregatemetric.mapper.AggregateDoubleMetricFieldMapper.subfieldName;
+
+public class AggregateMetricBackedAvgAggregatorTests extends AggregatorTestCase {
+
+    private static final String FIELD_NAME = "aggregate_metric_field";
+
+    public void testMatchesNumericDocValues() throws IOException {
+        testCase(new MatchAllDocsQuery(), iw -> {
+            iw.addDocument(
+                List.of(
+                    new NumericDocValuesField(subfieldName(FIELD_NAME, Metric.sum), Double.doubleToLongBits(20)),
+                    new NumericDocValuesField(subfieldName(FIELD_NAME, Metric.value_count), 2)
+                )
+            );
+
+            iw.addDocument(
+                List.of(
+                    new NumericDocValuesField(subfieldName(FIELD_NAME, Metric.sum), Double.doubleToLongBits(50)),
+                    new NumericDocValuesField(subfieldName(FIELD_NAME, Metric.value_count), 5)
+                )
+            );
+        }, avg -> {
+            assertEquals(10, avg.getValue(), 0d);
+            assertTrue(AggregationInspectionHelper.hasValue(avg));
+        });
+    }
+
+    public void testNoDocs() throws IOException {
+        testCase(new MatchAllDocsQuery(), iw -> {
+            // Intentionally not writing any docs
+        }, avg -> {
+            assertEquals(Double.NaN, avg.getValue(), 0d);
+            assertFalse(AggregationInspectionHelper.hasValue(avg));
+        });
+    }
+
+    public void testNoMatchingField() throws IOException {
+        testCase(new MatchAllDocsQuery(), iw -> {
+            iw.addDocument(singleton(new NumericDocValuesField("wrong_number", 7)));
+            iw.addDocument(singleton(new NumericDocValuesField("wrong_number", 1)));
+        }, avg -> {
+            assertEquals(Double.NaN, avg.getValue(), 0d);
+            assertFalse(AggregationInspectionHelper.hasValue(avg));
+        });
+    }
+
+    public void testQueryFiltering() throws IOException {
+        testCase(new TermQuery(new Term("match", "yes")), iw -> {
+            iw.addDocument(
+                List.of(
+                    new StringField("match", "yes", Field.Store.NO),
+                    new NumericDocValuesField(subfieldName(FIELD_NAME, Metric.sum), Double.doubleToLongBits(10)),
+                    new NumericDocValuesField(subfieldName(FIELD_NAME, Metric.value_count), 1)
+                )
+            );
+            iw.addDocument(
+                List.of(
+                    new StringField("match", "yes", Field.Store.NO),
+                    new NumericDocValuesField(subfieldName(FIELD_NAME, Metric.sum), Double.doubleToLongBits(50)),
+                    new NumericDocValuesField(subfieldName(FIELD_NAME, Metric.value_count), 5)
+                )
+            );
+            iw.addDocument(
+                List.of(
+                    new StringField("match", "no", Field.Store.NO),
+                    new NumericDocValuesField(subfieldName(FIELD_NAME, Metric.sum), Double.doubleToLongBits(40)),
+                    new NumericDocValuesField(subfieldName(FIELD_NAME, Metric.value_count), 5)
+                )
+            );
+        }, avg -> {
+            assertEquals(10d, avg.getValue(), 0d);
+            assertTrue(AggregationInspectionHelper.hasValue(avg));
+        });
+    }
+
+    /**
+     * Create a default aggregate_metric_double field type containing sum and a value_count metrics.
+     *
+     * @param fieldName the name of the field
+     * @return the created field type
+     */
+    private AggregateDoubleMetricFieldType createDefaultFieldType(String fieldName) {
+        AggregateDoubleMetricFieldType fieldType = new AggregateDoubleMetricFieldType(fieldName);
+
+        for (Metric m : List.of(Metric.value_count, Metric.sum)) {
+            String subfieldName = subfieldName(fieldName, m);
+            NumberFieldMapper.NumberFieldType subfield = new NumberFieldMapper.NumberFieldType(
+                subfieldName,
+                NumberFieldMapper.NumberType.DOUBLE
+            );
+            fieldType.addMetricField(m, subfield);
+        }
+        fieldType.setDefaultMetric(Metric.sum);
+        return fieldType;
+    }
+
+    private void testCase(Query query, CheckedConsumer<RandomIndexWriter, IOException> buildIndex, Consumer<InternalAvg> verify)
+        throws IOException {
+        MappedFieldType fieldType = createDefaultFieldType(FIELD_NAME);
+        AggregationBuilder aggregationBuilder = createAggBuilderForTypeTest(fieldType, FIELD_NAME);
+        testCase(aggregationBuilder, query, buildIndex, verify, fieldType);
+    }
+
+    @Override
+    protected List<SearchPlugin> getSearchPlugins() {
+        return List.of(new AggregateMetricMapperPlugin());
+    }
+
+    @Override
+    protected AggregationBuilder createAggBuilderForTypeTest(MappedFieldType fieldType, String fieldName) {
+        return new AvgAggregationBuilder("avg_agg").field(fieldName);
+    }
+
+    @Override
+    protected List<ValuesSourceType> getSupportedValuesSourceTypes() {
+        return List.of(
+            CoreValuesSourceType.NUMERIC,
+            CoreValuesSourceType.DATE,
+            CoreValuesSourceType.BOOLEAN,
+            AggregateMetricsValuesSourceType.AGGREGATE_METRIC
+        );
+    }
+
+}

x-pack/plugin/mapper-aggregate-metric/src/test/java/org/elasticsearch/xpack/aggregatemetric/aggregations/metrics/AggregateMetricBackedMaxAggregatorTests.java

@@ -0,0 +1,160 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.xpack.aggregatemetric.aggregations.metrics;
+
+import org.apache.lucene.document.Field;
+import org.apache.lucene.document.NumericDocValuesField;
+import org.apache.lucene.document.StringField;
+import org.apache.lucene.index.RandomIndexWriter;
+import org.apache.lucene.index.Term;
+import org.apache.lucene.search.MatchAllDocsQuery;
+import org.apache.lucene.search.Query;
+import org.apache.lucene.search.TermQuery;
+import org.elasticsearch.common.CheckedConsumer;
+import org.elasticsearch.index.mapper.MappedFieldType;
+import org.elasticsearch.index.mapper.NumberFieldMapper;
+import org.elasticsearch.plugins.SearchPlugin;
+import org.elasticsearch.search.aggregations.AggregationBuilder;
+import org.elasticsearch.search.aggregations.AggregatorTestCase;
+import org.elasticsearch.search.aggregations.metrics.InternalMax;
+import org.elasticsearch.search.aggregations.metrics.MaxAggregationBuilder;
+import org.elasticsearch.search.aggregations.support.AggregationInspectionHelper;
+import org.elasticsearch.search.aggregations.support.CoreValuesSourceType;
+import org.elasticsearch.search.aggregations.support.ValuesSourceType;
+import org.elasticsearch.xpack.aggregatemetric.AggregateMetricMapperPlugin;
+import org.elasticsearch.xpack.aggregatemetric.aggregations.support.AggregateMetricsValuesSourceType;
+import org.elasticsearch.xpack.aggregatemetric.mapper.AggregateDoubleMetricFieldMapper.AggregateDoubleMetricFieldType;
+import org.elasticsearch.xpack.aggregatemetric.mapper.AggregateDoubleMetricFieldMapper.Metric;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.function.Consumer;
+
+import static java.util.Collections.singleton;
+import static org.elasticsearch.xpack.aggregatemetric.mapper.AggregateDoubleMetricFieldMapper.subfieldName;
+
+public class AggregateMetricBackedMaxAggregatorTests extends AggregatorTestCase {
+
+    private static final String FIELD_NAME = "aggregate_metric_field";
+
+    public void testMatchesNumericDocValues() throws IOException {
+        testCase(new MatchAllDocsQuery(), iw -> {
+            iw.addDocument(
+                List.of(
+                    new NumericDocValuesField(subfieldName(FIELD_NAME, Metric.max), Double.doubleToLongBits(10)),
+                    new NumericDocValuesField(subfieldName(FIELD_NAME, Metric.min), Double.doubleToLongBits(2))
+                )
+            );
+
+            iw.addDocument(
+                List.of(
+                    new NumericDocValuesField(subfieldName(FIELD_NAME, Metric.max), Double.doubleToLongBits(50)),
+                    new NumericDocValuesField(subfieldName(FIELD_NAME, Metric.min), Double.doubleToLongBits(5))
+                )
+            );
+        }, max -> {
+            assertEquals(50, max.getValue(), 0d);
+            assertTrue(AggregationInspectionHelper.hasValue(max));
+        });
+    }
+
+    public void testNoDocs() throws IOException {
+        testCase(new MatchAllDocsQuery(), iw -> {
+            // Intentionally not writing any docs
+        }, max -> {
+            assertEquals(Double.NEGATIVE_INFINITY, max.getValue(), 0d);
+            assertFalse(AggregationInspectionHelper.hasValue(max));
+        });
+    }
+
+    public void testNoMatchingField() throws IOException {
+        testCase(new MatchAllDocsQuery(), iw -> {
+            iw.addDocument(singleton(new NumericDocValuesField("wrong_number", 7)));
+            iw.addDocument(singleton(new NumericDocValuesField("wrong_number", 1)));
+        }, max -> {
+            assertEquals(Double.NEGATIVE_INFINITY, max.getValue(), 0d);
+            assertFalse(AggregationInspectionHelper.hasValue(max));
+        });
+    }
+
+    public void testQueryFiltering() throws IOException {
+        testCase(new TermQuery(new Term("match", "yes")), iw -> {
+            iw.addDocument(
+                List.of(
+                    new StringField("match", "yes", Field.Store.NO),
+                    new NumericDocValuesField(subfieldName(FIELD_NAME, Metric.max), Double.doubleToLongBits(10)),
+                    new NumericDocValuesField(subfieldName(FIELD_NAME, Metric.min), Double.doubleToLongBits(2))
+                )
+            );
+            iw.addDocument(
+                List.of(
+                    new StringField("match", "yes", Field.Store.NO),
+                    new NumericDocValuesField(subfieldName(FIELD_NAME, Metric.max), Double.doubleToLongBits(20)),
+                    new NumericDocValuesField(subfieldName(FIELD_NAME, Metric.min), Double.doubleToLongBits(5))
+                )
+            );
+            iw.addDocument(
+                List.of(
+                    new StringField("match", "no", Field.Store.NO),
+                    new NumericDocValuesField(subfieldName(FIELD_NAME, Metric.max), Double.doubleToLongBits(40)),
+                    new NumericDocValuesField(subfieldName(FIELD_NAME, Metric.min), Double.doubleToLongBits(1))
+                )
+            );
+        }, max -> {
+            assertEquals(20L, max.getValue(), 0d);
+            assertTrue(AggregationInspectionHelper.hasValue(max));
+        });
+    }
+
+    /**
+     * Create a default aggregate_metric_double field type containing min and a max metrics.
+     *
+     * @param fieldName the name of the field
+     * @return the created field type
+     */
+    private AggregateDoubleMetricFieldType createDefaultFieldType(String fieldName) {
+        AggregateDoubleMetricFieldType fieldType = new AggregateDoubleMetricFieldType(fieldName);
+
+        for (Metric m : List.of(Metric.min, Metric.max)) {
+            String subfieldName = subfieldName(fieldName, m);
+            NumberFieldMapper.NumberFieldType subfield = new NumberFieldMapper.NumberFieldType(
+                subfieldName,
+                NumberFieldMapper.NumberType.DOUBLE
+            );
+            fieldType.addMetricField(m, subfield);
+        }
+        fieldType.setDefaultMetric(Metric.min);
+        return fieldType;
+    }
+
+    private void testCase(Query query, CheckedConsumer<RandomIndexWriter, IOException> buildIndex, Consumer<InternalMax> verify)
+        throws IOException {
+        MappedFieldType fieldType = createDefaultFieldType(FIELD_NAME);
+        AggregationBuilder aggregationBuilder = createAggBuilderForTypeTest(fieldType, FIELD_NAME);
+        testCase(aggregationBuilder, query, buildIndex, verify, fieldType);
+    }
+
+    @Override
+    protected List<SearchPlugin> getSearchPlugins() {
+        return List.of(new AggregateMetricMapperPlugin());
+    }
+
+    @Override
+    protected AggregationBuilder createAggBuilderForTypeTest(MappedFieldType fieldType, String fieldName) {
+        return new MaxAggregationBuilder("max_agg").field(fieldName);
+    }
+
+    @Override
+    protected List<ValuesSourceType> getSupportedValuesSourceTypes() {
+        return List.of(
+            CoreValuesSourceType.NUMERIC,
+            CoreValuesSourceType.DATE,
+            CoreValuesSourceType.BOOLEAN,
+            AggregateMetricsValuesSourceType.AGGREGATE_METRIC
+        );
+    }
+
+}

x-pack/plugin/mapper-aggregate-metric/src/test/java/org/elasticsearch/xpack/aggregatemetric/aggregations/metrics/AggregateMetricBackedMinAggregatorTests.java

@@ -0,0 +1,160 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.xpack.aggregatemetric.aggregations.metrics;
+
+import org.apache.lucene.document.Field;
+import org.apache.lucene.document.NumericDocValuesField;
+import org.apache.lucene.document.StringField;
+import org.apache.lucene.index.RandomIndexWriter;
+import org.apache.lucene.index.Term;
+import org.apache.lucene.search.MatchAllDocsQuery;
+import org.apache.lucene.search.Query;
+import org.apache.lucene.search.TermQuery;
+import org.elasticsearch.common.CheckedConsumer;
+import org.elasticsearch.index.mapper.MappedFieldType;
+import org.elasticsearch.index.mapper.NumberFieldMapper;
+import org.elasticsearch.plugins.SearchPlugin;
+import org.elasticsearch.search.aggregations.AggregationBuilder;
+import org.elasticsearch.search.aggregations.AggregatorTestCase;
+import org.elasticsearch.search.aggregations.metrics.InternalMin;
+import org.elasticsearch.search.aggregations.metrics.MinAggregationBuilder;
+import org.elasticsearch.search.aggregations.support.AggregationInspectionHelper;
+import org.elasticsearch.search.aggregations.support.CoreValuesSourceType;
+import org.elasticsearch.search.aggregations.support.ValuesSourceType;
+import org.elasticsearch.xpack.aggregatemetric.AggregateMetricMapperPlugin;
+import org.elasticsearch.xpack.aggregatemetric.aggregations.support.AggregateMetricsValuesSourceType;
+import org.elasticsearch.xpack.aggregatemetric.mapper.AggregateDoubleMetricFieldMapper.AggregateDoubleMetricFieldType;
+import org.elasticsearch.xpack.aggregatemetric.mapper.AggregateDoubleMetricFieldMapper.Metric;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.function.Consumer;
+
+import static java.util.Collections.singleton;
+import static org.elasticsearch.xpack.aggregatemetric.mapper.AggregateDoubleMetricFieldMapper.subfieldName;
+
+public class AggregateMetricBackedMinAggregatorTests extends AggregatorTestCase {
+
+    private static final String FIELD_NAME = "aggregate_metric_field";
+
+    public void testMatchesNumericDocValues() throws IOException {
+        testCase(new MatchAllDocsQuery(), iw -> {
+            iw.addDocument(
+                List.of(
+                    new NumericDocValuesField(subfieldName(FIELD_NAME, Metric.max), Double.doubleToLongBits(10)),
+                    new NumericDocValuesField(subfieldName(FIELD_NAME, Metric.min), Double.doubleToLongBits(2))
+                )
+            );
+
+            iw.addDocument(
+                List.of(
+                    new NumericDocValuesField(subfieldName(FIELD_NAME, Metric.max), Double.doubleToLongBits(50)),
+                    new NumericDocValuesField(subfieldName(FIELD_NAME, Metric.min), Double.doubleToLongBits(5))
+                )
+            );
+        }, min -> {
+            assertEquals(2, min.getValue(), 0d);
+            assertTrue(AggregationInspectionHelper.hasValue(min));
+        });
+    }
+
+    public void testNoDocs() throws IOException {
+        testCase(new MatchAllDocsQuery(), iw -> {
+            // Intentionally not writing any docs
+        }, min -> {
+            assertEquals(Double.POSITIVE_INFINITY, min.getValue(), 0d);
+            assertFalse(AggregationInspectionHelper.hasValue(min));
+        });
+    }
+
+    public void testNoMatchingField() throws IOException {
+        testCase(new MatchAllDocsQuery(), iw -> {
+            iw.addDocument(singleton(new NumericDocValuesField("wrong_number", 7)));
+            iw.addDocument(singleton(new NumericDocValuesField("wrong_number", 1)));
+        }, min -> {
+            assertEquals(Double.POSITIVE_INFINITY, min.getValue(), 0d);
+            assertFalse(AggregationInspectionHelper.hasValue(min));
+        });
+    }
+
+    public void testQueryFiltering() throws IOException {
+        testCase(new TermQuery(new Term("match", "yes")), iw -> {
+            iw.addDocument(
+                List.of(
+                    new StringField("match", "yes", Field.Store.NO),
+                    new NumericDocValuesField(subfieldName(FIELD_NAME, Metric.max), Double.doubleToLongBits(10)),
+                    new NumericDocValuesField(subfieldName(FIELD_NAME, Metric.min), Double.doubleToLongBits(2))
+                )
+            );
+            iw.addDocument(
+                List.of(
+                    new StringField("match", "yes", Field.Store.NO),
+                    new NumericDocValuesField(subfieldName(FIELD_NAME, Metric.max), Double.doubleToLongBits(20)),
+                    new NumericDocValuesField(subfieldName(FIELD_NAME, Metric.min), Double.doubleToLongBits(5))
+                )
+            );
+            iw.addDocument(
+                List.of(
+                    new StringField("match", "no", Field.Store.NO),
+                    new NumericDocValuesField(subfieldName(FIELD_NAME, Metric.max), Double.doubleToLongBits(40)),
+                    new NumericDocValuesField(subfieldName(FIELD_NAME, Metric.min), Double.doubleToLongBits(1))
+                )
+            );
+        }, min -> {
+            assertEquals(2L, min.getValue(), 0d);
+            assertTrue(AggregationInspectionHelper.hasValue(min));
+        });
+    }
+
+    /**
+     * Create a default aggregate_metric_double field type containing min and a max metrics.
+     *
+     * @param fieldName the name of the field
+     * @return the created field type
+     */
+    private AggregateDoubleMetricFieldType createDefaultFieldType(String fieldName) {
+        AggregateDoubleMetricFieldType fieldType = new AggregateDoubleMetricFieldType(fieldName);
+
+        for (Metric m : List.of(Metric.min, Metric.max)) {
+            String subfieldName = subfieldName(fieldName, m);
+            NumberFieldMapper.NumberFieldType subfield = new NumberFieldMapper.NumberFieldType(
+                subfieldName,
+                NumberFieldMapper.NumberType.DOUBLE
+            );
+            fieldType.addMetricField(m, subfield);
+        }
+        fieldType.setDefaultMetric(Metric.min);
+        return fieldType;
+    }
+
+    private void testCase(Query query, CheckedConsumer<RandomIndexWriter, IOException> buildIndex, Consumer<InternalMin> verify)
+        throws IOException {
+        MappedFieldType fieldType = createDefaultFieldType(FIELD_NAME);
+        AggregationBuilder aggregationBuilder = createAggBuilderForTypeTest(fieldType, FIELD_NAME);
+        testCase(aggregationBuilder, query, buildIndex, verify, fieldType);
+    }
+
+    @Override
+    protected List<SearchPlugin> getSearchPlugins() {
+        return List.of(new AggregateMetricMapperPlugin());
+    }
+
+    @Override
+    protected AggregationBuilder createAggBuilderForTypeTest(MappedFieldType fieldType, String fieldName) {
+        return new MinAggregationBuilder("min_agg").field(fieldName);
+    }
+
+    @Override
+    protected List<ValuesSourceType> getSupportedValuesSourceTypes() {
+        return List.of(
+            CoreValuesSourceType.NUMERIC,
+            CoreValuesSourceType.DATE,
+            CoreValuesSourceType.BOOLEAN,
+            AggregateMetricsValuesSourceType.AGGREGATE_METRIC
+        );
+    }
+
+}

x-pack/plugin/mapper-aggregate-metric/src/test/java/org/elasticsearch/xpack/aggregatemetric/aggregations/metrics/AggregateMetricBackedSumAggregatorTests.java

@@ -0,0 +1,160 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.xpack.aggregatemetric.aggregations.metrics;
+
+import org.apache.lucene.document.Field;
+import org.apache.lucene.document.NumericDocValuesField;
+import org.apache.lucene.document.StringField;
+import org.apache.lucene.index.RandomIndexWriter;
+import org.apache.lucene.index.Term;
+import org.apache.lucene.search.MatchAllDocsQuery;
+import org.apache.lucene.search.Query;
+import org.apache.lucene.search.TermQuery;
+import org.elasticsearch.common.CheckedConsumer;
+import org.elasticsearch.index.mapper.MappedFieldType;
+import org.elasticsearch.index.mapper.NumberFieldMapper;
+import org.elasticsearch.plugins.SearchPlugin;
+import org.elasticsearch.search.aggregations.AggregationBuilder;
+import org.elasticsearch.search.aggregations.AggregatorTestCase;
+import org.elasticsearch.search.aggregations.metrics.InternalSum;
+import org.elasticsearch.search.aggregations.metrics.SumAggregationBuilder;
+import org.elasticsearch.search.aggregations.support.AggregationInspectionHelper;
+import org.elasticsearch.search.aggregations.support.CoreValuesSourceType;
+import org.elasticsearch.search.aggregations.support.ValuesSourceType;
+import org.elasticsearch.xpack.aggregatemetric.AggregateMetricMapperPlugin;
+import org.elasticsearch.xpack.aggregatemetric.aggregations.support.AggregateMetricsValuesSourceType;
+import org.elasticsearch.xpack.aggregatemetric.mapper.AggregateDoubleMetricFieldMapper.AggregateDoubleMetricFieldType;
+import org.elasticsearch.xpack.aggregatemetric.mapper.AggregateDoubleMetricFieldMapper.Metric;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.function.Consumer;
+
+import static java.util.Collections.singleton;
+import static org.elasticsearch.xpack.aggregatemetric.mapper.AggregateDoubleMetricFieldMapper.subfieldName;
+
+public class AggregateMetricBackedSumAggregatorTests extends AggregatorTestCase {
+
+    private static final String FIELD_NAME = "aggregate_metric_field";
+
+    public void testMatchesNumericDocValues() throws IOException {
+        testCase(new MatchAllDocsQuery(), iw -> {
+            iw.addDocument(
+                List.of(
+                    new NumericDocValuesField(subfieldName(FIELD_NAME, Metric.sum), Double.doubleToLongBits(10)),
+                    new NumericDocValuesField(subfieldName(FIELD_NAME, Metric.value_count), Double.doubleToLongBits(2))
+                )
+            );
+
+            iw.addDocument(
+                List.of(
+                    new NumericDocValuesField(subfieldName(FIELD_NAME, Metric.sum), Double.doubleToLongBits(50)),
+                    new NumericDocValuesField(subfieldName(FIELD_NAME, Metric.value_count), Double.doubleToLongBits(5))
+                )
+            );
+        }, sum -> {
+            assertEquals(60, sum.getValue(), 0d);
+            assertTrue(AggregationInspectionHelper.hasValue(sum));
+        });
+    }
+
+    public void testNoDocs() throws IOException {
+        testCase(new MatchAllDocsQuery(), iw -> {
+            // Intentionally not writing any docs
+        }, sum -> {
+            assertEquals(0L, sum.getValue(), 0d);
+            assertFalse(AggregationInspectionHelper.hasValue(sum));
+        });
+    }
+
+    public void testNoMatchingField() throws IOException {
+        testCase(new MatchAllDocsQuery(), iw -> {
+            iw.addDocument(singleton(new NumericDocValuesField("wrong_number", 7)));
+            iw.addDocument(singleton(new NumericDocValuesField("wrong_number", 1)));
+        }, sum -> {
+            assertEquals(0L, sum.getValue(), 0d);
+            assertFalse(AggregationInspectionHelper.hasValue(sum));
+        });
+    }
+
+    public void testQueryFiltering() throws IOException {
+        testCase(new TermQuery(new Term("match", "yes")), iw -> {
+            iw.addDocument(
+                List.of(
+                    new StringField("match", "yes", Field.Store.NO),
+                    new NumericDocValuesField(subfieldName(FIELD_NAME, Metric.sum), Double.doubleToLongBits(10)),
+                    new NumericDocValuesField(subfieldName(FIELD_NAME, Metric.value_count), Double.doubleToLongBits(2))
+                )
+            );
+            iw.addDocument(
+                List.of(
+                    new StringField("match", "yes", Field.Store.NO),
+                    new NumericDocValuesField(subfieldName(FIELD_NAME, Metric.sum), Double.doubleToLongBits(20)),
+                    new NumericDocValuesField(subfieldName(FIELD_NAME, Metric.value_count), Double.doubleToLongBits(5))
+                )
+            );
+            iw.addDocument(
+                List.of(
+                    new StringField("match", "no", Field.Store.NO),
+                    new NumericDocValuesField(subfieldName(FIELD_NAME, Metric.sum), Double.doubleToLongBits(40)),
+                    new NumericDocValuesField(subfieldName(FIELD_NAME, Metric.value_count), Double.doubleToLongBits(5))
+                )
+            );
+        }, sum -> {
+            assertEquals(30L, sum.getValue(), 0d);
+            assertTrue(AggregationInspectionHelper.hasValue(sum));
+        });
+    }
+
+    /**
+     * Create a default aggregate_metric_double field type containing sum and a value_count metrics.
+     *
+     * @param fieldName the name of the field
+     * @return the created field type
+     */
+    private AggregateDoubleMetricFieldType createDefaultFieldType(String fieldName) {
+        AggregateDoubleMetricFieldType fieldType = new AggregateDoubleMetricFieldType(fieldName);
+
+        for (Metric m : List.of(Metric.value_count, Metric.sum)) {
+            String subfieldName = subfieldName(fieldName, m);
+            NumberFieldMapper.NumberFieldType subfield = new NumberFieldMapper.NumberFieldType(
+                subfieldName,
+                NumberFieldMapper.NumberType.DOUBLE
+            );
+            fieldType.addMetricField(m, subfield);
+        }
+        fieldType.setDefaultMetric(Metric.sum);
+        return fieldType;
+    }
+
+    private void testCase(Query query, CheckedConsumer<RandomIndexWriter, IOException> buildIndex, Consumer<InternalSum> verify)
+        throws IOException {
+        MappedFieldType fieldType = createDefaultFieldType(FIELD_NAME);
+        AggregationBuilder aggregationBuilder = createAggBuilderForTypeTest(fieldType, FIELD_NAME);
+        testCase(aggregationBuilder, query, buildIndex, verify, fieldType);
+    }
+
+    @Override
+    protected List<SearchPlugin> getSearchPlugins() {
+        return List.of(new AggregateMetricMapperPlugin());
+    }
+
+    @Override
+    protected AggregationBuilder createAggBuilderForTypeTest(MappedFieldType fieldType, String fieldName) {
+        return new SumAggregationBuilder("sum_agg").field(fieldName);
+    }
+
+    @Override
+    protected List<ValuesSourceType> getSupportedValuesSourceTypes() {
+        return List.of(
+            CoreValuesSourceType.NUMERIC,
+            CoreValuesSourceType.DATE,
+            CoreValuesSourceType.BOOLEAN,
+            AggregateMetricsValuesSourceType.AGGREGATE_METRIC
+        );
+    }
+
+}

x-pack/plugin/mapper-aggregate-metric/src/test/java/org/elasticsearch/xpack/aggregatemetric/aggregations/metrics/AggregateMetricBackedValueCountAggregatorTests.java

@@ -0,0 +1,163 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.xpack.aggregatemetric.aggregations.metrics;
+
+import org.apache.lucene.document.Field;
+import org.apache.lucene.document.NumericDocValuesField;
+import org.apache.lucene.document.StringField;
+import org.apache.lucene.index.RandomIndexWriter;
+import org.apache.lucene.index.Term;
+import org.apache.lucene.search.MatchAllDocsQuery;
+import org.apache.lucene.search.Query;
+import org.apache.lucene.search.TermQuery;
+import org.elasticsearch.common.CheckedConsumer;
+import org.elasticsearch.index.mapper.MappedFieldType;
+import org.elasticsearch.index.mapper.NumberFieldMapper;
+import org.elasticsearch.plugins.SearchPlugin;
+import org.elasticsearch.search.aggregations.AggregationBuilder;
+import org.elasticsearch.search.aggregations.AggregatorTestCase;
+import org.elasticsearch.search.aggregations.metrics.InternalValueCount;
+import org.elasticsearch.search.aggregations.metrics.ValueCountAggregationBuilder;
+import org.elasticsearch.search.aggregations.support.AggregationInspectionHelper;
+import org.elasticsearch.search.aggregations.support.CoreValuesSourceType;
+import org.elasticsearch.search.aggregations.support.ValuesSourceType;
+import org.elasticsearch.xpack.aggregatemetric.AggregateMetricMapperPlugin;
+import org.elasticsearch.xpack.aggregatemetric.aggregations.support.AggregateMetricsValuesSourceType;
+import org.elasticsearch.xpack.aggregatemetric.mapper.AggregateDoubleMetricFieldMapper.AggregateDoubleMetricFieldType;
+import org.elasticsearch.xpack.aggregatemetric.mapper.AggregateDoubleMetricFieldMapper.Metric;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.function.Consumer;
+
+import static java.util.Collections.singleton;
+import static org.elasticsearch.xpack.aggregatemetric.mapper.AggregateDoubleMetricFieldMapper.subfieldName;
+
+public class AggregateMetricBackedValueCountAggregatorTests extends AggregatorTestCase {
+
+    private static final String FIELD_NAME = "aggregate_metric_field";
+
+    public void testMatchesNumericDocValues() throws IOException {
+        testCase(new MatchAllDocsQuery(), iw -> {
+            iw.addDocument(
+                List.of(
+                    new NumericDocValuesField(subfieldName(FIELD_NAME, Metric.sum), Double.doubleToLongBits(10)),
+                    new NumericDocValuesField(subfieldName(FIELD_NAME, Metric.value_count), 2)
+                )
+            );
+            iw.addDocument(
+                List.of(
+                    new NumericDocValuesField(subfieldName(FIELD_NAME, Metric.sum), Double.doubleToLongBits(50)),
+                    new NumericDocValuesField(subfieldName(FIELD_NAME, Metric.value_count), 5)
+                )
+            );
+        }, valueCount -> {
+            assertEquals(7, valueCount.getValue(), 0d);
+            assertTrue(AggregationInspectionHelper.hasValue(valueCount));
+        });
+    }
+
+    public void testNoDocs() throws IOException {
+        testCase(new MatchAllDocsQuery(), iw -> {
+            // Intentionally not writing any docs
+        }, valueCount -> {
+            assertEquals(0L, valueCount.getValue(), 0d);
+            assertFalse(AggregationInspectionHelper.hasValue(valueCount));
+        });
+    }
+
+    public void testNoMatchingField() throws IOException {
+        testCase(new MatchAllDocsQuery(), iw -> {
+            iw.addDocument(singleton(new NumericDocValuesField("wrong_number", 7)));
+            iw.addDocument(singleton(new NumericDocValuesField("wrong_number", 1)));
+        }, sum -> {
+            assertEquals(0L, sum.getValue(), 0d);
+            assertFalse(AggregationInspectionHelper.hasValue(sum));
+        });
+    }
+
+    public void testQueryFiltering() throws IOException {
+        testCase(new TermQuery(new Term("match", "yes")), iw -> {
+            iw.addDocument(
+                List.of(
+                    new StringField("match", "yes", Field.Store.NO),
+                    new NumericDocValuesField(subfieldName(FIELD_NAME, Metric.sum), Double.doubleToLongBits(10)),
+                    new NumericDocValuesField(subfieldName(FIELD_NAME, Metric.value_count), 2)
+                )
+            );
+            iw.addDocument(
+                List.of(
+                    new StringField("match", "yes", Field.Store.NO),
+                    new NumericDocValuesField(subfieldName(FIELD_NAME, Metric.sum), Double.doubleToLongBits(20)),
+                    new NumericDocValuesField(subfieldName(FIELD_NAME, Metric.value_count), 5)
+                )
+            );
+            iw.addDocument(
+                List.of(
+                    new StringField("match", "no", Field.Store.NO),
+                    new NumericDocValuesField(subfieldName(FIELD_NAME, Metric.sum), Double.doubleToLongBits(40)),
+                    new NumericDocValuesField(subfieldName(FIELD_NAME, Metric.value_count), 5)
+                )
+            );
+        }, sum -> {
+            assertEquals(7L, sum.getValue(), 0d);
+            assertTrue(AggregationInspectionHelper.hasValue(sum));
+        });
+    }
+
+    /**
+     * Create a default aggregate_metric_double field type containing sum and a value_count metrics.
+     *
+     * @param fieldName the name of the field
+     * @return the created field type
+     */
+    private AggregateDoubleMetricFieldType createDefaultFieldType(String fieldName) {
+        AggregateDoubleMetricFieldType fieldType = new AggregateDoubleMetricFieldType(fieldName);
+
+        for (Metric m : List.of(Metric.value_count, Metric.sum)) {
+            String subfieldName = subfieldName(fieldName, m);
+            NumberFieldMapper.NumberFieldType subfield = new NumberFieldMapper.NumberFieldType(
+                subfieldName,
+                NumberFieldMapper.NumberType.DOUBLE
+            );
+            fieldType.addMetricField(m, subfield);
+        }
+        fieldType.setDefaultMetric(Metric.sum);
+        return fieldType;
+    }
+
+    private void testCase(Query query, CheckedConsumer<RandomIndexWriter, IOException> buildIndex, Consumer<InternalValueCount> verify)
+        throws IOException {
+        MappedFieldType fieldType = createDefaultFieldType(FIELD_NAME);
+        AggregationBuilder aggregationBuilder = createAggBuilderForTypeTest(fieldType, FIELD_NAME);
+        testCase(aggregationBuilder, query, buildIndex, verify, fieldType);
+    }
+
+    @Override
+    protected List<SearchPlugin> getSearchPlugins() {
+        return List.of(new AggregateMetricMapperPlugin());
+    }
+
+    @Override
+    protected AggregationBuilder createAggBuilderForTypeTest(MappedFieldType fieldType, String fieldName) {
+        return new ValueCountAggregationBuilder("value_count_agg").field(fieldName);
+    }
+
+    @Override
+    protected List<ValuesSourceType> getSupportedValuesSourceTypes() {
+        return List.of(
+            CoreValuesSourceType.NUMERIC,
+            CoreValuesSourceType.BYTES,
+            CoreValuesSourceType.GEOPOINT,
+            CoreValuesSourceType.RANGE,
+            CoreValuesSourceType.BOOLEAN,
+            CoreValuesSourceType.DATE,
+            CoreValuesSourceType.IP,
+            AggregateMetricsValuesSourceType.AGGREGATE_METRIC
+        );
+    }
+
+}

x-pack/plugin/mapper-aggregate-metric/src/test/java/org/elasticsearch/xpack/aggregatemetric/mapper/AggregateDoubleMetricFieldMapperTests.java

@@ -0,0 +1,530 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.xpack.aggregatemetric.mapper;
+
+import org.apache.lucene.search.DocValuesFieldExistsQuery;
+import org.apache.lucene.search.Query;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.XContentFactory;
+import org.elasticsearch.index.mapper.DocumentMapper;
+import org.elasticsearch.index.mapper.MappedFieldType;
+import org.elasticsearch.index.mapper.Mapper;
+import org.elasticsearch.index.mapper.MapperParsingException;
+import org.elasticsearch.index.mapper.MapperService;
+import org.elasticsearch.index.mapper.MapperTestCase;
+import org.elasticsearch.index.mapper.ParseContext;
+import org.elasticsearch.index.mapper.ParsedDocument;
+import org.elasticsearch.plugins.Plugin;
+import org.elasticsearch.xpack.aggregatemetric.AggregateMetricMapperPlugin;
+import org.hamcrest.Matchers;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+import static org.elasticsearch.xpack.aggregatemetric.mapper.AggregateDoubleMetricFieldMapper.Names.IGNORE_MALFORMED;
+import static org.elasticsearch.xpack.aggregatemetric.mapper.AggregateDoubleMetricFieldMapper.Names.METRICS;
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.notNullValue;
+import static org.hamcrest.Matchers.nullValue;
+import static org.hamcrest.core.IsInstanceOf.instanceOf;
+
+public class AggregateDoubleMetricFieldMapperTests extends MapperTestCase {
+
+    public static final String METRICS_FIELD = METRICS;
+    public static final String IGNORE_MALFORMED_FIELD = IGNORE_MALFORMED;
+    public static final String CONTENT_TYPE = AggregateDoubleMetricFieldMapper.CONTENT_TYPE;
+    public static final String DEFAULT_METRIC = AggregateDoubleMetricFieldMapper.Names.DEFAULT_METRIC;
+
+    @Override
+    protected Collection<? extends Plugin> getPlugins() {
+        return List.of(new AggregateMetricMapperPlugin());
+    }
+
+    @Override
+    protected void minimalMapping(XContentBuilder b) throws IOException {
+        b.field("type", CONTENT_TYPE).field(METRICS_FIELD, new String[] { "min", "max", "value_count" }).field(DEFAULT_METRIC, "max");
+    }
+
+    @Override
+    protected void registerParameters(ParameterChecker checker) throws IOException {
+        checker.registerUpdateCheck(
+            b -> b.field(IGNORE_MALFORMED_FIELD, true),
+            m -> assertTrue(((AggregateDoubleMetricFieldMapper) m).ignoreMalformed())
+        );
+
+        checker.registerConflictCheck(
+            DEFAULT_METRIC,
+            fieldMapping(this::minimalMapping),
+            fieldMapping(
+                b -> { b.field("type", CONTENT_TYPE).field(METRICS_FIELD, new String[] { "min", "max" }).field(DEFAULT_METRIC, "min"); }
+            )
+        );
+
+        checker.registerConflictCheck(
+            METRICS_FIELD,
+            fieldMapping(this::minimalMapping),
+            fieldMapping(
+                b -> { b.field("type", CONTENT_TYPE).field(METRICS_FIELD, new String[] { "min", "max" }).field(DEFAULT_METRIC, "max"); }
+            )
+        );
+
+        checker.registerConflictCheck(
+            METRICS_FIELD,
+            fieldMapping(this::minimalMapping),
+            fieldMapping(
+                b -> {
+                    b.field("type", CONTENT_TYPE)
+                        .field(METRICS_FIELD, new String[] { "min", "max", "value_count", "sum" })
+                        .field(DEFAULT_METRIC, "min");
+                }
+            )
+        );
+    }
+
+    @Override
+    protected Object getSampleValueForDocument() {
+        return Map.of("min", -10.1, "max", 50.0, "value_count", 14);
+    }
+
+    @Override
+    protected Object getSampleValueForQuery() {
+        return 50.0;
+    }
+
+    /**
+     * Test parsing field mapping and adding simple field
+     */
+    public void testParseValue() throws Exception {
+        DocumentMapper mapper = createDocumentMapper(fieldMapping(this::minimalMapping));
+        ParsedDocument doc = mapper.parse(
+            source(b -> b.startObject("field").field("min", -10.1).field("max", 50.0).field("value_count", 14).endObject())
+        );
+        assertEquals(-10.1, doc.rootDoc().getField("field.min").numericValue());
+
+        Mapper fieldMapper = mapper.mappers().getMapper("field");
+        assertThat(fieldMapper, instanceOf(AggregateDoubleMetricFieldMapper.class));
+    }
+
+    /**
+     * Test that invalid field mapping containing no metrics is not accepted
+     */
+    public void testInvalidMapping() throws Exception {
+        XContentBuilder mapping = XContentFactory.jsonBuilder()
+            .startObject()
+            .startObject("_doc")
+            .startObject("properties")
+            .startObject("metric")
+            .field("type", CONTENT_TYPE)
+            .endObject()
+            .endObject()
+            .endObject()
+            .endObject();
+
+        Exception e = expectThrows(MapperParsingException.class, () -> createDocumentMapper(mapping));
+        assertThat(e.getMessage(), containsString("Property [metrics] is required for field [metric]."));
+    }
+
+    /**
+     * Test parsing an aggregate_metric field that contains no values
+     */
+    public void testParseEmptyValue() throws Exception {
+        DocumentMapper mapper = createDocumentMapper(fieldMapping(this::minimalMapping));
+
+        Exception e = expectThrows(MapperParsingException.class, () -> mapper.parse(source(b -> b.startObject("field").endObject())));
+        assertThat(
+            e.getCause().getMessage(),
+            containsString("Aggregate metric field [field] must contain all metrics [min, max, value_count]")
+        );
+    }
+
+    /**
+     * Test parsing an aggregate_metric field that contains no values
+     * when ignore_malformed = true
+     */
+    public void testParseEmptyValueIgnoreMalformed() throws Exception {
+        DocumentMapper mapper = createDocumentMapper(
+            fieldMapping(
+                b -> b.field("type", CONTENT_TYPE)
+                    .field(METRICS_FIELD, new String[] { "min", "max", "value_count" })
+                    .field("ignore_malformed", true)
+                    .field(DEFAULT_METRIC, "max")
+            )
+        );
+        ParsedDocument doc = mapper.parse(source(b -> b.startObject("field").endObject()));
+        assertThat(doc.rootDoc().getField("field"), nullValue());
+    }
+
+    /**
+     * Test adding a metric that other than the supported ones (min, max, sum, value_count)
+     */
+    public void testUnsupportedMetric() throws Exception {
+        Exception e = expectThrows(
+            MapperParsingException.class,
+            () -> createDocumentMapper(
+                fieldMapping(b -> b.field("type", CONTENT_TYPE).field(METRICS_FIELD, new String[] { "min", "max", "unsupported" }))
+            )
+        );
+        assertThat(e.getMessage(), containsString("Metric [unsupported] is not supported."));
+    }
+
+    /**
+     * Test inserting a document containing a metric that has not been defined in the field mapping.
+     */
+    public void testUnmappedMetric() throws Exception {
+        DocumentMapper mapper = createDocumentMapper(fieldMapping(this::minimalMapping));
+        Exception e = expectThrows(
+            MapperParsingException.class,
+            () -> mapper.parse(
+                source(
+                    b -> b.startObject("field").field("min", -10.1).field("max", 50.0).field("value_count", 14).field("sum", 55).endObject()
+                )
+            )
+        );
+        assertThat(e.getCause().getMessage(), containsString("Aggregate metric [sum] does not exist in the mapping of field [field]"));
+    }
+
+    /**
+     * Test inserting a document containing a metric that has not been defined in the field mapping.
+     * Field will be ignored because config ignore_malformed has been set.
+     */
+    public void testUnmappedMetricWithIgnoreMalformed() throws Exception {
+        DocumentMapper mapper = createDocumentMapper(
+            fieldMapping(
+                b -> b.field("type", CONTENT_TYPE)
+                    .field(METRICS_FIELD, new String[] { "min", "max" })
+                    .field("ignore_malformed", true)
+                    .field(DEFAULT_METRIC, "max")
+            )
+        );
+
+        ParsedDocument doc = mapper.parse(
+            source(b -> b.startObject("field").field("min", -10.1).field("max", 50.0).field("sum", 55).endObject())
+        );
+        assertNull(doc.rootDoc().getField("metric.min"));
+    }
+
+    /**
+     * Test inserting a document containing less metrics than those defined in the field mapping.
+     * An exception will be thrown
+     */
+    public void testMissingMetric() throws Exception {
+        DocumentMapper mapper = createDocumentMapper(fieldMapping(this::minimalMapping));
+
+        Exception e = expectThrows(
+            MapperParsingException.class,
+            () -> mapper.parse(source(b -> b.startObject("field").field("min", -10.1).field("max", 50.0).endObject()))
+        );
+        assertThat(
+            e.getCause().getMessage(),
+            containsString("Aggregate metric field [field] must contain all metrics [min, max, value_count]")
+        );
+    }
+
+    /**
+     * Test inserting a document containing less metrics than those defined in the field mapping.
+     * Field will be ignored because config ignore_malformed has been set.
+     */
+    public void testMissingMetricWithIgnoreMalformed() throws Exception {
+        DocumentMapper mapper = createDocumentMapper(
+            fieldMapping(
+                b -> b.field("type", CONTENT_TYPE)
+                    .field(METRICS_FIELD, new String[] { "min", "max" })
+                    .field("ignore_malformed", true)
+                    .field(DEFAULT_METRIC, "max")
+            )
+        );
+
+        ParsedDocument doc = mapper.parse(source(b -> b.startObject("field").field("min", -10.1).field("max", 50.0).endObject()));
+
+        assertNull(doc.rootDoc().getField("metric.min"));
+    }
+
+    /**
+     * Test a metric that has an invalid value (string instead of number)
+     */
+    public void testInvalidMetricValue() throws Exception {
+        DocumentMapper mapper = createDocumentMapper(fieldMapping(this::minimalMapping));
+        Exception e = expectThrows(
+            MapperParsingException.class,
+            () -> mapper.parse(
+                source(b -> b.startObject("field").field("min", "10.0").field("max", 50.0).field("value_count", 14).endObject())
+            )
+        );
+
+        assertThat(
+            e.getCause().getMessage(),
+            containsString("Failed to parse object: expecting token of type [VALUE_NUMBER] but found [VALUE_STRING]")
+        );
+    }
+
+    /**
+     * Test a metric that has an invalid value (string instead of number)
+     * with ignore_malformed = true
+     */
+    public void testInvalidMetricValueIgnoreMalformed() throws Exception {
+        DocumentMapper mapper = createDocumentMapper(
+            fieldMapping(
+                b -> b.field("type", CONTENT_TYPE)
+                    .field(METRICS_FIELD, new String[] { "min", "max" })
+                    .field("ignore_malformed", true)
+                    .field(DEFAULT_METRIC, "max")
+            )
+        );
+        ParsedDocument doc = mapper.parse(source(b -> b.startObject("field").field("min", "10.0").field("max", 50.0).endObject()));
+        assertThat(doc.rootDoc().getField("metric"), nullValue());
+    }
+
+    /**
+     * Test a field that has a negative value for value_count
+     */
+    public void testNegativeValueCount() throws Exception {
+        DocumentMapper mapper = createDocumentMapper(fieldMapping(this::minimalMapping));
+        Exception e = expectThrows(
+            MapperParsingException.class,
+            () -> mapper.parse(
+                source(b -> b.startObject("field").field("min", 10.0).field("max", 50.0).field("value_count", -14).endObject())
+            )
+        );
+        assertThat(
+            e.getCause().getMessage(),
+            containsString("Aggregate metric [value_count] of field [field] cannot be a negative number")
+        );
+    }
+
+    /**
+     * Test a field that has a negative value for value_count with ignore_malformed = true
+     * No exception will be thrown but the field will be ignored
+     */
+    public void testNegativeValueCountIgnoreMalformed() throws Exception {
+        DocumentMapper mapper = createDocumentMapper(
+            fieldMapping(
+                b -> b.field("type", CONTENT_TYPE).field(METRICS_FIELD, new String[] { "value_count" }).field("ignore_malformed", true)
+            )
+        );
+
+        ParsedDocument doc = mapper.parse(source(b -> b.startObject("field").field("value_count", -14).endObject()));
+        assertThat(doc.rootDoc().getField("field.value_count"), nullValue());
+    }
+
+    /**
+     * Test parsing a value_count metric written as double with zero decimal digits
+     */
+    public void testValueCountDouble() throws Exception {
+        DocumentMapper mapper = createDocumentMapper(fieldMapping(this::minimalMapping));
+        ParsedDocument doc = mapper.parse(
+            source(b -> b.startObject("field").field("min", 10.0).field("max", 50.0).field("value_count", 77.0).endObject())
+        );
+        assertEquals(77, doc.rootDoc().getField("field.value_count").numericValue().longValue());
+    }
+
+    /**
+     * Test parsing a value_count metric written as double with some decimal digits
+     */
+    public void testInvalidDoubleValueCount() throws Exception {
+        DocumentMapper mapper = createDocumentMapper(fieldMapping(this::minimalMapping));
+        Exception e = expectThrows(
+            MapperParsingException.class,
+            () -> mapper.parse(
+                source(b -> b.startObject("field").field("min", 10.0).field("max", 50.0).field("value_count", 77.33).endObject())
+            )
+        );
+        assertThat(
+            e.getCause().getMessage(),
+            containsString("failed to parse field [field.value_count] of type [integer] in document with id '1'.")
+        );
+    }
+
+    /**
+     * Test inserting a document containing an array of metrics. An exception must be thrown.
+     */
+    public void testParseArrayValue() throws Exception {
+        DocumentMapper mapper = createDocumentMapper(fieldMapping(this::minimalMapping));
+        Exception e = expectThrows(
+            MapperParsingException.class,
+            () -> mapper.parse(
+                source(
+                    b -> b.startArray("field")
+                        .startObject()
+                        .field("min", 10.0)
+                        .field("max", 50.0)
+                        .field("value_count", 3)
+                        .endObject()
+                        .startObject()
+                        .field("min", 11.0)
+                        .field("max", 51.0)
+                        .field("value_count", 3)
+                        .endObject()
+                        .endArray()
+                )
+            )
+        );
+        assertThat(
+            e.getCause().getMessage(),
+            containsString(
+                "Field [field] of type [aggregate_metric_double] "
+                    + "does not support indexing multiple values for the same field in the same document"
+            )
+        );
+    }
+
+    /**
+     * Test setting the default_metric explicitly
+     */
+    public void testExplicitDefaultMetric() throws Exception {
+        DocumentMapper mapper = createDocumentMapper(
+            fieldMapping(
+                b -> b.field("type", CONTENT_TYPE).field(METRICS_FIELD, new String[] { "value_count", "sum" }).field(DEFAULT_METRIC, "sum")
+            )
+        );
+
+        Mapper fieldMapper = mapper.mappers().getMapper("field");
+        assertThat(fieldMapper, instanceOf(AggregateDoubleMetricFieldMapper.class));
+        assertEquals(AggregateDoubleMetricFieldMapper.Metric.sum, ((AggregateDoubleMetricFieldMapper) fieldMapper).defaultMetric());
+    }
+
+    /**
+     * Test the default_metric when not set explicitly. When only a single metric is contained, this is set as the default
+     */
+    public void testImplicitDefaultMetricSingleMetric() throws Exception {
+        DocumentMapper mapper = createDocumentMapper(
+            fieldMapping(b -> b.field("type", CONTENT_TYPE).field(METRICS_FIELD, new String[] { "value_count" }))
+        );
+
+        Mapper fieldMapper = mapper.mappers().getMapper("field");
+        assertThat(fieldMapper, instanceOf(AggregateDoubleMetricFieldMapper.class));
+        assertEquals(AggregateDoubleMetricFieldMapper.Metric.value_count, ((AggregateDoubleMetricFieldMapper) fieldMapper).defaultMetric);
+    }
+
+    /**
+     * Test the default_metric when not set explicitly, by default we have set it to be the max.
+     */
+    public void testImplicitDefaultMetric() throws Exception {
+        DocumentMapper mapper = createDocumentMapper(fieldMapping(this::minimalMapping));
+        Mapper fieldMapper = mapper.mappers().getMapper("field");
+        assertThat(fieldMapper, instanceOf(AggregateDoubleMetricFieldMapper.class));
+        assertEquals(AggregateDoubleMetricFieldMapper.Metric.max, ((AggregateDoubleMetricFieldMapper) fieldMapper).defaultMetric);
+    }
+
+    /**
+     * Test the default_metric when not set explicitly. When more than one metrics are contained
+     * and max is not one of them, an exception should be thrown.
+     */
+    public void testMissingDefaultMetric() {
+        Exception e = expectThrows(
+            MapperParsingException.class,
+            () -> createDocumentMapper(
+                fieldMapping(b -> b.field("type", CONTENT_TYPE).field(METRICS_FIELD, new String[] { "value_count", "sum" }))
+            )
+        );
+        assertThat(e.getMessage(), containsString("Property [default_metric] is required for field [field]."));
+    }
+
+    /**
+     * Test setting an invalid value for the default_metric. An exception must be thrown
+     */
+    public void testInvalidDefaultMetric() {
+        Exception e = expectThrows(
+            MapperParsingException.class,
+            () -> createDocumentMapper(
+                fieldMapping(
+                    b -> b.field("type", CONTENT_TYPE)
+                        .field(METRICS_FIELD, new String[] { "value_count", "sum" })
+                        .field(DEFAULT_METRIC, "invalid_metric")
+                )
+            )
+        );
+        assertThat(e.getMessage(), containsString("Metric [invalid_metric] is not supported."));
+    }
+
+    /**
+     * Test setting a value for the default_metric that is not contained in the "metrics" field.
+     * An exception must be thrown
+     */
+    public void testUndefinedDefaultMetric() {
+        Exception e = expectThrows(
+            MapperParsingException.class,
+            () -> createDocumentMapper(
+                fieldMapping(
+                    b -> b.field("type", CONTENT_TYPE)
+                        .field(METRICS_FIELD, new String[] { "value_count", "sum" })
+                        .field(DEFAULT_METRIC, "min")
+                )
+            )
+        );
+        assertThat(e.getMessage(), containsString("Default metric [min] is not defined in the metrics of field [field]."));
+    }
+
+    /**
+     * Test parsing field mapping and adding simple field
+     */
+    public void testParseNestedValue() throws Exception {
+        DocumentMapper mapper = createDocumentMapper(
+            fieldMapping(
+                b -> b.startObject("properties")
+                    .startObject("subfield")
+                    .field("type", CONTENT_TYPE)
+                    .field(METRICS_FIELD, new String[] { "min", "max", "sum", "value_count" })
+                    .field(DEFAULT_METRIC, "max")
+                    .endObject()
+                    .endObject()
+            )
+        );
+
+        Mapper fieldMapper = mapper.mappers().getMapper("field.subfield");
+        assertThat(fieldMapper, instanceOf(AggregateDoubleMetricFieldMapper.class));
+        ParsedDocument doc = mapper.parse(
+            source(
+                b -> b.startObject("field")
+                    .startObject("subfield")
+                    .field("min", 10.1)
+                    .field("max", 50.0)
+                    .field("sum", 43)
+                    .field("value_count", 14)
+                    .endObject()
+                    .endObject()
+            )
+        );
+        assertThat(doc.rootDoc().getField("field.subfield.min"), notNullValue());
+    }
+
+    /**
+     *  subfields of aggregate_metric_double should not be searchable or exposed in field_caps
+     */
+    public void testNoSubFieldsIterated() throws IOException {
+        AggregateDoubleMetricFieldMapper.Metric[] values = AggregateDoubleMetricFieldMapper.Metric.values();
+        List<AggregateDoubleMetricFieldMapper.Metric> subset = randomSubsetOf(randomIntBetween(1, values.length), values);
+        DocumentMapper mapper = createDocumentMapper(
+            fieldMapping(b -> b.field("type", CONTENT_TYPE).field(METRICS_FIELD, subset).field(DEFAULT_METRIC, subset.get(0)))
+        );
+        Iterator<Mapper> iterator = mapper.mappers().getMapper("field").iterator();
+        assertFalse(iterator.hasNext());
+    }
+
+    public void testFieldCaps() throws IOException {
+        MapperService aggMetricMapperService = createMapperService(fieldMapping(this::minimalMapping));
+        MappedFieldType fieldType = aggMetricMapperService.fieldType("field");
+        assertThat(fieldType.familyTypeName(), equalTo("double"));
+        assertTrue(fieldType.isSearchable());
+        assertTrue(fieldType.isAggregatable());
+    }
+
+    /*
+     * Since all queries for aggregate_metric_double fields are delegated to their default_metric numeric
+     *  sub-field, we override this method so that testExistsQueryMinimalMapping() passes successfully.
+     */
+    protected void assertExistsQuery(MappedFieldType fieldType, Query query, ParseContext.Document fields) {
+        assertThat(query, Matchers.instanceOf(DocValuesFieldExistsQuery.class));
+        DocValuesFieldExistsQuery fieldExistsQuery = (DocValuesFieldExistsQuery) query;
+        String defaultMetric = ((AggregateDoubleMetricFieldMapper.AggregateDoubleMetricFieldType) fieldType).getDefaultMetric().name();
+        assertEquals("field." + defaultMetric, fieldExistsQuery.getField());
+        assertDocValuesField(fields, "field." + defaultMetric);
+        assertNoFieldNamesField(fields);
+    }
+}

x-pack/plugin/mapper-aggregate-metric/src/test/java/org/elasticsearch/xpack/aggregatemetric/mapper/AggregateDoubleMetricFieldTypeTests.java

@@ -0,0 +1,149 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.xpack.aggregatemetric.mapper;
+
+import org.apache.lucene.document.DoublePoint;
+import org.apache.lucene.document.NumericDocValuesField;
+import org.apache.lucene.index.DirectoryReader;
+import org.apache.lucene.index.LeafReaderContext;
+import org.apache.lucene.index.RandomIndexWriter;
+import org.apache.lucene.search.IndexOrDocValuesQuery;
+import org.apache.lucene.search.IndexSearcher;
+import org.apache.lucene.search.MatchAllDocsQuery;
+import org.apache.lucene.search.Query;
+import org.apache.lucene.store.Directory;
+import org.elasticsearch.Version;
+import org.elasticsearch.common.lucene.search.function.ScriptScoreQuery;
+import org.elasticsearch.index.fielddata.ScriptDocValues;
+import org.elasticsearch.index.mapper.FieldTypeTestCase;
+import org.elasticsearch.index.mapper.MappedFieldType;
+import org.elasticsearch.index.mapper.NumberFieldMapper;
+import org.elasticsearch.index.query.QueryShardContext;
+import org.elasticsearch.script.ScoreScript;
+import org.elasticsearch.script.Script;
+import org.elasticsearch.search.lookup.SearchLookup;
+import org.elasticsearch.xpack.aggregatemetric.mapper.AggregateDoubleMetricFieldMapper.AggregateDoubleMetricFieldType;
+import org.elasticsearch.xpack.aggregatemetric.mapper.AggregateDoubleMetricFieldMapper.Metric;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import static java.util.Arrays.asList;
+import static org.elasticsearch.xpack.aggregatemetric.mapper.AggregateDoubleMetricFieldMapper.subfieldName;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.instanceOf;
+import static org.mockito.Matchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class AggregateDoubleMetricFieldTypeTests extends FieldTypeTestCase {
+
+    protected AggregateDoubleMetricFieldType createDefaultFieldType(String name, Map<String, String> meta, Metric defaultMetric) {
+        AggregateDoubleMetricFieldType fieldType = new AggregateDoubleMetricFieldType(name, meta);
+        for (AggregateDoubleMetricFieldMapper.Metric m : List.of(
+            AggregateDoubleMetricFieldMapper.Metric.min,
+            AggregateDoubleMetricFieldMapper.Metric.max
+        )) {
+            String subfieldName = subfieldName(fieldType.name(), m);
+            NumberFieldMapper.NumberFieldType subfield = new NumberFieldMapper.NumberFieldType(
+                subfieldName,
+                NumberFieldMapper.NumberType.DOUBLE
+            );
+            fieldType.addMetricField(m, subfield);
+        }
+        fieldType.setDefaultMetric(defaultMetric);
+        return fieldType;
+    }
+
+    public void testTermQuery() {
+        final MappedFieldType fieldType = createDefaultFieldType("foo", Collections.emptyMap(), Metric.max);
+        Query query = fieldType.termQuery(55.2, null);
+        assertThat(query, equalTo(DoublePoint.newRangeQuery("foo.max", 55.2, 55.2)));
+    }
+
+    public void testTermsQuery() {
+        final MappedFieldType fieldType = createDefaultFieldType("foo", Collections.emptyMap(), Metric.max);
+        Query query = fieldType.termsQuery(asList(55.2, 500.3), null);
+        assertThat(query, equalTo(DoublePoint.newSetQuery("foo.max", 55.2, 500.3)));
+    }
+
+    public void testRangeQuery() {
+        final MappedFieldType fieldType = createDefaultFieldType("foo", Collections.emptyMap(), Metric.max);
+        Query query = fieldType.rangeQuery(10.1, 100.1, true, true, null, null, null, null);
+        assertThat(query, instanceOf(IndexOrDocValuesQuery.class));
+    }
+
+    public void testFetchSourceValueWithOneMetric() throws IOException {
+        final MappedFieldType fieldType = createDefaultFieldType("field", Collections.emptyMap(), Metric.min);
+        final double defaultValue = 45.8;
+        final Map<String, Object> metric = Collections.singletonMap("min", defaultValue);
+        assertEquals(List.of(defaultValue), fetchSourceValue(fieldType, metric));
+    }
+
+    public void testFetchSourceValueWithMultipleMetrics() throws IOException {
+        final MappedFieldType fieldType = createDefaultFieldType("field", Collections.emptyMap(), Metric.max);
+        final double defaultValue = 45.8;
+        final Map<String, Object> metric = Map.of("min", 14.2, "max", defaultValue);
+        assertEquals(List.of(defaultValue), fetchSourceValue(fieldType, metric));
+    }
+
+    /** Tests that aggregate_metric_double uses the default_metric subfield's doc-values as values in scripts */
+    public void testUsedInScript() throws IOException {
+        final MappedFieldType mappedFieldType = createDefaultFieldType("field", Collections.emptyMap(), Metric.max);
+        try (Directory directory = newDirectory(); RandomIndexWriter iw = new RandomIndexWriter(random(), directory)) {
+            iw.addDocument(
+                List.of(
+                    new NumericDocValuesField(subfieldName("field", Metric.max), Double.doubleToLongBits(10)),
+                    new NumericDocValuesField(subfieldName("field", Metric.min), Double.doubleToLongBits(2))
+                )
+            );
+            iw.addDocument(
+                List.of(
+                    new NumericDocValuesField(subfieldName("field", Metric.max), Double.doubleToLongBits(4)),
+                    new NumericDocValuesField(subfieldName("field", Metric.min), Double.doubleToLongBits(1))
+                )
+            );
+            iw.addDocument(
+                List.of(
+                    new NumericDocValuesField(subfieldName("field", Metric.max), Double.doubleToLongBits(7)),
+                    new NumericDocValuesField(subfieldName("field", Metric.min), Double.doubleToLongBits(4))
+                )
+            );
+            try (DirectoryReader reader = iw.getReader()) {
+                QueryShardContext queryShardContext = mock(QueryShardContext.class);
+                when(queryShardContext.getFieldType(anyString())).thenReturn(mappedFieldType);
+                when(queryShardContext.allowExpensiveQueries()).thenReturn(true);
+                SearchLookup lookup = new SearchLookup(
+                    queryShardContext::getFieldType,
+                    (mft, lookupSupplier) -> mft.fielddataBuilder("test", lookupSupplier).build(null, null)
+                );
+                when(queryShardContext.lookup()).thenReturn(lookup);
+                IndexSearcher searcher = newSearcher(reader);
+                assertThat(searcher.count(new ScriptScoreQuery(new MatchAllDocsQuery(), new Script("test"), new ScoreScript.LeafFactory() {
+                    @Override
+                    public boolean needs_score() {
+                        return false;
+                    }
+
+                    @Override
+                    public ScoreScript newInstance(LeafReaderContext ctx) {
+                        return new ScoreScript(Map.of(), queryShardContext.lookup(), ctx) {
+                            @Override
+                            public double execute(ExplanationHolder explanation) {
+                                Map<String, ScriptDocValues<?>> doc = getDoc();
+                                ScriptDocValues.Doubles doubles = (ScriptDocValues.Doubles) doc.get("field");
+                                return doubles.get(0);
+                            }
+                        };
+                    }
+                }, 7f, "test", 0, Version.CURRENT)), equalTo(2));
+            }
+        }
+    }
+
+}

x-pack/plugin/src/test/resources/rest-api-spec/test/aggregate-metrics/10_basic.yml

@@ -0,0 +1,292 @@
+---
+"Test exists query on aggregate metric field":
+  - skip:
+      version: " - 7.99.99"
+      reason: "Aggregate metric fields are currently only implemented in 8.0."
+
+  - do:
+      indices.create:
+        index:  aggregate_metric_test
+        body:
+          mappings:
+            properties:
+              metric:
+                type: aggregate_metric_double
+                metrics: [min, max]
+                default_metric: max
+
+  - do:
+      index:
+        index:  aggregate_metric_test
+        id:     1
+        body:
+          metric:
+            min: 18.2
+            max: 100
+        refresh: true
+
+  - do:
+      search:
+        index: aggregate_metric_test
+        body:
+          query:
+            exists:
+              field: metric
+
+  - match: { hits.total.value: 1 }
+
+  - do:
+      search:
+        index: aggregate_metric_test
+        body:
+          query:
+            exists:
+              field: metric.min
+
+  - match: { hits.total.value: 0 }
+
+  - do:
+      search:
+        index: aggregate_metric_test
+        body:
+          query:
+            exists:
+              field: metric.nonexistent_key
+
+  - match: { hits.total.value: 0 }
+
+---
+"Test term query on aggregate metric field":
+  - skip:
+      version: " - 7.99.99"
+      reason: "Aggregate metric fields are currently only implemented in 8.0."
+
+  - do:
+      indices.create:
+        index:  test
+        body:
+          mappings:
+            properties:
+              metric:
+                type: aggregate_metric_double
+                metrics: [min, max]
+                default_metric: max
+
+  - do:
+      index:
+        index:  test
+        id:     1
+        body:
+          metric:
+            min: 18.2
+            max: 100
+        refresh: true
+
+  - do:
+      index:
+        index:  test
+        id:     2
+        body:
+          metric:
+            min: 50
+            max: 1000
+        refresh: true
+
+  - do:
+      search:
+        index: test
+        body:
+          query:
+            term:
+              metric:
+                value: 1000
+
+  - match: { hits.total.value: 1 }
+  - length: { hits.hits: 1 }
+  - match: { hits.hits.0._id: "2" }
+
+  - do:
+      search:
+        index: test
+        body:
+          query:
+            term:
+              metric:
+                value: 100
+
+  - match: { hits.total.value: 1 }
+  - length: { hits.hits: 1 }
+  - match: { hits.hits.0._id: "1" }
+
+
+  - do:
+      search:
+        index: test
+        body:
+          query:
+            term:
+              metric.min:
+                value: 50
+
+  - match: { hits.total.value: 0 }
+
+
+---
+"Test range query on aggregate metric field":
+  - skip:
+      version: " - 7.99.99"
+      reason: "Aggregate metric fields are currently only implemented in 8.0."
+
+  - do:
+      indices.create:
+        index:  test
+        body:
+          mappings:
+            properties:
+              metric:
+                type: aggregate_metric_double
+                metrics: [min, max]
+                default_metric: max
+
+  - do:
+      index:
+        index:  test
+        id:     1
+        body:
+          metric:
+            min: 18.2
+            max: 100
+        refresh: true
+
+  - do:
+      index:
+        index:  test
+        id:     2
+        body:
+          metric:
+            min: 50
+            max: 1000
+        refresh: true
+
+  - do:
+      search:
+        index: test
+        body:
+          query:
+            range:
+              metric:
+                gte: 900
+                lte: 1000
+
+  - match: { hits.total.value: 1 }
+  - length: { hits.hits: 1 }
+  - match: { hits.hits.0._id: "2" }
+
+  - do:
+      search:
+        index: test
+        body:
+          query:
+            range:
+              metric:
+                gte: 90
+                lte: 200
+
+  - match: { hits.total.value: 1 }
+  - length: { hits.hits: 1 }
+  - match: { hits.hits.0._id: "1" }
+
+  - do:
+      search:
+        index: test
+        body:
+          query:
+            range:
+              metric.min:
+                gte: 10
+                lte: 40
+
+  - match: { hits.total.value: 0 }
+
+---
+"Sort":
+  - skip:
+      version: " - 7.99.99"
+      reason: "Aggregate metric fields are currently only implemented in 8.0."
+
+  - do:
+      indices.create:
+        index:  test
+        body:
+          mappings:
+            properties:
+              metric:
+                type: aggregate_metric_double
+                metrics: [min, max]
+                default_metric: max
+
+  - do:
+      index:
+        index:  test
+        id:     1
+        body:
+          metric:
+            min: 18.2
+            max: 3000
+        refresh: true
+
+  - do:
+      index:
+        index:  test
+        id:     2
+        body:
+          metric:
+            min: 50
+            max: 1000
+        refresh: true
+
+  - do:
+      index:
+        index:  test
+        id:     3
+        body:
+          metric:
+            min: 150
+            max: 5000
+        refresh: true
+
+  - do:
+      search:
+        index: test
+        body:
+          sort: [ { metric: asc } ]
+
+  - match: { "hits.total.value": 3 }
+  - match: { hits.hits.0._id: "2" }
+  - match: { hits.hits.1._id: "1" }
+  - match: { hits.hits.2._id: "3" }
+
+  - do:
+      search:
+        index: test
+        body:
+          sort: [ { metric: desc } ]
+
+  - match: { "hits.total.value": 3 }
+  - match: { hits.hits.0._id: "3" }
+  - match: { hits.hits.1._id: "1" }
+  - match: { hits.hits.2._id: "2" }
+
+  - do:
+      catch: bad_request
+      search:
+        index: test
+        body:
+          sort: [ { metric.max: asc } ]
+
+  - do:
+      catch: bad_request
+      search:
+        index: test
+        body:
+          sort: [ { metric.min: desc } ]

x-pack/plugin/src/test/resources/rest-api-spec/test/aggregate-metrics/20_min_max_agg.yml

@@ -0,0 +1,81 @@
+###############################################################################################################
+# Test min and max aggregations on an index that contains aggregate_metric_double fields
+###############################################################################################################
+
+setup:
+  - do:
+      indices.create:
+        index:  aggregate_metric_test
+        body:
+          mappings:
+            properties:
+              metric:
+                type: aggregate_metric_double
+                metrics: [min, max]
+                default_metric: max
+
+  - do:
+      bulk:
+        index: aggregate_metric_test
+        refresh: true
+        body:
+          - '{"index": {}}'
+          - '{"metric": {"min": 0, "max": 100} }'
+          - '{"index": {}}'
+          - '{"metric": {"min": 60, "max": 100} }'
+          - '{"index": {}}'
+          - '{"metric": {"min": -400.50, "max": 1000} }'
+          - '{"index": {}}'
+          - '{"metric": {"min": 1, "max": 99.3} }'
+          - '{"index": {}}'
+          - '{"metric": {"min": -100, "max": -40} }'
+---
+"Test min_max aggs":
+  - skip:
+      version: " - 7.99.99"
+      reason: "Aggregate metric fields are currently only implemented in 8.0."
+
+  - do:
+      search:
+        index: aggregate_metric_test
+        size: 0
+        body:
+          aggs:
+            max_agg:
+              max:
+                field: metric
+            min_agg:
+              min:
+                field: metric
+
+  - match: { hits.total.value: 5 }
+  - match: { aggregations.max_agg.value: 1000}
+  - match: { aggregations.min_agg.value: -400.50}
+
+
+---
+"Test min_max aggs with query":
+  - skip:
+      version: " - 7.99.99"
+      reason: "Aggregate metric fields are currently only implemented in 8.0."
+
+  - do:
+      search:
+        index: aggregate_metric_test
+        size: 0
+        body:
+          query:
+            term:
+              metric:
+                value: 100
+          aggs:
+            max_agg:
+              max:
+                field: metric
+            min_agg:
+              min:
+                field: metric
+
+  - match: { hits.total.value: 2 }
+  - match: { aggregations.max_agg.value: 100}
+  - match: { aggregations.min_agg.value: 0}

x-pack/plugin/src/test/resources/rest-api-spec/test/aggregate-metrics/30_sum_agg.yml

@@ -0,0 +1,68 @@
+setup:
+  - do:
+      indices.create:
+        index:  aggregate_metric_test
+        body:
+          mappings:
+            properties:
+              metric:
+                type: aggregate_metric_double
+                metrics: [sum, value_count]
+                default_metric: value_count
+
+  - do:
+      bulk:
+        index: aggregate_metric_test
+        refresh: true
+        body:
+          - '{"index": {}}'
+          - '{"metric": {"sum": 10000.45, "value_count": 100} }'
+          - '{"index": {}}'
+          - '{"metric": {"sum": 60, "value_count": 20} }'
+          - '{"index": {}}'
+          - '{"metric": {"sum": -400.50, "value_count": 1000} }'
+          - '{"index": {}}'
+          - '{"metric": {"sum": 1, "value_count": 20} }'
+          - '{"index": {}}'
+          - '{"metric": {"sum": -100, "value_count": 40} }'
+---
+"Test sum agg":
+  - skip:
+      version: " - 7.99.99"
+      reason: "Aggregate metric fields are currently only implemented in 8.0."
+
+  - do:
+      search:
+        index: aggregate_metric_test
+        size: 0
+        body:
+          aggs:
+            sum_agg:
+              sum:
+                field: metric
+
+  - match: { hits.total.value: 5 }
+  - match: { aggregations.sum_agg.value: 9560.95}
+
+---
+"Test sum agg with query":
+  - skip:
+      version: " - 7.99.99"
+      reason: "Aggregate metric fields are currently only implemented in 8.0."
+
+  - do:
+      search:
+        index: aggregate_metric_test
+        size: 0
+        body:
+          query:
+            term:
+              metric:
+                value: 20
+          aggs:
+            sum_agg:
+              sum:
+                field: metric
+
+  - match: { hits.total.value: 2 }
+  - match: { aggregations.sum_agg.value: 61}

x-pack/plugin/src/test/resources/rest-api-spec/test/aggregate-metrics/40_avg_agg.yml

@@ -0,0 +1,68 @@
+setup:
+  - do:
+      indices.create:
+        index:  aggregate_metric_test
+        body:
+          mappings:
+            properties:
+              metric:
+                type: aggregate_metric_double
+                metrics: [sum, value_count]
+                default_metric: value_count
+
+  - do:
+      bulk:
+        index: aggregate_metric_test
+        refresh: true
+        body:
+          - '{"index": {}}'
+          - '{"metric": {"sum": 10000, "value_count": 100} }'
+          - '{"index": {}}'
+          - '{"metric": {"sum": 60, "value_count": 20} }'
+          - '{"index": {}}'
+          - '{"metric": {"sum": -400, "value_count": 780} }'
+          - '{"index": {}}'
+          - '{"metric": {"sum": 40, "value_count": 20} }'
+          - '{"index": {}}'
+          - '{"metric": {"sum": -100, "value_count": 40} }'
+---
+"Test avg agg":
+  - skip:
+      version: " - 7.99.99"
+      reason: "Aggregate metric fields are currently only implemented in 8.0."
+
+  - do:
+      search:
+        index: aggregate_metric_test
+        size: 0
+        body:
+          aggs:
+            avg_agg:
+              avg:
+                field: metric
+
+  - match: { hits.total.value: 5 }
+  - match: { aggregations.avg_agg.value: 10} # = (9600 / 960)
+
+---
+"Test avg agg with query":
+  - skip:
+      version: " - 7.99.99"
+      reason: "Aggregate metric fields are currently only implemented in 8.0."
+
+  - do:
+      search:
+        index: aggregate_metric_test
+        size: 0
+        body:
+          query:
+            term:
+              metric:
+                value: 20
+          aggs:
+            avg_agg:
+              avg:
+                field: metric
+
+  - match: { hits.total.value: 2 }
+  - match: { aggregations.avg_agg.value: 2.5}

x-pack/plugin/src/test/resources/rest-api-spec/test/aggregate-metrics/50_value_count_agg.yml

@@ -0,0 +1,68 @@
+setup:
+  - do:
+      indices.create:
+        index:  aggregate_metric_test
+        body:
+          mappings:
+            properties:
+              metric:
+                type: aggregate_metric_double
+                metrics: [sum, value_count]
+                default_metric: value_count
+
+  - do:
+      bulk:
+        index: aggregate_metric_test
+        refresh: true
+        body:
+          - '{"index": {}}'
+          - '{"metric": {"sum": 10000, "value_count": 100} }'
+          - '{"index": {}}'
+          - '{"metric": {"sum": 60, "value_count": 20} }'
+          - '{"index": {}}'
+          - '{"metric": {"sum": -400, "value_count": 780} }'
+          - '{"index": {}}'
+          - '{"metric": {"sum": 40, "value_count": 20} }'
+          - '{"index": {}}'
+          - '{"metric": {"sum": -100, "value_count": 40} }'
+---
+"Test value_count agg":
+  - skip:
+      version: " - 7.99.99"
+      reason: "Aggregate metric fields are currently only implemented in 8.0."
+
+  - do:
+      search:
+        index: aggregate_metric_test
+        size: 0
+        body:
+          aggs:
+            value_count_agg:
+              value_count:
+                field: metric
+
+  - match: { hits.total.value: 5 }
+  - match: { aggregations.value_count_agg.value: 960}
+
+---
+"Test value_count agg with query":
+  - skip:
+      version: " - 7.99.99"
+      reason: "Aggregate metric fields are currently only implemented in 8.0."
+
+  - do:
+      search:
+        index: aggregate_metric_test
+        size: 0
+        body:
+          query:
+            term:
+              metric:
+                value: 20
+          aggs:
+            value_count_agg:
+              value_count:
+                field: metric
+
+  - match: { hits.total.value: 2 }
+  - match: { aggregations.value_count_agg.value: 40}

x-pack/plugin/src/test/resources/rest-api-spec/test/aggregate-metrics/80_raw_agg_metric_aggs.yml

@@ -0,0 +1,131 @@
+###############################################################################################################
+# Test aggregations on an aggregate index containing aggregate_metric_double fields and a raw index containing
+# raw data.
+###############################################################################################################
+
+setup:
+  # Setup aggregate index with field metric of type aggregate_metric_double
+  - do:
+      indices.create:
+        index:  test_metric_aggregate
+        body:
+          mappings:
+            properties:
+              t:
+                type: long
+              metric:
+                type: aggregate_metric_double
+                metrics: [min, max, sum, value_count]
+                default_metric: max
+  - do:
+      bulk:
+        index: test_metric_aggregate
+        refresh: true
+        body:
+          - '{"index": {}}'
+          - '{"t": 10, "metric": {"min": 10.999, "max": 400.11, "sum": 400, "value_count": 6} }'
+          - '{"index": {}}'
+          - '{"t": 20, "metric": {"min": 200, "max": 300, "sum": 1000, "value_count": 5} }'
+          - '{"index": {}}'
+          - '{"t": 30, "metric": {"min": 10, "max": 50, "sum": 2000, "value_count": 5} }'
+          - '{"index": {}}'
+          - '{"t": 50, "metric": {"min": 10, "max": 400, "sum": 3000, "value_count": 5} }'
+
+  # Setup raw index with field metric of type double
+  - do:
+      indices.create:
+        index:  test_metric_raw
+        body:
+          mappings:
+            properties:
+              t:
+                type: long
+              metric:
+                type: double
+  - do:
+      bulk:
+        index: test_metric_raw
+        refresh: true
+        body:
+          - '{"index": {}}'
+          - '{"t": 10, "metric": 50}'
+          - '{"index": {}}'
+          - '{"t": 10, "metric": 100}'
+          - '{"index": {}}'
+          - '{"t": 20, "metric": 200}'
+          - '{"index": {}}'
+          - '{"t": 30, "metric": 600}'
+
+---
+"Test mixed aggregations":
+  - skip:
+      version: " - 7.99.99"
+      reason: "Aggregate metric fields are currently only implemented in 8.0."
+
+  - do:
+      search:
+        index: test_metric_*
+        size: 0
+        body:
+          aggs:
+            max_agg:
+              max:
+                field: metric
+            min_agg:
+              min:
+                field: metric
+            sum_agg:
+              sum:
+                field: metric
+            avg_agg:
+              avg:
+                field: metric
+            value_count_agg:
+              value_count:
+                field: metric
+
+  - match: { hits.total.value: 8 }
+  - match: { aggregations.max_agg.value: 600}
+  - match: { aggregations.min_agg.value: 10}
+  - match: { aggregations.sum_agg.value: 7350}
+  - match: { aggregations.avg_agg.value: 294}
+  - match: { aggregations.value_count_agg.value: 25}
+
+---
+"Test mixed aggregations with query":
+  - skip:
+      version: " - 7.99.99"
+      reason: "Aggregate metric fields are currently only implemented in 8.0."
+
+  - do:
+      search:
+        index: test_metric_*
+        size: 0
+        body:
+          query:
+            term:
+              t:
+                value: 10
+          aggs:
+            max_agg:
+              max:
+                field: metric
+            min_agg:
+              min:
+                field: metric
+            sum_agg:
+              sum:
+                field: metric
+            avg_agg:
+              avg:
+                field: metric
+            value_count_agg:
+              value_count:
+                field: metric
+
+  - match: { hits.total.value: 3 }
+  - match: { aggregations.max_agg.value: 400.11}
+  - match: { aggregations.min_agg.value: 10.999}
+  - match: { aggregations.sum_agg.value: 550}
+  - match: { aggregations.avg_agg.value: 68.75}
+  - match: { aggregations.value_count_agg.value: 8}