Make ValuesSourceConfig behave like a config object (#57762) (#58012)

modules/aggs-matrix-stats/src/main/java/org/elasticsearch/search/aggregations/support/ArrayValuesSourceAggregatorFactory.java

@@ -50,9 +50,9 @@ public abstract class ArrayValuesSourceAggregatorFactory
         HashMap<String, ValuesSource> valuesSources = new HashMap<>();
 
         for (Map.Entry<String, ValuesSourceConfig> config : configs.entrySet()) {
-            ValuesSource vs = config.getValue().toValuesSource();
-            if (vs != null) {
-                valuesSources.put(config.getKey(), vs);
+            ValuesSourceConfig vsc = config.getValue();
+            if (vsc.hasValues()) {
+                valuesSources.put(config.getKey(), vsc.getValuesSource());
             }
         }
         if (valuesSources.isEmpty()) {

modules/aggs-matrix-stats/src/test/java/org/elasticsearch/search/aggregations/matrix/stats/MatrixStatsAggregatorTests.java

@@ -61,6 +61,27 @@ public class MatrixStatsAggregatorTests extends AggregatorTestCase {
         }
     }
 
+    public void testUnmapped() throws Exception {
+        MappedFieldType ft =
+            new NumberFieldMapper.NumberFieldType(NumberFieldMapper.NumberType.DOUBLE);
+        ft.setName("field");
+
+        try (Directory directory = newDirectory();
+             RandomIndexWriter indexWriter = new RandomIndexWriter(random(), directory)) {
+            if (randomBoolean()) {
+                indexWriter.addDocument(Collections.singleton(new StringField("another_field", "value", Field.Store.NO)));
+            }
+            try (IndexReader reader = indexWriter.getReader()) {
+                IndexSearcher searcher = new IndexSearcher(reader);
+                MatrixStatsAggregationBuilder aggBuilder = new MatrixStatsAggregationBuilder("my_agg")
+                    .fields(Collections.singletonList("bogus"));
+                InternalMatrixStats stats = search(searcher, new MatchAllDocsQuery(), aggBuilder, ft);
+                assertNull(stats.getStats());
+                assertFalse(MatrixAggregationInspectionHelper.hasValue(stats));
+            }
+        }
+    }
+
     public void testTwoFields() throws Exception {
         String fieldA = "a";
         MappedFieldType ftA = new NumberFieldMapper.NumberFieldType(NumberFieldMapper.NumberType.DOUBLE);

modules/parent-join/src/main/java/org/elasticsearch/join/aggregations/ChildrenAggregatorFactory.java

@@ -68,11 +68,11 @@ public class ChildrenAggregatorFactory extends ValuesSourceAggregatorFactory {
     }
 
     @Override
-    protected Aggregator doCreateInternal(ValuesSource rawValuesSource,
-                                          SearchContext searchContext, Aggregator parent,
+    protected Aggregator doCreateInternal(SearchContext searchContext, Aggregator parent,
                                           boolean collectsFromSingleBucket,
                                           Map<String, Object> metadata) throws IOException {
 
+        ValuesSource rawValuesSource = config.getValuesSource();
         if (rawValuesSource instanceof WithOrdinals == false) {
             throw new AggregationExecutionException("ValuesSource type " + rawValuesSource.toString() +
                 "is not supported for aggregation " + this.name());

modules/parent-join/src/main/java/org/elasticsearch/join/aggregations/ParentAggregatorFactory.java

@@ -69,11 +69,11 @@ public class ParentAggregatorFactory extends ValuesSourceAggregatorFactory {
     }
 
     @Override
-    protected Aggregator doCreateInternal(ValuesSource rawValuesSource,
-                                          SearchContext searchContext, Aggregator children,
+    protected Aggregator doCreateInternal(SearchContext searchContext, Aggregator children,
                                           boolean collectsFromSingleBucket,
                                           Map<String, Object> metadata) throws IOException {
 
+        ValuesSource rawValuesSource = config.getValuesSource();
         if (rawValuesSource instanceof WithOrdinals == false) {
             throw new AggregationExecutionException("ValuesSource type " + rawValuesSource.toString() +
                 "is not supported for aggregation " + this.name());

server/src/main/java/org/elasticsearch/search/aggregations/bucket/composite/DateHistogramValuesSourceBuilder.java

@@ -249,7 +249,7 @@ public class DateHistogramValuesSourceBuilder
     @Override
     protected CompositeValuesSourceConfig innerBuild(QueryShardContext queryShardContext, ValuesSourceConfig config) throws IOException {
         Rounding rounding = dateHistogramInterval.createRounding(timeZone(), offset);
-        ValuesSource orig = config.toValuesSource();
+        ValuesSource orig = config.hasValues() ? config.getValuesSource() : null;
         if (orig == null) {
             orig = ValuesSource.Numeric.EMPTY;
         }

server/src/main/java/org/elasticsearch/search/aggregations/bucket/composite/GeoTileGridValuesSourceBuilder.java

@@ -129,7 +129,7 @@ public class GeoTileGridValuesSourceBuilder extends CompositeValuesSourceBuilder
 
     @Override
     protected CompositeValuesSourceConfig innerBuild(QueryShardContext queryShardContext, ValuesSourceConfig config) throws IOException {
-        ValuesSource orig = config.toValuesSource();
+        ValuesSource orig = config.hasValues() ? config.getValuesSource() : null;
         if (orig == null) {
             orig = ValuesSource.GeoPoint.EMPTY;
         }

server/src/main/java/org/elasticsearch/search/aggregations/bucket/composite/HistogramValuesSourceBuilder.java

@@ -111,7 +111,7 @@ public class HistogramValuesSourceBuilder extends CompositeValuesSourceBuilder<H
 
     @Override
     protected CompositeValuesSourceConfig innerBuild(QueryShardContext queryShardContext, ValuesSourceConfig config) throws IOException {
-        ValuesSource orig = config.toValuesSource();
+        ValuesSource orig = config.hasValues() ? config.getValuesSource() : null;
         if (orig == null) {
             orig = ValuesSource.Numeric.EMPTY;
         }

server/src/main/java/org/elasticsearch/search/aggregations/bucket/composite/TermsValuesSourceBuilder.java

@@ -71,7 +71,7 @@ public class TermsValuesSourceBuilder extends CompositeValuesSourceBuilder<Terms
 
     @Override
     protected CompositeValuesSourceConfig innerBuild(QueryShardContext queryShardContext, ValuesSourceConfig config) throws IOException {
-        ValuesSource vs = config.toValuesSource();
+        ValuesSource vs = config.hasValues() ? config.getValuesSource() : null;
         if (vs == null) {
             // The field is unmapped so we use a value source that can parse any type of values.
             // This is needed because the after values are parsed even when there are no values to process.

server/src/main/java/org/elasticsearch/search/aggregations/bucket/geogrid/GeoHashGridAggregatorFactory.java

@@ -74,18 +74,17 @@ public class GeoHashGridAggregatorFactory extends ValuesSourceAggregatorFactory
     }
 
     @Override
-    protected Aggregator doCreateInternal(final ValuesSource valuesSource,
-                                            SearchContext searchContext,
-                                            Aggregator parent,
-                                            boolean collectsFromSingleBucket,
-                                            Map<String, Object> metadata) throws IOException {
+    protected Aggregator doCreateInternal(SearchContext searchContext,
+                                          Aggregator parent,
+                                          boolean collectsFromSingleBucket,
+                                          Map<String, Object> metadata) throws IOException {
         AggregatorSupplier aggregatorSupplier = queryShardContext.getValuesSourceRegistry()
             .getAggregator(config, GeoHashGridAggregationBuilder.NAME);
         if (aggregatorSupplier instanceof GeoGridAggregatorSupplier == false) {
             throw new AggregationExecutionException("Registry miss-match - expected "
                 + GeoGridAggregatorSupplier.class.getName() + ", found [" + aggregatorSupplier.getClass().toString() + "]");
         }
-        return ((GeoGridAggregatorSupplier) aggregatorSupplier).build(name, factories, valuesSource, precision, geoBoundingBox,
+        return ((GeoGridAggregatorSupplier) aggregatorSupplier).build(name, factories, config.getValuesSource(), precision, geoBoundingBox,
             requiredSize, shardSize, searchContext, parent, collectsFromSingleBucket, metadata);
     }
 

server/src/main/java/org/elasticsearch/search/aggregations/bucket/geogrid/GeoTileGridAggregatorFactory.java

@@ -72,18 +72,17 @@ public class GeoTileGridAggregatorFactory extends ValuesSourceAggregatorFactory
     }
 
     @Override
-    protected Aggregator doCreateInternal(final ValuesSource valuesSource,
-                                            SearchContext searchContext,
-                                            Aggregator parent,
-                                            boolean collectsFromSingleBucket,
-                                            Map<String, Object> metadata) throws IOException {
+    protected Aggregator doCreateInternal(SearchContext searchContext,
+                                          Aggregator parent,
+                                          boolean collectsFromSingleBucket,
+                                          Map<String, Object> metadata) throws IOException {
         AggregatorSupplier aggregatorSupplier = queryShardContext.getValuesSourceRegistry()
             .getAggregator(config, GeoTileGridAggregationBuilder.NAME);
         if (aggregatorSupplier instanceof GeoGridAggregatorSupplier == false) {
             throw new AggregationExecutionException("Registry miss-match - expected "
                 + GeoGridAggregatorSupplier.class.getName() + ", found [" + aggregatorSupplier.getClass().toString() + "]");
         }
-        return ((GeoGridAggregatorSupplier) aggregatorSupplier).build(name, factories, valuesSource, precision, geoBoundingBox,
+        return ((GeoGridAggregatorSupplier) aggregatorSupplier).build(name, factories, config.getValuesSource(), precision, geoBoundingBox,
             requiredSize, shardSize, searchContext, parent, collectsFromSingleBucket, metadata);
     }
 

server/src/main/java/org/elasticsearch/search/aggregations/bucket/histogram/AutoDateHistogramAggregator.java

@@ -22,7 +22,6 @@ import org.apache.lucene.index.LeafReaderContext;
 import org.apache.lucene.index.SortedNumericDocValues;
 import org.apache.lucene.search.ScoreMode;
 import org.apache.lucene.util.CollectionUtil;
-import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.Rounding;
 import org.elasticsearch.common.lease.Releasables;
 import org.elasticsearch.common.util.LongHash;
@@ -38,6 +37,7 @@ import org.elasticsearch.search.aggregations.bucket.DeferringBucketCollector;
 import org.elasticsearch.search.aggregations.bucket.MergingBucketsDeferringCollector;
 import org.elasticsearch.search.aggregations.bucket.histogram.AutoDateHistogramAggregationBuilder.RoundingInfo;
 import org.elasticsearch.search.aggregations.support.ValuesSource;
+import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
 import org.elasticsearch.search.internal.SearchContext;
 
 import java.io.IOException;
@@ -48,7 +48,7 @@ import java.util.function.Function;
 /**
  * An aggregator for date values that attempts to return a specific number of
  * buckets, reconfiguring how it rounds dates to buckets on the fly as new
- * data arrives. 
+ * data arrives.
  */
 class AutoDateHistogramAggregator extends DeferableBucketAggregator {
 
@@ -63,14 +63,23 @@ class AutoDateHistogramAggregator extends DeferableBucketAggregator {
     private int targetBuckets;
     private MergingBucketsDeferringCollector deferringCollector;
 
-    AutoDateHistogramAggregator(String name, AggregatorFactories factories, int numBuckets, RoundingInfo[] roundingInfos,
-        Function<Rounding, Rounding.Prepared> roundingPreparer, @Nullable ValuesSource valuesSource, DocValueFormat formatter,
-        SearchContext aggregationContext, Aggregator parent, Map<String, Object> metadata) throws IOException {
+    AutoDateHistogramAggregator(
+        String name,
+        AggregatorFactories factories,
+        int numBuckets,
+        RoundingInfo[] roundingInfos,
+        Function<Rounding, Rounding.Prepared> roundingPreparer,
+        ValuesSourceConfig valuesSourceConfig,
+        SearchContext aggregationContext,
+        Aggregator parent,
+        Map<String, Object> metadata
+    ) throws IOException {
 
         super(name, factories, aggregationContext, parent, metadata);
         this.targetBuckets = numBuckets;
-        this.valuesSource = (ValuesSource.Numeric) valuesSource;
-        this.formatter = formatter;
+        // TODO: Remove null usage here, by using a different aggregator for create
+        this.valuesSource = valuesSourceConfig.hasValues() ? (ValuesSource.Numeric) valuesSourceConfig.getValuesSource() : null;
+        this.formatter = valuesSourceConfig.format();
         this.roundingInfos = roundingInfos;
         this.roundingPreparer = roundingPreparer;
         preparedRounding = roundingPreparer.apply(roundingInfos[roundingIdx].rounding);

server/src/main/java/org/elasticsearch/search/aggregations/bucket/histogram/AutoDateHistogramAggregatorFactory.java

@@ -28,7 +28,6 @@ import org.elasticsearch.search.aggregations.AggregatorFactory;
 import org.elasticsearch.search.aggregations.bucket.histogram.AutoDateHistogramAggregationBuilder.RoundingInfo;
 import org.elasticsearch.search.aggregations.support.AggregatorSupplier;
 import org.elasticsearch.search.aggregations.support.CoreValuesSourceType;
-import org.elasticsearch.search.aggregations.support.ValuesSource;
 import org.elasticsearch.search.aggregations.support.ValuesSourceAggregatorFactory;
 import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
 import org.elasticsearch.search.aggregations.support.ValuesSourceRegistry;
@@ -64,11 +63,10 @@ public final class AutoDateHistogramAggregatorFactory extends ValuesSourceAggreg
     }
 
     @Override
-    protected Aggregator doCreateInternal(ValuesSource valuesSource,
-                                            SearchContext searchContext,
-                                            Aggregator parent,
-                                            boolean collectsFromSingleBucket,
-                                            Map<String, Object> metadata) throws IOException {
+    protected Aggregator doCreateInternal(SearchContext searchContext,
+                                          Aggregator parent,
+                                          boolean collectsFromSingleBucket,
+                                          Map<String, Object> metadata) throws IOException {
         if (collectsFromSingleBucket == false) {
             return asMultiBucketAggregator(this, searchContext, parent);
         }
@@ -79,16 +77,16 @@ public final class AutoDateHistogramAggregatorFactory extends ValuesSourceAggreg
                 aggregatorSupplier.getClass().toString() + "]");
         }
         Function<Rounding, Rounding.Prepared> roundingPreparer =
-                valuesSource.roundingPreparer(searchContext.getQueryShardContext().getIndexReader());
+                config.getValuesSource().roundingPreparer(searchContext.getQueryShardContext().getIndexReader());
         return ((AutoDateHistogramAggregatorSupplier) aggregatorSupplier).build(name, factories, numBuckets, roundingInfos,
-                roundingPreparer, valuesSource, config.format(), searchContext, parent, metadata);
+                roundingPreparer, config, searchContext, parent, metadata);
     }
 
     @Override
     protected Aggregator createUnmapped(SearchContext searchContext,
                                             Aggregator parent,
                                             Map<String, Object> metadata) throws IOException {
-        return new AutoDateHistogramAggregator(name, factories, numBuckets, roundingInfos, Rounding::prepareForUnknown, null,
-                config.format(), searchContext, parent, metadata);
+        return new AutoDateHistogramAggregator(name, factories, numBuckets, roundingInfos, Rounding::prepareForUnknown,
+                config, searchContext, parent, metadata);
     }
 }

server/src/main/java/org/elasticsearch/search/aggregations/bucket/histogram/AutoDateHistogramAggregatorSupplier.java

@@ -21,11 +21,10 @@ package org.elasticsearch.search.aggregations.bucket.histogram;
 
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.Rounding;
-import org.elasticsearch.search.DocValueFormat;
 import org.elasticsearch.search.aggregations.Aggregator;
 import org.elasticsearch.search.aggregations.AggregatorFactories;
 import org.elasticsearch.search.aggregations.support.AggregatorSupplier;
-import org.elasticsearch.search.aggregations.support.ValuesSource;
+import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
 import org.elasticsearch.search.internal.SearchContext;
 
 import java.io.IOException;
@@ -41,8 +40,7 @@ public interface AutoDateHistogramAggregatorSupplier extends AggregatorSupplier
         AutoDateHistogramAggregationBuilder.RoundingInfo[] roundingInfos,
         @Nullable
         Function<Rounding, Rounding.Prepared> roundingPreparer,
-        @Nullable ValuesSource valuesSource,
-        DocValueFormat formatter,
+        ValuesSourceConfig valuesSourceConfig,
         SearchContext aggregationContext,
         Aggregator parent,
         Map<String, Object> metadata

server/src/main/java/org/elasticsearch/search/aggregations/bucket/histogram/DateHistogramAggregationSupplier.java

@@ -21,12 +21,11 @@ package org.elasticsearch.search.aggregations.bucket.histogram;
 
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.Rounding;
-import org.elasticsearch.search.DocValueFormat;
 import org.elasticsearch.search.aggregations.Aggregator;
 import org.elasticsearch.search.aggregations.AggregatorFactories;
 import org.elasticsearch.search.aggregations.BucketOrder;
 import org.elasticsearch.search.aggregations.support.AggregatorSupplier;
-import org.elasticsearch.search.aggregations.support.ValuesSource;
+import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
 import org.elasticsearch.search.internal.SearchContext;
 
 import java.io.IOException;
@@ -42,8 +41,7 @@ public interface DateHistogramAggregationSupplier extends AggregatorSupplier {
                      boolean keyed,
                      long minDocCount,
                      @Nullable ExtendedBounds extendedBounds,
-                     @Nullable ValuesSource valuesSource,
-                     DocValueFormat formatter,
+                     ValuesSourceConfig valuesSourceConfig,
                      SearchContext aggregationContext,
                      Aggregator parent,
                      boolean collectsFromSingleBucket,

server/src/main/java/org/elasticsearch/search/aggregations/bucket/histogram/DateHistogramAggregator.java

@@ -35,6 +35,7 @@ import org.elasticsearch.search.aggregations.LeafBucketCollectorBase;
 import org.elasticsearch.search.aggregations.bucket.BucketsAggregator;
 import org.elasticsearch.search.aggregations.bucket.terms.LongKeyedBucketOrds;
 import org.elasticsearch.search.aggregations.support.ValuesSource;
+import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
 import org.elasticsearch.search.internal.SearchContext;
 
 import java.io.IOException;
@@ -65,11 +66,21 @@ class DateHistogramAggregator extends BucketsAggregator {
 
     private final LongKeyedBucketOrds bucketOrds;
 
-    DateHistogramAggregator(String name, AggregatorFactories factories, Rounding rounding, Rounding.Prepared preparedRounding,
-            BucketOrder order, boolean keyed,
-            long minDocCount, @Nullable ExtendedBounds extendedBounds, @Nullable ValuesSource valuesSource,
-            DocValueFormat formatter, SearchContext aggregationContext,
-            Aggregator parent, boolean collectsFromSingleBucket, Map<String, Object> metadata) throws IOException {
+    DateHistogramAggregator(
+        String name,
+        AggregatorFactories factories,
+        Rounding rounding,
+        Rounding.Prepared preparedRounding,
+        BucketOrder order,
+        boolean keyed,
+        long minDocCount,
+        @Nullable ExtendedBounds extendedBounds,
+        ValuesSourceConfig valuesSourceConfig,
+        SearchContext aggregationContext,
+        Aggregator parent,
+        boolean collectsFromSingleBucket,
+        Map<String, Object> metadata
+    ) throws IOException {
 
         super(name, factories, aggregationContext, parent, metadata);
         this.rounding = rounding;
@@ -79,8 +90,9 @@ class DateHistogramAggregator extends BucketsAggregator {
         this.keyed = keyed;
         this.minDocCount = minDocCount;
         this.extendedBounds = extendedBounds;
-        this.valuesSource = (ValuesSource.Numeric) valuesSource;
-        this.formatter = formatter;
+        // TODO: Stop using null here
+        this.valuesSource = valuesSourceConfig.hasValues() ? (ValuesSource.Numeric) valuesSourceConfig.getValuesSource() : null;
+        this.formatter = valuesSourceConfig.format();
 
         bucketOrds = LongKeyedBucketOrds.build(context.bigArrays(), collectsFromSingleBucket);
     }
@@ -129,7 +141,7 @@ class DateHistogramAggregator extends BucketsAggregator {
 
     @Override
     public InternalAggregation[] buildAggregations(long[] owningBucketOrds) throws IOException {
-        return buildAggregationsForVariableBuckets(owningBucketOrds, bucketOrds, 
+        return buildAggregationsForVariableBuckets(owningBucketOrds, bucketOrds,
             (bucketValue, docCount, subAggregationResults) -> {
                 return new InternalDateHistogram.Bucket(bucketValue, docCount, keyed, formatter, subAggregationResults);
             }, buckets -> {

server/src/main/java/org/elasticsearch/search/aggregations/bucket/histogram/DateHistogramAggregatorFactory.java

@@ -29,7 +29,6 @@ import org.elasticsearch.search.aggregations.AggregatorFactory;
 import org.elasticsearch.search.aggregations.BucketOrder;
 import org.elasticsearch.search.aggregations.support.AggregatorSupplier;
 import org.elasticsearch.search.aggregations.support.CoreValuesSourceType;
-import org.elasticsearch.search.aggregations.support.ValuesSource;
 import org.elasticsearch.search.aggregations.support.ValuesSourceAggregatorFactory;
 import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
 import org.elasticsearch.search.aggregations.support.ValuesSourceRegistry;
@@ -76,18 +75,21 @@ public final class DateHistogramAggregatorFactory extends ValuesSourceAggregator
     }
 
     @Override
-    protected Aggregator doCreateInternal(ValuesSource valuesSource,
-                                            SearchContext searchContext,
-                                            Aggregator parent,
-                                            boolean collectsFromSingleBucket,
-                                            Map<String, Object> metadata) throws IOException {
+    protected Aggregator doCreateInternal(
+        SearchContext searchContext,
+        Aggregator parent,
+        boolean collectsFromSingleBucket,
+        Map<String, Object> metadata) throws IOException {
         AggregatorSupplier aggregatorSupplier = queryShardContext.getValuesSourceRegistry().getAggregator(config,
             DateHistogramAggregationBuilder.NAME);
         if (aggregatorSupplier instanceof DateHistogramAggregationSupplier == false) {
             throw new AggregationExecutionException("Registry miss-match - expected DateHistogramAggregationSupplier, found [" +
                 aggregatorSupplier.getClass().toString() + "]");
         }
-        Rounding.Prepared preparedRounding = valuesSource.roundingPreparer(queryShardContext.getIndexReader()).apply(shardRounding);
+        // TODO: Is there a reason not to get the prepared rounding in the supplier itself?
+        Rounding.Prepared preparedRounding = config.getValuesSource()
+            .roundingPreparer(queryShardContext.getIndexReader())
+            .apply(shardRounding);
         return ((DateHistogramAggregationSupplier) aggregatorSupplier).build(
             name,
             factories,
@@ -97,8 +99,7 @@ public final class DateHistogramAggregatorFactory extends ValuesSourceAggregator
             keyed,
             minDocCount,
             extendedBounds,
-            valuesSource,
-            config.format(),
+            config,
             searchContext,
             parent,
             collectsFromSingleBucket,
@@ -111,6 +112,6 @@ public final class DateHistogramAggregatorFactory extends ValuesSourceAggregator
                                             Aggregator parent,
                                             Map<String, Object> metadata) throws IOException {
         return new DateHistogramAggregator(name, factories, rounding, null, order, keyed, minDocCount, extendedBounds,
-            null, config.format(), searchContext, parent, false, metadata);
+            config, searchContext, parent, false, metadata);
     }
 }

server/src/main/java/org/elasticsearch/search/aggregations/bucket/histogram/DateRangeHistogramAggregator.java

@@ -38,6 +38,7 @@ import org.elasticsearch.search.aggregations.LeafBucketCollectorBase;
 import org.elasticsearch.search.aggregations.bucket.BucketsAggregator;
 import org.elasticsearch.search.aggregations.bucket.terms.LongKeyedBucketOrds;
 import org.elasticsearch.search.aggregations.support.ValuesSource;
+import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
 import org.elasticsearch.search.internal.SearchContext;
 
 import java.io.IOException;
@@ -69,11 +70,21 @@ class DateRangeHistogramAggregator extends BucketsAggregator {
 
     private final LongKeyedBucketOrds bucketOrds;
 
-    DateRangeHistogramAggregator(String name, AggregatorFactories factories, Rounding rounding, Rounding.Prepared preparedRounding,
-                                 BucketOrder order, boolean keyed,
-                                 long minDocCount, @Nullable ExtendedBounds extendedBounds, @Nullable ValuesSource valuesSource,
-                                 DocValueFormat formatter, SearchContext aggregationContext,
-                                 Aggregator parent, boolean collectsFromSingleBucket, Map<String, Object> metadata) throws IOException {
+    DateRangeHistogramAggregator(
+        String name,
+        AggregatorFactories factories,
+        Rounding rounding,
+        Rounding.Prepared preparedRounding,
+        BucketOrder order,
+        boolean keyed,
+        long minDocCount,
+        @Nullable ExtendedBounds extendedBounds,
+        ValuesSourceConfig valuesSourceConfig,
+        SearchContext aggregationContext,
+        Aggregator parent,
+        boolean collectsFromSingleBucket,
+        Map<String, Object> metadata
+    ) throws IOException {
 
         super(name, factories, aggregationContext, parent, metadata);
         this.rounding = rounding;
@@ -83,8 +94,9 @@ class DateRangeHistogramAggregator extends BucketsAggregator {
         this.keyed = keyed;
         this.minDocCount = minDocCount;
         this.extendedBounds = extendedBounds;
-        this.valuesSource = (ValuesSource.Range) valuesSource;
-        this.formatter = formatter;
+        // TODO: Stop using null here
+        this.valuesSource = valuesSourceConfig.hasValues() ? (ValuesSource.Range) valuesSourceConfig.getValuesSource() : null;
+        this.formatter = valuesSourceConfig.format();
         if (this.valuesSource.rangeType() != RangeType.DATE) {
             throw new IllegalArgumentException("Expected date range type but found range type [" + this.valuesSource.rangeType().name
                 + "]");

server/src/main/java/org/elasticsearch/search/aggregations/bucket/histogram/HistogramAggregatorFactory.java

@@ -29,7 +29,6 @@ import org.elasticsearch.search.aggregations.BucketOrder;
 import org.elasticsearch.search.aggregations.support.AggregatorSupplier;
 import org.elasticsearch.search.aggregations.support.CoreValuesSourceType;
 import org.elasticsearch.search.aggregations.support.HistogramAggregatorSupplier;
-import org.elasticsearch.search.aggregations.support.ValuesSource;
 import org.elasticsearch.search.aggregations.support.ValuesSourceAggregatorFactory;
 import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
 import org.elasticsearch.search.aggregations.support.ValuesSourceRegistry;
@@ -87,11 +86,10 @@ public final class HistogramAggregatorFactory extends ValuesSourceAggregatorFact
     }
 
     @Override
-    protected Aggregator doCreateInternal(ValuesSource valuesSource,
-                                            SearchContext searchContext,
-                                            Aggregator parent,
-                                            boolean collectsFromSingleBucket,
-                                            Map<String, Object> metadata) throws IOException {
+    protected Aggregator doCreateInternal(SearchContext searchContext,
+                                          Aggregator parent,
+                                          boolean collectsFromSingleBucket,
+                                          Map<String, Object> metadata) throws IOException {
         AggregatorSupplier aggregatorSupplier = queryShardContext.getValuesSourceRegistry().getAggregator(config,
             HistogramAggregationBuilder.NAME);
         if (aggregatorSupplier instanceof HistogramAggregatorSupplier == false) {
@@ -100,7 +98,7 @@ public final class HistogramAggregatorFactory extends ValuesSourceAggregatorFact
         }
         HistogramAggregatorSupplier histogramAggregatorSupplier = (HistogramAggregatorSupplier) aggregatorSupplier;
         return histogramAggregatorSupplier.build(name, factories, interval, offset, order, keyed, minDocCount, minBound, maxBound,
-                valuesSource, config.format(), searchContext, parent, collectsFromSingleBucket, metadata);
+            config, searchContext, parent, collectsFromSingleBucket, metadata);
     }
 
     @Override
@@ -108,6 +106,6 @@ public final class HistogramAggregatorFactory extends ValuesSourceAggregatorFact
                                             Aggregator parent,
                                             Map<String, Object> metadata) throws IOException {
         return new NumericHistogramAggregator(name, factories, interval, offset, order, keyed, minDocCount, minBound, maxBound,
-            null, config.format(), searchContext, parent, false, metadata);
+            config, searchContext, parent, false, metadata);
     }
 }

server/src/main/java/org/elasticsearch/search/aggregations/bucket/histogram/NumericHistogramAggregator.java

@@ -21,15 +21,14 @@ package org.elasticsearch.search.aggregations.bucket.histogram;
 
 import org.apache.lucene.index.LeafReaderContext;
 import org.apache.lucene.search.ScoreMode;
-import org.elasticsearch.common.Nullable;
 import org.elasticsearch.index.fielddata.SortedNumericDoubleValues;
-import org.elasticsearch.search.DocValueFormat;
 import org.elasticsearch.search.aggregations.Aggregator;
 import org.elasticsearch.search.aggregations.AggregatorFactories;
 import org.elasticsearch.search.aggregations.BucketOrder;
 import org.elasticsearch.search.aggregations.LeafBucketCollector;
 import org.elasticsearch.search.aggregations.LeafBucketCollectorBase;
 import org.elasticsearch.search.aggregations.support.ValuesSource;
+import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
 import org.elasticsearch.search.internal.SearchContext;
 
 import java.io.IOException;
@@ -54,8 +53,7 @@ public class NumericHistogramAggregator extends AbstractHistogramAggregator {
         long minDocCount,
         double minBound,
         double maxBound,
-        @Nullable ValuesSource valuesSource,
-        DocValueFormat formatter,
+        ValuesSourceConfig valuesSourceConfig,
         SearchContext context,
         Aggregator parent,
         boolean collectsFromSingleBucket,
@@ -71,13 +69,14 @@ public class NumericHistogramAggregator extends AbstractHistogramAggregator {
             minDocCount,
             minBound,
             maxBound,
-            formatter,
+            valuesSourceConfig.format(),
             context,
             parent,
             collectsFromSingleBucket,
             metadata
         );
-        this.valuesSource = (ValuesSource.Numeric) valuesSource;
+        // TODO: Stop using null here
+        this.valuesSource = valuesSourceConfig.hasValues() ? (ValuesSource.Numeric) valuesSourceConfig.getValuesSource() : null;
     }
 
     @Override

server/src/main/java/org/elasticsearch/search/aggregations/bucket/histogram/RangeHistogramAggregator.java

@@ -21,17 +21,16 @@ package org.elasticsearch.search.aggregations.bucket.histogram;
 
 import org.apache.lucene.index.LeafReaderContext;
 import org.apache.lucene.util.BytesRef;
-import org.elasticsearch.common.Nullable;
 import org.elasticsearch.index.fielddata.SortedBinaryDocValues;
 import org.elasticsearch.index.mapper.RangeFieldMapper;
 import org.elasticsearch.index.mapper.RangeType;
-import org.elasticsearch.search.DocValueFormat;
 import org.elasticsearch.search.aggregations.Aggregator;
 import org.elasticsearch.search.aggregations.AggregatorFactories;
 import org.elasticsearch.search.aggregations.BucketOrder;
 import org.elasticsearch.search.aggregations.LeafBucketCollector;
 import org.elasticsearch.search.aggregations.LeafBucketCollectorBase;
 import org.elasticsearch.search.aggregations.support.ValuesSource;
+import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
 import org.elasticsearch.search.internal.SearchContext;
 
 import java.io.IOException;
@@ -51,8 +50,7 @@ public class RangeHistogramAggregator extends AbstractHistogramAggregator {
         long minDocCount,
         double minBound,
         double maxBound,
-        @Nullable ValuesSource valuesSource,
-        DocValueFormat formatter,
+        ValuesSourceConfig valuesSourceConfig,
         SearchContext context,
         Aggregator parent,
         boolean collectsFromSingleBucket,
@@ -68,13 +66,14 @@ public class RangeHistogramAggregator extends AbstractHistogramAggregator {
             minDocCount,
             minBound,
             maxBound,
-            formatter,
+            valuesSourceConfig.format(),
             context,
             parent,
             collectsFromSingleBucket,
             metadata
         );
-        this.valuesSource = (ValuesSource.Range) valuesSource;
+        // TODO: Stop using nulls here
+        this.valuesSource = valuesSourceConfig.hasValues() ? (ValuesSource.Range) valuesSourceConfig.getValuesSource() : null;
         if (this.valuesSource.rangeType().isNumeric() == false) {
             throw new IllegalArgumentException(
                 "Expected numeric range type but found non-numeric range [" + this.valuesSource.rangeType().name + "]"

server/src/main/java/org/elasticsearch/search/aggregations/bucket/missing/MissingAggregator.java

@@ -28,6 +28,7 @@ import org.elasticsearch.search.aggregations.LeafBucketCollectorBase;
 import org.elasticsearch.search.aggregations.bucket.BucketsAggregator;
 import org.elasticsearch.search.aggregations.bucket.SingleBucketAggregator;
 import org.elasticsearch.search.aggregations.support.ValuesSource;
+import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
 import org.elasticsearch.search.internal.SearchContext;
 
 import java.io.IOException;
@@ -40,12 +41,13 @@ public class MissingAggregator extends BucketsAggregator implements SingleBucket
     public MissingAggregator(
             String name,
             AggregatorFactories factories,
-            ValuesSource valuesSource,
+            ValuesSourceConfig valuesSourceConfig,
             SearchContext aggregationContext,
             Aggregator parent,
             Map<String, Object> metadata) throws IOException {
         super(name, factories, aggregationContext, parent, metadata);
-        this.valuesSource = valuesSource;
+        // TODO: Stop using nulls here
+        this.valuesSource = valuesSourceConfig.hasValues() ? valuesSourceConfig.getValuesSource() : null;
     }
 
     @Override

server/src/main/java/org/elasticsearch/search/aggregations/bucket/missing/MissingAggregatorFactory.java

@@ -26,7 +26,6 @@ import org.elasticsearch.search.aggregations.AggregatorFactories;
 import org.elasticsearch.search.aggregations.AggregatorFactory;
 import org.elasticsearch.search.aggregations.support.AggregatorSupplier;
 import org.elasticsearch.search.aggregations.support.CoreValuesSourceType;
-import org.elasticsearch.search.aggregations.support.ValuesSource;
 import org.elasticsearch.search.aggregations.support.ValuesSourceAggregatorFactory;
 import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
 import org.elasticsearch.search.aggregations.support.ValuesSourceRegistry;
@@ -52,15 +51,14 @@ public class MissingAggregatorFactory extends ValuesSourceAggregatorFactory {
     protected MissingAggregator createUnmapped(SearchContext searchContext,
                                                 Aggregator parent,
                                                 Map<String, Object> metadata) throws IOException {
-        return new MissingAggregator(name, factories, null, searchContext, parent, metadata);
+        return new MissingAggregator(name, factories, config, searchContext, parent, metadata);
     }
 
     @Override
-    protected MissingAggregator doCreateInternal(ValuesSource valuesSource,
-                                                    SearchContext searchContext,
-                                                    Aggregator parent,
-                                                    boolean collectsFromSingleBucket,
-                                                    Map<String, Object> metadata) throws IOException {
+    protected MissingAggregator doCreateInternal(SearchContext searchContext,
+                                          Aggregator parent,
+                                          boolean collectsFromSingleBucket,
+                                          Map<String, Object> metadata) throws IOException {
         final AggregatorSupplier aggregatorSupplier = queryShardContext.getValuesSourceRegistry()
             .getAggregator(config, MissingAggregationBuilder.NAME);
         if (aggregatorSupplier instanceof MissingAggregatorSupplier == false) {
@@ -68,8 +66,8 @@ public class MissingAggregatorFactory extends ValuesSourceAggregatorFactory {
                 aggregatorSupplier.getClass().toString() + "]");
         }
 
-        return (MissingAggregator) ((MissingAggregatorSupplier) aggregatorSupplier)
-            .build(name, factories, valuesSource, searchContext, parent, metadata);
+        return (MissingAggregator) ((MissingAggregatorSupplier) aggregatorSupplier).build(name, factories, config, searchContext, parent,
+            metadata);
     }
 
 }

server/src/main/java/org/elasticsearch/search/aggregations/bucket/missing/MissingAggregatorSupplier.java

@@ -19,11 +19,10 @@
 
 package org.elasticsearch.search.aggregations.bucket.missing;
 
-import org.elasticsearch.common.Nullable;
 import org.elasticsearch.search.aggregations.Aggregator;
 import org.elasticsearch.search.aggregations.AggregatorFactories;
 import org.elasticsearch.search.aggregations.support.AggregatorSupplier;
-import org.elasticsearch.search.aggregations.support.ValuesSource;
+import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
 import org.elasticsearch.search.internal.SearchContext;
 
 import java.io.IOException;
@@ -34,7 +33,7 @@ public interface MissingAggregatorSupplier extends AggregatorSupplier {
 
     Aggregator build(String name,
                      AggregatorFactories factories,
-                     @Nullable ValuesSource valuesSource,
+                     ValuesSourceConfig valuesSourceConfig,
                      SearchContext aggregationContext,
                      Aggregator parent,
                      Map<String, Object> metadata) throws IOException;

server/src/main/java/org/elasticsearch/search/aggregations/bucket/range/AbstractRangeAggregatorFactory.java

@@ -28,7 +28,6 @@ import org.elasticsearch.search.aggregations.bucket.range.RangeAggregator.Range;
 import org.elasticsearch.search.aggregations.bucket.range.RangeAggregator.Unmapped;
 import org.elasticsearch.search.aggregations.support.AggregatorSupplier;
 import org.elasticsearch.search.aggregations.support.CoreValuesSourceType;
-import org.elasticsearch.search.aggregations.support.ValuesSource;
 import org.elasticsearch.search.aggregations.support.ValuesSource.Numeric;
 import org.elasticsearch.search.aggregations.support.ValuesSourceAggregatorFactory;
 import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
@@ -77,11 +76,10 @@ public class AbstractRangeAggregatorFactory<R extends Range> extends ValuesSourc
     }
 
     @Override
-    protected Aggregator doCreateInternal(ValuesSource valuesSource,
-                                            SearchContext searchContext,
-                                            Aggregator parent,
-                                            boolean collectsFromSingleBucket,
-                                            Map<String, Object> metadata) throws IOException {
+    protected Aggregator doCreateInternal(SearchContext searchContext,
+                                          Aggregator parent,
+                                          boolean collectsFromSingleBucket,
+                                          Map<String, Object> metadata) throws IOException {
 
         AggregatorSupplier aggregatorSupplier = queryShardContext.getValuesSourceRegistry().getAggregator(config,
             aggregationTypeName);
@@ -89,7 +87,17 @@ public class AbstractRangeAggregatorFactory<R extends Range> extends ValuesSourc
             throw new AggregationExecutionException("Registry miss-match - expected RangeAggregatorSupplier, found [" +
                 aggregatorSupplier.getClass().toString() + "]");
         }
-        return ((RangeAggregatorSupplier)aggregatorSupplier).build(name, factories, (Numeric) valuesSource, config.format(), rangeFactory,
-            ranges, keyed, searchContext, parent, metadata);
+        return ((RangeAggregatorSupplier) aggregatorSupplier).build(
+            name,
+            factories,
+            (Numeric) config.getValuesSource(),
+            config.format(),
+            rangeFactory,
+            ranges,
+            keyed,
+            searchContext,
+            parent,
+            metadata
+        );
     }
 }

server/src/main/java/org/elasticsearch/search/aggregations/bucket/range/BinaryRangeAggregatorFactory.java

@@ -25,7 +25,6 @@ import org.elasticsearch.search.aggregations.AggregatorFactories.Builder;
 import org.elasticsearch.search.aggregations.AggregatorFactory;
 import org.elasticsearch.search.aggregations.support.AggregatorSupplier;
 import org.elasticsearch.search.aggregations.support.CoreValuesSourceType;
-import org.elasticsearch.search.aggregations.support.ValuesSource;
 import org.elasticsearch.search.aggregations.support.ValuesSourceAggregatorFactory;
 import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
 import org.elasticsearch.search.aggregations.support.ValuesSourceRegistry;
@@ -62,8 +61,7 @@ public class BinaryRangeAggregatorFactory extends ValuesSourceAggregatorFactory
     }
 
     @Override
-    protected Aggregator doCreateInternal(ValuesSource valuesSource,
-                                          SearchContext searchContext, Aggregator parent,
+    protected Aggregator doCreateInternal(SearchContext searchContext, Aggregator parent,
                                           boolean collectsFromSingleBucket,
                                           Map<String, Object> metadata) throws IOException {
         AggregatorSupplier aggregatorSupplier = queryShardContext.getValuesSourceRegistry().getAggregator(config,
@@ -73,7 +71,7 @@ public class BinaryRangeAggregatorFactory extends ValuesSourceAggregatorFactory
             throw new AggregationExecutionException("Registry miss-match - expected IpRangeAggregatorSupplier, found [" +
                 aggregatorSupplier.getClass().toString() + "]");
         }
-        return ((IpRangeAggregatorSupplier) aggregatorSupplier).build(name, factories, valuesSource, config.format(),
+        return ((IpRangeAggregatorSupplier) aggregatorSupplier).build(name, factories, config.getValuesSource(), config.format(),
                 ranges, keyed, searchContext, parent, metadata);
     }
 

server/src/main/java/org/elasticsearch/search/aggregations/bucket/range/GeoDistanceRangeAggregatorFactory.java

@@ -85,11 +85,10 @@ public class GeoDistanceRangeAggregatorFactory extends ValuesSourceAggregatorFac
     }
 
     @Override
-    protected Aggregator doCreateInternal(final ValuesSource valuesSource,
-                                            SearchContext searchContext,
-                                            Aggregator parent,
-                                            boolean collectsFromSingleBucket,
-                                            Map<String, Object> metadata) throws IOException {
+    protected Aggregator doCreateInternal(SearchContext searchContext,
+                                          Aggregator parent,
+                                          boolean collectsFromSingleBucket,
+                                          Map<String, Object> metadata) throws IOException {
         AggregatorSupplier aggregatorSupplier = queryShardContext.getValuesSourceRegistry()
             .getAggregator(config, GeoDistanceAggregationBuilder.NAME);
         if (aggregatorSupplier instanceof GeoDistanceAggregatorSupplier == false) {
@@ -97,7 +96,7 @@ public class GeoDistanceRangeAggregatorFactory extends ValuesSourceAggregatorFac
                 + GeoDistanceAggregatorSupplier.class.getName() + ", found [" + aggregatorSupplier.getClass().toString() + "]");
         }
         return ((GeoDistanceAggregatorSupplier) aggregatorSupplier).build(name,  factories, distanceType,  origin,
-            unit, config.toValuesSource(), config.format(), rangeFactory, ranges, keyed, searchContext, parent, metadata);
+            unit, config.getValuesSource(), config.format(), rangeFactory, ranges, keyed, searchContext, parent, metadata);
     }
 
     private static class DistanceSource extends ValuesSource.Numeric {

server/src/main/java/org/elasticsearch/search/aggregations/bucket/sampler/DiversifiedAggregatorFactory.java

@@ -29,7 +29,6 @@ import org.elasticsearch.search.aggregations.NonCollectingAggregator;
 import org.elasticsearch.search.aggregations.bucket.sampler.SamplerAggregator.ExecutionMode;
 import org.elasticsearch.search.aggregations.support.AggregatorSupplier;
 import org.elasticsearch.search.aggregations.support.CoreValuesSourceType;
-import org.elasticsearch.search.aggregations.support.ValuesSource;
 import org.elasticsearch.search.aggregations.support.ValuesSourceAggregatorFactory;
 import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
 import org.elasticsearch.search.aggregations.support.ValuesSourceRegistry;
@@ -47,15 +46,15 @@ public class DiversifiedAggregatorFactory extends ValuesSourceAggregatorFactory
                 CoreValuesSourceType.DATE,
                 CoreValuesSourceType.BOOLEAN),
             (DiversifiedAggregatorSupplier) (String name, int shardSize, AggregatorFactories factories, SearchContext context,
-                                             Aggregator parent, Map<String, Object> metadata, ValuesSource valuesSource,
+                                             Aggregator parent, Map<String, Object> metadata, ValuesSourceConfig valuesSourceConfig,
                                              int maxDocsPerValue, String executionHint) ->
-                new DiversifiedNumericSamplerAggregator(name, shardSize, factories, context, parent, metadata, valuesSource,
+                new DiversifiedNumericSamplerAggregator(name, shardSize, factories, context, parent, metadata, valuesSourceConfig,
                     maxDocsPerValue)
         );
 
         builder.register(DiversifiedAggregationBuilder.NAME, CoreValuesSourceType.BYTES,
             (DiversifiedAggregatorSupplier) (String name, int shardSize, AggregatorFactories factories, SearchContext context,
-                                             Aggregator parent, Map<String, Object> metadata, ValuesSource valuesSource,
+                                             Aggregator parent, Map<String, Object> metadata, ValuesSourceConfig valuesSourceConfig,
                                              int maxDocsPerValue, String executionHint) -> {
                 ExecutionMode execution = null;
                 if (executionHint != null) {
@@ -66,10 +65,10 @@ public class DiversifiedAggregatorFactory extends ValuesSourceAggregatorFactory
                 if (execution == null) {
                     execution = ExecutionMode.GLOBAL_ORDINALS;
                 }
-                if ((execution.needsGlobalOrdinals()) && (!(valuesSource instanceof ValuesSource.Bytes.WithOrdinals))) {
+                if ((execution.needsGlobalOrdinals()) && (valuesSourceConfig.hasGlobalOrdinals() == false)) {
                     execution = ExecutionMode.MAP;
                 }
-                return execution.create(name, factories, shardSize, maxDocsPerValue, valuesSource, context, parent, metadata);
+                return execution.create(name, factories, shardSize, maxDocsPerValue, valuesSourceConfig, context, parent, metadata);
         });
 
     }
@@ -88,20 +87,18 @@ public class DiversifiedAggregatorFactory extends ValuesSourceAggregatorFactory
     }
 
     @Override
-    protected Aggregator doCreateInternal(ValuesSource valuesSource,
-                                            SearchContext searchContext,
-                                            Aggregator parent,
-                                            boolean collectsFromSingleBucket,
-                                            Map<String, Object> metadata) throws IOException {
+    protected Aggregator doCreateInternal(SearchContext searchContext,
+                                          Aggregator parent,
+                                          boolean collectsFromSingleBucket,
+                                          Map<String, Object> metadata) throws IOException {
 
-        AggregatorSupplier supplier = queryShardContext.getValuesSourceRegistry().getAggregator(config,
-            DiversifiedAggregationBuilder.NAME);
+        AggregatorSupplier supplier = queryShardContext.getValuesSourceRegistry().getAggregator(config, DiversifiedAggregationBuilder.NAME);
         if (supplier instanceof DiversifiedAggregatorSupplier == false) {
             throw new AggregationExecutionException("Registry miss-match - expected " + DiversifiedAggregatorSupplier.class.toString() +
                 ", found [" + supplier.getClass().toString() + "]");
         }
-        return ((DiversifiedAggregatorSupplier) supplier).build(name, shardSize, factories, searchContext, parent, metadata, valuesSource,
-            maxDocsPerValue, executionHint);
+        return ((DiversifiedAggregatorSupplier) supplier).build(name, shardSize, factories, searchContext, parent, metadata,
+            config, maxDocsPerValue, executionHint);
     }
 
     @Override

server/src/main/java/org/elasticsearch/search/aggregations/bucket/sampler/DiversifiedAggregatorSupplier.java

@@ -22,7 +22,7 @@ package org.elasticsearch.search.aggregations.bucket.sampler;
 import org.elasticsearch.search.aggregations.Aggregator;
 import org.elasticsearch.search.aggregations.AggregatorFactories;
 import org.elasticsearch.search.aggregations.support.AggregatorSupplier;
-import org.elasticsearch.search.aggregations.support.ValuesSource;
+import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
 import org.elasticsearch.search.internal.SearchContext;
 
 import java.io.IOException;
@@ -36,7 +36,7 @@ public interface DiversifiedAggregatorSupplier extends AggregatorSupplier {
         SearchContext context,
         Aggregator parent,
         Map<String, Object> metadata,
-        ValuesSource valuesSource,
+        ValuesSourceConfig valuesSourceConfig,
         int maxDocsPerValue,
         String executionHint) throws IOException;
 }

server/src/main/java/org/elasticsearch/search/aggregations/bucket/sampler/DiversifiedBytesHashSamplerAggregator.java

@@ -32,6 +32,7 @@ import org.elasticsearch.search.aggregations.Aggregator;
 import org.elasticsearch.search.aggregations.AggregatorFactories;
 import org.elasticsearch.search.aggregations.bucket.DeferringBucketCollector;
 import org.elasticsearch.search.aggregations.support.ValuesSource;
+import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
 import org.elasticsearch.search.internal.SearchContext;
 
 import java.io.IOException;
@@ -47,12 +48,19 @@ public class DiversifiedBytesHashSamplerAggregator extends SamplerAggregator {
     private ValuesSource valuesSource;
     private int maxDocsPerValue;
 
-    DiversifiedBytesHashSamplerAggregator(String name, int shardSize, AggregatorFactories factories,
-            SearchContext context, Aggregator parent, Map<String, Object> metadata,
-            ValuesSource valuesSource,
-            int maxDocsPerValue) throws IOException {
+    DiversifiedBytesHashSamplerAggregator(
+        String name,
+        int shardSize,
+        AggregatorFactories factories,
+        SearchContext context,
+        Aggregator parent,
+        Map<String, Object> metadata,
+        ValuesSourceConfig valuesSourceConfig,
+        int maxDocsPerValue
+    ) throws IOException {
         super(name, shardSize, factories, context, parent, metadata);
-        this.valuesSource = valuesSource;
+        assert valuesSourceConfig.hasValues();
+        this.valuesSource = valuesSourceConfig.getValuesSource();
         this.maxDocsPerValue = maxDocsPerValue;
     }
 

server/src/main/java/org/elasticsearch/search/aggregations/bucket/sampler/DiversifiedMapSamplerAggregator.java

@@ -34,6 +34,7 @@ import org.elasticsearch.search.aggregations.Aggregator;
 import org.elasticsearch.search.aggregations.AggregatorFactories;
 import org.elasticsearch.search.aggregations.bucket.DeferringBucketCollector;
 import org.elasticsearch.search.aggregations.support.ValuesSource;
+import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
 import org.elasticsearch.search.internal.SearchContext;
 
 import java.io.IOException;
@@ -46,11 +47,19 @@ public class DiversifiedMapSamplerAggregator extends SamplerAggregator {
     private int maxDocsPerValue;
     private BytesRefHash bucketOrds;
 
-    DiversifiedMapSamplerAggregator(String name, int shardSize, AggregatorFactories factories,
-            SearchContext context, Aggregator parent, Map<String, Object> metadata,
-            ValuesSource valuesSource, int maxDocsPerValue) throws IOException {
+    DiversifiedMapSamplerAggregator(
+        String name,
+        int shardSize,
+        AggregatorFactories factories,
+        SearchContext context,
+        Aggregator parent,
+        Map<String, Object> metadata,
+        ValuesSourceConfig valuesSourceConfig,
+        int maxDocsPerValue
+    ) throws IOException {
         super(name, shardSize, factories, context, parent, metadata);
-        this.valuesSource = valuesSource;
+        assert valuesSourceConfig.hasValues();
+        this.valuesSource = valuesSourceConfig.getValuesSource();
         this.maxDocsPerValue = maxDocsPerValue;
         // Need to use super class shardSize since it is limited to maxDoc
         bucketOrds = new BytesRefHash(this.shardSize, context.bigArrays());

server/src/main/java/org/elasticsearch/search/aggregations/bucket/sampler/DiversifiedNumericSamplerAggregator.java

@@ -31,6 +31,7 @@ import org.elasticsearch.search.aggregations.Aggregator;
 import org.elasticsearch.search.aggregations.AggregatorFactories;
 import org.elasticsearch.search.aggregations.bucket.DeferringBucketCollector;
 import org.elasticsearch.search.aggregations.support.ValuesSource;
+import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
 import org.elasticsearch.search.internal.SearchContext;
 
 import java.io.IOException;
@@ -42,11 +43,19 @@ public class DiversifiedNumericSamplerAggregator extends SamplerAggregator {
     private ValuesSource.Numeric valuesSource;
     private int maxDocsPerValue;
 
-    DiversifiedNumericSamplerAggregator(String name, int shardSize, AggregatorFactories factories,
-            SearchContext context, Aggregator parent, Map<String, Object> metadata,
-            ValuesSource valuesSource, int maxDocsPerValue) throws IOException {
+    DiversifiedNumericSamplerAggregator(
+        String name,
+        int shardSize,
+        AggregatorFactories factories,
+        SearchContext context,
+        Aggregator parent,
+        Map<String, Object> metadata,
+        ValuesSourceConfig valuesSourceConfig,
+        int maxDocsPerValue
+    ) throws IOException {
         super(name, shardSize, factories, context, parent, metadata);
-        this.valuesSource = (ValuesSource.Numeric) valuesSource;
+        assert valuesSourceConfig.hasValues();
+        this.valuesSource = (ValuesSource.Numeric) valuesSourceConfig.getValuesSource();
         this.maxDocsPerValue = maxDocsPerValue;
     }
 

server/src/main/java/org/elasticsearch/search/aggregations/bucket/sampler/DiversifiedOrdinalsSamplerAggregator.java

@@ -32,6 +32,7 @@ import org.elasticsearch.search.aggregations.Aggregator;
 import org.elasticsearch.search.aggregations.AggregatorFactories;
 import org.elasticsearch.search.aggregations.bucket.DeferringBucketCollector;
 import org.elasticsearch.search.aggregations.support.ValuesSource;
+import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
 import org.elasticsearch.search.internal.SearchContext;
 
 import java.io.IOException;
@@ -43,11 +44,19 @@ public class DiversifiedOrdinalsSamplerAggregator extends SamplerAggregator {
     private ValuesSource.Bytes.WithOrdinals.FieldData valuesSource;
     private int maxDocsPerValue;
 
-    DiversifiedOrdinalsSamplerAggregator(String name, int shardSize, AggregatorFactories factories,
-            SearchContext context, Aggregator parent, Map<String, Object> metadata,
-            ValuesSource.Bytes.WithOrdinals.FieldData valuesSource, int maxDocsPerValue) throws IOException {
+    DiversifiedOrdinalsSamplerAggregator(
+        String name,
+        int shardSize,
+        AggregatorFactories factories,
+        SearchContext context,
+        Aggregator parent,
+        Map<String, Object> metadata,
+        ValuesSourceConfig valuesSourceConfig,
+        int maxDocsPerValue
+    ) throws IOException {
         super(name, shardSize, factories, context, parent, metadata);
-        this.valuesSource = valuesSource;
+        assert valuesSourceConfig.hasValues();
+        this.valuesSource = (ValuesSource.Bytes.WithOrdinals.FieldData) valuesSourceConfig.getValuesSource();
         this.maxDocsPerValue = maxDocsPerValue;
     }
 

server/src/main/java/org/elasticsearch/search/aggregations/bucket/sampler/SamplerAggregator.java

@@ -33,7 +33,7 @@ import org.elasticsearch.search.aggregations.LeafBucketCollector;
 import org.elasticsearch.search.aggregations.bucket.DeferableBucketAggregator;
 import org.elasticsearch.search.aggregations.bucket.DeferringBucketCollector;
 import org.elasticsearch.search.aggregations.bucket.SingleBucketAggregator;
-import org.elasticsearch.search.aggregations.support.ValuesSource;
+import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
 import org.elasticsearch.search.internal.SearchContext;
 
 import java.io.IOException;
@@ -63,12 +63,19 @@ public class SamplerAggregator extends DeferableBucketAggregator implements Sing
         MAP(new ParseField("map")) {
 
             @Override
-            Aggregator create(String name, AggregatorFactories factories, int shardSize, int maxDocsPerValue, ValuesSource valuesSource,
-                    SearchContext context, Aggregator parent, Map<String, Object> metadata) throws IOException {
+            Aggregator create(
+                String name,
+                AggregatorFactories factories,
+                int shardSize,
+                int maxDocsPerValue,
+                ValuesSourceConfig valuesSourceConfig,
+                SearchContext context,
+                Aggregator parent,
+                Map<String, Object> metadata
+            ) throws IOException {
 
                 return new DiversifiedMapSamplerAggregator(name, shardSize, factories, context, parent, metadata,
-                        valuesSource,
-                        maxDocsPerValue);
+                        valuesSourceConfig, maxDocsPerValue);
             }
 
             @Override
@@ -80,12 +87,19 @@ public class SamplerAggregator extends DeferableBucketAggregator implements Sing
         BYTES_HASH(new ParseField("bytes_hash")) {
 
             @Override
-            Aggregator create(String name, AggregatorFactories factories, int shardSize, int maxDocsPerValue, ValuesSource valuesSource,
-                    SearchContext context, Aggregator parent, Map<String, Object> metadata) throws IOException {
+            Aggregator create(
+                String name,
+                AggregatorFactories factories,
+                int shardSize,
+                int maxDocsPerValue,
+                ValuesSourceConfig valuesSourceConfig,
+                SearchContext context,
+                Aggregator parent,
+                Map<String, Object> metadata
+            ) throws IOException {
 
                 return new DiversifiedBytesHashSamplerAggregator(name, shardSize, factories, context, parent, metadata,
-                        valuesSource,
-                        maxDocsPerValue);
+                        valuesSourceConfig, maxDocsPerValue);
             }
 
             @Override
@@ -97,10 +111,18 @@ public class SamplerAggregator extends DeferableBucketAggregator implements Sing
         GLOBAL_ORDINALS(new ParseField("global_ordinals")) {
 
             @Override
-            Aggregator create(String name, AggregatorFactories factories, int shardSize, int maxDocsPerValue, ValuesSource valuesSource,
-                    SearchContext context, Aggregator parent, Map<String, Object> metadata) throws IOException {
+            Aggregator create(
+                String name,
+                AggregatorFactories factories,
+                int shardSize,
+                int maxDocsPerValue,
+                ValuesSourceConfig valuesSourceConfig,
+                SearchContext context,
+                Aggregator parent,
+                Map<String, Object> metadata
+            ) throws IOException {
                 return new DiversifiedOrdinalsSamplerAggregator(name, shardSize, factories, context, parent, metadata,
-                        (ValuesSource.Bytes.WithOrdinals.FieldData) valuesSource, maxDocsPerValue);
+                         valuesSourceConfig, maxDocsPerValue);
             }
 
             @Override
@@ -125,8 +147,16 @@ public class SamplerAggregator extends DeferableBucketAggregator implements Sing
             this.parseField = parseField;
         }
 
-        abstract Aggregator create(String name, AggregatorFactories factories, int shardSize, int maxDocsPerValue,
-                ValuesSource valuesSource, SearchContext context, Aggregator parent, Map<String, Object> metadata) throws IOException;
+        abstract Aggregator create(
+            String name,
+            AggregatorFactories factories,
+            int shardSize,
+            int maxDocsPerValue,
+            ValuesSourceConfig valuesSourceConfig,
+            SearchContext context,
+            Aggregator parent,
+            Map<String, Object> metadata
+        ) throws IOException;
 
         abstract boolean needsGlobalOrdinals();
 

server/src/main/java/org/elasticsearch/search/aggregations/bucket/terms/RareTermsAggregatorFactory.java

@@ -151,11 +151,10 @@ public class RareTermsAggregatorFactory extends ValuesSourceAggregatorFactory {
     }
 
     @Override
-    protected Aggregator doCreateInternal(ValuesSource valuesSource,
-                                            SearchContext searchContext,
-                                            Aggregator parent,
-                                            boolean collectsFromSingleBucket,
-                                            Map<String, Object> metadata) throws IOException {
+    protected Aggregator doCreateInternal(SearchContext searchContext,
+                                          Aggregator parent,
+                                          boolean collectsFromSingleBucket,
+                                          Map<String, Object> metadata) throws IOException {
         if (collectsFromSingleBucket == false) {
             return asMultiBucketAggregator(this, searchContext, parent);
         }
@@ -167,7 +166,7 @@ public class RareTermsAggregatorFactory extends ValuesSourceAggregatorFactory {
                 aggregatorSupplier.getClass().toString() + "]");
         }
 
-        return ((RareTermsAggregatorSupplier) aggregatorSupplier).build(name, factories, valuesSource, config.format(),
+        return ((RareTermsAggregatorSupplier) aggregatorSupplier).build(name, factories, config.getValuesSource(), config.format(),
             maxDocCount, precision, includeExclude, searchContext, parent, metadata);
     }
 

server/src/main/java/org/elasticsearch/search/aggregations/bucket/terms/SignificantTermsAggregatorFactory.java

@@ -187,14 +187,14 @@ public class SignificantTermsAggregatorFactory extends ValuesSourceAggregatorFac
                                       Map<String, Object> metadata) throws IOException {
         super(name, config, queryShardContext, parent, subFactoriesBuilder, metadata);
 
-        if (config.unmapped() == false) {
+        if (config.hasValues()) {
             if (config.fieldContext().fieldType().isSearchable() == false) {
                 throw new IllegalArgumentException("SignificantText aggregation requires fields to be searchable, but ["
                     + config.fieldContext().fieldType().name() + "] is not");
             }
         }
 
-        if (config.unmapped() == false) {
+        if (config.hasValues()) {
             if (config.fieldContext().fieldType().isSearchable() == false) {
                 throw new IllegalArgumentException("SignificantText aggregation requires fields to be searchable, but ["
                     + config.fieldContext().fieldType().name() + "] is not");
@@ -293,11 +293,10 @@ public class SignificantTermsAggregatorFactory extends ValuesSourceAggregatorFac
     }
 
     @Override
-    protected Aggregator doCreateInternal(ValuesSource valuesSource,
-                                            SearchContext searchContext,
-                                            Aggregator parent,
-                                            boolean collectsFromSingleBucket,
-                                            Map<String, Object> metadata) throws IOException {
+    protected Aggregator doCreateInternal(SearchContext searchContext,
+                                          Aggregator parent,
+                                          boolean collectsFromSingleBucket,
+                                          Map<String, Object> metadata) throws IOException {
         AggregatorSupplier aggregatorSupplier = queryShardContext.getValuesSourceRegistry().getAggregator(config,
             SignificantTermsAggregationBuilder.NAME);
         if (aggregatorSupplier instanceof SignificantTermsAggregatorSupplier == false) {
@@ -325,7 +324,7 @@ public class SignificantTermsAggregatorFactory extends ValuesSourceAggregatorFac
         }
 
         // TODO we should refactor so that we don't need to use this Factory as a singleton (e.g. stop passing `this` to the aggregators)
-        return sigTermsAggregatorSupplier.build(name, factories, valuesSource, config.format(),
+        return sigTermsAggregatorSupplier.build(name, factories, config.getValuesSource(), config.format(),
             bucketCountThresholds, includeExclude, executionHint, searchContext, parent,
             significanceHeuristic, this, collectsFromSingleBucket, metadata);
     }
@@ -398,7 +397,7 @@ public class SignificantTermsAggregatorFactory extends ValuesSourceAggregatorFac
                      **/
                     remapGlobalOrd = false;
                 }
-                
+
                 return new GlobalOrdinalsStringTermsAggregator(
                     name,
                     factories,

server/src/main/java/org/elasticsearch/search/aggregations/bucket/terms/TermsAggregatorFactory.java

@@ -146,7 +146,7 @@ public class TermsAggregatorFactory extends ValuesSourceAggregatorFactory {
                     subAggCollectMode = pickSubAggColectMode(factories, bucketCountThresholds.getShardSize(), -1);
                 }
 
-                ValuesSource.Numeric numericValuesSource = (ValuesSource.Numeric) valuesSource; 
+                ValuesSource.Numeric numericValuesSource = (ValuesSource.Numeric) valuesSource;
                 IncludeExclude.LongFilter longFilter = null;
                 Function<NumericTermsAggregator, ResultStrategy<?, ?>> resultStrategy;
                 if (numericValuesSource.isFloatingPoint()) {
@@ -223,8 +223,7 @@ public class TermsAggregatorFactory extends ValuesSourceAggregatorFactory {
     }
 
     @Override
-    protected Aggregator doCreateInternal(ValuesSource valuesSource,
-                                          SearchContext searchContext,
+    protected Aggregator doCreateInternal(SearchContext searchContext,
                                           Aggregator parent,
                                           boolean collectsFromSingleBucket,
                                           Map<String, Object> metadata) throws IOException {
@@ -246,7 +245,7 @@ public class TermsAggregatorFactory extends ValuesSourceAggregatorFactory {
         }
         bucketCountThresholds.ensureValidity();
 
-        return termsAggregatorSupplier.build(name, factories, valuesSource, order, config.format(),
+        return termsAggregatorSupplier.build(name, factories, config.getValuesSource(), order, config.format(),
             bucketCountThresholds, includeExclude, executionHint, searchContext, parent, collectMode,
             showTermDocCountError, collectsFromSingleBucket, metadata);
     }

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

@@ -46,10 +46,11 @@ class AvgAggregator extends NumericMetricsAggregator.SingleValue {
     DoubleArray compensations;
     DocValueFormat format;
 
-    AvgAggregator(String name, ValuesSourceConfig valuesSourceConfig, ValuesSource valuesSource,  SearchContext context,
+    AvgAggregator(String name, ValuesSourceConfig valuesSourceConfig, SearchContext context,
                   Aggregator parent, Map<String, Object> metadata) throws IOException {
         super(name, context, parent, metadata);
-        this.valuesSource = (ValuesSource.Numeric) valuesSource;
+        // TODO Stop expecting nulls here
+        this.valuesSource = valuesSourceConfig.hasValues() ? (ValuesSource.Numeric) valuesSourceConfig.getValuesSource() : null;
         this.format = valuesSourceConfig.format();
         if (valuesSource != null) {
             final BigArrays bigArrays = context.bigArrays();

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

@@ -26,7 +26,6 @@ import org.elasticsearch.search.aggregations.AggregatorFactories;
 import org.elasticsearch.search.aggregations.AggregatorFactory;
 import org.elasticsearch.search.aggregations.support.AggregatorSupplier;
 import org.elasticsearch.search.aggregations.support.CoreValuesSourceType;
-import org.elasticsearch.search.aggregations.support.ValuesSource;
 import org.elasticsearch.search.aggregations.support.ValuesSourceAggregatorFactory;
 import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
 import org.elasticsearch.search.aggregations.support.ValuesSourceRegistry;
@@ -54,15 +53,14 @@ class AvgAggregatorFactory extends ValuesSourceAggregatorFactory {
     protected Aggregator createUnmapped(SearchContext searchContext,
                                             Aggregator parent,
                                             Map<String, Object> metadata) throws IOException {
-        return new AvgAggregator(name, config, null, searchContext, parent, metadata);
+        return new AvgAggregator(name, config, searchContext, parent, metadata);
     }
 
     @Override
-    protected Aggregator doCreateInternal(ValuesSource valuesSource,
-                                            SearchContext searchContext,
-                                            Aggregator parent,
-                                            boolean collectsFromSingleBucket,
-                                            Map<String, Object> metadata) throws IOException {
+    protected Aggregator doCreateInternal(SearchContext searchContext,
+                                          Aggregator parent,
+                                          boolean collectsFromSingleBucket,
+                                          Map<String, Object> metadata) throws IOException {
         AggregatorSupplier aggregatorSupplier = queryShardContext.getValuesSourceRegistry().getAggregator(config,
             AvgAggregationBuilder.NAME);
 
@@ -70,6 +68,6 @@ class AvgAggregatorFactory extends ValuesSourceAggregatorFactory {
             throw new AggregationExecutionException("Registry miss-match - expected MetricAggregatorSupplier, found [" +
                 aggregatorSupplier.getClass().toString() + "]");
         }
-        return ((MetricAggregatorSupplier) aggregatorSupplier).build(name, config, valuesSource, searchContext, parent, metadata);
+        return ((MetricAggregatorSupplier) aggregatorSupplier).build(name, config, searchContext, parent, metadata);
     }
 }

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

@@ -41,6 +41,7 @@ import org.elasticsearch.search.aggregations.Aggregator;
 import org.elasticsearch.search.aggregations.InternalAggregation;
 import org.elasticsearch.search.aggregations.LeafBucketCollector;
 import org.elasticsearch.search.aggregations.support.ValuesSource;
+import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
 import org.elasticsearch.search.internal.SearchContext;
 
 import java.io.IOException;
@@ -62,13 +63,14 @@ public class CardinalityAggregator extends NumericMetricsAggregator.SingleValue
 
     public CardinalityAggregator(
             String name,
-            ValuesSource valuesSource,
+            ValuesSourceConfig valuesSourceConfig,
             int precision,
             SearchContext context,
             Aggregator parent,
             Map<String, Object> metadata) throws IOException {
         super(name, context, parent, metadata);
-        this.valuesSource = valuesSource;
+        // TODO: Stop using nulls here
+        this.valuesSource = valuesSourceConfig.hasValues() ? valuesSourceConfig.getValuesSource() : null;
         this.precision = precision;
         this.counts = valuesSource == null ? null : new HyperLogLogPlusPlus(precision, context.bigArrays(), 1);
     }

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

@@ -26,7 +26,6 @@ import org.elasticsearch.search.aggregations.AggregatorFactories;
 import org.elasticsearch.search.aggregations.AggregatorFactory;
 import org.elasticsearch.search.aggregations.support.AggregatorSupplier;
 import org.elasticsearch.search.aggregations.support.CoreValuesSourceType;
-import org.elasticsearch.search.aggregations.support.ValuesSource;
 import org.elasticsearch.search.aggregations.support.ValuesSourceAggregatorFactory;
 import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
 import org.elasticsearch.search.aggregations.support.ValuesSourceRegistry;
@@ -58,15 +57,14 @@ class CardinalityAggregatorFactory extends ValuesSourceAggregatorFactory {
     protected Aggregator createUnmapped(SearchContext searchContext,
                                             Aggregator parent,
                                             Map<String, Object> metadata) throws IOException {
-        return new CardinalityAggregator(name, null, precision(), searchContext, parent, metadata);
+        return new CardinalityAggregator(name, config, precision(), searchContext, parent, metadata);
     }
 
     @Override
-    protected Aggregator doCreateInternal(ValuesSource valuesSource,
-                                            SearchContext searchContext,
-                                            Aggregator parent,
-                                            boolean collectsFromSingleBucket,
-                                            Map<String, Object> metadata) throws IOException {
+    protected Aggregator doCreateInternal(SearchContext searchContext,
+                                          Aggregator parent,
+                                          boolean collectsFromSingleBucket,
+                                          Map<String, Object> metadata) throws IOException {
         AggregatorSupplier aggregatorSupplier = queryShardContext.getValuesSourceRegistry().getAggregator(config,
             CardinalityAggregationBuilder.NAME);
         if (aggregatorSupplier instanceof CardinalityAggregatorSupplier == false) {
@@ -74,7 +72,7 @@ class CardinalityAggregatorFactory extends ValuesSourceAggregatorFactory {
                 aggregatorSupplier.getClass().toString() + "]");
         }
         CardinalityAggregatorSupplier cardinalityAggregatorSupplier = (CardinalityAggregatorSupplier) aggregatorSupplier;
-        return cardinalityAggregatorSupplier.build(name, valuesSource, precision(), searchContext, parent, metadata);
+        return cardinalityAggregatorSupplier.build(name, config, precision(), searchContext, parent, metadata);
     }
 
     private int precision() {

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

@@ -21,7 +21,7 @@ package org.elasticsearch.search.aggregations.metrics;
 
 import org.elasticsearch.search.aggregations.Aggregator;
 import org.elasticsearch.search.aggregations.support.AggregatorSupplier;
-import org.elasticsearch.search.aggregations.support.ValuesSource;
+import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
 import org.elasticsearch.search.internal.SearchContext;
 
 import java.io.IOException;
@@ -29,7 +29,7 @@ import java.util.Map;
 
 public interface CardinalityAggregatorSupplier extends AggregatorSupplier {
     Aggregator build(String name,
-                     ValuesSource valuesSource,
+                     ValuesSourceConfig valuesSourceConfig,
                      int precision,
                      SearchContext context,
                      Aggregator parent,

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

@@ -32,6 +32,7 @@ import org.elasticsearch.search.aggregations.InternalAggregation;
 import org.elasticsearch.search.aggregations.LeafBucketCollector;
 import org.elasticsearch.search.aggregations.LeafBucketCollectorBase;
 import org.elasticsearch.search.aggregations.support.ValuesSource;
+import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
 import org.elasticsearch.search.internal.SearchContext;
 
 import java.io.IOException;
@@ -53,11 +54,18 @@ class ExtendedStatsAggregator extends NumericMetricsAggregator.MultiValue {
     DoubleArray sumOfSqrs;
     DoubleArray compensationOfSqrs;
 
-    ExtendedStatsAggregator(String name, ValuesSource.Numeric valuesSource, DocValueFormat formatter,
-            SearchContext context, Aggregator parent, double sigma, Map<String, Object> metadata) throws IOException {
+    ExtendedStatsAggregator(
+        String name,
+        ValuesSourceConfig valuesSourceConfig,
+        SearchContext context,
+        Aggregator parent,
+        double sigma,
+        Map<String, Object> metadata
+    ) throws IOException {
         super(name, context, parent, metadata);
-        this.valuesSource = valuesSource;
-        this.format = formatter;
+        // TODO: stop depending on nulls here
+        this.valuesSource = valuesSourceConfig.hasValues() ? (ValuesSource.Numeric) valuesSourceConfig.getValuesSource() : null;
+        this.format = valuesSourceConfig.format();
         this.sigma = sigma;
         if (valuesSource != null) {
             final BigArrays bigArrays = context.bigArrays();

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

@@ -26,8 +26,6 @@ import org.elasticsearch.search.aggregations.AggregatorFactories;
 import org.elasticsearch.search.aggregations.AggregatorFactory;
 import org.elasticsearch.search.aggregations.support.AggregatorSupplier;
 import org.elasticsearch.search.aggregations.support.CoreValuesSourceType;
-import org.elasticsearch.search.aggregations.support.ValuesSource;
-import org.elasticsearch.search.aggregations.support.ValuesSource.Numeric;
 import org.elasticsearch.search.aggregations.support.ValuesSourceAggregatorFactory;
 import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
 import org.elasticsearch.search.aggregations.support.ValuesSourceRegistry;
@@ -62,12 +60,11 @@ class ExtendedStatsAggregatorFactory extends ValuesSourceAggregatorFactory {
     protected Aggregator createUnmapped(SearchContext searchContext,
                                             Aggregator parent,
                                             Map<String, Object> metadata) throws IOException {
-        return new ExtendedStatsAggregator(name, null, config.format(), searchContext, parent, sigma, metadata);
+        return new ExtendedStatsAggregator(name, config, searchContext, parent, sigma, metadata);
     }
 
     @Override
-    protected Aggregator doCreateInternal(ValuesSource valuesSource,
-                                          SearchContext searchContext,
+    protected Aggregator doCreateInternal(SearchContext searchContext,
                                           Aggregator parent,
                                           boolean collectsFromSingleBucket,
                                           Map<String, Object> metadata) throws IOException {
@@ -78,7 +75,6 @@ class ExtendedStatsAggregatorFactory extends ValuesSourceAggregatorFactory {
             throw new AggregationExecutionException("Registry miss-match - expected ExtendedStatsAggregatorProvider, found [" +
                 aggregatorSupplier.getClass().toString() + "]");
         }
-        return new ExtendedStatsAggregator(name, (Numeric) valuesSource, config.format(), searchContext,
-            parent, sigma, metadata);
+        return new ExtendedStatsAggregator(name, config, searchContext, parent, sigma, metadata);
     }
 }

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

@@ -18,10 +18,9 @@
  */
 package org.elasticsearch.search.aggregations.metrics;
 
-import org.elasticsearch.search.DocValueFormat;
 import org.elasticsearch.search.aggregations.Aggregator;
 import org.elasticsearch.search.aggregations.support.AggregatorSupplier;
-import org.elasticsearch.search.aggregations.support.ValuesSource;
+import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
 import org.elasticsearch.search.internal.SearchContext;
 
 import java.io.IOException;
@@ -30,8 +29,7 @@ import java.util.Map;
 public interface ExtendedStatsAggregatorProvider extends AggregatorSupplier {
 
     Aggregator build(String name,
-                     ValuesSource.Numeric valuesSource,
-                     DocValueFormat formatter,
+                     ValuesSourceConfig valuesSourceConfig,
                      SearchContext context,
                      Aggregator parent,
                      double sigma,

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

@@ -31,6 +31,7 @@ import org.elasticsearch.search.aggregations.InternalAggregation;
 import org.elasticsearch.search.aggregations.LeafBucketCollector;
 import org.elasticsearch.search.aggregations.LeafBucketCollectorBase;
 import org.elasticsearch.search.aggregations.support.ValuesSource;
+import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
 import org.elasticsearch.search.internal.SearchContext;
 
 import java.io.IOException;
@@ -49,10 +50,17 @@ final class GeoBoundsAggregator extends MetricsAggregator {
     DoubleArray negLefts;
     DoubleArray negRights;
 
-    GeoBoundsAggregator(String name, SearchContext aggregationContext, Aggregator parent,
-            ValuesSource valuesSource, boolean wrapLongitude, Map<String, Object> metadata) throws IOException {
+    GeoBoundsAggregator(
+        String name,
+        SearchContext aggregationContext,
+        Aggregator parent,
+        ValuesSourceConfig valuesSourceConfig,
+        boolean wrapLongitude,
+        Map<String, Object> metadata
+    ) throws IOException {
         super(name, aggregationContext, parent, metadata);
-        this.valuesSource = (ValuesSource.GeoPoint) valuesSource;
+        // TODO: stop expecting nulls here
+        this.valuesSource = valuesSourceConfig.hasValues() ? (ValuesSource.GeoPoint) valuesSourceConfig.getValuesSource() : null;
         this.wrapLongitude = wrapLongitude;
         if (valuesSource != null) {
             final BigArrays bigArrays = context.bigArrays();

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

@@ -26,7 +26,6 @@ import org.elasticsearch.search.aggregations.AggregatorFactories;
 import org.elasticsearch.search.aggregations.AggregatorFactory;
 import org.elasticsearch.search.aggregations.support.AggregatorSupplier;
 import org.elasticsearch.search.aggregations.support.CoreValuesSourceType;
-import org.elasticsearch.search.aggregations.support.ValuesSource;
 import org.elasticsearch.search.aggregations.support.ValuesSourceAggregatorFactory;
 import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
 import org.elasticsearch.search.aggregations.support.ValuesSourceRegistry;
@@ -54,15 +53,14 @@ class GeoBoundsAggregatorFactory extends ValuesSourceAggregatorFactory {
     protected Aggregator createUnmapped(SearchContext searchContext,
                                             Aggregator parent,
                                             Map<String, Object> metadata) throws IOException {
-        return new GeoBoundsAggregator(name, searchContext, parent, null, wrapLongitude, metadata);
+        return new GeoBoundsAggregator(name, searchContext, parent, config, wrapLongitude, metadata);
     }
 
     @Override
-    protected Aggregator doCreateInternal(ValuesSource valuesSource,
-                                            SearchContext searchContext,
-                                            Aggregator parent,
-                                            boolean collectsFromSingleBucket,
-                                            Map<String, Object> metadata) throws IOException {
+    protected Aggregator doCreateInternal(SearchContext searchContext,
+                                          Aggregator parent,
+                                          boolean collectsFromSingleBucket,
+                                          Map<String, Object> metadata) throws IOException {
         AggregatorSupplier aggregatorSupplier = queryShardContext.getValuesSourceRegistry()
             .getAggregator(config, GeoBoundsAggregationBuilder.NAME);
 
@@ -71,7 +69,7 @@ class GeoBoundsAggregatorFactory extends ValuesSourceAggregatorFactory {
                 + GeoBoundsAggregatorSupplier.class.getName() + ", found [" + aggregatorSupplier.getClass().toString() + "]");
         }
 
-        return ((GeoBoundsAggregatorSupplier) aggregatorSupplier).build(name, searchContext, parent, valuesSource, wrapLongitude, metadata);
+        return ((GeoBoundsAggregatorSupplier) aggregatorSupplier).build(name, searchContext, parent, config, wrapLongitude, metadata);
     }
 
     static void registerAggregators(ValuesSourceRegistry.Builder builder) {

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

@@ -21,7 +21,7 @@ package org.elasticsearch.search.aggregations.metrics;
 
 import org.elasticsearch.search.aggregations.Aggregator;
 import org.elasticsearch.search.aggregations.support.AggregatorSupplier;
-import org.elasticsearch.search.aggregations.support.ValuesSource;
+import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
 import org.elasticsearch.search.internal.SearchContext;
 
 import java.io.IOException;
@@ -30,7 +30,11 @@ import java.util.Map;
 @FunctionalInterface
 public interface GeoBoundsAggregatorSupplier extends AggregatorSupplier {
 
-    MetricsAggregator build(String name, SearchContext aggregationContext, Aggregator parent,
-                        ValuesSource valuesSource, boolean wrapLongitude,
-                        Map<String, Object> metadata) throws IOException;
+    MetricsAggregator build(
+        String name,
+        SearchContext aggregationContext,
+        Aggregator parent,
+        ValuesSourceConfig valuesSourceConfig,
+        boolean wrapLongitude,
+        Map<String, Object> metadata) throws IOException;
 }

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

@@ -31,6 +31,7 @@ import org.elasticsearch.search.aggregations.InternalAggregation;
 import org.elasticsearch.search.aggregations.LeafBucketCollector;
 import org.elasticsearch.search.aggregations.LeafBucketCollectorBase;
 import org.elasticsearch.search.aggregations.support.ValuesSource;
+import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
 import org.elasticsearch.search.internal.SearchContext;
 
 import java.io.IOException;
@@ -44,10 +45,16 @@ final class GeoCentroidAggregator extends MetricsAggregator {
     private DoubleArray lonSum, lonCompensations, latSum, latCompensations;
     private LongArray counts;
 
-    GeoCentroidAggregator(String name, SearchContext context, Aggregator parent,
-                                    ValuesSource valuesSource, Map<String, Object> metadata) throws IOException {
+    GeoCentroidAggregator(
+        String name,
+        ValuesSourceConfig valuesSourceConfig,
+        SearchContext context,
+        Aggregator parent,
+        Map<String, Object> metadata
+    ) throws IOException {
         super(name, context, parent, metadata);
-        this.valuesSource = (ValuesSource.GeoPoint) valuesSource;
+        // TODO: Stop expecting nulls here
+        this.valuesSource = valuesSourceConfig.hasValues() ? (ValuesSource.GeoPoint) valuesSourceConfig.getValuesSource() : null;
         if (valuesSource != null) {
             final BigArrays bigArrays = context.bigArrays();
             lonSum = bigArrays.newDoubleArray(1, true);

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

@@ -26,7 +26,6 @@ import org.elasticsearch.search.aggregations.AggregatorFactories;
 import org.elasticsearch.search.aggregations.AggregatorFactory;
 import org.elasticsearch.search.aggregations.support.AggregatorSupplier;
 import org.elasticsearch.search.aggregations.support.CoreValuesSourceType;
-import org.elasticsearch.search.aggregations.support.ValuesSource;
 import org.elasticsearch.search.aggregations.support.ValuesSourceAggregatorFactory;
 import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
 import org.elasticsearch.search.aggregations.support.ValuesSourceRegistry;
@@ -50,27 +49,26 @@ class GeoCentroidAggregatorFactory extends ValuesSourceAggregatorFactory {
     protected Aggregator createUnmapped(SearchContext searchContext,
                                             Aggregator parent,
                                             Map<String, Object> metadata) throws IOException {
-        return new GeoCentroidAggregator(name, searchContext, parent, null, metadata);
+        return new GeoCentroidAggregator(name, config, searchContext, parent, metadata);
     }
 
     @Override
-    protected Aggregator doCreateInternal(ValuesSource valuesSource,
-                                            SearchContext searchContext,
-                                            Aggregator parent,
-                                            boolean collectsFromSingleBucket,
-                                            Map<String, Object> metadata) throws IOException {
+    protected Aggregator doCreateInternal(SearchContext searchContext,
+                                          Aggregator parent,
+                                          boolean collectsFromSingleBucket,
+                                          Map<String, Object> metadata) throws IOException {
 
         AggregatorSupplier aggregatorSupplier = queryShardContext.getValuesSourceRegistry().getAggregator(config,
             GeoCentroidAggregationBuilder.NAME);
-        if (aggregatorSupplier instanceof GeoCentroidAggregatorSupplier == false) {
+        if (aggregatorSupplier instanceof MetricAggregatorSupplier == false) {
             throw new AggregationExecutionException("Registry miss-match - expected "
-                + GeoCentroidAggregatorSupplier.class.getName() + ", found [" + aggregatorSupplier.getClass().toString() + "]");
+                + MetricAggregatorSupplier.class.getName() + ", found [" + aggregatorSupplier.getClass().toString() + "]");
         }
-        return ((GeoCentroidAggregatorSupplier) aggregatorSupplier).build(name, searchContext, parent, valuesSource, metadata);
+        return ((MetricAggregatorSupplier) aggregatorSupplier).build(name, config, searchContext, parent, metadata);
     }
 
     static void registerAggregators(ValuesSourceRegistry.Builder builder) {
         builder.register(GeoCentroidAggregationBuilder.NAME, CoreValuesSourceType.GEOPOINT,
-            (GeoCentroidAggregatorSupplier) GeoCentroidAggregator::new);
+            (MetricAggregatorSupplier) GeoCentroidAggregator::new);
     }
 }

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

@@ -1,36 +0,0 @@
-/*
- * Licensed to Elasticsearch under one or more contributor
- * license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright
- * ownership. Elasticsearch licenses this file to you under
- * the Apache License, Version 2.0 (the "License"); you may
- * not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied.  See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.elasticsearch.search.aggregations.metrics;
-
-import org.elasticsearch.search.aggregations.Aggregator;
-import org.elasticsearch.search.aggregations.support.AggregatorSupplier;
-import org.elasticsearch.search.aggregations.support.ValuesSource;
-import org.elasticsearch.search.internal.SearchContext;
-
-import java.io.IOException;
-import java.util.Map;
-
-@FunctionalInterface
-public interface GeoCentroidAggregatorSupplier extends AggregatorSupplier {
-
-    MetricsAggregator build(String name, SearchContext context, Aggregator parent,
-                          ValuesSource valuesSource,
-                          Map<String, Object> metadata) throws IOException;
-}

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

@@ -58,11 +58,11 @@ class MaxAggregator extends NumericMetricsAggregator.SingleValue {
 
     MaxAggregator(String name,
                     ValuesSourceConfig config,
-                    ValuesSource valuesSource,
                     SearchContext context,
                     Aggregator parent, Map<String, Object> metadata) throws IOException {
         super(name, context, parent, metadata);
-        this.valuesSource = (ValuesSource.Numeric) valuesSource;
+        // TODO stop expecting nulls here
+        this.valuesSource = config.hasValues() ? (ValuesSource.Numeric) config.getValuesSource() : null;
         if (valuesSource != null) {
             maxes = context.bigArrays().newDoubleArray(1, false);
             maxes.fill(0, maxes.size(), Double.NEGATIVE_INFINITY);

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

@@ -26,7 +26,6 @@ import org.elasticsearch.search.aggregations.AggregatorFactories;
 import org.elasticsearch.search.aggregations.AggregatorFactory;
 import org.elasticsearch.search.aggregations.support.AggregatorSupplier;
 import org.elasticsearch.search.aggregations.support.CoreValuesSourceType;
-import org.elasticsearch.search.aggregations.support.ValuesSource;
 import org.elasticsearch.search.aggregations.support.ValuesSourceAggregatorFactory;
 import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
 import org.elasticsearch.search.aggregations.support.ValuesSourceRegistry;
@@ -53,15 +52,14 @@ class MaxAggregatorFactory extends ValuesSourceAggregatorFactory {
     protected Aggregator createUnmapped(SearchContext searchContext,
                                             Aggregator parent,
                                             Map<String, Object> metadata) throws IOException {
-        return new MaxAggregator(name, config, null, searchContext, parent, metadata);
+        return new MaxAggregator(name, config, searchContext, parent, metadata);
     }
 
     @Override
-    protected Aggregator doCreateInternal(ValuesSource valuesSource,
-                                            SearchContext searchContext,
-                                            Aggregator parent,
-                                            boolean collectsFromSingleBucket,
-                                            Map<String, Object> metadata) throws IOException {
+    protected Aggregator doCreateInternal(SearchContext searchContext,
+                                          Aggregator parent,
+                                          boolean collectsFromSingleBucket,
+                                          Map<String, Object> metadata) throws IOException {
         AggregatorSupplier aggregatorSupplier = queryShardContext.getValuesSourceRegistry().getAggregator(config,
             MaxAggregationBuilder.NAME);
 
@@ -69,6 +67,6 @@ class MaxAggregatorFactory extends ValuesSourceAggregatorFactory {
             throw new AggregationExecutionException("Registry miss-match - expected MetricAggregatorSupplier, found [" +
                 aggregatorSupplier.getClass().toString() + "]");
         }
-        return ((MetricAggregatorSupplier) aggregatorSupplier).build(name, config, valuesSource, searchContext, parent, metadata);
+        return ((MetricAggregatorSupplier) aggregatorSupplier).build(name, config, searchContext, parent, metadata);
     }
 }

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

@@ -26,7 +26,6 @@ import org.elasticsearch.search.aggregations.AggregatorFactories;
 import org.elasticsearch.search.aggregations.AggregatorFactory;
 import org.elasticsearch.search.aggregations.support.AggregatorSupplier;
 import org.elasticsearch.search.aggregations.support.CoreValuesSourceType;
-import org.elasticsearch.search.aggregations.support.ValuesSource;
 import org.elasticsearch.search.aggregations.support.ValuesSourceAggregatorFactory;
 import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
 import org.elasticsearch.search.aggregations.support.ValuesSourceRegistry;
@@ -73,11 +72,10 @@ public class MedianAbsoluteDeviationAggregatorFactory extends ValuesSourceAggreg
     }
 
     @Override
-    protected Aggregator doCreateInternal(ValuesSource valuesSource,
-                                            SearchContext searchContext,
-                                            Aggregator parent,
-                                            boolean collectsFromSingleBucket,
-                                            Map<String, Object> metadata) throws IOException {
+    protected Aggregator doCreateInternal(SearchContext searchContext,
+                                          Aggregator parent,
+                                          boolean collectsFromSingleBucket,
+                                          Map<String, Object> metadata) throws IOException {
         AggregatorSupplier aggregatorSupplier = queryShardContext.getValuesSourceRegistry().getAggregator(config,
             MedianAbsoluteDeviationAggregationBuilder.NAME);
 
@@ -85,7 +83,7 @@ public class MedianAbsoluteDeviationAggregatorFactory extends ValuesSourceAggreg
             throw new AggregationExecutionException("Registry miss-match - expected MedianAbsoluteDeviationAggregatorSupplier, found [" +
                 aggregatorSupplier.getClass().toString() + "]");
         }
-        return ((MedianAbsoluteDeviationAggregatorSupplier) aggregatorSupplier).build(name, valuesSource, config.format(),
+        return ((MedianAbsoluteDeviationAggregatorSupplier) aggregatorSupplier).build(name, config.getValuesSource(), config.format(),
             searchContext, parent, metadata, compression);
     }
 }

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

@@ -20,7 +20,6 @@ package org.elasticsearch.search.aggregations.metrics;
 
 import org.elasticsearch.search.aggregations.Aggregator;
 import org.elasticsearch.search.aggregations.support.AggregatorSupplier;
-import org.elasticsearch.search.aggregations.support.ValuesSource;
 import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
 import org.elasticsearch.search.internal.SearchContext;
 
@@ -30,7 +29,6 @@ import java.util.Map;
 public interface MetricAggregatorSupplier extends AggregatorSupplier {
     Aggregator build(String name,
                      ValuesSourceConfig valuesSourceConfig,
-                     ValuesSource valuesSource,
                      SearchContext context,
                      Aggregator parent,
                      Map<String, Object> metadata) throws IOException;

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

@@ -61,12 +61,12 @@ class MinAggregator extends NumericMetricsAggregator.SingleValue {
 
     MinAggregator(String name,
                     ValuesSourceConfig config,
-                    ValuesSource valuesSource,
                     SearchContext context,
                     Aggregator parent,
                     Map<String, Object> metadata) throws IOException {
         super(name, context, parent, metadata);
-        this.valuesSource = (ValuesSource.Numeric) valuesSource;
+        // TODO: Stop using nulls here
+        this.valuesSource = config.hasValues() ? (ValuesSource.Numeric) config.getValuesSource() : null;
         if (valuesSource != null) {
             mins = context.bigArrays().newDoubleArray(1, false);
             mins.fill(0, mins.size(), Double.POSITIVE_INFINITY);

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

@@ -26,7 +26,6 @@ import org.elasticsearch.search.aggregations.AggregatorFactories;
 import org.elasticsearch.search.aggregations.AggregatorFactory;
 import org.elasticsearch.search.aggregations.support.AggregatorSupplier;
 import org.elasticsearch.search.aggregations.support.CoreValuesSourceType;
-import org.elasticsearch.search.aggregations.support.ValuesSource;
 import org.elasticsearch.search.aggregations.support.ValuesSourceAggregatorFactory;
 import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
 import org.elasticsearch.search.aggregations.support.ValuesSourceRegistry;
@@ -53,15 +52,14 @@ class MinAggregatorFactory extends ValuesSourceAggregatorFactory {
     protected Aggregator createUnmapped(SearchContext searchContext,
                                             Aggregator parent,
                                             Map<String, Object> metadata) throws IOException {
-        return new MinAggregator(name, config, null, searchContext, parent, metadata);
+        return new MinAggregator(name, config, searchContext, parent, metadata);
     }
 
     @Override
-    protected Aggregator doCreateInternal(ValuesSource valuesSource,
-                                            SearchContext searchContext,
-                                            Aggregator parent,
-                                            boolean collectsFromSingleBucket,
-                                            Map<String, Object> metadata) throws IOException {
+    protected Aggregator doCreateInternal(SearchContext searchContext,
+                                          Aggregator parent,
+                                          boolean collectsFromSingleBucket,
+                                          Map<String, Object> metadata) throws IOException {
         AggregatorSupplier aggregatorSupplier = queryShardContext.getValuesSourceRegistry().getAggregator(config,
             MinAggregationBuilder.NAME);
 
@@ -69,6 +67,6 @@ class MinAggregatorFactory extends ValuesSourceAggregatorFactory {
             throw new AggregationExecutionException("Registry miss-match - expected MetricAggregatorSupplier, found [" +
                 aggregatorSupplier.getClass().toString() + "]");
         }
-        return ((MetricAggregatorSupplier) aggregatorSupplier).build(name, config, valuesSource, searchContext, parent, metadata);
+        return ((MetricAggregatorSupplier) aggregatorSupplier).build(name, config, searchContext, parent, metadata);
     }
 }

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

@@ -84,8 +84,7 @@ class PercentileRanksAggregatorFactory extends ValuesSourceAggregatorFactory {
     }
 
     @Override
-    protected Aggregator doCreateInternal(ValuesSource valuesSource,
-                                          SearchContext searchContext,
+    protected Aggregator doCreateInternal(SearchContext searchContext,
                                           Aggregator parent,
                                           boolean collectsFromSingleBucket,
                                           Map<String, Object> metadata) throws IOException {
@@ -97,7 +96,16 @@ class PercentileRanksAggregatorFactory extends ValuesSourceAggregatorFactory {
                 aggregatorSupplier.getClass().toString() + "]");
         }
         PercentilesAggregatorSupplier percentilesAggregatorSupplier = (PercentilesAggregatorSupplier) aggregatorSupplier;
-        return percentilesAggregatorSupplier.build(name, valuesSource, searchContext, parent, percents, percentilesConfig, keyed,
-            config.format(), metadata);
+        return percentilesAggregatorSupplier.build(
+            name,
+            config.getValuesSource(),
+            searchContext,
+            parent,
+            percents,
+            percentilesConfig,
+            keyed,
+            config.format(),
+            metadata
+        );
     }
 }

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

@@ -83,8 +83,7 @@ class PercentilesAggregatorFactory extends ValuesSourceAggregatorFactory {
     }
 
     @Override
-    protected Aggregator doCreateInternal(ValuesSource valuesSource,
-                                          SearchContext searchContext,
+    protected Aggregator doCreateInternal(SearchContext searchContext,
                                           Aggregator parent,
                                           boolean collectsFromSingleBucket,
                                           Map<String, Object> metadata) throws IOException {
@@ -97,7 +96,16 @@ class PercentilesAggregatorFactory extends ValuesSourceAggregatorFactory {
                 aggregatorSupplier.getClass().toString() + "]");
         }
         PercentilesAggregatorSupplier percentilesAggregatorSupplier = (PercentilesAggregatorSupplier) aggregatorSupplier;
-        return percentilesAggregatorSupplier.build(name, valuesSource, searchContext, parent, percents, percentilesConfig, keyed,
-            config.format(), metadata);
+        return percentilesAggregatorSupplier.build(
+            name,
+            config.getValuesSource(),
+            searchContext,
+            parent,
+            percents,
+            percentilesConfig,
+            keyed,
+            config.format(),
+            metadata
+        );
     }
 }

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

@@ -48,10 +48,16 @@ class StatsAggregator extends NumericMetricsAggregator.MultiValue {
     DoubleArray mins;
     DoubleArray maxes;
 
-    StatsAggregator(String name, ValuesSourceConfig valuesSourceConfig, ValuesSource valuesSource,
-                    SearchContext context, Aggregator parent, Map<String, Object> metadata) throws IOException {
+    StatsAggregator(
+        String name,
+        ValuesSourceConfig valuesSourceConfig,
+        SearchContext context,
+        Aggregator parent,
+        Map<String, Object> metadata
+    ) throws IOException {
         super(name, context, parent, metadata);
-        this.valuesSource = (ValuesSource.Numeric) valuesSource;
+        // TODO: stop using nulls here
+        this.valuesSource = valuesSourceConfig.hasValues() ? (ValuesSource.Numeric) valuesSourceConfig.getValuesSource() : null;
         if (valuesSource != null) {
             final BigArrays bigArrays = context.bigArrays();
             counts = bigArrays.newLongArray(1, true);

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

@@ -26,7 +26,6 @@ import org.elasticsearch.search.aggregations.AggregatorFactories;
 import org.elasticsearch.search.aggregations.AggregatorFactory;
 import org.elasticsearch.search.aggregations.support.AggregatorSupplier;
 import org.elasticsearch.search.aggregations.support.CoreValuesSourceType;
-import org.elasticsearch.search.aggregations.support.ValuesSource;
 import org.elasticsearch.search.aggregations.support.ValuesSourceAggregatorFactory;
 import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
 import org.elasticsearch.search.aggregations.support.ValuesSourceRegistry;
@@ -56,15 +55,14 @@ class StatsAggregatorFactory extends ValuesSourceAggregatorFactory {
     protected Aggregator createUnmapped(SearchContext searchContext,
                                             Aggregator parent,
                                             Map<String, Object> metadata) throws IOException {
-        return new StatsAggregator(name, config, null, searchContext, parent, metadata);
+        return new StatsAggregator(name, config, searchContext, parent, metadata);
     }
 
     @Override
-    protected Aggregator doCreateInternal(ValuesSource valuesSource,
-                                            SearchContext searchContext,
-                                            Aggregator parent,
-                                            boolean collectsFromSingleBucket,
-                                            Map<String, Object> metadata) throws IOException {
+    protected Aggregator doCreateInternal(SearchContext searchContext,
+                                          Aggregator parent,
+                                          boolean collectsFromSingleBucket,
+                                          Map<String, Object> metadata) throws IOException {
         AggregatorSupplier aggregatorSupplier = queryShardContext.getValuesSourceRegistry().getAggregator(config,
             StatsAggregationBuilder.NAME);
 
@@ -72,6 +70,6 @@ class StatsAggregatorFactory extends ValuesSourceAggregatorFactory {
             throw new AggregationExecutionException("Registry miss-match - expected MetricAggregatorSupplier, found [" +
                 aggregatorSupplier.getClass().toString() + "]");
         }
-        return ((MetricAggregatorSupplier) aggregatorSupplier).build(name, config, valuesSource, searchContext, parent, metadata);
+        return ((MetricAggregatorSupplier) aggregatorSupplier).build(name, config, searchContext, parent, metadata);
     }
 }

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

@@ -44,10 +44,16 @@ public class SumAggregator extends NumericMetricsAggregator.SingleValue {
     private DoubleArray sums;
     private DoubleArray compensations;
 
-    SumAggregator(String name, ValuesSourceConfig valuesSourceConfig, ValuesSource valuesSource,  SearchContext context,
-                  Aggregator parent, Map<String, Object> metadata) throws IOException {
+    SumAggregator(
+        String name,
+        ValuesSourceConfig valuesSourceConfig,
+        SearchContext context,
+        Aggregator parent,
+        Map<String, Object> metadata
+    ) throws IOException {
         super(name, context, parent, metadata);
-        this.valuesSource = (ValuesSource.Numeric) valuesSource;
+        // TODO: stop expecting nulls here
+        this.valuesSource = valuesSourceConfig.hasValues() ? (ValuesSource.Numeric) valuesSourceConfig.getValuesSource() : null;
         this.format = valuesSourceConfig.format();
         if (valuesSource != null) {
             sums = context.bigArrays().newDoubleArray(1, true);

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

@@ -26,7 +26,6 @@ import org.elasticsearch.search.aggregations.AggregatorFactories;
 import org.elasticsearch.search.aggregations.AggregatorFactory;
 import org.elasticsearch.search.aggregations.support.AggregatorSupplier;
 import org.elasticsearch.search.aggregations.support.CoreValuesSourceType;
-import org.elasticsearch.search.aggregations.support.ValuesSource;
 import org.elasticsearch.search.aggregations.support.ValuesSourceAggregatorFactory;
 import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
 import org.elasticsearch.search.aggregations.support.ValuesSourceRegistry;
@@ -57,15 +56,14 @@ class SumAggregatorFactory extends ValuesSourceAggregatorFactory {
                                             Aggregator parent,
                                             Map<String, Object> metadata)
             throws IOException {
-        return new SumAggregator(name, config, null, searchContext, parent, metadata);
+        return new SumAggregator(name, config, searchContext, parent, metadata);
     }
 
     @Override
-    protected Aggregator doCreateInternal(ValuesSource valuesSource,
-                                            SearchContext searchContext,
-                                            Aggregator parent,
-                                            boolean collectsFromSingleBucket,
-                                            Map<String, Object> metadata) throws IOException {
+    protected Aggregator doCreateInternal(SearchContext searchContext,
+                                          Aggregator parent,
+                                          boolean collectsFromSingleBucket,
+                                          Map<String, Object> metadata) throws IOException {
         AggregatorSupplier aggregatorSupplier = queryShardContext.getValuesSourceRegistry().getAggregator(config,
             SumAggregationBuilder.NAME);
 
@@ -73,6 +71,6 @@ class SumAggregatorFactory extends ValuesSourceAggregatorFactory {
             throw new AggregationExecutionException("Registry miss-match - expected MetricAggregatorSupplier, found [" +
                 aggregatorSupplier.getClass().toString() + "]");
         }
-        return ((MetricAggregatorSupplier) aggregatorSupplier).build(name, config, valuesSource, searchContext, parent, metadata);
+        return ((MetricAggregatorSupplier) aggregatorSupplier).build(name, config, searchContext, parent, metadata);
     }
 }

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

@@ -30,6 +30,7 @@ import org.elasticsearch.search.aggregations.InternalAggregation;
 import org.elasticsearch.search.aggregations.LeafBucketCollector;
 import org.elasticsearch.search.aggregations.LeafBucketCollectorBase;
 import org.elasticsearch.search.aggregations.support.ValuesSource;
+import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
 import org.elasticsearch.search.internal.SearchContext;
 
 import java.io.IOException;
@@ -50,12 +51,13 @@ public class ValueCountAggregator extends NumericMetricsAggregator.SingleValue {
 
     public ValueCountAggregator(
             String name,
-            ValuesSource valuesSource,
+            ValuesSourceConfig valuesSourceConfig,
             SearchContext aggregationContext,
             Aggregator parent,
             Map<String, Object> metadata) throws IOException {
         super(name, aggregationContext, parent, metadata);
-        this.valuesSource = valuesSource;
+        // TODO: stop expecting nulls here
+        this.valuesSource = valuesSourceConfig.hasValues() ? valuesSourceConfig.getValuesSource() : null;
         if (valuesSource != null) {
             counts = context.bigArrays().newLongArray(1, true);
         }

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

@@ -26,7 +26,6 @@ import org.elasticsearch.search.aggregations.AggregatorFactories;
 import org.elasticsearch.search.aggregations.AggregatorFactory;
 import org.elasticsearch.search.aggregations.support.AggregatorSupplier;
 import org.elasticsearch.search.aggregations.support.CoreValuesSourceType;
-import org.elasticsearch.search.aggregations.support.ValuesSource;
 import org.elasticsearch.search.aggregations.support.ValuesSourceAggregatorFactory;
 import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
 import org.elasticsearch.search.aggregations.support.ValuesSourceRegistry;
@@ -39,7 +38,7 @@ class ValueCountAggregatorFactory extends ValuesSourceAggregatorFactory {
 
     public static void registerAggregators(ValuesSourceRegistry.Builder builder) {
         builder.register(ValueCountAggregationBuilder.NAME, CoreValuesSourceType.ALL_CORE,
-            (ValueCountAggregatorSupplier) ValueCountAggregator::new);
+            (MetricAggregatorSupplier) ValueCountAggregator::new);
     }
 
     ValueCountAggregatorFactory(String name, ValuesSourceConfig config, QueryShardContext queryShardContext,
@@ -52,22 +51,21 @@ class ValueCountAggregatorFactory extends ValuesSourceAggregatorFactory {
     protected Aggregator createUnmapped(SearchContext searchContext,
                                             Aggregator parent,
                                             Map<String, Object> metadata) throws IOException {
-        return new ValueCountAggregator(name, null, searchContext, parent, metadata);
+        return new ValueCountAggregator(name, config, searchContext, parent, metadata);
     }
 
     @Override
-    protected Aggregator doCreateInternal(ValuesSource valuesSource,
-                                            SearchContext searchContext,
-                                            Aggregator parent,
-                                            boolean collectsFromSingleBucket,
-                                            Map<String, Object> metadata) throws IOException {
+    protected Aggregator doCreateInternal(SearchContext searchContext,
+                                          Aggregator parent,
+                                          boolean collectsFromSingleBucket,
+                                          Map<String, Object> metadata) throws IOException {
         AggregatorSupplier aggregatorSupplier = queryShardContext.getValuesSourceRegistry().getAggregator(config,
             ValueCountAggregationBuilder.NAME);
-        if (aggregatorSupplier instanceof ValueCountAggregatorSupplier == false) {
-            throw new AggregationExecutionException("Registry miss-match - expected ValueCountAggregatorSupplier, found [" +
+        if (aggregatorSupplier instanceof MetricAggregatorSupplier == false) {
+            throw new AggregationExecutionException("Registry miss-match - expected MetricAggregatorSupplier, found [" +
                 aggregatorSupplier.getClass().toString() + "]");
         }
-        return ((ValueCountAggregatorSupplier) aggregatorSupplier)
-            .build(name, valuesSource, searchContext, parent, metadata);
+        return ((MetricAggregatorSupplier) aggregatorSupplier)
+            .build(name, config, searchContext, parent, metadata);
     }
 }

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

@@ -1,35 +0,0 @@
-/*
- * Licensed to Elasticsearch under one or more contributor
- * license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright
- * ownership. Elasticsearch licenses this file to you under
- * the Apache License, Version 2.0 (the "License"); you may
- * not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied.  See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.elasticsearch.search.aggregations.metrics;
-
-import org.elasticsearch.search.aggregations.Aggregator;
-import org.elasticsearch.search.aggregations.support.AggregatorSupplier;
-import org.elasticsearch.search.aggregations.support.ValuesSource;
-import org.elasticsearch.search.internal.SearchContext;
-
-import java.io.IOException;
-import java.util.Map;
-
-public interface ValueCountAggregatorSupplier extends AggregatorSupplier {
-    Aggregator build(String name,
-                     ValuesSource valuesSource,
-                     SearchContext aggregationContext,
-                     Aggregator parent,
-                     Map<String, Object> metadata) throws IOException;
-}

server/src/main/java/org/elasticsearch/search/aggregations/support/AggregatorSupplier.java

@@ -18,5 +18,21 @@
  */
 package org.elasticsearch.search.aggregations.support;
 
+/**
+ * {@link AggregatorSupplier} serves as a marker for what the {@link ValuesSourceRegistry} holds to construct aggregator instances.
+ * The aggregators for each aggregation should all share a signature, and that signature should be used to create an AggregatorSupplier for
+ * that aggregation.  Alternatively, if an existing supplier has a matching signature, please re-use that.
+ *
+ * In many cases, this can be a simple wrapper over the aggregator constructor.  If that is sufficient, please just use a reference to the
+ * constructor.  Implementing a supplier (typically as a lambda) should only be necessary when factors besides the {@link ValuesSourceType}
+ * are necessary for selecting the correct aggregator implementation.  This happens in terms for example where we make decisions based on
+ * the availability of global ordinals.
+ *
+ * The suppliers are responsible for any casting of {@link ValuesSource} that needs to happen.  They must accept a base {@link ValuesSource}
+ * instance.  The suppliers may perform additional logic to configure the aggregator as needed, such as in
+ * {@link org.elasticsearch.search.aggregations.bucket.terms.TermsAggregatorFactory} deciding the execution mode.
+ *
+ * There is ongoing work to  normalize aggregator constructor signatures, and thus reduce the number of AggregatorSupplier interfaces.
+ */
 public interface AggregatorSupplier {
 }

server/src/main/java/org/elasticsearch/search/aggregations/support/HistogramAggregatorSupplier.java

@@ -18,11 +18,11 @@
  */
 package org.elasticsearch.search.aggregations.support;
 
-import org.elasticsearch.common.Nullable;
-import org.elasticsearch.search.DocValueFormat;
 import org.elasticsearch.search.aggregations.Aggregator;
 import org.elasticsearch.search.aggregations.AggregatorFactories;
 import org.elasticsearch.search.aggregations.BucketOrder;
+import org.elasticsearch.search.aggregations.support.AggregatorSupplier;
+import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
 import org.elasticsearch.search.internal.SearchContext;
 
 import java.io.IOException;
@@ -39,8 +39,7 @@ public interface HistogramAggregatorSupplier extends AggregatorSupplier {
         long minDocCount,
         double minBound,
         double maxBound,
-        @Nullable ValuesSource valuesSource,
-        DocValueFormat formatter,
+        ValuesSourceConfig valuesSourceConfig,
         SearchContext context,
         Aggregator parent,
         boolean collectsFromSingleBucket,

server/src/main/java/org/elasticsearch/search/aggregations/support/MultiValuesSource.java

@@ -39,7 +39,7 @@ public abstract class MultiValuesSource <VS extends ValuesSource> {
                                         QueryShardContext context) {
             values = new HashMap<>(valuesSourceConfigs.size());
             for (Map.Entry<String, ValuesSourceConfig> entry : valuesSourceConfigs.entrySet()) {
-                final ValuesSource valuesSource = entry.getValue().toValuesSource();
+                final ValuesSource valuesSource = entry.getValue().getValuesSource();
                 if (valuesSource instanceof ValuesSource.Numeric == false) {
                     throw new AggregationExecutionException("ValuesSource type " + valuesSource.toString() +
                         "is not supported for multi-valued aggregation");

server/src/main/java/org/elasticsearch/search/aggregations/support/ValuesSource.java

@@ -78,6 +78,13 @@ public abstract class ValuesSource {
      */
     public abstract Function<Rounding, Rounding.Prepared> roundingPreparer(IndexReader reader) throws IOException;
 
+    /**
+     * Check if this values source supports using global ordinals
+     */
+    public boolean hasGlobalOrdinals() {
+        return false;
+    }
+
     public static class Range extends ValuesSource {
         private final RangeType rangeType;
         protected final IndexFieldData<?> indexFieldData;
@@ -166,6 +173,11 @@ public abstract class ValuesSource {
                 return true;
             }
 
+            @Override
+            public boolean hasGlobalOrdinals() {
+                return true;
+            }
+
             /** Returns a mapping from segment ordinals to global ordinals. */
             public abstract LongUnaryOperator globalOrdinalsMapping(LeafReaderContext context)
                     throws IOException;

server/src/main/java/org/elasticsearch/search/aggregations/support/ValuesSourceAggregatorFactory.java

@@ -42,19 +42,17 @@ public abstract class ValuesSourceAggregatorFactory extends AggregatorFactory {
     @Override
     public Aggregator createInternal(SearchContext searchContext, Aggregator parent, boolean collectsFromSingleBucket,
                                      Map<String, Object> metadata) throws IOException {
-        ValuesSource vs = config.toValuesSource();
-        if (vs == null) {
+        if (config.hasValues() == false) {
             return createUnmapped(searchContext, parent, metadata);
         }
-        return doCreateInternal(vs, searchContext, parent, collectsFromSingleBucket, metadata);
+        return doCreateInternal(searchContext, parent, collectsFromSingleBucket, metadata);
     }
 
     protected abstract Aggregator createUnmapped(SearchContext searchContext,
                                                  Aggregator parent,
                                                  Map<String, Object> metadata) throws IOException;
 
-    protected abstract Aggregator doCreateInternal(ValuesSource valuesSource,
-                                                   SearchContext searchContext,
+    protected abstract Aggregator doCreateInternal(SearchContext searchContext,
                                                    Aggregator parent,
                                                    boolean collectsFromSingleBucket,
                                                    Map<String, Object> metadata) throws IOException;

server/src/main/java/org/elasticsearch/search/aggregations/support/ValuesSourceConfig.java

@@ -154,10 +154,18 @@ public class ValuesSourceConfig {
         if (valuesSourceType == null) {
             valuesSourceType = defaultValueSourceType;
         }
-        config = new ValuesSourceConfig(valuesSourceType, fieldContext, unmapped, aggregationScript, scriptValueType, context::nowInMillis);
-        config.format(resolveFormat(format, valuesSourceType, timeZone, fieldType));
-        config.missing(missing);
-        config.timezone(timeZone);
+        DocValueFormat docValueFormat = resolveFormat(format, valuesSourceType, timeZone, fieldType);
+        config = new ValuesSourceConfig(
+            valuesSourceType,
+            fieldContext,
+            unmapped,
+            aggregationScript,
+            scriptValueType,
+            missing,
+            timeZone,
+            docValueFormat,
+            context::nowInMillis
+        );
         return config;
     }
 
@@ -231,6 +239,9 @@ public class ValuesSourceConfig {
             false,
             null,
             null,
+            null,
+            null,
+            null,
             queryShardContext::nowInMillis
         );
     }
@@ -239,19 +250,23 @@ public class ValuesSourceConfig {
      * Convenience method for creating unmapped configs
      */
     public static ValuesSourceConfig resolveUnmapped(ValuesSourceType valuesSourceType, QueryShardContext queryShardContext) {
-        return new ValuesSourceConfig(valuesSourceType, null, true, null, null, queryShardContext::nowInMillis);
+        return new ValuesSourceConfig(valuesSourceType, null, true, null, null, null, null, null, queryShardContext::nowInMillis);
     }
 
     private final ValuesSourceType valuesSourceType;
-    private FieldContext fieldContext;
-    private AggregationScript.LeafFactory script;
-    private ValueType scriptValueType;
-    private boolean unmapped;
-    private DocValueFormat format = DocValueFormat.RAW;
-    private Object missing;
-    private ZoneId timeZone;
-    private LongSupplier nowSupplier;
+    private final FieldContext fieldContext;
+    private final AggregationScript.LeafFactory script;
+    private final ValueType scriptValueType;
+    private final boolean unmapped;
+    private final DocValueFormat format;
+    private final Object missing;
+    private final ZoneId timeZone;
+    private final LongSupplier nowSupplier;
+    private final ValuesSource valuesSource;
 
+    private ValuesSourceConfig() {
+        throw new UnsupportedOperationException();
+    }
 
     public ValuesSourceConfig(
         ValuesSourceType valuesSourceType,
@@ -259,6 +274,9 @@ public class ValuesSourceConfig {
         boolean unmapped,
         AggregationScript.LeafFactory script,
         ValueType scriptValueType,
+        Object missing,
+        ZoneId timeZone,
+        DocValueFormat format,
         LongSupplier nowSupplier
     ) {
         if (unmapped && fieldContext != null) {
@@ -269,8 +287,38 @@ public class ValuesSourceConfig {
         this.unmapped = unmapped;
         this.script = script;
         this.scriptValueType = scriptValueType;
+        this.missing = missing;
+        this.timeZone = timeZone;
+        this.format = format == null ? DocValueFormat.RAW : format;
         this.nowSupplier = nowSupplier;
 
+        if (!valid()) {
+            // TODO: resolve no longer generates invalid configs.  Once VSConfig is immutable, we can drop this check
+            throw new IllegalStateException(
+                "value source config is invalid; must have either a field context or a script or marked as unwrapped");
+        }
+        valuesSource = ConstructValuesSource(missing, format, nowSupplier);
+    }
+
+    private ValuesSource ConstructValuesSource(Object missing, DocValueFormat format, LongSupplier nowSupplier) {
+        final ValuesSource vs;
+        if (this.unmapped) {
+            vs = valueSourceType().getEmpty();
+        } else {
+            if (fieldContext() == null) {
+                // Script case
+                vs = valueSourceType().getScript(script(), scriptValueType());
+            } else {
+                // Field or Value Script case
+                vs = valueSourceType().getField(fieldContext(), script());
+            }
+        }
+
+        if (missing() != null) {
+            return valueSourceType().replaceMissing(vs, missing, format, nowSupplier);
+        } else {
+            return vs;
+        }
     }
 
     public ValuesSourceType valueSourceType() {
@@ -285,8 +333,12 @@ public class ValuesSourceConfig {
         return script;
     }
 
-    public boolean unmapped() {
-        return unmapped;
+    /**
+     * Returns true if the values source configured by this object can yield values.  We might not be able to yield values if, for example,
+     * the specified field does not exist on this index.
+     */
+    public boolean hasValues() {
+        return fieldContext != null || script != null || missing != null;
     }
 
     public boolean valid() {
@@ -297,25 +349,10 @@ public class ValuesSourceConfig {
         return this.scriptValueType;
     }
 
-    private ValuesSourceConfig format(final DocValueFormat format) {
-        this.format = format;
-        return this;
-    }
-
-    private ValuesSourceConfig missing(final Object missing) {
-        this.missing = missing;
-        return this;
-    }
-
     public Object missing() {
         return this.missing;
     }
 
-    private ValuesSourceConfig timezone(final ZoneId timeZone) {
-        this.timeZone = timeZone;
-        return this;
-    }
-
     public ZoneId timezone() {
         return this.timeZone;
     }
@@ -324,41 +361,11 @@ public class ValuesSourceConfig {
         return format;
     }
 
-    /**
-     * Transform the {@link ValuesSourceType} we selected in resolve into the specific {@link ValuesSource} instance to use for this shard
-     * @return - A {@link ValuesSource} ready to be read from by an aggregator
-     */
-    @Nullable
-    public ValuesSource toValuesSource() {
-        if (!valid()) {
-            // TODO: resolve no longer generates invalid configs.  Once VSConfig is immutable, we can drop this check
-            throw new IllegalStateException(
-                "value source config is invalid; must have either a field context or a script or marked as unwrapped");
-        }
+    public ValuesSource getValuesSource() {
+        return valuesSource;
+    }
 
-        final ValuesSource vs;
-        if (unmapped()) {
-            if (missing() == null) {
-                /* Null values source signals to the AggregationBuilder to use the createUnmapped method, which aggregator factories can
-                 * override to provide an aggregator optimized to return empty values
-                 */
-                vs = null;
-            } else {
-                vs = valueSourceType().getEmpty();
-            }
-        } else {
-            if (fieldContext() == null) {
-                // Script case
-                vs = valueSourceType().getScript(script(), scriptValueType());
-            } else {
-                // Field or Value Script case
-                vs = valueSourceType().getField(fieldContext(), script());
-            }
-        }
-
-        if (missing() == null) {
-            return vs;
-        }
-        return valueSourceType().replaceMissing(vs, missing, format, nowSupplier);
+    public boolean hasGlobalOrdinals() {
+        return valuesSource.hasGlobalOrdinals();
     }
 }

server/src/test/java/org/elasticsearch/search/aggregations/metrics/WeightedAvgAggregatorTests.java

@@ -83,6 +83,36 @@ public class WeightedAvgAggregatorTests extends AggregatorTestCase {
         });
     }
 
+    public void testUnmappedWeight() throws IOException {
+        MultiValuesSourceFieldConfig valueConfig = new MultiValuesSourceFieldConfig.Builder().setFieldName("value_field").build();
+        MultiValuesSourceFieldConfig weightConfig = new MultiValuesSourceFieldConfig.Builder().setFieldName("weight_field").build();
+        WeightedAvgAggregationBuilder aggregationBuilder = new WeightedAvgAggregationBuilder("_name")
+            .value(valueConfig)
+            .weight(weightConfig);
+        testCase(new MatchAllDocsQuery(), aggregationBuilder, iw -> {
+            iw.addDocument(singleton(new SortedNumericDocValuesField("value_field", 7)));
+            iw.addDocument(singleton(new SortedNumericDocValuesField("value_field", 3)));
+        }, avg -> {
+            assertEquals(Double.NaN, avg.getValue(), 0);
+            assertFalse(AggregationInspectionHelper.hasValue(avg));
+        });
+    }
+
+    public void testUnmappedValue() throws IOException {
+        MultiValuesSourceFieldConfig valueConfig = new MultiValuesSourceFieldConfig.Builder().setFieldName("value_field").build();
+        MultiValuesSourceFieldConfig weightConfig = new MultiValuesSourceFieldConfig.Builder().setFieldName("weight_field").build();
+        WeightedAvgAggregationBuilder aggregationBuilder = new WeightedAvgAggregationBuilder("_name")
+            .value(valueConfig)
+            .weight(weightConfig);
+        testCase(new MatchAllDocsQuery(), aggregationBuilder, iw -> {
+            iw.addDocument(singleton(new SortedNumericDocValuesField("weight_field", 7)));
+            iw.addDocument(singleton(new SortedNumericDocValuesField("weight_field", 3)));
+        }, avg -> {
+            assertEquals(Double.NaN, avg.getValue(), 0);
+            assertFalse(AggregationInspectionHelper.hasValue(avg));
+        });
+    }
+
     public void testSomeMatchesSortedNumericDocValuesNoWeight() throws IOException {
         MultiValuesSourceFieldConfig valueConfig = new MultiValuesSourceFieldConfig.Builder().setFieldName("value_field").build();
         MultiValuesSourceFieldConfig weightConfig = new MultiValuesSourceFieldConfig.Builder().setFieldName("weight_field").build();

server/src/test/java/org/elasticsearch/search/aggregations/support/ValuesSourceConfigTests.java

@@ -47,7 +47,7 @@ public class ValuesSourceConfigTests extends ESSingleNodeTestCase {
 
             ValuesSourceConfig config = ValuesSourceConfig.resolve(
                     context, null, "bytes", null, null, null, null, CoreValuesSourceType.BYTES, null);
-            ValuesSource.Bytes valuesSource = (ValuesSource.Bytes) config.toValuesSource();
+            ValuesSource.Bytes valuesSource = (ValuesSource.Bytes) config.getValuesSource();
             LeafReaderContext ctx = searcher.getIndexReader().leaves().get(0);
             SortedBinaryDocValues values = valuesSource.bytesValues(ctx);
             assertTrue(values.advanceExact(0));
@@ -69,14 +69,14 @@ public class ValuesSourceConfigTests extends ESSingleNodeTestCase {
 
             ValuesSourceConfig config = ValuesSourceConfig.resolve(
                     context, null, "bytes", null, null, null, null, CoreValuesSourceType.BYTES, null);
-            ValuesSource.Bytes valuesSource = (ValuesSource.Bytes) config.toValuesSource();
+            ValuesSource.Bytes valuesSource = (ValuesSource.Bytes) config.getValuesSource();
             LeafReaderContext ctx = searcher.getIndexReader().leaves().get(0);
             SortedBinaryDocValues values = valuesSource.bytesValues(ctx);
             assertFalse(values.advanceExact(0));
 
             config = ValuesSourceConfig.resolve(
                     context, null, "bytes", null, "abc", null, null, CoreValuesSourceType.BYTES, null);
-            valuesSource = (ValuesSource.Bytes) config.toValuesSource();
+            valuesSource = (ValuesSource.Bytes) config.getValuesSource();
             values = valuesSource.bytesValues(ctx);
             assertTrue(values.advanceExact(0));
             assertEquals(1, values.docValueCount());
@@ -95,12 +95,13 @@ public class ValuesSourceConfigTests extends ESSingleNodeTestCase {
             QueryShardContext context = indexService.newQueryShardContext(0, searcher, () -> 42L, null);
             ValuesSourceConfig config = ValuesSourceConfig.resolve(
                     context, ValueType.STRING, "bytes", null, null, null, null, CoreValuesSourceType.BYTES, null);
-            ValuesSource.Bytes valuesSource = (ValuesSource.Bytes) config.toValuesSource();
-            assertNull(valuesSource);
+            ValuesSource.Bytes valuesSource = (ValuesSource.Bytes) config.getValuesSource();
+            assertNotNull(valuesSource);
+            assertFalse(config.hasValues());
 
             config = ValuesSourceConfig.resolve(
                     context, ValueType.STRING, "bytes", null, "abc", null, null, CoreValuesSourceType.BYTES, null);
-            valuesSource = (ValuesSource.Bytes) config.toValuesSource();
+            valuesSource = (ValuesSource.Bytes) config.getValuesSource();
             LeafReaderContext ctx = searcher.getIndexReader().leaves().get(0);
             SortedBinaryDocValues values = valuesSource.bytesValues(ctx);
             assertTrue(values.advanceExact(0));
@@ -122,7 +123,7 @@ public class ValuesSourceConfigTests extends ESSingleNodeTestCase {
 
             ValuesSourceConfig config = ValuesSourceConfig.resolve(
                     context, null, "long", null, null, null, null, CoreValuesSourceType.BYTES, null);
-            ValuesSource.Numeric valuesSource = (ValuesSource.Numeric) config.toValuesSource();
+            ValuesSource.Numeric valuesSource = (ValuesSource.Numeric) config.getValuesSource();
             LeafReaderContext ctx = searcher.getIndexReader().leaves().get(0);
             SortedNumericDocValues values = valuesSource.longValues(ctx);
             assertTrue(values.advanceExact(0));
@@ -144,14 +145,14 @@ public class ValuesSourceConfigTests extends ESSingleNodeTestCase {
 
             ValuesSourceConfig config = ValuesSourceConfig.resolve(
                     context, null, "long", null, null, null, null, CoreValuesSourceType.BYTES, null);
-            ValuesSource.Numeric valuesSource = (ValuesSource.Numeric) config.toValuesSource();
+            ValuesSource.Numeric valuesSource = (ValuesSource.Numeric) config.getValuesSource();
             LeafReaderContext ctx = searcher.getIndexReader().leaves().get(0);
             SortedNumericDocValues values = valuesSource.longValues(ctx);
             assertFalse(values.advanceExact(0));
 
             config = ValuesSourceConfig.resolve(
                     context, null, "long", null, 42, null, null, CoreValuesSourceType.BYTES, null);
-            valuesSource = (ValuesSource.Numeric) config.toValuesSource();
+            valuesSource = (ValuesSource.Numeric) config.getValuesSource();
             values = valuesSource.longValues(ctx);
             assertTrue(values.advanceExact(0));
             assertEquals(1, values.docValueCount());
@@ -171,12 +172,13 @@ public class ValuesSourceConfigTests extends ESSingleNodeTestCase {
 
             ValuesSourceConfig config = ValuesSourceConfig.resolve(
                     context, ValueType.NUMBER, "long", null, null, null, null, CoreValuesSourceType.BYTES, null);
-            ValuesSource.Numeric valuesSource = (ValuesSource.Numeric) config.toValuesSource();
-            assertNull(valuesSource);
+            ValuesSource.Numeric valuesSource = (ValuesSource.Numeric) config.getValuesSource();
+            assertNotNull(valuesSource);
+            assertFalse(config.hasValues());
 
             config = ValuesSourceConfig.resolve(
                     context, ValueType.NUMBER, "long", null, 42, null, null, CoreValuesSourceType.BYTES, null);
-            valuesSource = (ValuesSource.Numeric) config.toValuesSource();
+            valuesSource = (ValuesSource.Numeric) config.getValuesSource();
             LeafReaderContext ctx = searcher.getIndexReader().leaves().get(0);
             SortedNumericDocValues values = valuesSource.longValues(ctx);
             assertTrue(values.advanceExact(0));
@@ -198,7 +200,7 @@ public class ValuesSourceConfigTests extends ESSingleNodeTestCase {
 
             ValuesSourceConfig config = ValuesSourceConfig.resolve(
                     context, null, "bool", null, null, null, null, CoreValuesSourceType.BYTES, null);
-            ValuesSource.Numeric valuesSource = (ValuesSource.Numeric) config.toValuesSource();
+            ValuesSource.Numeric valuesSource = (ValuesSource.Numeric) config.getValuesSource();
             LeafReaderContext ctx = searcher.getIndexReader().leaves().get(0);
             SortedNumericDocValues values = valuesSource.longValues(ctx);
             assertTrue(values.advanceExact(0));
@@ -220,14 +222,14 @@ public class ValuesSourceConfigTests extends ESSingleNodeTestCase {
 
             ValuesSourceConfig config = ValuesSourceConfig.resolve(
                     context, null, "bool", null, null, null, null, CoreValuesSourceType.BYTES, null);
-            ValuesSource.Numeric valuesSource = (ValuesSource.Numeric) config.toValuesSource();
+            ValuesSource.Numeric valuesSource = (ValuesSource.Numeric) config.getValuesSource();
             LeafReaderContext ctx = searcher.getIndexReader().leaves().get(0);
             SortedNumericDocValues values = valuesSource.longValues(ctx);
             assertFalse(values.advanceExact(0));
 
             config = ValuesSourceConfig.resolve(
                     context, null, "bool", null, true, null, null, CoreValuesSourceType.BYTES, null);
-            valuesSource = (ValuesSource.Numeric) config.toValuesSource();
+            valuesSource = (ValuesSource.Numeric) config.getValuesSource();
             values = valuesSource.longValues(ctx);
             assertTrue(values.advanceExact(0));
             assertEquals(1, values.docValueCount());
@@ -247,12 +249,13 @@ public class ValuesSourceConfigTests extends ESSingleNodeTestCase {
 
             ValuesSourceConfig config = ValuesSourceConfig.resolve(
                     context, ValueType.BOOLEAN, "bool", null, null, null, null, CoreValuesSourceType.BYTES, null);
-            ValuesSource.Numeric valuesSource = (ValuesSource.Numeric) config.toValuesSource();
-            assertNull(valuesSource);
+            ValuesSource.Numeric valuesSource = (ValuesSource.Numeric) config.getValuesSource();
+            assertNotNull(valuesSource);
+            assertFalse(config.hasValues());
 
             config = ValuesSourceConfig.resolve(
                     context, ValueType.BOOLEAN, "bool", null, true, null, null, CoreValuesSourceType.BYTES, null);
-            valuesSource = (ValuesSource.Numeric) config.toValuesSource();
+            valuesSource = (ValuesSource.Numeric) config.getValuesSource();
             LeafReaderContext ctx = searcher.getIndexReader().leaves().get(0);
             SortedNumericDocValues values = valuesSource.longValues(ctx);
             assertTrue(values.advanceExact(0));
@@ -284,7 +287,7 @@ public class ValuesSourceConfigTests extends ESSingleNodeTestCase {
             QueryShardContext context = indexService.newQueryShardContext(0, searcher, () -> 42L, null);
             ValuesSourceConfig config = ValuesSourceConfig.resolve(
                 context, ValueType.STRING, "alias", null, null, null, null, CoreValuesSourceType.BYTES, null);
-            ValuesSource.Bytes valuesSource = (ValuesSource.Bytes) config.toValuesSource();
+            ValuesSource.Bytes valuesSource = (ValuesSource.Bytes) config.getValuesSource();
 
             LeafReaderContext ctx = searcher.getIndexReader().leaves().get(0);
             SortedBinaryDocValues values = valuesSource.bytesValues(ctx);

x-pack/plugin/analytics/src/main/java/org/elasticsearch/xpack/analytics/aggregations/metrics/AnalyticsAggregatorFactory.java

@@ -14,10 +14,8 @@ import org.elasticsearch.search.aggregations.metrics.PercentilesConfig;
 import org.elasticsearch.search.aggregations.metrics.PercentilesMethod;
 import org.elasticsearch.search.aggregations.metrics.SumAggregationBuilder;
 import org.elasticsearch.search.aggregations.metrics.ValueCountAggregationBuilder;
-import org.elasticsearch.search.aggregations.metrics.ValueCountAggregatorSupplier;
 import org.elasticsearch.search.aggregations.support.ValuesSourceRegistry;
 import org.elasticsearch.xpack.analytics.aggregations.support.AnalyticsValuesSourceType;
-import org.elasticsearch.xpack.analytics.aggregations.support.HistogramValuesSource;
 
 public class AnalyticsAggregatorFactory {
 
@@ -68,36 +66,21 @@ public class AnalyticsAggregatorFactory {
     public static void registerHistoBackedSumAggregator(ValuesSourceRegistry.Builder builder) {
         builder.register(SumAggregationBuilder.NAME,
             AnalyticsValuesSourceType.HISTOGRAM,
-            (MetricAggregatorSupplier) (name, valuesSourceConfig, valuesSource, context, parent, metadata) -> new HistoBackedSumAggregator(
-                name,
-                (HistogramValuesSource.Histogram) valuesSource,
-                valuesSourceConfig.format(),
-                context,
-                parent,
-                metadata
-            )
+            (MetricAggregatorSupplier) HistoBackedSumAggregator::new
         );
     }
 
     public static void registerHistoBackedValueCountAggregator(ValuesSourceRegistry.Builder builder) {
         builder.register(ValueCountAggregationBuilder.NAME,
             AnalyticsValuesSourceType.HISTOGRAM,
-            (ValueCountAggregatorSupplier) (name, valuesSource, context, parent, metadata) ->
-                new HistoBackedValueCountAggregator(name, (HistogramValuesSource.Histogram) valuesSource, context, parent, metadata)
+            (MetricAggregatorSupplier) HistoBackedValueCountAggregator::new
         );
     }
 
     public static void registerHistoBackedAverageAggregator(ValuesSourceRegistry.Builder builder) {
         builder.register(AvgAggregationBuilder.NAME,
             AnalyticsValuesSourceType.HISTOGRAM,
-            (MetricAggregatorSupplier) (name, valuesSourceConfig, valuesSource, context, parent, metadata) -> new HistoBackedAvgAggregator(
-                name,
-                (HistogramValuesSource.Histogram) valuesSource,
-                valuesSourceConfig.format(),
-                context,
-                parent,
-                metadata
-            )
+            (MetricAggregatorSupplier) HistoBackedAvgAggregator::new
         );
     }
 }

x-pack/plugin/analytics/src/main/java/org/elasticsearch/xpack/analytics/aggregations/metrics/HistoBackedAvgAggregator.java

@@ -21,6 +21,7 @@ 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.analytics.aggregations.support.HistogramValuesSource;
 
@@ -40,11 +41,17 @@ class HistoBackedAvgAggregator extends NumericMetricsAggregator.SingleValue {
     DoubleArray compensations;
     DocValueFormat format;
 
-    HistoBackedAvgAggregator(String name, HistogramValuesSource.Histogram valuesSource, DocValueFormat formatter, SearchContext context,
-            Aggregator parent, Map<String, Object> metadata) throws IOException {
+    HistoBackedAvgAggregator(
+        String name,
+        ValuesSourceConfig valuesSourceConfig,
+        SearchContext context,
+        Aggregator parent,
+        Map<String, Object> metadata
+    ) throws IOException {
         super(name, context, parent, metadata);
-        this.valuesSource = valuesSource;
-        this.format = formatter;
+        // TODO: Stop depending on nulls here
+        this.valuesSource = valuesSourceConfig.hasValues() ? (HistogramValuesSource.Histogram) valuesSourceConfig.getValuesSource() : null;
+        this.format = valuesSourceConfig.format();
         if (valuesSource != null) {
             final BigArrays bigArrays = context.bigArrays();
             counts = bigArrays.newLongArray(1, true);

x-pack/plugin/analytics/src/main/java/org/elasticsearch/xpack/analytics/aggregations/metrics/HistoBackedSumAggregator.java

@@ -20,6 +20,7 @@ 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.analytics.aggregations.support.HistogramValuesSource;
 
@@ -40,11 +41,17 @@ class HistoBackedSumAggregator extends NumericMetricsAggregator.SingleValue {
     private DoubleArray sums;
     private DoubleArray compensations;
 
-    HistoBackedSumAggregator(String name, HistogramValuesSource.Histogram valuesSource, DocValueFormat formatter, SearchContext context,
-            Aggregator parent, Map<String, Object> metadata) throws IOException {
+    HistoBackedSumAggregator(
+        String name,
+        ValuesSourceConfig valuesSourceConfig,
+        SearchContext context,
+        Aggregator parent,
+        Map<String, Object> metadata
+    ) throws IOException {
         super(name, context, parent, metadata);
-        this.valuesSource = valuesSource;
-        this.format = formatter;
+        // TODO: stop expecting a null here
+        this.valuesSource = valuesSourceConfig.hasValues() ? (HistogramValuesSource.Histogram) valuesSourceConfig.getValuesSource() : null;
+        this.format = valuesSourceConfig.format();
         if (valuesSource != null) {
             sums = context.bigArrays().newDoubleArray(1, true);
             compensations = context.bigArrays().newDoubleArray(1, true);

x-pack/plugin/analytics/src/main/java/org/elasticsearch/xpack/analytics/aggregations/metrics/HistoBackedValueCountAggregator.java

@@ -17,6 +17,7 @@ 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.analytics.aggregations.support.HistogramValuesSource;
 
@@ -37,12 +38,13 @@ public class HistoBackedValueCountAggregator extends NumericMetricsAggregator.Si
 
     public HistoBackedValueCountAggregator(
             String name,
-            HistogramValuesSource.Histogram valuesSource,
+            ValuesSourceConfig valuesSourceConfig,
             SearchContext aggregationContext,
             Aggregator parent,
             Map<String, Object> metadata) throws IOException {
         super(name, aggregationContext, parent, metadata);
-        this.valuesSource = valuesSource;
+        // TODO: stop using nulls here
+        this.valuesSource = valuesSourceConfig.hasValues() ? (HistogramValuesSource.Histogram) valuesSourceConfig.getValuesSource() : null;
         if (valuesSource != null) {
             counts = context.bigArrays().newLongArray(1, true);
         }

x-pack/plugin/analytics/src/main/java/org/elasticsearch/xpack/analytics/boxplot/BoxplotAggregatorFactory.java

@@ -13,7 +13,6 @@ import org.elasticsearch.search.aggregations.AggregatorFactories;
 import org.elasticsearch.search.aggregations.AggregatorFactory;
 import org.elasticsearch.search.aggregations.support.AggregatorSupplier;
 import org.elasticsearch.search.aggregations.support.CoreValuesSourceType;
-import org.elasticsearch.search.aggregations.support.ValuesSource;
 import org.elasticsearch.search.aggregations.support.ValuesSourceAggregatorFactory;
 import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
 import org.elasticsearch.search.aggregations.support.ValuesSourceRegistry;
@@ -54,8 +53,7 @@ public class BoxplotAggregatorFactory extends ValuesSourceAggregatorFactory {
     }
 
     @Override
-    protected Aggregator doCreateInternal(ValuesSource valuesSource,
-                                          SearchContext searchContext,
+    protected Aggregator doCreateInternal(SearchContext searchContext,
                                           Aggregator parent,
                                           boolean collectsFromSingleBucket,
                                           Map<String, Object> metadata) throws IOException {
@@ -66,7 +64,7 @@ public class BoxplotAggregatorFactory extends ValuesSourceAggregatorFactory {
             throw new AggregationExecutionException("Registry miss-match - expected BoxplotAggregatorSupplier, found [" +
                 aggregatorSupplier.getClass().toString() + "]");
         }
-        return ((BoxplotAggregatorSupplier) aggregatorSupplier).build(name, valuesSource, config.format(), compression,
+        return ((BoxplotAggregatorSupplier) aggregatorSupplier).build(name, config.getValuesSource(), config.format(), compression,
             searchContext, parent, metadata);
     }
 }

x-pack/plugin/analytics/src/main/java/org/elasticsearch/xpack/analytics/stringstats/StringStatsAggregatorFactory.java

@@ -13,7 +13,6 @@ import org.elasticsearch.search.aggregations.AggregatorFactories;
 import org.elasticsearch.search.aggregations.AggregatorFactory;
 import org.elasticsearch.search.aggregations.support.AggregatorSupplier;
 import org.elasticsearch.search.aggregations.support.CoreValuesSourceType;
-import org.elasticsearch.search.aggregations.support.ValuesSource;
 import org.elasticsearch.search.aggregations.support.ValuesSourceAggregatorFactory;
 import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
 import org.elasticsearch.search.aggregations.support.ValuesSourceRegistry;
@@ -48,8 +47,7 @@ class StringStatsAggregatorFactory extends ValuesSourceAggregatorFactory {
     }
 
     @Override
-    protected Aggregator doCreateInternal(ValuesSource valuesSource,
-                                          SearchContext searchContext,
+    protected Aggregator doCreateInternal(SearchContext searchContext,
                                           Aggregator parent,
                                           boolean collectsFromSingleBucket,
                                           Map<String, Object> metadata) throws IOException {
@@ -60,7 +58,7 @@ class StringStatsAggregatorFactory extends ValuesSourceAggregatorFactory {
             throw new AggregationExecutionException("Registry miss-match - expected StringStatsAggregatorSupplier, found [" +
                 aggregatorSupplier.getClass().toString() + "]");
         }
-        return ((StringStatsAggregatorSupplier) aggregatorSupplier).build(name, valuesSource, showDistribution, config.format(),
+        return ((StringStatsAggregatorSupplier) aggregatorSupplier).build(name, config.getValuesSource(), showDistribution, config.format(),
             searchContext, parent, metadata);
     }
 

x-pack/plugin/analytics/src/main/java/org/elasticsearch/xpack/analytics/topmetrics/TopMetricsAggregatorFactory.java

@@ -63,7 +63,7 @@ public class TopMetricsAggregatorFactory extends AggregatorFactory {
                             config.getFieldName(), config.getScript(), config.getMissing(), config.getTimeZone(), null,
                         CoreValuesSourceType.NUMERIC, TopMetricsAggregationBuilder.NAME);
                     return new TopMetricsAggregator.MetricSource(config.getFieldName(), resolved.format(),
-                        (ValuesSource.Numeric) resolved.toValuesSource());
+                        (ValuesSource.Numeric) resolved.getValuesSource());
                 }).collect(toList());
         return new TopMetricsAggregator(name, searchContext, parent, metadata, size, sortBuilders.get(0), metricSources);
     }

x-pack/plugin/spatial/src/main/java/org/elasticsearch/xpack/spatial/SpatialPlugin.java

@@ -22,11 +22,10 @@ import org.elasticsearch.search.aggregations.metrics.CardinalityAggregatorSuppli
 import org.elasticsearch.search.aggregations.metrics.GeoBoundsAggregationBuilder;
 import org.elasticsearch.search.aggregations.metrics.GeoBoundsAggregatorSupplier;
 import org.elasticsearch.search.aggregations.metrics.GeoCentroidAggregationBuilder;
-import org.elasticsearch.search.aggregations.metrics.GeoCentroidAggregatorSupplier;
 import org.elasticsearch.search.aggregations.metrics.GeoGridAggregatorSupplier;
+import org.elasticsearch.search.aggregations.metrics.MetricAggregatorSupplier;
 import org.elasticsearch.search.aggregations.metrics.ValueCountAggregationBuilder;
 import org.elasticsearch.search.aggregations.metrics.ValueCountAggregator;
-import org.elasticsearch.search.aggregations.metrics.ValueCountAggregatorSupplier;
 import org.elasticsearch.search.aggregations.support.ValuesSourceRegistry;
 import org.elasticsearch.xpack.core.XPackPlugin;
 import org.elasticsearch.xpack.spatial.aggregations.metrics.GeoShapeCentroidAggregator;
@@ -101,17 +100,15 @@ public class SpatialPlugin extends GeoPlugin implements MapperPlugin, SearchPlug
 
     private static void registerGeoShapeBoundsAggregator(ValuesSourceRegistry.Builder builder) {
         builder.register(GeoBoundsAggregationBuilder.NAME, GeoShapeValuesSourceType.instance(),
-            (GeoBoundsAggregatorSupplier) (name, aggregationContext, parent, valuesSource, wrapLongitude, metadata)
-                -> new GeoShapeBoundsAggregator(name, aggregationContext, parent, (GeoShapeValuesSource) valuesSource,
-                wrapLongitude, metadata));
+            (GeoBoundsAggregatorSupplier) GeoShapeBoundsAggregator::new);
     }
 
     private void registerGeoShapeCentroidAggregator(ValuesSourceRegistry.Builder builder) {
         builder.register(GeoCentroidAggregationBuilder.NAME, GeoShapeValuesSourceType.instance(),
-            (GeoCentroidAggregatorSupplier) (name, aggregationContext, parent, valuesSource, metadata)
+            (MetricAggregatorSupplier) (name, valuesSourceConfig, aggregationContext, parent, metadata)
                 -> {
                 if (getLicenseState().isAllowed(XPackLicenseState.Feature.SPATIAL_GEO_CENTROID)) {
-                    return new GeoShapeCentroidAggregator(name, aggregationContext, parent, (GeoShapeValuesSource) valuesSource, metadata);
+                    return new GeoShapeCentroidAggregator(name, aggregationContext, parent, valuesSourceConfig, metadata);
                 }
                 throw LicenseUtils.newComplianceException("geo_centroid aggregation on geo_shape fields");
             });
@@ -161,7 +158,7 @@ public class SpatialPlugin extends GeoPlugin implements MapperPlugin, SearchPlug
 
     private static void registerValueCountAggregator(ValuesSourceRegistry.Builder builder) {
         builder.register(ValueCountAggregationBuilder.NAME, GeoShapeValuesSourceType.instance(),
-            (ValueCountAggregatorSupplier) ValueCountAggregator::new
+            (MetricAggregatorSupplier) ValueCountAggregator::new
         );
     }
 

x-pack/plugin/spatial/src/main/java/org/elasticsearch/xpack/spatial/aggregations/metrics/GeoShapeCentroidAggregator.java

@@ -21,6 +21,7 @@ import org.elasticsearch.search.aggregations.LeafBucketCollectorBase;
 import org.elasticsearch.search.aggregations.metrics.CompensatedSum;
 import org.elasticsearch.search.aggregations.metrics.InternalGeoCentroid;
 import org.elasticsearch.search.aggregations.metrics.MetricsAggregator;
+import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
 import org.elasticsearch.search.internal.SearchContext;
 import org.elasticsearch.xpack.spatial.index.fielddata.DimensionalShapeType;
 import org.elasticsearch.xpack.spatial.index.fielddata.MultiGeoShapeValues;
@@ -38,10 +39,16 @@ public final class GeoShapeCentroidAggregator extends MetricsAggregator {
     private LongArray counts;
     private ByteArray dimensionalShapeTypes;
 
-    public GeoShapeCentroidAggregator(String name, SearchContext context, Aggregator parent,
-                               GeoShapeValuesSource valuesSource, Map<String, Object> metadata) throws IOException {
+    public GeoShapeCentroidAggregator(
+        String name,
+        SearchContext context,
+        Aggregator parent,
+        ValuesSourceConfig valuesSourceConfig,
+        Map<String, Object> metadata
+    ) throws IOException {
         super(name, context, parent, metadata);
-        this.valuesSource = valuesSource;
+        // TODO: stop expecting nulls here
+        this.valuesSource = valuesSourceConfig.hasValues() ? (GeoShapeValuesSource) valuesSourceConfig.getValuesSource() : null;
         if (valuesSource != null) {
             final BigArrays bigArrays = context.bigArrays();
             lonSum = bigArrays.newDoubleArray(1, true);

x-pack/plugin/spatial/src/main/java/org/elasticsearch/xpack/spatial/search/aggregations/metrics/GeoShapeBoundsAggregator.java

@@ -16,6 +16,7 @@ import org.elasticsearch.search.aggregations.LeafBucketCollector;
 import org.elasticsearch.search.aggregations.LeafBucketCollectorBase;
 import org.elasticsearch.search.aggregations.metrics.InternalGeoBounds;
 import org.elasticsearch.search.aggregations.metrics.MetricsAggregator;
+import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
 import org.elasticsearch.search.internal.SearchContext;
 import org.elasticsearch.xpack.spatial.index.fielddata.MultiGeoShapeValues;
 import org.elasticsearch.xpack.spatial.search.aggregations.support.GeoShapeValuesSource;
@@ -33,10 +34,16 @@ public final class GeoShapeBoundsAggregator extends MetricsAggregator {
     private DoubleArray negLefts;
     private DoubleArray negRights;
 
-    public GeoShapeBoundsAggregator(String name, SearchContext aggregationContext, Aggregator parent,
-                             GeoShapeValuesSource valuesSource, boolean wrapLongitude, Map<String, Object> metadata) throws IOException {
+    public GeoShapeBoundsAggregator(
+        String name,
+        SearchContext aggregationContext,
+        Aggregator parent,
+        ValuesSourceConfig valuesSourceConfig,
+        boolean wrapLongitude,
+        Map<String, Object> metadata
+    ) throws IOException {
         super(name, aggregationContext, parent, metadata);
-        this.valuesSource = valuesSource;
+        this.valuesSource = valuesSourceConfig.hasValues() ? (GeoShapeValuesSource) valuesSourceConfig.getValuesSource() : null;
         this.wrapLongitude = wrapLongitude;
         if (valuesSource != null) {
             final BigArrays bigArrays = context.bigArrays();

x-pack/plugin/spatial/src/test/java/org/elasticsearch/xpack/spatial/SpatialPluginTests.java

@@ -12,8 +12,8 @@ import org.elasticsearch.license.XPackLicenseState;
 import org.elasticsearch.search.aggregations.bucket.geogrid.GeoHashGridAggregationBuilder;
 import org.elasticsearch.search.aggregations.bucket.geogrid.GeoTileGridAggregationBuilder;
 import org.elasticsearch.search.aggregations.metrics.GeoCentroidAggregationBuilder;
-import org.elasticsearch.search.aggregations.metrics.GeoCentroidAggregatorSupplier;
 import org.elasticsearch.search.aggregations.metrics.GeoGridAggregatorSupplier;
+import org.elasticsearch.search.aggregations.metrics.MetricAggregatorSupplier;
 import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
 import org.elasticsearch.search.aggregations.support.ValuesSourceRegistry;
 import org.elasticsearch.test.ESTestCase;
@@ -35,8 +35,8 @@ public class SpatialPluginTests extends ESTestCase {
             List<Consumer<ValuesSourceRegistry.Builder>> registrar = plugin.getAggregationExtentions();
             registrar.forEach(c -> c.accept(registryBuilder));
             ValuesSourceRegistry registry = registryBuilder.build();
-            GeoCentroidAggregatorSupplier centroidSupplier = (GeoCentroidAggregatorSupplier) registry.getAggregator(
-                new ValuesSourceConfig(GeoShapeValuesSourceType.instance(), null, false, null, null, null),
+            MetricAggregatorSupplier centroidSupplier = (MetricAggregatorSupplier) registry.getAggregator(
+                new ValuesSourceConfig(GeoShapeValuesSourceType.instance(), null, true, null, null, null, null, null, null),
                  GeoCentroidAggregationBuilder.NAME);
             if (License.OperationMode.TRIAL != operationMode &&
                     License.OperationMode.compare(operationMode, License.OperationMode.GOLD) < 0) {
@@ -57,7 +57,7 @@ public class SpatialPluginTests extends ESTestCase {
                 registrar.forEach(c -> c.accept(registryBuilder));
                 ValuesSourceRegistry registry = registryBuilder.build();
                 GeoGridAggregatorSupplier supplier = (GeoGridAggregatorSupplier) registry.getAggregator(
-                    new ValuesSourceConfig(GeoShapeValuesSourceType.instance(), null, false, null, null, null),
+                    new ValuesSourceConfig(GeoShapeValuesSourceType.instance(), null, true, null, null, null, null, null, null),
                     builderName);
                 if (License.OperationMode.TRIAL != operationMode &&
                     License.OperationMode.compare(operationMode, License.OperationMode.GOLD) < 0) {