Skip to content

Commit 4b6565f

Browse files
imotovimotov
committed
Add support for histogram fields to rate aggregation (elastic#63289)
The rate aggregation now supports histogram fields. At the moment only sum is supported. Closes elastic#62939
1 parent 0aebb59 commit 4b6565f

File tree

7 files changed

+232
-70
lines changed

7 files changed

+232
-70
lines changed

‎docs/reference/aggregations/metrics/rate-aggregation.asciidoc

+2-1
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,8 @@
44
=== Rate Aggregation
55

66
A `rate` metrics aggregation can be used only inside a `date_histogram` and calculates a rate of documents or a field in each
7-
`date_histogram` bucket.
7+
`date_histogram` bucket. The field values can be generated by a provided script or extracted from specific numeric or
8+
<<histogram,histogram fields>> in the documents.
89

910
==== Syntax
1011

‎x-pack/plugin/analytics/src/main/java/org/elasticsearch/xpack/analytics/rate/RateAggregator.java renamed to ‎x-pack/plugin/analytics/src/main/java/org/elasticsearch/xpack/analytics/rate/AbstractRateAggregator.java

+9-47
Original file line numberDiff line numberDiff line change
@@ -5,39 +5,33 @@
55
*/
66
package org.elasticsearch.xpack.analytics.rate;
77

8-
import org.apache.lucene.index.LeafReaderContext;
8+
import java.io.IOException;
9+
import java.util.Map;
10+
911
import org.apache.lucene.search.ScoreMode;
1012
import org.elasticsearch.common.Rounding;
1113
import org.elasticsearch.common.lease.Releasables;
12-
import org.elasticsearch.common.util.BigArrays;
1314
import org.elasticsearch.common.util.DoubleArray;
14-
import org.elasticsearch.index.fielddata.SortedNumericDoubleValues;
1515
import org.elasticsearch.search.DocValueFormat;
1616
import org.elasticsearch.search.aggregations.Aggregator;
1717
import org.elasticsearch.search.aggregations.InternalAggregation;
18-
import org.elasticsearch.search.aggregations.LeafBucketCollector;
19-
import org.elasticsearch.search.aggregations.LeafBucketCollectorBase;
2018
import org.elasticsearch.search.aggregations.bucket.histogram.SizedBucketAggregator;
21-
import org.elasticsearch.search.aggregations.metrics.CompensatedSum;
2219
import org.elasticsearch.search.aggregations.metrics.NumericMetricsAggregator;
2320
import org.elasticsearch.search.aggregations.support.ValuesSource;
2421
import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
2522
import org.elasticsearch.search.internal.SearchContext;
2623

27-
import java.io.IOException;
28-
import java.util.Map;
29-
30-
public class RateAggregator extends NumericMetricsAggregator.SingleValue {
24+
public abstract class AbstractRateAggregator extends NumericMetricsAggregator.SingleValue {
3125

32-
private final ValuesSource.Numeric valuesSource;
26+
protected final ValuesSource valuesSource;
3327
private final DocValueFormat format;
3428
private final Rounding.DateTimeUnit rateUnit;
3529
private final SizedBucketAggregator sizedBucketAggregator;
3630

37-
private DoubleArray sums;
38-
private DoubleArray compensations;
31+
protected DoubleArray sums;
32+
protected DoubleArray compensations;
3933

40-
public RateAggregator(
34+
public AbstractRateAggregator(
4135
String name,
4236
ValuesSourceConfig valuesSourceConfig,
4337
Rounding.DateTimeUnit rateUnit,
@@ -46,7 +40,7 @@ public RateAggregator(
4640
Map<String, Object> metadata
4741
) throws IOException {
4842
super(name, context, parent, metadata);
49-
this.valuesSource = (ValuesSource.Numeric) valuesSourceConfig.getValuesSource();
43+
this.valuesSource = valuesSourceConfig.getValuesSource();
5044
this.format = valuesSourceConfig.format();
5145
if (valuesSource != null) {
5246
sums = context.bigArrays().newDoubleArray(1, true);
@@ -75,38 +69,6 @@ public ScoreMode scoreMode() {
7569
return valuesSource != null && valuesSource.needsScores() ? ScoreMode.COMPLETE : ScoreMode.COMPLETE_NO_SCORES;
7670
}
7771

78-
@Override
79-
public LeafBucketCollector getLeafCollector(LeafReaderContext ctx, final LeafBucketCollector sub) throws IOException {
80-
final BigArrays bigArrays = context.bigArrays();
81-
final SortedNumericDoubleValues values = valuesSource.doubleValues(ctx);
82-
final CompensatedSum kahanSummation = new CompensatedSum(0, 0);
83-
84-
return new LeafBucketCollectorBase(sub, values) {
85-
@Override
86-
public void collect(int doc, long bucket) throws IOException {
87-
sums = bigArrays.grow(sums, bucket + 1);
88-
compensations = bigArrays.grow(compensations, bucket + 1);
89-
90-
if (values.advanceExact(doc)) {
91-
final int valuesCount = values.docValueCount();
92-
// Compute the sum of double values with Kahan summation algorithm which is more
93-
// accurate than naive summation.
94-
double sum = sums.get(bucket);
95-
double compensation = compensations.get(bucket);
96-
kahanSummation.reset(sum, compensation);
97-
98-
for (int i = 0; i < valuesCount; i++) {
99-
double value = values.nextValue();
100-
kahanSummation.add(value);
101-
}
102-
103-
compensations.set(bucket, kahanSummation.delta());
104-
sums.set(bucket, kahanSummation.value());
105-
}
106-
}
107-
};
108-
}
109-
11072
@Override
11173
public double metric(long owningBucketOrd) {
11274
if (sizedBucketAggregator == null || valuesSource == null || owningBucketOrd >= sums.size()) {

‎x-pack/plugin/analytics/src/main/java/org/elasticsearch/xpack/analytics/rate/HistogramRateAggregator.java

+62
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,62 @@
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+
7+
package org.elasticsearch.xpack.analytics.rate;
8+
9+
import java.io.IOException;
10+
import java.util.Map;
11+
12+
import org.apache.lucene.index.LeafReaderContext;
13+
import org.elasticsearch.common.Rounding;
14+
import org.elasticsearch.common.util.BigArrays;
15+
import org.elasticsearch.index.fielddata.HistogramValue;
16+
import org.elasticsearch.index.fielddata.HistogramValues;
17+
import org.elasticsearch.search.aggregations.Aggregator;
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.support.ValuesSourceConfig;
22+
import org.elasticsearch.search.internal.SearchContext;
23+
import org.elasticsearch.xpack.analytics.aggregations.support.HistogramValuesSource;
24+
25+
public class HistogramRateAggregator extends AbstractRateAggregator {
26+
public HistogramRateAggregator(
27+
String name,
28+
ValuesSourceConfig valuesSourceConfig,
29+
Rounding.DateTimeUnit rateUnit,
30+
SearchContext context,
31+
Aggregator parent,
32+
Map<String, Object> metadata
33+
) throws IOException {
34+
super(name, valuesSourceConfig, rateUnit, context, parent, metadata);
35+
}
36+
37+
@Override
38+
public LeafBucketCollector getLeafCollector(LeafReaderContext ctx, final LeafBucketCollector sub) throws IOException {
39+
final BigArrays bigArrays = context.bigArrays();
40+
final CompensatedSum kahanSummation = new CompensatedSum(0, 0);
41+
final HistogramValues values = ((HistogramValuesSource.Histogram) valuesSource).getHistogramValues(ctx);
42+
return new LeafBucketCollectorBase(sub, values) {
43+
@Override
44+
public void collect(int doc, long bucket) throws IOException {
45+
sums = bigArrays.grow(sums, bucket + 1);
46+
compensations = bigArrays.grow(compensations, bucket + 1);
47+
48+
if (values.advanceExact(doc)) {
49+
final HistogramValue sketch = values.histogram();
50+
while (sketch.next()) {
51+
double sum = sums.get(bucket);
52+
double compensation = compensations.get(bucket);
53+
kahanSummation.reset(sum, compensation);
54+
kahanSummation.add(sketch.value());
55+
compensations.set(bucket, kahanSummation.delta());
56+
sums.set(bucket, kahanSummation.value());
57+
}
58+
}
59+
}
60+
};
61+
}
62+
}

‎x-pack/plugin/analytics/src/main/java/org/elasticsearch/xpack/analytics/rate/NumericRateAggregator.java

+66
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,66 @@
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+
7+
package org.elasticsearch.xpack.analytics.rate;
8+
9+
import java.io.IOException;
10+
import java.util.Map;
11+
12+
import org.apache.lucene.index.LeafReaderContext;
13+
import org.elasticsearch.common.Rounding;
14+
import org.elasticsearch.common.util.BigArrays;
15+
import org.elasticsearch.index.fielddata.SortedNumericDoubleValues;
16+
import org.elasticsearch.search.aggregations.Aggregator;
17+
import org.elasticsearch.search.aggregations.LeafBucketCollector;
18+
import org.elasticsearch.search.aggregations.LeafBucketCollectorBase;
19+
import org.elasticsearch.search.aggregations.metrics.CompensatedSum;
20+
import org.elasticsearch.search.aggregations.support.ValuesSource;
21+
import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
22+
import org.elasticsearch.search.internal.SearchContext;
23+
24+
public class NumericRateAggregator extends AbstractRateAggregator {
25+
public NumericRateAggregator(
26+
String name,
27+
ValuesSourceConfig valuesSourceConfig,
28+
Rounding.DateTimeUnit rateUnit,
29+
SearchContext context,
30+
Aggregator parent,
31+
Map<String, Object> metadata
32+
) throws IOException {
33+
super(name, valuesSourceConfig, rateUnit, context, parent, metadata);
34+
}
35+
36+
@Override
37+
public LeafBucketCollector getLeafCollector(LeafReaderContext ctx, final LeafBucketCollector sub) throws IOException {
38+
final BigArrays bigArrays = context.bigArrays();
39+
final CompensatedSum kahanSummation = new CompensatedSum(0, 0);
40+
final SortedNumericDoubleValues values = ((ValuesSource.Numeric) valuesSource).doubleValues(ctx);
41+
return new LeafBucketCollectorBase(sub, values) {
42+
@Override
43+
public void collect(int doc, long bucket) throws IOException {
44+
sums = bigArrays.grow(sums, bucket + 1);
45+
compensations = bigArrays.grow(compensations, bucket + 1);
46+
47+
if (values.advanceExact(doc)) {
48+
final int valuesCount = values.docValueCount();
49+
// Compute the sum of double values with Kahan summation algorithm which is more
50+
// accurate than naive summation.
51+
double sum = sums.get(bucket);
52+
double compensation = compensations.get(bucket);
53+
kahanSummation.reset(sum, compensation);
54+
55+
for (int i = 0; i < valuesCount; i++) {
56+
double value = values.nextValue();
57+
kahanSummation.add(value);
58+
}
59+
60+
compensations.set(bucket, kahanSummation.delta());
61+
sums.set(bucket, kahanSummation.value());
62+
}
63+
}
64+
};
65+
}
66+
}

‎x-pack/plugin/analytics/src/main/java/org/elasticsearch/xpack/analytics/rate/RateAggregationBuilder.java

+1-1
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,7 @@
2828
import java.util.Map;
2929
import java.util.Objects;
3030

31-
public class RateAggregationBuilder extends ValuesSourceAggregationBuilder.LeafOnly<ValuesSource.Numeric, RateAggregationBuilder> {
31+
public class RateAggregationBuilder extends ValuesSourceAggregationBuilder.LeafOnly<ValuesSource, RateAggregationBuilder> {
3232
public static final String NAME = "rate";
3333
public static final ParseField UNIT_FIELD = new ParseField("unit");
3434
public static final ValuesSourceRegistry.RegistryKey<RateAggregatorSupplier> REGISTRY_KEY = new ValuesSourceRegistry.RegistryKey<>(

‎x-pack/plugin/analytics/src/main/java/org/elasticsearch/xpack/analytics/rate/RateAggregatorFactory.java

+14-6
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,10 @@
66

77
package org.elasticsearch.xpack.analytics.rate;
88

9+
import java.io.IOException;
10+
import java.util.Collections;
11+
import java.util.Map;
12+
913
import org.apache.lucene.index.LeafReaderContext;
1014
import org.elasticsearch.common.Rounding;
1115
import org.elasticsearch.index.query.QueryShardContext;
@@ -20,9 +24,7 @@
2024
import org.elasticsearch.search.aggregations.support.ValuesSourceRegistry;
2125
import org.elasticsearch.search.internal.SearchContext;
2226

23-
import java.io.IOException;
24-
import java.util.Arrays;
25-
import java.util.Map;
27+
import org.elasticsearch.xpack.analytics.aggregations.support.AnalyticsValuesSourceType;
2628

2729
class RateAggregatorFactory extends ValuesSourceAggregatorFactory {
2830

@@ -44,15 +46,21 @@ class RateAggregatorFactory extends ValuesSourceAggregatorFactory {
4446
static void registerAggregators(ValuesSourceRegistry.Builder builder) {
4547
builder.register(
4648
RateAggregationBuilder.REGISTRY_KEY,
47-
Arrays.asList(CoreValuesSourceType.NUMERIC, CoreValuesSourceType.BOOLEAN),
48-
RateAggregator::new,
49+
Collections.singletonList(CoreValuesSourceType.NUMERIC),
50+
NumericRateAggregator::new,
51+
true
52+
);
53+
builder.register(
54+
RateAggregationBuilder.REGISTRY_KEY,
55+
Collections.singletonList(AnalyticsValuesSourceType.HISTOGRAM),
56+
HistogramRateAggregator::new,
4957
true
5058
);
5159
}
5260

5361
@Override
5462
protected Aggregator createUnmapped(SearchContext searchContext, Aggregator parent, Map<String, Object> metadata) throws IOException {
55-
return new RateAggregator(name, config, rateUnit, searchContext, parent, metadata) {
63+
return new AbstractRateAggregator(name, config, rateUnit, searchContext, parent, metadata) {
5664
@Override
5765
public LeafBucketCollector getLeafCollector(LeafReaderContext ctx, LeafBucketCollector sub) {
5866
return LeafBucketCollector.NO_OP_COLLECTOR;

0 commit comments

Comments
 (0)