ValuesSource Refactor: move histo VSType into XPack module (#53298)

- Introduces a new API (`getBareAggregatorRegistrar()`) which allows plugins to register aggregations against existing agg definitions defined in Core.
- This moves the histogram VSType over to XPack where it belongs. `getHistogramValues()` still remains as a Core concept
- Moves the histo-specific bits over to xpack (e.g. the actual aggregator logic). This requires extra boilerplate since we need to create a new "Analytics" Percentile/Rank aggregators to deal with the histo field. Doubly-so since percentiles/ranks are extra boiler-plate'y... should be much lighter for other aggs

Author: polyfractal

server/src/main/java/org/elasticsearch/plugins/SearchPlugin.java

@@ -114,6 +114,13 @@ default List<QuerySpec<?>> getQueries() {
     default List<AggregationSpec> getAggregations() {
         return emptyList();
     }
+    /**
+     * Allows plugins to register new aggregations using aggregation names that are already defined
+     * in Core, as long as the new aggregations target different ValuesSourceTypes
+     */
+    default List<Consumer<ValuesSourceRegistry>> getBareAggregatorRegistrar() {
+        return emptyList();
+    }
     /**
      * The new {@link PipelineAggregator}s added by this plugin.
      */

server/src/main/java/org/elasticsearch/search/SearchModule.java

@@ -458,6 +458,9 @@ private void registerAggregations(List<SearchPlugin> plugins) {
         registerAggregation((new AggregationSpec(CompositeAggregationBuilder.NAME, CompositeAggregationBuilder::new,
                 CompositeAggregationBuilder.PARSER).addResultReader(InternalComposite::new)));
         registerFromPlugin(plugins, SearchPlugin::getAggregations, this::registerAggregation);
+
+        // after aggs have been registered, see if there are any new VSTypes that need to be linked to core fields
+        registerFromPlugin(plugins, SearchPlugin::getBareAggregatorRegistrar, this::registerBareAggregatorRegistrar);
     }
 
     private void registerAggregation(AggregationSpec spec) {
@@ -478,6 +481,12 @@ private void registerAggregation(AggregationSpec spec) {
         }
     }
 
+    private void registerBareAggregatorRegistrar(Consumer<ValuesSourceRegistry> registrar) {
+        if (registrar != null) {
+            registrar.accept(this.valuesSourceRegistry);
+        }
+    }
+
     private void registerPipelineAggregations(List<SearchPlugin> plugins) {
         registerPipelineAggregation(new PipelineAggregationSpec(
                 DerivativePipelineAggregationBuilder.NAME,

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

@@ -26,8 +26,6 @@
 import org.elasticsearch.common.util.ArrayUtils;
 import org.elasticsearch.common.util.BigArrays;
 import org.elasticsearch.common.util.ObjectArray;
-import org.elasticsearch.index.fielddata.HistogramValue;
-import org.elasticsearch.index.fielddata.HistogramValues;
 import org.elasticsearch.index.fielddata.SortedNumericDoubleValues;
 import org.elasticsearch.search.DocValueFormat;
 import org.elasticsearch.search.aggregations.Aggregator;
@@ -78,18 +76,8 @@ public LeafBucketCollector getLeafCollector(LeafReaderContext ctx,
             return LeafBucketCollector.NO_OP_COLLECTOR;
         }
         final BigArrays bigArrays = context.bigArrays();
-        if (valuesSource instanceof ValuesSource.Histogram) {
-            final HistogramValues values = ((ValuesSource.Histogram)valuesSource).getHistogramValues(ctx);
-            return collectHistogramValues(values, bigArrays, sub);
-        } else {
-            final SortedNumericDoubleValues values = ((ValuesSource.Numeric)valuesSource).doubleValues(ctx);
-            return collectNumeric(values, bigArrays, sub);
-        }
-
-    }
 
-    private LeafBucketCollector collectNumeric(final SortedNumericDoubleValues values,
-                                               final BigArrays bigArrays, final LeafBucketCollector sub) {
+        final SortedNumericDoubleValues values = ((ValuesSource.Numeric)valuesSource).doubleValues(ctx);
         return new LeafBucketCollectorBase(sub, values) {
             @Override
             public void collect(int doc, long bucket) throws IOException {
@@ -102,35 +90,21 @@ public void collect(int doc, long bucket) throws IOException {
                 }
             }
         };
-    }
 
-    private LeafBucketCollector collectHistogramValues(final HistogramValues values,
-                                                       final BigArrays bigArrays, final LeafBucketCollector sub) {
-        return new LeafBucketCollectorBase(sub, values) {
-            @Override
-            public void collect(int doc, long bucket) throws IOException {
-                DoubleHistogram state = getExistingOrNewHistogram(bigArrays, bucket);
-                if (values.advanceExact(doc)) {
-                    final HistogramValue sketch = values.histogram();
-                    while (sketch.next()) {
-                        state.recordValueWithCount(sketch.value(), sketch.count());
-                    }
-                }
-            }
-        };
     }
 
     private DoubleHistogram getExistingOrNewHistogram(final BigArrays bigArrays, long bucket) {
         states = bigArrays.grow(states, bucket + 1);
         DoubleHistogram state = states.get(bucket);
         if (state == null) {
             state = new DoubleHistogram(numberOfSignificantValueDigits);
-            // Set the histogram to autosize so it can resize itself as
-            // the data range increases. Resize operations should be
-            // rare as the histogram buckets are exponential (on the top
-            // level). In the future we could expose the range as an
-            // option on the request so the histogram can be fixed at
-            // initialisation and doesn't need resizing.
+            /* Set the histogram to autosize so it can resize itself as
+               the data range increases. Resize operations should be
+               rare as the histogram buckets are exponential (on the top
+               level). In the future we could expose the range as an
+               option on the request so the histogram can be fixed at
+               initialisation and doesn't need resizing.
+             */
             state.setAutoResize(true);
             states.set(bucket, state);
         }

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

@@ -25,8 +25,6 @@
 import org.elasticsearch.common.util.ArrayUtils;
 import org.elasticsearch.common.util.BigArrays;
 import org.elasticsearch.common.util.ObjectArray;
-import org.elasticsearch.index.fielddata.HistogramValue;
-import org.elasticsearch.index.fielddata.HistogramValues;
 import org.elasticsearch.index.fielddata.SortedNumericDoubleValues;
 import org.elasticsearch.search.DocValueFormat;
 import org.elasticsearch.search.aggregations.Aggregator;
@@ -77,18 +75,7 @@ public LeafBucketCollector getLeafCollector(LeafReaderContext ctx,
             return LeafBucketCollector.NO_OP_COLLECTOR;
         }
         final BigArrays bigArrays = context.bigArrays();
-        if (valuesSource instanceof ValuesSource.Histogram) {
-            final HistogramValues values = ((ValuesSource.Histogram)valuesSource).getHistogramValues(ctx);
-            return collectHistogramValues(values, bigArrays, sub);
-        } else {
-            final SortedNumericDoubleValues values = ((ValuesSource.Numeric)valuesSource).doubleValues(ctx);
-            return collectNumeric(values, bigArrays, sub);
-        }
-
-    }
-
-    private LeafBucketCollector collectNumeric(final SortedNumericDoubleValues values,
-                                               final BigArrays bigArrays, final LeafBucketCollector sub) {
+        final SortedNumericDoubleValues values = ((ValuesSource.Numeric)valuesSource).doubleValues(ctx);
         return new LeafBucketCollectorBase(sub, values) {
             @Override
             public void collect(int doc, long bucket) throws IOException {
@@ -103,22 +90,6 @@ public void collect(int doc, long bucket) throws IOException {
         };
     }
 
-    private LeafBucketCollector collectHistogramValues(final HistogramValues values,
-                                                       final BigArrays bigArrays, final LeafBucketCollector sub) {
-        return new LeafBucketCollectorBase(sub, values) {
-            @Override
-            public void collect(int doc, long bucket) throws IOException {
-                TDigestState state = getExistingOrNewHistogram(bigArrays, bucket);
-                if (values.advanceExact(doc)) {
-                    final HistogramValue sketch = values.histogram();
-                    while(sketch.next()) {
-                        state.add(sketch.value(), sketch.count());
-                    }
-                }
-            }
-        };
-    }
-
     private TDigestState getExistingOrNewHistogram(final BigArrays bigArrays, long bucket) {
         states = bigArrays.grow(states, bucket + 1);
         TDigestState state = states.get(bucket);

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

@@ -31,8 +31,8 @@
 public class InternalHDRPercentileRanks extends AbstractInternalHDRPercentiles implements PercentileRanks {
     public static final String NAME = "hdr_percentile_ranks";
 
-    InternalHDRPercentileRanks(String name, double[] cdfValues, DoubleHistogram state, boolean keyed, DocValueFormat formatter,
-            List<PipelineAggregator> pipelineAggregators, Map<String, Object> metaData) {
+    public InternalHDRPercentileRanks(String name, double[] cdfValues, DoubleHistogram state, boolean keyed, DocValueFormat formatter,
+                                      List<PipelineAggregator> pipelineAggregators, Map<String, Object> metaData) {
         super(name, cdfValues, state, keyed, formatter, pipelineAggregators, metaData);
     }
 
@@ -74,7 +74,7 @@ protected AbstractInternalHDRPercentiles createReduced(String name, double[] key
         return new InternalHDRPercentileRanks(name, keys, merged, keyed, format, pipelineAggregators, metaData);
     }
 
-    static double percentileRank(DoubleHistogram state, double value) {
+    public static double percentileRank(DoubleHistogram state, double value) {
         if (state.getTotalCount() == 0) {
             return Double.NaN;
         }

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

@@ -31,8 +31,8 @@
 public class InternalHDRPercentiles extends AbstractInternalHDRPercentiles implements Percentiles {
     public static final String NAME = "hdr_percentiles";
 
-    InternalHDRPercentiles(String name, double[] percents, DoubleHistogram state, boolean keyed, DocValueFormat formatter,
-            List<PipelineAggregator> pipelineAggregators, Map<String, Object> metaData) {
+    public InternalHDRPercentiles(String name, double[] percents, DoubleHistogram state, boolean keyed, DocValueFormat formatter,
+                                  List<PipelineAggregator> pipelineAggregators, Map<String, Object> metaData) {
         super(name, percents, state, keyed, formatter, pipelineAggregators, metaData);
     }
 

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

@@ -30,8 +30,8 @@
 public class InternalTDigestPercentileRanks extends AbstractInternalTDigestPercentiles implements PercentileRanks {
     public static final String NAME = "tdigest_percentile_ranks";
 
-    InternalTDigestPercentileRanks(String name, double[] cdfValues, TDigestState state, boolean keyed, DocValueFormat formatter,
-                                    List<PipelineAggregator> pipelineAggregators, Map<String, Object> metaData) {
+    public InternalTDigestPercentileRanks(String name, double[] cdfValues, TDigestState state, boolean keyed, DocValueFormat formatter,
+                                          List<PipelineAggregator> pipelineAggregators, Map<String, Object> metaData) {
         super(name, cdfValues, state, keyed, formatter, pipelineAggregators, metaData);
     }
 
@@ -73,7 +73,7 @@ protected AbstractInternalTDigestPercentiles createReduced(String name, double[]
         return new InternalTDigestPercentileRanks(name, keys, merged, keyed, format, pipelineAggregators, metaData);
     }
 
-    static double percentileRank(TDigestState state, double value) {
+    public static double percentileRank(TDigestState state, double value) {
         double percentileRank = state.cdf(value);
         if (percentileRank < 0) {
             percentileRank = 0;

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

@@ -30,8 +30,8 @@
 public class InternalTDigestPercentiles extends AbstractInternalTDigestPercentiles implements Percentiles {
     public static final String NAME = "tdigest_percentiles";
 
-    InternalTDigestPercentiles(String name, double[] percents, TDigestState state, boolean keyed, DocValueFormat formatter,
-            List<PipelineAggregator> pipelineAggregators, Map<String, Object> metaData) {
+    public InternalTDigestPercentiles(String name, double[] percents, TDigestState state, boolean keyed, DocValueFormat formatter,
+                                      List<PipelineAggregator> pipelineAggregators, Map<String, Object> metaData) {
         super(name, percents, state, keyed, formatter, pipelineAggregators, metaData);
     }
 

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

@@ -46,7 +46,7 @@ class PercentileRanksAggregatorFactory extends ValuesSourceAggregatorFactory {
 
     static void registerAggregators(ValuesSourceRegistry valuesSourceRegistry) {
         valuesSourceRegistry.register(PercentileRanksAggregationBuilder.NAME,
-            List.of(CoreValuesSourceType.NUMERIC, CoreValuesSourceType.HISTOGRAM, CoreValuesSourceType.DATE, CoreValuesSourceType.BOOLEAN),
+            List.of(CoreValuesSourceType.NUMERIC, CoreValuesSourceType.DATE, CoreValuesSourceType.BOOLEAN),
             new PercentilesAggregatorSupplier() {
                 @Override
                 public Aggregator build(String name, ValuesSource valuesSource, SearchContext context, Aggregator parent,

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

@@ -50,7 +50,7 @@ class PercentilesAggregatorFactory extends ValuesSourceAggregatorFactory {
 
     static void registerAggregators(ValuesSourceRegistry valuesSourceRegistry) {
         valuesSourceRegistry.register(PercentilesAggregationBuilder.NAME,
-            List.of(CoreValuesSourceType.NUMERIC, CoreValuesSourceType.HISTOGRAM, CoreValuesSourceType.DATE, CoreValuesSourceType.BOOLEAN),
+            List.of(CoreValuesSourceType.NUMERIC, CoreValuesSourceType.DATE, CoreValuesSourceType.BOOLEAN),
             new PercentilesAggregatorSupplier() {
                 @Override
                 public Aggregator build(String name, ValuesSource valuesSource, SearchContext context, Aggregator parent,