Skip to content

Commit d89f9f3

Browse files
csoulioscsoulios
authored
Implement aggregations on aggregate metrics (#53986)
Following the implementation of the aggregate_metric_double field mapper(#49830) we are implementing the Min, Max, ValueCount, Sum and Average aggregations on aggregate metrics. The code builds on the excellent work done for #42949 and uses the extensible ValuesSources infrastructure to wire up common metric aggregation on the aggregate_metric_double field type. This PR is part of the rollups v2 refactoring as described in meta issue #42720
1 parent 046118c commit d89f9f3

30 files changed

+2421
-69
lines changed

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -32,7 +32,7 @@
3232
public class InternalSum extends InternalNumericMetricsAggregation.SingleValue implements Sum {
3333
private final double sum;
3434

35-
public InternalSum(String name, double sum, DocValueFormat formatter, Map<String, Object> metadata) {
35+
public InternalSum(String name, double sum, DocValueFormat formatter, Map<String, Object> metadata) {
3636
super(name, metadata);
3737
this.sum = sum;
3838
this.format = formatter;

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

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -48,7 +48,7 @@
4848
import java.util.Map;
4949
import java.util.function.Function;
5050

51-
class MinAggregator extends NumericMetricsAggregator.SingleValue {
51+
public class MinAggregator extends NumericMetricsAggregator.SingleValue {
5252
private static final int MAX_BKD_LOOKUPS = 1024;
5353

5454
final ValuesSource.Numeric valuesSource;
@@ -168,7 +168,7 @@ public void doClose() {
168168
* @param parent The parent aggregator.
169169
* @param config The config for the values source metric.
170170
*/
171-
static Function<byte[], Number> getPointReaderOrNull(SearchContext context, Aggregator parent,
171+
public static Function<byte[], Number> getPointReaderOrNull(SearchContext context, Aggregator parent,
172172
ValuesSourceConfig config) {
173173
if (context.query() != null &&
174174
context.query().getClass() != MatchAllDocsQuery.class) {

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,7 @@
77
* not use this file except in compliance with the License.
88
* You may obtain a copy of the License at
99
*
10-
* http://www.apache.org/licenses/LICENSE-2.0
10+
* http://www.apache.org/licenses/LICENSE-2.0
1111
*
1212
* Unless required by applicable law or agreed to in writing,
1313
* software distributed under the License is distributed on an

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

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,6 @@
77
evaluationDependsOn(xpackModule('core'))
88

99
apply plugin: 'elasticsearch.esplugin'
10-
1110
esplugin {
1211
name 'x-pack-aggregate-metric'
1312
description 'Module for the aggregate_metric field type, which allows pre-aggregated fields to be stored a single field.'
@@ -16,7 +15,12 @@ esplugin {
1615
}
1716
archivesBaseName = 'x-pack-aggregate-metric'
1817

18+
compileJava.options.compilerArgs << "-Xlint:-rawtypes"
19+
compileTestJava.options.compilerArgs << "-Xlint:-rawtypes"
20+
1921
dependencies {
22+
compileOnly project(":server")
23+
2024
compileOnly project(path: xpackModule('core'), configuration: 'default')
2125
testCompile project(path: xpackModule('core'), configuration: 'testArtifacts')
2226
}

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

Lines changed: 15 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -12,17 +12,21 @@
1212
import org.elasticsearch.plugins.ActionPlugin;
1313
import org.elasticsearch.plugins.MapperPlugin;
1414
import org.elasticsearch.plugins.Plugin;
15+
import org.elasticsearch.plugins.SearchPlugin;
16+
import org.elasticsearch.search.aggregations.support.ValuesSourceRegistry;
17+
import org.elasticsearch.xpack.aggregatemetric.aggregations.metrics.AggregateMetricsAggregatorsRegistrar;
1518
import org.elasticsearch.xpack.aggregatemetric.mapper.AggregateDoubleMetricFieldMapper;
1619
import org.elasticsearch.xpack.core.action.XPackInfoFeatureAction;
1720
import org.elasticsearch.xpack.core.action.XPackUsageFeatureAction;
1821

1922
import java.util.Arrays;
2023
import java.util.List;
2124
import java.util.Map;
25+
import java.util.function.Consumer;
2226

2327
import static java.util.Collections.singletonMap;
2428

25-
public class AggregateMetricMapperPlugin extends Plugin implements MapperPlugin, ActionPlugin {
29+
public class AggregateMetricMapperPlugin extends Plugin implements MapperPlugin, ActionPlugin, SearchPlugin {
2630

2731
@Override
2832
public Map<String, Mapper.TypeParser> getMappers() {
@@ -37,4 +41,14 @@ public Map<String, Mapper.TypeParser> getMappers() {
3741
);
3842
}
3943

44+
@Override
45+
public List<Consumer<ValuesSourceRegistry.Builder>> getAggregationExtentions() {
46+
return List.of(
47+
AggregateMetricsAggregatorsRegistrar::registerSumAggregator,
48+
AggregateMetricsAggregatorsRegistrar::registerAvgAggregator,
49+
AggregateMetricsAggregatorsRegistrar::registerMinAggregator,
50+
AggregateMetricsAggregatorsRegistrar::registerMaxAggregator,
51+
AggregateMetricsAggregatorsRegistrar::registerValueCountAggregator
52+
);
53+
}
4054
}

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

Lines changed: 135 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,135 @@
1+
/*
2+
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
3+
* or more contributor license agreements. Licensed under the Elastic License;
4+
* you may not use this file except in compliance with the Elastic License.
5+
*/
6+
package org.elasticsearch.xpack.aggregatemetric.aggregations.metrics;
7+
8+
import org.apache.lucene.index.LeafReaderContext;
9+
import org.apache.lucene.search.ScoreMode;
10+
import org.elasticsearch.common.lease.Releasables;
11+
import org.elasticsearch.common.util.BigArrays;
12+
import org.elasticsearch.common.util.DoubleArray;
13+
import org.elasticsearch.common.util.LongArray;
14+
import org.elasticsearch.index.fielddata.SortedNumericDoubleValues;
15+
import org.elasticsearch.search.DocValueFormat;
16+
import org.elasticsearch.search.aggregations.Aggregator;
17+
import org.elasticsearch.search.aggregations.InternalAggregation;
18+
import org.elasticsearch.search.aggregations.LeafBucketCollector;
19+
import org.elasticsearch.search.aggregations.LeafBucketCollectorBase;
20+
import org.elasticsearch.search.aggregations.metrics.CompensatedSum;
21+
import org.elasticsearch.search.aggregations.metrics.InternalAvg;
22+
import org.elasticsearch.search.aggregations.metrics.NumericMetricsAggregator;
23+
import org.elasticsearch.search.internal.SearchContext;
24+
import org.elasticsearch.xpack.aggregatemetric.aggregations.support.AggregateMetricsValuesSource;
25+
import org.elasticsearch.xpack.aggregatemetric.mapper.AggregateDoubleMetricFieldMapper.Metric;
26+
27+
import java.io.IOException;
28+
import java.util.Map;
29+
30+
class AggregateMetricBackedAvgAggregator extends NumericMetricsAggregator.SingleValue {
31+
32+
final AggregateMetricsValuesSource.AggregateDoubleMetric valuesSource;
33+
34+
LongArray counts;
35+
DoubleArray sums;
36+
DoubleArray compensations;
37+
DocValueFormat format;
38+
39+
AggregateMetricBackedAvgAggregator(
40+
String name,
41+
AggregateMetricsValuesSource.AggregateDoubleMetric valuesSource,
42+
DocValueFormat formatter,
43+
SearchContext context,
44+
Aggregator parent,
45+
Map<String, Object> metadata
46+
) throws IOException {
47+
super(name, context, parent, metadata);
48+
this.valuesSource = valuesSource;
49+
this.format = formatter;
50+
if (valuesSource != null) {
51+
final BigArrays bigArrays = context.bigArrays();
52+
counts = bigArrays.newLongArray(1, true);
53+
sums = bigArrays.newDoubleArray(1, true);
54+
compensations = bigArrays.newDoubleArray(1, true);
55+
}
56+
}
57+
58+
@Override
59+
public ScoreMode scoreMode() {
60+
return valuesSource != null && valuesSource.needsScores() ? ScoreMode.COMPLETE : ScoreMode.COMPLETE_NO_SCORES;
61+
}
62+
63+
@Override
64+
public LeafBucketCollector getLeafCollector(LeafReaderContext ctx, final LeafBucketCollector sub) throws IOException {
65+
if (valuesSource == null) {
66+
return LeafBucketCollector.NO_OP_COLLECTOR;
67+
}
68+
final BigArrays bigArrays = context.bigArrays();
69+
// Retrieve aggregate values for metrics sum and value_count
70+
final SortedNumericDoubleValues aggregateSums = valuesSource.getAggregateMetricValues(ctx, Metric.sum);
71+
final SortedNumericDoubleValues aggregateValueCounts = valuesSource.getAggregateMetricValues(ctx, Metric.value_count);
72+
final CompensatedSum kahanSummation = new CompensatedSum(0, 0);
73+
return new LeafBucketCollectorBase(sub, sums) {
74+
@Override
75+
public void collect(int doc, long bucket) throws IOException {
76+
sums = bigArrays.grow(sums, bucket + 1);
77+
compensations = bigArrays.grow(compensations, bucket + 1);
78+
79+
// Read aggregate values for sums
80+
if (aggregateSums.advanceExact(doc)) {
81+
// Compute the sum of double values with Kahan summation algorithm which is more
82+
// accurate than naive summation.
83+
double sum = sums.get(bucket);
84+
double compensation = compensations.get(bucket);
85+
86+
kahanSummation.reset(sum, compensation);
87+
for (int i = 0; i < aggregateSums.docValueCount(); i++) {
88+
double value = aggregateSums.nextValue();
89+
kahanSummation.add(value);
90+
}
91+
92+
sums.set(bucket, kahanSummation.value());
93+
compensations.set(bucket, kahanSummation.delta());
94+
}
95+
96+
counts = bigArrays.grow(counts, bucket + 1);
97+
// Read aggregate values for value_count
98+
if (aggregateValueCounts.advanceExact(doc)) {
99+
for (int i = 0; i < aggregateValueCounts.docValueCount(); i++) {
100+
double d = aggregateValueCounts.nextValue();
101+
long value = Double.valueOf(d).longValue();
102+
counts.increment(bucket, value);
103+
}
104+
}
105+
}
106+
};
107+
}
108+
109+
@Override
110+
public double metric(long owningBucketOrd) {
111+
if (valuesSource == null || owningBucketOrd >= sums.size()) {
112+
return Double.NaN;
113+
}
114+
return sums.get(owningBucketOrd) / counts.get(owningBucketOrd);
115+
}
116+
117+
@Override
118+
public InternalAggregation buildAggregation(long bucket) {
119+
if (valuesSource == null || bucket >= sums.size()) {
120+
return buildEmptyAggregation();
121+
}
122+
return new InternalAvg(name, sums.get(bucket), counts.get(bucket), format, metadata());
123+
}
124+
125+
@Override
126+
public InternalAggregation buildEmptyAggregation() {
127+
return new InternalAvg(name, 0.0, 0L, format, metadata());
128+
}
129+
130+
@Override
131+
public void doClose() {
132+
Releasables.close(counts, sums, compensations);
133+
}
134+
135+
}

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

Lines changed: 118 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,118 @@
1+
/*
2+
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
3+
* or more contributor license agreements. Licensed under the Elastic License;
4+
* you may not use this file except in compliance with the Elastic License.
5+
*/
6+
package org.elasticsearch.xpack.aggregatemetric.aggregations.metrics;
7+
8+
import org.apache.lucene.index.LeafReaderContext;
9+
import org.apache.lucene.search.CollectionTerminatedException;
10+
import org.apache.lucene.search.ScoreMode;
11+
import org.elasticsearch.common.lease.Releasables;
12+
import org.elasticsearch.common.util.BigArrays;
13+
import org.elasticsearch.common.util.DoubleArray;
14+
import org.elasticsearch.index.fielddata.NumericDoubleValues;
15+
import org.elasticsearch.index.fielddata.SortedNumericDoubleValues;
16+
import org.elasticsearch.search.DocValueFormat;
17+
import org.elasticsearch.search.MultiValueMode;
18+
import org.elasticsearch.search.aggregations.Aggregator;
19+
import org.elasticsearch.search.aggregations.InternalAggregation;
20+
import org.elasticsearch.search.aggregations.LeafBucketCollector;
21+
import org.elasticsearch.search.aggregations.LeafBucketCollectorBase;
22+
import org.elasticsearch.search.aggregations.metrics.InternalMax;
23+
import org.elasticsearch.search.aggregations.metrics.NumericMetricsAggregator;
24+
import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
25+
import org.elasticsearch.search.internal.SearchContext;
26+
import org.elasticsearch.xpack.aggregatemetric.aggregations.support.AggregateMetricsValuesSource;
27+
import org.elasticsearch.xpack.aggregatemetric.mapper.AggregateDoubleMetricFieldMapper.Metric;
28+
29+
import java.io.IOException;
30+
import java.util.Map;
31+
32+
class AggregateMetricBackedMaxAggregator extends NumericMetricsAggregator.SingleValue {
33+
34+
private final AggregateMetricsValuesSource.AggregateDoubleMetric valuesSource;
35+
final DocValueFormat formatter;
36+
DoubleArray maxes;
37+
38+
AggregateMetricBackedMaxAggregator(
39+
String name,
40+
ValuesSourceConfig config,
41+
AggregateMetricsValuesSource.AggregateDoubleMetric valuesSource,
42+
SearchContext context,
43+
Aggregator parent,
44+
Map<String, Object> metadata
45+
) throws IOException {
46+
super(name, context, parent, metadata);
47+
this.valuesSource = valuesSource;
48+
if (valuesSource != null) {
49+
maxes = context.bigArrays().newDoubleArray(1, false);
50+
maxes.fill(0, maxes.size(), Double.NEGATIVE_INFINITY);
51+
}
52+
this.formatter = config.format();
53+
}
54+
55+
@Override
56+
public ScoreMode scoreMode() {
57+
return valuesSource != null && valuesSource.needsScores() ? ScoreMode.COMPLETE : ScoreMode.COMPLETE_NO_SCORES;
58+
}
59+
60+
@Override
61+
public LeafBucketCollector getLeafCollector(LeafReaderContext ctx, final LeafBucketCollector sub) throws IOException {
62+
if (valuesSource == null) {
63+
if (parent != null) {
64+
return LeafBucketCollector.NO_OP_COLLECTOR;
65+
} else {
66+
// we have no parent and the values source is empty so we can skip collecting hits.
67+
throw new CollectionTerminatedException();
68+
}
69+
}
70+
71+
final BigArrays bigArrays = context.bigArrays();
72+
final SortedNumericDoubleValues allValues = valuesSource.getAggregateMetricValues(ctx, Metric.max);
73+
final NumericDoubleValues values = MultiValueMode.MAX.select(allValues);
74+
return new LeafBucketCollectorBase(sub, allValues) {
75+
76+
@Override
77+
public void collect(int doc, long bucket) throws IOException {
78+
if (bucket >= maxes.size()) {
79+
long from = maxes.size();
80+
maxes = bigArrays.grow(maxes, bucket + 1);
81+
maxes.fill(from, maxes.size(), Double.NEGATIVE_INFINITY);
82+
}
83+
if (values.advanceExact(doc)) {
84+
final double value = values.doubleValue();
85+
double max = maxes.get(bucket);
86+
max = Math.max(max, value);
87+
maxes.set(bucket, max);
88+
}
89+
}
90+
};
91+
}
92+
93+
@Override
94+
public double metric(long owningBucketOrd) {
95+
if (valuesSource == null || owningBucketOrd >= maxes.size()) {
96+
return Double.NEGATIVE_INFINITY;
97+
}
98+
return maxes.get(owningBucketOrd);
99+
}
100+
101+
@Override
102+
public InternalAggregation buildAggregation(long bucket) {
103+
if (valuesSource == null || bucket >= maxes.size()) {
104+
return buildEmptyAggregation();
105+
}
106+
return new InternalMax(name, maxes.get(bucket), formatter, metadata());
107+
}
108+
109+
@Override
110+
public InternalAggregation buildEmptyAggregation() {
111+
return new InternalMax(name, Double.NEGATIVE_INFINITY, formatter, metadata());
112+
}
113+
114+
@Override
115+
public void doClose() {
116+
Releasables.close(maxes);
117+
}
118+
}

0 commit comments

Comments
 (0)