You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by jo...@apache.org on 2019/01/17 22:51:21 UTC
[incubator-druid] branch master updated: Fixed buckets histogram
aggregator (#6638)
This is an automated email from the ASF dual-hosted git repository.
jonwei pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-druid.git
The following commit(s) were added to refs/heads/master by this push:
new 68f744e Fixed buckets histogram aggregator (#6638)
68f744e is described below
commit 68f744ec0ae573f17fc140f2e468869605e77fb6
Author: Jonathan Wei <jo...@users.noreply.github.com>
AuthorDate: Thu Jan 17 14:51:16 2019 -0800
Fixed buckets histogram aggregator (#6638)
* Fixed buckets histogram aggregator
* PR comments
* More PR comments
* Checkstyle
* TeamCity
* More TeamCity
* PR comment
* PR comment
* Fix doc formatting
---
NOTICE | 6 +
benchmarks/pom.xml | 22 +
.../benchmark/FixedHistogramAddBenchmark.java | 119 ++
.../druid/benchmark/FixedHistogramBenchmark.java | 201 +++
.../extensions-core/approximate-histograms.md | 165 ++-
extensions-core/datasketches/pom.xml | 16 +
extensions-core/histogram/pom.xml | 17 +
.../histogram/ApproximateHistogramDruidModule.java | 7 +-
.../histogram/FixedBucketsHistogram.java | 1364 ++++++++++++++++++
.../histogram/FixedBucketsHistogramAggregator.java | 127 ++
.../FixedBucketsHistogramAggregatorFactory.java | 334 +++++
.../FixedBucketsHistogramBufferAggregator.java | 125 ++
.../histogram/FixedBucketsHistogramSerde.java | 189 +++
.../aggregation/histogram/MaxPostAggregator.java | 10 +-
.../aggregation/histogram/MinPostAggregator.java | 12 +-
.../histogram/QuantilePostAggregator.java | 11 +-
.../histogram/QuantilesPostAggregator.java | 24 +-
.../FixedBucketsHistogramAggregationTest.java | 162 +++
.../FixedBucketsHistogramBufferAggregatorTest.java | 82 ++
.../FixedBucketsHistogramGroupByQueryTest.java | 262 ++++
.../histogram/FixedBucketsHistogramTest.java | 1453 ++++++++++++++++++++
.../FixedBucketsHistogramTopNQueryTest.java | 247 ++++
.../druid/query/aggregation/AggregatorUtil.java | 3 +
.../segment/incremental/IncrementalIndex.java | 2 +-
.../segment/serde/ComplexMetricExtractor.java | 7 +
...argeColumnSupportedComplexColumnSerializer.java | 1 -
26 files changed, 4942 insertions(+), 26 deletions(-)
diff --git a/NOTICE b/NOTICE
index fb92a6d..5ed39d8 100644
--- a/NOTICE
+++ b/NOTICE
@@ -92,3 +92,9 @@ This product contains modified versions of the Dockerfile and related configurat
* https://github.com/sequenceiq/hadoop-docker/
* COMMIT TAG:
* update this when this patch is committed
+
+This product contains fixed bins histogram percentile computation code adapted from Netflix Spectator:
+ * LICENSE:
+ * https://github.com/Netflix/spectator/blob/master/LICENSE (Apache License, Version 2.0)
+ * HOMEPAGE:
+ * https://github.com/Netflix/spectator
diff --git a/benchmarks/pom.xml b/benchmarks/pom.xml
index 9e2bd8d..9bd28e9 100644
--- a/benchmarks/pom.xml
+++ b/benchmarks/pom.xml
@@ -62,6 +62,28 @@
<version>${project.parent.version}</version>
</dependency>
<dependency>
+ <groupId>org.apache.druid.extensions</groupId>
+ <artifactId>druid-datasketches</artifactId>
+ <version>${project.parent.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.druid.extensions</groupId>
+ <artifactId>druid-datasketches</artifactId>
+ <version>${project.parent.version}</version>
+ <type>test-jar</type>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.druid.extensions</groupId>
+ <artifactId>druid-histogram</artifactId>
+ <version>${project.parent.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.druid.extensions</groupId>
+ <artifactId>druid-histogram</artifactId>
+ <version>${project.parent.version}</version>
+ <type>test-jar</type>
+ </dependency>
+ <dependency>
<groupId>org.apache.druid</groupId>
<artifactId>druid-processing</artifactId>
<version>${project.parent.version}</version>
diff --git a/benchmarks/src/main/java/org/apache/druid/benchmark/FixedHistogramAddBenchmark.java b/benchmarks/src/main/java/org/apache/druid/benchmark/FixedHistogramAddBenchmark.java
new file mode 100644
index 0000000..39ef26c
--- /dev/null
+++ b/benchmarks/src/main/java/org/apache/druid/benchmark/FixedHistogramAddBenchmark.java
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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.apache.druid.benchmark;
+
+import org.apache.commons.math3.distribution.NormalDistribution;
+import org.apache.druid.query.aggregation.histogram.FixedBucketsHistogram;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.infra.Blackhole;
+
+import java.util.Random;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
+
+@State(Scope.Benchmark)
+@Fork(value = 1)
+@Warmup(iterations = 10)
+@Measurement(iterations = 25)
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.MICROSECONDS)
+public class FixedHistogramAddBenchmark
+{
+ private static final int LOWER_LIMIT = 0;
+ private static final int UPPER_LIMIT = 100000;
+
+ // Number of samples
+ @Param({"100000", "1000000"})
+ int numEvents;
+
+ // Number of buckets
+ @Param({"10", "100", "1000", "10000", "100000"})
+ int numBuckets;
+
+ private FixedBucketsHistogram fixedHistogramForAdds;
+ private int[] randomValues;
+
+ private float[] normalDistributionValues;
+
+ @Setup
+ public void setup()
+ {
+ randomValues = new int[numEvents];
+ Random r = ThreadLocalRandom.current();
+ for (int i = 0; i < numEvents; i++) {
+ randomValues[i] = r.nextInt(UPPER_LIMIT);
+ }
+
+ fixedHistogramForAdds = new FixedBucketsHistogram(
+ LOWER_LIMIT,
+ UPPER_LIMIT,
+ numBuckets,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW
+ );
+
+ NormalDistribution normalDistribution = new NormalDistribution(50000, 10000);
+ normalDistributionValues = new float[numEvents];
+ for (int i = 0; i < numEvents; i++) {
+ normalDistributionValues[i] = (float) normalDistribution.sample();
+ }
+ }
+
+ @Benchmark
+ public void addFixedHisto(Blackhole bh)
+ {
+ fixedHistogramForAdds = new FixedBucketsHistogram(
+ LOWER_LIMIT,
+ UPPER_LIMIT,
+ numBuckets,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW
+ );
+
+ for (int i = 0; i < numEvents; i++) {
+ fixedHistogramForAdds.add(randomValues[i]);
+ }
+ bh.consume(fixedHistogramForAdds);
+ }
+
+ @Benchmark
+ public void addFixedHistoNormal(Blackhole bh)
+ {
+ fixedHistogramForAdds = new FixedBucketsHistogram(
+ LOWER_LIMIT,
+ UPPER_LIMIT,
+ numBuckets,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW
+ );
+
+ for (int i = 0; i < numEvents; i++) {
+ fixedHistogramForAdds.add(normalDistributionValues[i]);
+ }
+ bh.consume(fixedHistogramForAdds);
+ }
+}
diff --git a/benchmarks/src/main/java/org/apache/druid/benchmark/FixedHistogramBenchmark.java b/benchmarks/src/main/java/org/apache/druid/benchmark/FixedHistogramBenchmark.java
new file mode 100644
index 0000000..8e0e6de
--- /dev/null
+++ b/benchmarks/src/main/java/org/apache/druid/benchmark/FixedHistogramBenchmark.java
@@ -0,0 +1,201 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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.apache.druid.benchmark;
+
+import org.apache.druid.java.util.common.logger.Logger;
+import org.apache.druid.query.aggregation.histogram.FixedBucketsHistogram;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.infra.Blackhole;
+
+import java.util.Random;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
+
+@State(Scope.Benchmark)
+@Fork(value = 1)
+@Warmup(iterations = 10)
+@Measurement(iterations = 25)
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.MICROSECONDS)
+public class FixedHistogramBenchmark
+{
+ private static final Logger log = new Logger(FixedHistogramBenchmark.class);
+
+ private static final int LOWER_LIMIT = 0;
+ private static final int UPPER_LIMIT = 100000;
+
+ // Number of samples
+ @Param({"1000000"})
+ int numEvents;
+
+ // Number of buckets
+ @Param({"10", "100", "1000", "10000", "100000"})
+ int numBuckets;
+
+ private FixedBucketsHistogram fixedHistogram;
+ private FixedBucketsHistogram fixedHistogram2;
+ private FixedBucketsHistogram fixedHistogram3;
+ private FixedBucketsHistogram fixedHistogramForSparseLower;
+ private FixedBucketsHistogram fixedHistogramForSparseUpper;
+
+ private int[] randomValues;
+
+ private byte[] fixedFullSerializedAlready;
+ private byte[] fixedSparseLowerSerialized;
+ private byte[] fixedSparseUpperSerialized;
+
+ private double[] percentilesForFixed = new double[]{12.5, 25, 50, 75, 98};
+
+ @Setup
+ public void setup()
+ {
+ fixedHistogram = new FixedBucketsHistogram(
+ LOWER_LIMIT,
+ UPPER_LIMIT,
+ numBuckets,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW
+ );
+
+ fixedHistogram2 = new FixedBucketsHistogram(
+ LOWER_LIMIT,
+ UPPER_LIMIT,
+ (int) Math.round(numBuckets * 1.5),
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW
+ );
+
+ fixedHistogram3 = new FixedBucketsHistogram(
+ LOWER_LIMIT,
+ UPPER_LIMIT,
+ numBuckets,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW
+ );
+
+ fixedHistogramForSparseLower = new FixedBucketsHistogram(
+ LOWER_LIMIT,
+ UPPER_LIMIT,
+ numBuckets,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW
+ );
+
+ fixedHistogramForSparseUpper = new FixedBucketsHistogram(
+ LOWER_LIMIT,
+ UPPER_LIMIT,
+ numBuckets,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW
+ );
+
+ randomValues = new int[numEvents];
+ Random r = ThreadLocalRandom.current();
+ for (int i = 0; i < numEvents; i++) {
+ randomValues[i] = r.nextInt(UPPER_LIMIT);
+ fixedHistogram.add(randomValues[i]);
+ fixedHistogram2.add(randomValues[i]);
+ fixedHistogram3.add(randomValues[i]);
+
+ if (randomValues[i] < UPPER_LIMIT * 0.4) {
+ fixedHistogramForSparseLower.add(randomValues[i]);
+ }
+
+
+ if (randomValues[i] > UPPER_LIMIT * 0.6) {
+ fixedHistogramForSparseUpper.add(randomValues[i]);
+ }
+ }
+
+ fixedFullSerializedAlready = fixedHistogram.toBytesFull(true);
+ fixedSparseLowerSerialized = fixedHistogramForSparseLower.toBytesSparse(fixedHistogram.getNonEmptyBucketCount());
+ fixedSparseUpperSerialized = fixedHistogramForSparseUpper.toBytesSparse(fixedHistogram.getNonEmptyBucketCount());
+ }
+
+ @Benchmark
+ public void mergeFixedDifferentBuckets(Blackhole bh)
+ {
+ FixedBucketsHistogram copy = fixedHistogram.getCopy();
+ copy.combineHistogram(fixedHistogram2);
+ bh.consume(copy);
+ }
+
+ @Benchmark
+ public void mergeFixedSameBuckets(Blackhole bh)
+ {
+ FixedBucketsHistogram copy = fixedHistogram.getCopy();
+ copy.combineHistogram(fixedHistogram3);
+ bh.consume(copy);
+ }
+
+ @Benchmark
+ public void getPercentilesFixed(Blackhole bh)
+ {
+ float[] percentiles = fixedHistogram.percentilesFloat(percentilesForFixed);
+ bh.consume(percentiles);
+ }
+
+ @Benchmark
+ public void serializeFixedSparseLower(Blackhole bh)
+ {
+ byte[] sparseSerialized = fixedHistogramForSparseLower.toBytesSparse(fixedHistogramForSparseUpper.getNonEmptyBucketCount());
+ bh.consume(sparseSerialized);
+ }
+
+ @Benchmark
+ public void deserializeFixedSparseLower(Blackhole bh)
+ {
+ FixedBucketsHistogram fixedBucketsHistogram = FixedBucketsHistogram.fromBytes(fixedSparseLowerSerialized);
+ bh.consume(fixedBucketsHistogram);
+ }
+
+ @Benchmark
+ public void serializeFixedSparseUpper(Blackhole bh)
+ {
+ byte[] sparseSerialized = fixedHistogramForSparseUpper.toBytesSparse(fixedHistogramForSparseUpper.getNonEmptyBucketCount());
+ bh.consume(sparseSerialized);
+ }
+
+ @Benchmark
+ public void deserializeFixedSparseUpper(Blackhole bh)
+ {
+ FixedBucketsHistogram fixedBucketsHistogram = FixedBucketsHistogram.fromBytes(fixedSparseUpperSerialized);
+ bh.consume(fixedBucketsHistogram);
+ }
+
+ @Benchmark
+ public void serializeFixedFull(Blackhole bh)
+ {
+ byte[] fullSerialized = fixedHistogram.toBytesFull(true);
+ bh.consume(fullSerialized);
+ }
+
+ @Benchmark
+ public void deserializeFixedFull(Blackhole bh)
+ {
+ FixedBucketsHistogram fixedBucketsHistogram = FixedBucketsHistogram.fromBytes(fixedFullSerializedAlready);
+ bh.consume(fixedBucketsHistogram);
+ }
+}
diff --git a/docs/content/development/extensions-core/approximate-histograms.md b/docs/content/development/extensions-core/approximate-histograms.md
index b8c899d..ea9e17b 100644
--- a/docs/content/development/extensions-core/approximate-histograms.md
+++ b/docs/content/development/extensions-core/approximate-histograms.md
@@ -1,6 +1,6 @@
---
layout: doc_page
-title: "Approximate Histogram aggregator"
+title: "Approximate Histogram aggregators"
---
<!--
@@ -22,10 +22,14 @@ title: "Approximate Histogram aggregator"
~ under the License.
-->
-# Approximate Histogram aggregator
+# Approximate Histogram aggregators
Make sure to [include](../../operations/including-extensions.html) `druid-histogram` as an extension.
+The `druid-histogram` extension provides an approximate histogram aggregator and a fixed buckets histogram aggregator.
+
+## Approximate Histogram aggregator
+
This aggregator is based on
[http://jmlr.org/papers/volume11/ben-haim10a/ben-haim10a.pdf](http://jmlr.org/papers/volume11/ben-haim10a/ben-haim10a.pdf)
to compute approximate histograms, with the following modifications:
@@ -92,17 +96,148 @@ query.
|`numBuckets` |Number of output buckets for the resulting histogram. Bucket intervals are dynamic, based on the range of the underlying data. Use a post-aggregator to have finer control over the bucketing scheme|7|
|`lowerLimit`/`upperLimit`|Restrict the approximation to the given range. The values outside this range will be aggregated into two centroids. Counts of values outside this range are still maintained. |-INF/+INF|
+## Fixed Buckets Histogram
+
+The fixed buckets histogram aggregator builds a histogram on a numeric column, with evenly-sized buckets across a specified value range. Values outside of the range are handled based on a user-specified outlier handling mode.
+
+This histogram supports the min/max/quantiles post-aggregators but does not support the bucketing post-aggregators.
+
+### When to use
+
+The accuracy/usefulness of the fixed buckets histogram is extremely data-dependent; it is provided to support special use cases where the user has a great deal of prior information about the data being aggregated and knows that a fixed buckets implementation is suitable.
+
+For general histogram and quantile use cases, the [DataSketches Quantiles Sketch](../extensions-core/datasketches-quantiles.html) extension is recommended.
+
+### Properties
+
+
+|Property |Description |Default |
+|-------------------------|------------------------------|----------------------------------|
+|`type`|Type of the aggregator. Must `fixedBucketsHistogram`.|No default, must be specified|
+|`name`|Column name for the aggregator.|No default, must be specified|
+|`fieldName`|Column name of the input to the aggregator.|No default, must be specified|
+|`lowerLimit`|Lower limit of the histogram. |No default, must be specified|
+|`upperLimit`|Upper limit of the histogram. |No default, must be specified|
+|`numBuckets`|Number of buckets for the histogram. The range [lowerLimit, upperLimit] will be divided into `numBuckets` intervals of equal size.|10|
+|`outlierHandlingMode`|Specifies how values outside of [lowerLimit, upperLimit] will be handled. Supported modes are "ignore", "overflow", and "clip". See [outlier handling modes](#outlier-handling-modes) for more details.|No default, must be specified|
+
+An example aggregator spec is shown below:
+
+```json
+{
+ "type" : "fixedBucketsHistogram",
+ "name" : <output_name>,
+ "fieldName" : <metric_name>,
+ "numBuckets" : <integer>,
+ "lowerLimit" : <double>,
+ "upperLimit" : <double>,
+ "outlierHandlingMode": <mode>
+}
+```
+
+### Outlier handling modes
+
+The outlier handling mode specifies what should be done with values outside of the histogram's range. There are three supported modes:
+
+- `ignore`: Throw away outlier values.
+- `overflow`: A count of outlier values will be tracked by the histogram, available in the `lowerOutlierCount` and `upperOutlierCount` fields.
+- `clip`: Outlier values will be clipped to the `lowerLimit` or the `upperLimit` and included in the histogram.
+
+If you don't care about outliers, `ignore` is the cheapest option performance-wise. There is currently no difference in storage size among the modes.
+
+### Output fields
-### Approximate Histogram post-aggregators
+The histogram aggregator's output object has the following fields:
+
+- `lowerLimit`: Lower limit of the histogram
+- `upperLimit`: Upper limit of the histogram
+- `numBuckets`: Number of histogram buckets
+- `outlierHandlingMode`: Outlier handling mode
+- `count`: Total number of values contained in the histgram, excluding outliers
+- `lowerOutlierCount`: Count of outlier values below `lowerLimit`. Only used if the outlier mode is `overflow`.
+- `upperOutlierCount`: Count of outlier values above `upperLimit`. Only used if the outlier mode is `overflow`.
+- `missingValueCount`: Count of null values seen by the histogram.
+- `max`: Max value seen by the histogram. This does not include outlier values.
+- `min`: Min value seen by the histogram. This does not include outlier values.
+- `histogram`: An array of longs with size `numBuckets`, containing the bucket counts
+
+### Ingesting existing histograms
+
+It is also possible to ingest existing fixed buckets histograms. The input must be a Base64 string encoding a byte array that contains a serialized histogram object. Both "full" and "sparse" formats can be used. Please see [Serialization formats](#serialization-formats) below for details.
+
+### Serialization formats
+
+#### Full serialization format
+
+This format includes the full histogram bucket count array in the serialization format.
+
+```
+byte: serialization version, must be 0x01
+byte: encoding mode, 0x01 for full
+double: lowerLimit
+double: upperLimit
+int: numBuckets
+byte: outlier handling mode (0x00 for `ignore`, 0x01 for `overflow`, and 0x02 for `clip`)
+long: count, total number of values contained in the histogram, excluding outliers
+long: lowerOutlierCount
+long: upperOutlierCount
+long: missingValueCount
+double: max
+double: min
+array of longs: bucket counts for the histogram
+```
+
+#### Sparse serialization format
+
+This format represents the histogram bucket counts as (bucketNum, count) pairs. This serialization format is used when less than half of the histogram's buckets have values.
+
+```
+byte: serialization version, must be 0x01
+byte: encoding mode, 0x02 for sparse
+double: lowerLimit
+double: upperLimit
+int: numBuckets
+byte: outlier handling mode (0x00 for `ignore`, 0x01 for `overflow`, and 0x02 for `clip`)
+long: count, total number of values contained in the histogram, excluding outliers
+long: lowerOutlierCount
+long: upperOutlierCount
+long: missingValueCount
+double: max
+double: min
+int: number of following (bucketNum, count) pairs
+sequence of (int, long) pairs:
+ int: bucket number
+ count: bucket count
+```
+
+### Combining histograms with different bucketing schemes
+
+It is possible to combine two histograms with different bucketing schemes (lowerLimit, upperLimit, numBuckets) together.
+
+The bucketing scheme of the "left hand" histogram will be preserved (i.e., when running a query, the bucketing schemes specified in the query's histogram aggregators will be preserved).
+
+When merging, we assume that values are evenly distributed within the buckets of the "right hand" histogram.
+
+When the right-hand histogram contains outliers (when using `overflow` mode), we assume that all of the outliers counted in the right-hand histogram will be outliers in the left-hand histogram as well.
+
+For performance and accuracy reasons, we recommend avoiding aggregation of histograms with different bucketing schemes if possible.
+
+### Null handling
+
+If `druid.generic.useDefaultValueForNull` is false, null values will be tracked in the `missingValueCount` field of the histogram.
+
+If `druid.generic.useDefaultValueForNull` is true, null values will be added to the histogram as the default 0.0 value.
+
+## Histogram post-aggregators
Post-aggregators are used to transform opaque approximate histogram sketches
into bucketed histogram representations, as well as to compute various
distribution metrics such as quantiles, min, and max.
-#### Equal buckets post-aggregator
+### Equal buckets post-aggregator
Computes a visual representation of the approximate histogram with a given number of equal-sized bins.
-Bucket intervals are based on the range of the underlying data.
+Bucket intervals are based on the range of the underlying data. This aggregator is not supported for the fixed buckets histogram.
```json
{
@@ -113,7 +248,7 @@ Bucket intervals are based on the range of the underlying data.
}
```
-#### Buckets post-aggregator
+### Buckets post-aggregator
Computes a visual representation given an initial breakpoint, offset, and a bucket size.
@@ -121,6 +256,8 @@ Bucket size determines the width of the binning interval.
Offset determines the value on which those interval bins align.
+This aggregator is not supported for the fixed buckets histogram.
+
```json
{
"type": "buckets",
@@ -131,26 +268,28 @@ Offset determines the value on which those interval bins align.
}
```
-#### Custom buckets post-aggregator
+### Custom buckets post-aggregator
Computes a visual representation of the approximate histogram with bins laid out according to the given breaks.
+This aggregator is not supported for the fixed buckets histogram.
+
```json
{ "type" : "customBuckets", "name" : <output_name>, "fieldName" : <aggregator_name>,
"breaks" : [ <value>, <value>, ... ] }
```
-#### min post-aggregator
+### min post-aggregator
-Returns the minimum value of the underlying approximate histogram aggregator
+Returns the minimum value of the underlying approximate or fixed buckets histogram aggregator
```json
{ "type" : "min", "name" : <output_name>, "fieldName" : <aggregator_name> }
```
-#### max post-aggregator
+### max post-aggregator
-Returns the maximum value of the underlying approximate histogram aggregator
+Returns the maximum value of the underlying approximate or fixed buckets histogram aggregator
```json
{ "type" : "max", "name" : <output_name>, "fieldName" : <aggregator_name> }
@@ -158,7 +297,7 @@ Returns the maximum value of the underlying approximate histogram aggregator
#### quantile post-aggregator
-Computes a single quantile based on the underlying approximate histogram aggregator
+Computes a single quantile based on the underlying approximate or fixed buckets histogram aggregator
```json
{ "type" : "quantile", "name" : <output_name>, "fieldName" : <aggregator_name>,
@@ -167,7 +306,7 @@ Computes a single quantile based on the underlying approximate histogram aggrega
#### quantiles post-aggregator
-Computes an array of quantiles based on the underlying approximate histogram aggregator
+Computes an array of quantiles based on the underlying approximate or fixed buckets histogram aggregator
```json
{ "type" : "quantiles", "name" : <output_name>, "fieldName" : <aggregator_name>,
diff --git a/extensions-core/datasketches/pom.xml b/extensions-core/datasketches/pom.xml
index ef44977..e3c10e9 100644
--- a/extensions-core/datasketches/pom.xml
+++ b/extensions-core/datasketches/pom.xml
@@ -133,4 +133,20 @@
</dependency>
</dependencies>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-jar-plugin</artifactId>
+ <executions>
+ <execution>
+ <goals>
+ <goal>test-jar</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+
</project>
diff --git a/extensions-core/histogram/pom.xml b/extensions-core/histogram/pom.xml
index f30bb39..bd689c7 100644
--- a/extensions-core/histogram/pom.xml
+++ b/extensions-core/histogram/pom.xml
@@ -87,4 +87,21 @@
</dependency>
</dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-jar-plugin</artifactId>
+ <executions>
+ <execution>
+ <goals>
+ <goal>test-jar</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+
</project>
diff --git a/extensions-core/histogram/src/main/java/org/apache/druid/query/aggregation/histogram/ApproximateHistogramDruidModule.java b/extensions-core/histogram/src/main/java/org/apache/druid/query/aggregation/histogram/ApproximateHistogramDruidModule.java
index b71e215..167f0c2 100644
--- a/extensions-core/histogram/src/main/java/org/apache/druid/query/aggregation/histogram/ApproximateHistogramDruidModule.java
+++ b/extensions-core/histogram/src/main/java/org/apache/druid/query/aggregation/histogram/ApproximateHistogramDruidModule.java
@@ -47,7 +47,8 @@ public class ApproximateHistogramDruidModule implements DruidModule
QuantilesPostAggregator.class,
QuantilePostAggregator.class,
MinPostAggregator.class,
- MaxPostAggregator.class
+ MaxPostAggregator.class,
+ FixedBucketsHistogramAggregatorFactory.class
)
);
}
@@ -59,6 +60,10 @@ public class ApproximateHistogramDruidModule implements DruidModule
ComplexMetrics.registerSerde("approximateHistogram", new ApproximateHistogramFoldingSerde());
}
+ if (ComplexMetrics.getSerdeForType(FixedBucketsHistogramAggregator.TYPE_NAME) == null) {
+ ComplexMetrics.registerSerde(FixedBucketsHistogramAggregator.TYPE_NAME, new FixedBucketsHistogramSerde());
+ }
+
if (binder != null) {
// Binder is null in some tests.
SqlBindings.addAggregator(binder, QuantileSqlAggregator.class);
diff --git a/extensions-core/histogram/src/main/java/org/apache/druid/query/aggregation/histogram/FixedBucketsHistogram.java b/extensions-core/histogram/src/main/java/org/apache/druid/query/aggregation/histogram/FixedBucketsHistogram.java
new file mode 100644
index 0000000..5ddafeb
--- /dev/null
+++ b/extensions-core/histogram/src/main/java/org/apache/druid/query/aggregation/histogram/FixedBucketsHistogram.java
@@ -0,0 +1,1364 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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.apache.druid.query.aggregation.histogram;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonValue;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import org.apache.commons.codec.binary.Base64;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.StringUtils;
+
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.Objects;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+public class FixedBucketsHistogram
+{
+ public static final byte SERIALIZATION_VERSION = 0x01;
+
+ /**
+ * Serialization header format:
+ *
+ * byte: serialization version, must be 0x01
+ * byte: encoding mode, 0x01 for full, 0x02 for sparse
+ */
+ public static final int SERDE_HEADER_SIZE = Byte.BYTES +
+ Byte.BYTES;
+
+ /**
+ * Common serialization fields format:
+ *
+ * double: lowerLimit
+ * double: upperLimit
+ * int: numBuckets
+ * byte: outlier handling mode (0x00 for `ignore`, 0x01 for `overflow`, and 0x02 for `clip`)
+ * long: count, total number of values contained in the histogram, excluding outliers
+ * long: lowerOutlierCount
+ * long: upperOutlierCount
+ * long: missingValueCount
+ * double: max
+ * double: min
+ */
+ public static final int COMMON_FIELDS_SIZE = Double.BYTES +
+ Double.BYTES +
+ Integer.BYTES +
+ Byte.BYTES +
+ Long.BYTES * 4 +
+ Double.BYTES +
+ Double.BYTES;
+
+ /**
+ * Full serialization format:
+ *
+ * serialization header
+ * common fields
+ * array of longs: bucket counts for the histogram
+ */
+ public static final byte FULL_ENCODING_MODE = 0x01;
+
+ /**
+ * Sparse serialization format:
+ *
+ * serialization header
+ * common fields
+ * int: number of following (bucketNum, count) pairs
+ * sequence of (int, long) pairs:
+ * int: bucket number
+ * count: bucket count
+ */
+ public static final byte SPARSE_ENCODING_MODE = 0x02;
+
+ /**
+ * Determines how the the histogram handles outliers.
+ *
+ * Ignore: do not track outliers at all
+ * Overflow: track outlier counts in upperOutlierCount and lowerOutlierCount.
+ * The min and max do not take outlier values into account.
+ * Clip: Clip outlier values to either the start or end of the histogram's range, adding them to the first or
+ * last bucket, respectively. The min and max are affected by such clipped outlier values.
+ */
+ public enum OutlierHandlingMode
+ {
+ IGNORE,
+ OVERFLOW,
+ CLIP;
+
+ @JsonValue
+ @Override
+ public String toString()
+ {
+ return StringUtils.toLowerCase(this.name());
+ }
+
+ @JsonCreator
+ public static OutlierHandlingMode fromString(String name)
+ {
+ return valueOf(StringUtils.toUpperCase(name));
+ }
+
+ public byte[] getCacheKey()
+ {
+ return new byte[] {(byte) this.ordinal()};
+ }
+ }
+
+ @JsonIgnore
+ private final OutlierHandler outlierHandler;
+
+ /**
+ * Locking is needed when the non-buffer aggregator is used within a realtime ingestion task.
+ *
+ * The following areas are locked:
+ * - Add value
+ * - Merge histograms
+ * - Compute percentiles
+ * - Serialization
+ */
+ @JsonIgnore
+ private final ReadWriteLock readWriteLock;
+
+ private double lowerLimit;
+ private double upperLimit;
+ private int numBuckets;
+
+ private long upperOutlierCount = 0;
+ private long lowerOutlierCount = 0;
+ private long missingValueCount = 0;
+ private long[] histogram;
+ private double bucketSize;
+ private OutlierHandlingMode outlierHandlingMode;
+
+ private long count = 0;
+ private double max = Double.NEGATIVE_INFINITY;
+ private double min = Double.POSITIVE_INFINITY;
+
+ public FixedBucketsHistogram(
+ double lowerLimit,
+ double upperLimit,
+ int numBuckets,
+ OutlierHandlingMode outlierHandlingMode
+ )
+ {
+ Preconditions.checkArgument(
+ upperLimit > lowerLimit,
+ "Upper limit [%s] must be greater than lower limit [%s].",
+ upperLimit,
+ lowerLimit
+ );
+ Preconditions.checkArgument(
+ numBuckets > 0,
+ "numBuckets must be > 0, got [%s] instead.",
+ numBuckets
+ );
+ this.lowerLimit = lowerLimit;
+ this.upperLimit = upperLimit;
+ this.numBuckets = numBuckets;
+ this.outlierHandlingMode = outlierHandlingMode;
+
+ this.histogram = new long[numBuckets];
+ this.bucketSize = (upperLimit - lowerLimit) / numBuckets;
+ this.readWriteLock = new ReentrantReadWriteLock(true);
+
+ this.outlierHandler = makeOutlierHandler();
+ }
+
+ @VisibleForTesting
+ protected FixedBucketsHistogram(
+ double lowerLimit,
+ double upperLimit,
+ int numBuckets,
+ OutlierHandlingMode outlierHandlingMode,
+ long[] histogram,
+ long count,
+ double max,
+ double min,
+ long lowerOutlierCount,
+ long upperOutlierCount,
+ long missingValueCount
+ )
+ {
+ this.lowerLimit = lowerLimit;
+ this.upperLimit = upperLimit;
+ this.numBuckets = numBuckets;
+ this.outlierHandlingMode = outlierHandlingMode;
+ this.histogram = histogram;
+ this.count = count;
+ this.max = max;
+ this.min = min;
+ this.upperOutlierCount = upperOutlierCount;
+ this.lowerOutlierCount = lowerOutlierCount;
+ this.missingValueCount = missingValueCount;
+
+ this.bucketSize = (upperLimit - lowerLimit) / numBuckets;
+ this.readWriteLock = new ReentrantReadWriteLock(true);
+ this.outlierHandler = makeOutlierHandler();
+ }
+
+ @VisibleForTesting
+ public FixedBucketsHistogram getCopy()
+ {
+ // only used for tests/benchmarks
+ return new FixedBucketsHistogram(
+ lowerLimit,
+ upperLimit,
+ numBuckets,
+ outlierHandlingMode,
+ Arrays.copyOf(histogram, histogram.length),
+ count,
+ max,
+ min,
+ lowerOutlierCount,
+ upperOutlierCount,
+ missingValueCount
+ );
+ }
+
+ @JsonProperty
+ public double getLowerLimit()
+ {
+ return lowerLimit;
+ }
+
+ @JsonProperty
+ public double getUpperLimit()
+ {
+ return upperLimit;
+ }
+
+ @JsonProperty
+ public int getNumBuckets()
+ {
+ return numBuckets;
+ }
+
+ @JsonProperty
+ public long getUpperOutlierCount()
+ {
+ return upperOutlierCount;
+ }
+
+ @JsonProperty
+ public long getLowerOutlierCount()
+ {
+ return lowerOutlierCount;
+ }
+
+ @JsonProperty
+ public long getMissingValueCount()
+ {
+ return missingValueCount;
+ }
+
+ @JsonProperty
+ public long[] getHistogram()
+ {
+ return histogram;
+ }
+
+ @JsonProperty
+ public double getBucketSize()
+ {
+ return bucketSize;
+ }
+
+ @JsonProperty
+ public long getCount()
+ {
+ return count;
+ }
+
+ @JsonProperty
+ public double getMax()
+ {
+ return max;
+ }
+
+ @JsonProperty
+ public double getMin()
+ {
+ return min;
+ }
+
+ @JsonProperty
+ public OutlierHandlingMode getOutlierHandlingMode()
+ {
+ return outlierHandlingMode;
+ }
+
+ @Override
+ public String toString()
+ {
+ return "{" +
+ "lowerLimit=" + lowerLimit +
+ ", upperLimit=" + upperLimit +
+ ", numBuckets=" + numBuckets +
+ ", upperOutlierCount=" + upperOutlierCount +
+ ", lowerOutlierCount=" + lowerOutlierCount +
+ ", missingValueCount=" + missingValueCount +
+ ", histogram=" + Arrays.toString(histogram) +
+ ", outlierHandlingMode=" + outlierHandlingMode +
+ ", count=" + count +
+ ", max=" + max +
+ ", min=" + min +
+ '}';
+ }
+
+ @Override
+ public boolean equals(Object o)
+ {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ FixedBucketsHistogram that = (FixedBucketsHistogram) o;
+ return Double.compare(that.getLowerLimit(), getLowerLimit()) == 0 &&
+ Double.compare(that.getUpperLimit(), getUpperLimit()) == 0 &&
+ getNumBuckets() == that.getNumBuckets() &&
+ getUpperOutlierCount() == that.getUpperOutlierCount() &&
+ getLowerOutlierCount() == that.getLowerOutlierCount() &&
+ getMissingValueCount() == that.getMissingValueCount() &&
+ Double.compare(that.getBucketSize(), getBucketSize()) == 0 &&
+ getCount() == that.getCount() &&
+ Double.compare(that.getMax(), getMax()) == 0 &&
+ Double.compare(that.getMin(), getMin()) == 0 &&
+ Arrays.equals(getHistogram(), that.getHistogram()) &&
+ getOutlierHandlingMode() == that.getOutlierHandlingMode();
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return Objects.hash(
+ getLowerLimit(),
+ getUpperLimit(),
+ getNumBuckets(),
+ getUpperOutlierCount(),
+ getLowerOutlierCount(),
+ getMissingValueCount(),
+ Arrays.hashCode(getHistogram()),
+ getBucketSize(),
+ getOutlierHandlingMode(),
+ getCount(),
+ getMax(),
+ getMin()
+ );
+ }
+
+ public ReadWriteLock getReadWriteLock()
+ {
+ return readWriteLock;
+ }
+
+ /**
+ * Add a value to the histogram, using the outlierHandler to account for outliers.
+ *
+ * @param value value to be added.
+ */
+ public void add(
+ double value
+ )
+ {
+ readWriteLock.writeLock().lock();
+
+ try {
+ if (value < lowerLimit) {
+ outlierHandler.handleOutlierAdd(false);
+ return;
+ } else if (value >= upperLimit) {
+ outlierHandler.handleOutlierAdd(true);
+ return;
+ }
+
+ count += 1;
+ if (value > max) {
+ max = value;
+ }
+ if (value < min) {
+ min = value;
+ }
+
+ double valueRelativeToRange = value - lowerLimit;
+ int targetBucket = (int) (valueRelativeToRange / bucketSize);
+
+ if (targetBucket >= histogram.length) {
+ targetBucket = histogram.length - 1;
+ }
+
+ histogram[targetBucket] += 1;
+ }
+ finally {
+ readWriteLock.writeLock().unlock();
+ }
+ }
+
+ /**
+ * Called when the histogram encounters a null value.
+ */
+ public void incrementMissing()
+ {
+ readWriteLock.writeLock().lock();
+
+ try {
+ missingValueCount++;
+ }
+ finally {
+ readWriteLock.writeLock().unlock();
+ }
+ }
+
+ /**
+ * Merge another histogram into this one. Only the state of this histogram is updated.
+ *
+ * If the two histograms have identical buckets, a simpler algorithm is used.
+ *
+ * @param otherHistogram
+ */
+ public void combineHistogram(FixedBucketsHistogram otherHistogram)
+ {
+ if (otherHistogram == null) {
+ return;
+ }
+
+ readWriteLock.writeLock().lock();
+ otherHistogram.getReadWriteLock().readLock().lock();
+
+ try {
+ missingValueCount += otherHistogram.getMissingValueCount();
+
+ if (bucketSize == otherHistogram.getBucketSize() &&
+ lowerLimit == otherHistogram.getLowerLimit() &&
+ upperLimit == otherHistogram.getUpperLimit()) {
+ combineHistogramSameBuckets(otherHistogram);
+ } else {
+ combineHistogramDifferentBuckets(otherHistogram);
+ }
+ }
+ finally {
+ readWriteLock.writeLock().unlock();
+ otherHistogram.getReadWriteLock().readLock().unlock();
+ }
+ }
+
+ /**
+ * Merge another histogram that has the same range and same buckets.
+ *
+ * @param otherHistogram
+ */
+ private void combineHistogramSameBuckets(FixedBucketsHistogram otherHistogram)
+ {
+ long[] otherHistogramArray = otherHistogram.getHistogram();
+ for (int i = 0; i < numBuckets; i++) {
+ histogram[i] += otherHistogramArray[i];
+ }
+
+ count += otherHistogram.getCount();
+ max = Math.max(max, otherHistogram.getMax());
+ min = Math.min(min, otherHistogram.getMin());
+
+ outlierHandler.handleOutliersForCombineSameBuckets(otherHistogram);
+ }
+
+ /**
+ * Merge another histogram that has different buckets from mine.
+ *
+ * First, check if the other histogram falls entirely outside of my defined range. If so, call the appropriate
+ * function for optimized handling.
+ *
+ * Otherwise, this merges the histograms with a more general function.
+ *
+ * @param otherHistogram
+ */
+ private void combineHistogramDifferentBuckets(FixedBucketsHistogram otherHistogram)
+ {
+ if (otherHistogram.getLowerLimit() >= upperLimit) {
+ outlierHandler.handleOutliersCombineDifferentBucketsAllUpper(otherHistogram);
+ } else if (otherHistogram.getUpperLimit() <= lowerLimit) {
+ outlierHandler.handleOutliersCombineDifferentBucketsAllLower(otherHistogram);
+ } else {
+ simpleInterpolateMerge(otherHistogram);
+ }
+ }
+
+ /**
+ * Get a sum of bucket counts from either the start of a histogram's range or end, up to a specified cutoff value.
+ *
+ * For example, if I have the following histogram with a range of 0-40, with 4 buckets and
+ * per-bucket counts of 5, 2, 10, and 7:
+ *
+ * | 5 | 2 | 24 | 7 |
+ * 0 10 20 30 40
+ *
+ * Calling this function with a cutoff of 25 and fromStart = true would:
+ * - Sum the first two bucket counts 5 + 2
+ * - Since the cutoff falls in the third bucket, multiply the third bucket's count by the fraction of the bucket range
+ * covered by the cutoff, in this case the fraction is ((25 - 20) / 10) = 0.5
+ * - The total count returned is 5 + 2 + 12
+ *
+ * @param cutoff Cutoff point within the histogram's range
+ * @param fromStart If true, sum the bucket counts starting from the beginning of the histogram range.
+ * If false, sum from the other direction, starting from the end of the histogram range.
+ * @return Sum of bucket counts up to the cutoff point
+ */
+ private double getCumulativeCount(double cutoff, boolean fromStart)
+ {
+ int cutoffBucket = (int) ((cutoff - lowerLimit) / bucketSize);
+ double count = 0;
+
+ if (fromStart) {
+ for (int i = 0; i <= cutoffBucket; i++) {
+ if (i == cutoffBucket) {
+ double bucketStart = i * bucketSize + lowerLimit;
+ double partialCount = ((cutoff - bucketStart) / bucketSize) * histogram[i];
+ count += partialCount;
+ } else {
+ count += histogram[i];
+ }
+ }
+ } else {
+ for (int i = cutoffBucket; i < histogram.length; i++) {
+ if (i == cutoffBucket) {
+ double bucketEnd = ((i + 1) * bucketSize) + lowerLimit;
+ double partialCount = ((bucketEnd - cutoff) / bucketSize) * histogram[i];
+ count += partialCount;
+ } else {
+ count += histogram[i];
+ }
+ }
+ }
+ return count;
+ }
+
+ /**
+ * Combines this histogram with another histogram by "rebucketing" the other histogram to match our bucketing scheme,
+ * assuming that the original input values are uniformly distributed within each bucket in the other histogram.
+ *
+ * Suppose we have the following histograms:
+ *
+ * | 0 | 0 | 0 | 0 | 0 | 0 |
+ * 0 1 2 3 4 5 6
+ *
+ * | 4 | 4 | 0 | 0 |
+ * 0 3 6 9 12
+ *
+ * We will preserve the bucketing scheme of our histogram, and determine what cut-off points within the other
+ * histogram align with our bucket boundaries.
+ *
+ * Using this example, we would effectively rebucket the second histogram to the following:
+ *
+ * | 1.333 | 1.333 | 1.333 | 1.333 | 1.333 | 1.333 | 0 | 0 | 0 | 0 | 0 | 0 |
+ * 0 1 2 3 4 5 6 7 8 9 10 11 12
+ *
+ * The 0-3 bucket in the second histogram is rebucketed across new buckets with size 1, with the original frequency 4
+ * multiplied by the fraction (new bucket size / original bucket size), in this case 1/3.
+ *
+ * These new rebucketed counts are then added to the base histogram's buckets, with rounding, resulting in:
+ *
+ * | 1 | 1 | 1 | 1 | 1 | 1 |
+ * 0 1 2 3 4 5 6
+ *
+ * @param otherHistogram other histogram to be merged
+ */
+ private void simpleInterpolateMerge(FixedBucketsHistogram otherHistogram)
+ {
+ double rangeStart = Math.max(lowerLimit, otherHistogram.getLowerLimit());
+ double rangeEnd = Math.min(upperLimit, otherHistogram.getUpperLimit());
+
+ int myCurBucket = (int) ((rangeStart - lowerLimit) / bucketSize);
+ double myNextCursorBoundary = (myCurBucket + 1) * bucketSize + lowerLimit;
+ double myCursor = rangeStart;
+
+ int theirCurBucket = (int) ((rangeStart - otherHistogram.getLowerLimit()) / otherHistogram.getBucketSize());
+ double theirNextCursorBoundary = (theirCurBucket + 1) * otherHistogram.getBucketSize() + otherHistogram.getLowerLimit();
+ double theirCursor = rangeStart;
+
+ double intraBucketStride = otherHistogram.getBucketSize() / otherHistogram.getHistogram()[theirCurBucket];
+
+ myNextCursorBoundary = Math.min(myNextCursorBoundary, rangeEnd);
+ theirNextCursorBoundary = Math.min(theirNextCursorBoundary, rangeEnd);
+
+ outlierHandler.simpleInterpolateMergeHandleOutliers(otherHistogram, rangeStart, rangeEnd);
+
+ double theirCurrentLowerBucketBoundary = theirCurBucket * otherHistogram.getBucketSize() + otherHistogram.getLowerLimit();
+
+ while (myCursor < rangeEnd) {
+ double needToConsume = myNextCursorBoundary - myCursor;
+ double canConsumeFromOtherBucket = theirNextCursorBoundary - theirCursor;
+ double toConsume = Math.min(needToConsume, canConsumeFromOtherBucket);
+
+ // consume one of our bucket's worth of data from the other histogram
+ while (needToConsume > 0) {
+ double fractional = toConsume / otherHistogram.getBucketSize();
+ double fractionalCount = otherHistogram.getHistogram()[theirCurBucket] * fractional;
+
+ if (Math.round(fractionalCount) > 0) {
+ // in this chunk from the other histogram bucket, calculate min and max interpolated values that would be added
+ double minStride = Math.ceil(
+ (theirCursor - (theirCurBucket * otherHistogram.getBucketSize() + otherHistogram.getLowerLimit()))
+ / intraBucketStride
+ );
+
+ minStride *= intraBucketStride;
+ minStride += theirCurrentLowerBucketBoundary;
+ if (minStride >= theirCursor) {
+ min = Math.min(minStride, min);
+ max = Math.max(minStride, max);
+ }
+
+ double maxStride = Math.floor(
+ (theirCursor + toConsume - (theirCurBucket * otherHistogram.getBucketSize()
+ + otherHistogram.getLowerLimit())) / intraBucketStride
+ );
+ maxStride *= intraBucketStride;
+ maxStride += theirCurrentLowerBucketBoundary;
+ if (maxStride < theirCursor + toConsume) {
+ max = Math.max(maxStride, max);
+ min = Math.min(maxStride, min);
+ }
+ }
+
+ histogram[myCurBucket] += Math.round(fractionalCount);
+ count += Math.round(fractionalCount);
+
+ needToConsume -= toConsume;
+ theirCursor += toConsume;
+ myCursor += toConsume;
+
+
+ if (theirCursor >= rangeEnd) {
+ break;
+ }
+
+ // we've crossed buckets in the other histogram
+ if (theirCursor >= theirNextCursorBoundary) {
+ theirCurBucket += 1;
+ intraBucketStride = otherHistogram.getBucketSize() / otherHistogram.getHistogram()[theirCurBucket];
+ theirCurrentLowerBucketBoundary = theirCurBucket * otherHistogram.getBucketSize() + otherHistogram.getLowerLimit();
+ theirNextCursorBoundary = (theirCurBucket + 1) * otherHistogram.getBucketSize() + otherHistogram.getLowerLimit();
+ theirNextCursorBoundary = Math.min(theirNextCursorBoundary, rangeEnd);
+ }
+
+ canConsumeFromOtherBucket = theirNextCursorBoundary - theirCursor;
+ toConsume = Math.min(needToConsume, canConsumeFromOtherBucket);
+ }
+
+ if (theirCursor >= rangeEnd) {
+ break;
+ }
+
+ myCurBucket += 1;
+ myNextCursorBoundary = (myCurBucket + 1) * bucketSize + lowerLimit;
+ myNextCursorBoundary = Math.min(myNextCursorBoundary, rangeEnd);
+ }
+ }
+
+
+ /**
+ * Estimate percentiles from the histogram bucket counts, using the following process:
+ *
+ * Suppose we have the following histogram with range 0-10, with bucket size = 2, and the following counts:
+ *
+ * | 0 | 1 | 2 | 4 | 0 |
+ * 0 2 4 6 8 10
+ *
+ * If we wanted to estimate the 75th percentile:
+ * 1. Find the frequency cut-off for the 75th percentile: 0.75 * 7 (the total count for the entire histogram) = 5.25
+ * 2. Find the bucket for the cut-off point: in this case, the 6-8 bucket, since the cumulative frequency
+ * from 0-6 is 3 and cumulative frequency from 0-8 is 7
+ * 3. The percentile estimate is L + F * W, where:
+ *
+ * L = lower boundary of target bucket
+ * F = (frequency cut-off - cumulative frequency up to target bucket) / frequency of target bucket
+ * W = bucket size
+ *
+ * In this case:
+ * 75th percentile estimate = 6 + ((5.25 - 3) / 4) * 2 = 7.125
+ *
+ * Based off PercentileBuckets code from Netflix Spectator:
+ * - https://github.com/Netflix/spectator/blob/0d083da3a60221de9cc710e314b0749e47a40e67/spectator-api/src/main/java/com/netflix/spectator/api/histogram/PercentileBuckets.java#L105
+ *
+ * @param pcts Array of percentiles to be estimated. Must be sorted in ascending order.
+ * @return Estimates of percentile values, in the same order as the provided input.
+ */
+ public float[] percentilesFloat(double[] pcts)
+ {
+ readWriteLock.readLock().lock();
+
+ try {
+ float[] results = new float[pcts.length];
+ long total = count;
+
+ int pctIdx = 0;
+
+ long prev = 0;
+ double prevP = 0.0;
+ double prevB = lowerLimit;
+ for (int i = 0; i < numBuckets; ++i) {
+ long next = prev + histogram[i];
+ double nextP = 100.0 * next / total;
+ double nextB = (i + 1) * bucketSize + lowerLimit;
+ while (pctIdx < pcts.length && nextP >= pcts[pctIdx]) {
+ double f = (pcts[pctIdx] - prevP) / (nextP - prevP);
+ results[pctIdx] = (float) (f * (nextB - prevB) + prevB);
+ ++pctIdx;
+ }
+ if (pctIdx >= pcts.length) {
+ break;
+ }
+ prev = next;
+ prevP = nextP;
+ prevB = nextB;
+ }
+
+ return results;
+ }
+ finally {
+ readWriteLock.readLock().unlock();
+ }
+ }
+
+ /**
+ * Encode the serialized form generated by toBytes() in Base64, used as the JSON serialization format.
+ *
+ * @return Base64 serialization
+ */
+ @JsonValue
+ public String toBase64()
+ {
+ byte[] asBytes = toBytes();
+ return StringUtils.fromUtf8(Base64.encodeBase64(asBytes));
+ }
+
+ /**
+ * Serialize the histogram, with two possible encodings chosen based on the number of filled buckets:
+ *
+ * Full: Store the histogram buckets as an array of bucket counts, with size numBuckets
+ * Sparse: Store the histogram buckets as a list of (bucketNum, count) pairs.
+ */
+ public byte[] toBytes()
+ {
+ readWriteLock.readLock().lock();
+
+ try {
+ int nonEmptyBuckets = getNonEmptyBucketCount();
+ if (nonEmptyBuckets < (numBuckets / 2)) {
+ return toBytesSparse(nonEmptyBuckets);
+ } else {
+ return toBytesFull(true);
+ }
+ }
+ finally {
+ readWriteLock.readLock().unlock();
+ }
+ }
+
+ /**
+ * Write a serialization header containing the serde version byte and full/sparse encoding mode byte.
+ *
+ * This header is not needed when serializing the histogram for localized internal use within the
+ * buffer aggregator implementation.
+ *
+ * @param buf Destination buffer
+ * @param mode Full or sparse mode
+ */
+ private void writeByteBufferSerdeHeader(ByteBuffer buf, byte mode)
+ {
+ buf.put(SERIALIZATION_VERSION);
+ buf.put(mode);
+ }
+
+ /**
+ * Serializes histogram fields that are common to both the full and sparse encoding modes.
+ *
+ * @param buf Destination buffer
+ */
+ private void writeByteBufferCommonFields(ByteBuffer buf)
+ {
+ buf.putDouble(lowerLimit);
+ buf.putDouble(upperLimit);
+ buf.putInt(numBuckets);
+ buf.put((byte) outlierHandlingMode.ordinal());
+
+ buf.putLong(count);
+ buf.putLong(lowerOutlierCount);
+ buf.putLong(upperOutlierCount);
+ buf.putLong(missingValueCount);
+
+ buf.putDouble(max);
+ buf.putDouble(min);
+ }
+
+ /**
+ * Serialize the histogram in full encoding mode.
+ *
+ * For efficiency, the header is not written when the histogram
+ * is serialized for localized internal use within the buffer aggregator implementation.
+ *
+ * @param withHeader If true, include the serialization header
+ * @return Serialized histogram with full encoding
+ */
+ public byte[] toBytesFull(boolean withHeader)
+ {
+ int size = getFullStorageSize(numBuckets);
+ if (withHeader) {
+ size += SERDE_HEADER_SIZE;
+ }
+ ByteBuffer buf = ByteBuffer.allocate(size);
+ writeByteBufferFull(buf, withHeader);
+ return buf.array();
+ }
+
+ /**
+ * Helper method for toBytesFull
+ *
+ * @param buf Destination buffer
+ * @param withHeader If true, include the serialization header
+ */
+ private void writeByteBufferFull(ByteBuffer buf, boolean withHeader)
+ {
+ if (withHeader) {
+ writeByteBufferSerdeHeader(buf, FULL_ENCODING_MODE);
+ }
+
+ writeByteBufferCommonFields(buf);
+
+ buf.asLongBuffer().put(histogram);
+ buf.position(buf.position() + Long.BYTES * histogram.length);
+ }
+
+ /**
+ * Serialize the histogram in sparse encoding mode.
+ *
+ * The serialization header is always written, since the sparse encoding is only used in situations where the
+ * header is required.
+ *
+ * @param nonEmptyBuckets Number of non-empty buckets in the histogram
+ * @return Serialized histogram with sparse encoding
+ */
+ public byte[] toBytesSparse(int nonEmptyBuckets)
+ {
+ int size = SERDE_HEADER_SIZE + getSparseStorageSize(nonEmptyBuckets);
+ ByteBuffer buf = ByteBuffer.allocate(size);
+ writeByteBufferSparse(buf, nonEmptyBuckets);
+ return buf.array();
+ }
+
+ /**
+ * Helper method for toBytesSparse
+ *
+ * @param buf Destination buffer
+ * @param nonEmptyBuckets Number of non-empty buckets in the histogram
+ */
+ public void writeByteBufferSparse(ByteBuffer buf, int nonEmptyBuckets)
+ {
+ writeByteBufferSerdeHeader(buf, SPARSE_ENCODING_MODE);
+ writeByteBufferCommonFields(buf);
+
+ buf.putInt(nonEmptyBuckets);
+ int bucketsWritten = 0;
+ for (int i = 0; i < numBuckets; i++) {
+ if (histogram[i] > 0) {
+ buf.putInt(i);
+ buf.putLong(histogram[i]);
+ bucketsWritten += 1;
+ }
+ if (bucketsWritten == nonEmptyBuckets) {
+ break;
+ }
+ }
+ }
+
+ /**
+ * Deserialize a Base64-encoded histogram, used when deserializing from JSON (such as when transferring query results)
+ * or when ingesting a pre-computed histogram.
+ *
+ * @param encodedHistogram Base64-encoded histogram
+ * @return Deserialized object
+ */
+ public static FixedBucketsHistogram fromBase64(String encodedHistogram)
+ {
+ byte[] asBytes = Base64.decodeBase64(encodedHistogram.getBytes(StandardCharsets.UTF_8));
+ return fromBytes(asBytes);
+ }
+
+ /**
+ * General deserialization method for FixedBucketsHistogram.
+ *
+ * @param bytes
+ * @return
+ */
+ public static FixedBucketsHistogram fromBytes(byte[] bytes)
+ {
+ ByteBuffer buf = ByteBuffer.wrap(bytes);
+ return fromByteBuffer(buf);
+ }
+
+ /**
+ * Deserialization helper method
+ *
+ * @param buf Source buffer containing serialized histogram
+ * @return Deserialized object
+ */
+ public static FixedBucketsHistogram fromByteBuffer(ByteBuffer buf)
+ {
+ byte serializationVersion = buf.get();
+ Preconditions.checkArgument(
+ serializationVersion == SERIALIZATION_VERSION,
+ StringUtils.format("Only serialization version %s is supported.", SERIALIZATION_VERSION)
+ );
+ byte mode = buf.get();
+ if (mode == FULL_ENCODING_MODE) {
+ return fromByteBufferFullNoSerdeHeader(buf);
+ } else if (mode == SPARSE_ENCODING_MODE) {
+ return fromBytesSparse(buf);
+ } else {
+ throw new ISE("Invalid histogram serde mode: %s", mode);
+ }
+ }
+
+ /**
+ * Helper method for deserializing histograms with full encoding mode. Assumes that the serialization header is not
+ * present or has already been read.
+ *
+ * @param buf Source buffer containing serialized full-encoding histogram.
+ * @return Deserialized object
+ */
+ protected static FixedBucketsHistogram fromByteBufferFullNoSerdeHeader(ByteBuffer buf)
+ {
+ double lowerLimit = buf.getDouble();
+ double upperLimit = buf.getDouble();
+ int numBuckets = buf.getInt();
+ OutlierHandlingMode outlierHandlingMode = OutlierHandlingMode.values()[buf.get()];
+
+ long count = buf.getLong();
+ long lowerOutlierCount = buf.getLong();
+ long upperOutlierCount = buf.getLong();
+ long missingValueCount = buf.getLong();
+
+ double max = buf.getDouble();
+ double min = buf.getDouble();
+
+ long histogram[] = new long[numBuckets];
+ buf.asLongBuffer().get(histogram);
+ buf.position(buf.position() + Long.BYTES * histogram.length);
+
+ return new FixedBucketsHistogram(
+ lowerLimit,
+ upperLimit,
+ numBuckets,
+ outlierHandlingMode,
+ histogram,
+ count,
+ max,
+ min,
+ lowerOutlierCount,
+ upperOutlierCount,
+ missingValueCount
+ );
+ }
+
+ /**
+ * Helper method for deserializing histograms with sparse encoding mode. Assumes that the serialization header is not
+ * present or has already been read.
+ *
+ * @param buf Source buffer containing serialized sparse-encoding histogram.
+ * @return Deserialized object
+ */
+ private static FixedBucketsHistogram fromBytesSparse(ByteBuffer buf)
+ {
+ double lowerLimit = buf.getDouble();
+ double upperLimit = buf.getDouble();
+ int numBuckets = buf.getInt();
+ OutlierHandlingMode outlierHandlingMode = OutlierHandlingMode.values()[buf.get()];
+
+ long count = buf.getLong();
+ long lowerOutlierCount = buf.getLong();
+ long upperOutlierCount = buf.getLong();
+ long missingValueCount = buf.getLong();
+
+ double max = buf.getDouble();
+ double min = buf.getDouble();
+
+ int nonEmptyBuckets = buf.getInt();
+ long histogram[] = new long[numBuckets];
+ for (int i = 0; i < nonEmptyBuckets; i++) {
+ int bucket = buf.getInt();
+ long bucketCount = buf.getLong();
+ histogram[bucket] = bucketCount;
+ }
+
+ return new FixedBucketsHistogram(
+ lowerLimit,
+ upperLimit,
+ numBuckets,
+ outlierHandlingMode,
+ histogram,
+ count,
+ max,
+ min,
+ lowerOutlierCount,
+ upperOutlierCount,
+ missingValueCount
+ );
+ }
+
+ /**
+ * Compute the size in bytes of a full-encoding serialized histogram, without the serialization header
+ *
+ * @param numBuckets number of buckets
+ * @return full serialized size in bytes
+ */
+ public static int getFullStorageSize(int numBuckets)
+ {
+ return COMMON_FIELDS_SIZE +
+ Long.BYTES * numBuckets;
+ }
+
+ /**
+ * Compute the size in bytes of a sparse-encoding serialized histogram, without the serialization header
+ *
+ * @param nonEmptyBuckets number of non-empty buckets
+ * @return sparse serialized size in bytes
+ */
+ public static int getSparseStorageSize(int nonEmptyBuckets)
+ {
+ return COMMON_FIELDS_SIZE +
+ Integer.BYTES +
+ (Integer.BYTES + Long.BYTES) * nonEmptyBuckets;
+ }
+
+ @VisibleForTesting
+ public int getNonEmptyBucketCount()
+ {
+ int count = 0;
+ for (int i = 0; i < numBuckets; i++) {
+ if (histogram[i] != 0) {
+ count++;
+ }
+ }
+ return count;
+ }
+
+ /**
+ * Updates the histogram state when an outlier value is encountered. An implementation is provided for each
+ * supported outlierHandlingMode.
+ */
+ private interface OutlierHandler
+ {
+ /**
+ * Handle an outlier when a single numeric value is added to the histogram.
+ *
+ * @param exceededMax If true, the value was an upper outlier. If false, the value was a lower outlier.
+ */
+ void handleOutlierAdd(boolean exceededMax);
+
+ /**
+ * Merge outlier information from another histogram.
+ *
+ * Called when merging two histograms that have overlapping ranges and potentially different bucket sizes.
+ *
+ * @param otherHistogram
+ * @param rangeStart
+ * @param rangeEnd
+ */
+ void simpleInterpolateMergeHandleOutliers(
+ FixedBucketsHistogram otherHistogram,
+ double rangeStart,
+ double rangeEnd
+ );
+
+ /**
+ * Merge outlier information from another histogram.
+ *
+ * Called when merging two histograms with identical buckets (same range and number of buckets)
+ *
+ * @param otherHistogram other histogram being merged
+ */
+ void handleOutliersForCombineSameBuckets(FixedBucketsHistogram otherHistogram);
+
+ /**
+ * Merge outlier information from another histogram.
+ *
+ * Called when merging two histograms, where the histogram to be merged has lowerLimit greater than my upperLimit
+ * @param otherHistogram other histogram being merged
+ */
+ void handleOutliersCombineDifferentBucketsAllUpper(FixedBucketsHistogram otherHistogram);
+
+ /**
+ * Merge outlier information from another histogram.
+ *
+ * Called when merging two histograms, where the histogram to be merged has upperLimit less than my lowerLimit
+ * @param otherHistogram other histogram being merged
+ */
+
+ void handleOutliersCombineDifferentBucketsAllLower(FixedBucketsHistogram otherHistogram);
+ }
+
+ private OutlierHandler makeOutlierHandler()
+ {
+ switch (outlierHandlingMode) {
+ case IGNORE:
+ return new IgnoreOutlierHandler();
+ case OVERFLOW:
+ return new OverflowOutlierHandler();
+ case CLIP:
+ return new ClipOutlierHandler();
+ default:
+ throw new ISE("Unknown outlier handling mode: %s", outlierHandlingMode);
+ }
+ }
+
+ private static class IgnoreOutlierHandler implements OutlierHandler
+ {
+ @Override
+ public void handleOutlierAdd(boolean exceededMax)
+ {
+ }
+
+ @Override
+ public void simpleInterpolateMergeHandleOutliers(
+ FixedBucketsHistogram otherHistogram,
+ double rangeStart,
+ double rangeEnd
+ )
+ {
+ }
+
+ @Override
+ public void handleOutliersForCombineSameBuckets(FixedBucketsHistogram otherHistogram)
+ {
+ }
+
+ @Override
+ public void handleOutliersCombineDifferentBucketsAllUpper(FixedBucketsHistogram otherHistogram)
+ {
+ }
+
+ @Override
+ public void handleOutliersCombineDifferentBucketsAllLower(FixedBucketsHistogram otherHistogram)
+ {
+ }
+ }
+
+ private class OverflowOutlierHandler implements OutlierHandler
+ {
+ @Override
+ public void handleOutlierAdd(boolean exceededMax)
+ {
+ if (exceededMax) {
+ upperOutlierCount += 1;
+ } else {
+ lowerOutlierCount += 1;
+ }
+ }
+
+ @Override
+ public void simpleInterpolateMergeHandleOutliers(
+ FixedBucketsHistogram otherHistogram,
+ double rangeStart,
+ double rangeEnd
+ )
+ {
+ // They contain me
+ if (lowerLimit == rangeStart && upperLimit == rangeEnd) {
+ long lowerCountFromOther = Math.round(otherHistogram.getCumulativeCount(rangeStart, true));
+ long upperCountFromOther = Math.round(otherHistogram.getCumulativeCount(rangeEnd, false));
+ upperOutlierCount += otherHistogram.getUpperOutlierCount() + upperCountFromOther;
+ lowerOutlierCount += otherHistogram.getLowerOutlierCount() + lowerCountFromOther;
+ } else if (rangeStart == lowerLimit) {
+ // assume that none of the other histogram's outliers fall within our range
+ // how many values were there in the portion of the other histogram that falls under my lower limit?
+ long lowerCountFromOther = Math.round(otherHistogram.getCumulativeCount(rangeStart, true));
+ lowerOutlierCount += lowerCountFromOther;
+ lowerOutlierCount += otherHistogram.getLowerOutlierCount();
+ upperOutlierCount += otherHistogram.getUpperOutlierCount();
+ } else if (rangeEnd == upperLimit) {
+ // how many values were there in the portion of the other histogram that is greater than my upper limit?
+ long upperCountFromOther = Math.round(otherHistogram.getCumulativeCount(rangeEnd, false));
+ upperOutlierCount += upperCountFromOther;
+ upperOutlierCount += otherHistogram.getUpperOutlierCount();
+ lowerOutlierCount += otherHistogram.getLowerOutlierCount();
+ } else if (rangeStart > lowerLimit && rangeEnd < upperLimit) {
+ upperOutlierCount += otherHistogram.getUpperOutlierCount();
+ lowerOutlierCount += otherHistogram.getLowerOutlierCount();
+ }
+ }
+
+ @Override
+ public void handleOutliersForCombineSameBuckets(FixedBucketsHistogram otherHistogram)
+ {
+ lowerOutlierCount += otherHistogram.getLowerOutlierCount();
+ upperOutlierCount += otherHistogram.getUpperOutlierCount();
+ }
+
+ @Override
+ public void handleOutliersCombineDifferentBucketsAllUpper(FixedBucketsHistogram otherHistogram)
+ {
+ // ignore any lower outliers in the other histogram, we're not sure where those outliers would fall
+ // within our range.
+ upperOutlierCount += otherHistogram.getCount() + otherHistogram.getUpperOutlierCount();
+ }
+
+ @Override
+ public void handleOutliersCombineDifferentBucketsAllLower(FixedBucketsHistogram otherHistogram)
+ {
+ // ignore any upper outliers in the other histogram, we're not sure where those outliers would fall
+ // within our range.
+ lowerOutlierCount += otherHistogram.getCount() + otherHistogram.getLowerOutlierCount();
+ }
+ }
+
+ private class ClipOutlierHandler implements OutlierHandler
+ {
+ @Override
+ public void handleOutlierAdd(boolean exceededMax)
+ {
+ double clippedValue;
+ count += 1;
+ if (exceededMax) {
+ clippedValue = upperLimit;
+ histogram[histogram.length - 1] += 1;
+ } else {
+ clippedValue = lowerLimit;
+ histogram[0] += 1;
+ }
+ if (clippedValue > max) {
+ max = clippedValue;
+ }
+ if (clippedValue < min) {
+ min = clippedValue;
+ }
+ }
+
+ @Override
+ public void simpleInterpolateMergeHandleOutliers(
+ FixedBucketsHistogram otherHistogram,
+ double rangeStart,
+ double rangeEnd
+ )
+ {
+ // They contain me
+ if (lowerLimit == rangeStart && upperLimit == rangeEnd) {
+ long lowerCountFromOther = Math.round(otherHistogram.getCumulativeCount(rangeStart, true));
+ long upperCountFromOther = Math.round(otherHistogram.getCumulativeCount(rangeEnd, false));
+ histogram[histogram.length - 1] += otherHistogram.getUpperOutlierCount() + upperCountFromOther;
+ histogram[0] += otherHistogram.getLowerOutlierCount() + lowerCountFromOther;
+ count += otherHistogram.getUpperOutlierCount() + otherHistogram.getLowerOutlierCount();
+ count += upperCountFromOther + lowerCountFromOther;
+ if (otherHistogram.getUpperOutlierCount() + upperCountFromOther > 0) {
+ max = Math.max(max, upperLimit);
+ }
+ if (otherHistogram.getLowerOutlierCount() + lowerCountFromOther > 0) {
+ min = Math.min(min, lowerLimit);
+ }
+ } else if (rangeStart == lowerLimit) {
+ // assume that none of the other histogram's outliers fall within our range
+ // how many values were there in the portion of the other histogram that falls under my lower limit?
+ long lowerCountFromOther = Math.round(otherHistogram.getCumulativeCount(rangeStart, true));
+
+ histogram[0] += lowerCountFromOther;
+ histogram[0] += otherHistogram.getLowerOutlierCount();
+ histogram[histogram.length - 1] += otherHistogram.getUpperOutlierCount();
+ count += lowerCountFromOther + otherHistogram.getLowerOutlierCount() + otherHistogram.getUpperOutlierCount();
+ if (lowerCountFromOther + otherHistogram.getLowerOutlierCount() > 0) {
+ min = lowerLimit;
+ }
+ if (otherHistogram.getUpperOutlierCount() > 0) {
+ max = upperLimit;
+ }
+ } else if (rangeEnd == upperLimit) {
+ // how many values were there in the portion of the other histogram that is greater than my upper limit?
+ long upperCountFromOther = Math.round(otherHistogram.getCumulativeCount(rangeEnd, false));
+
+ histogram[histogram.length - 1] += upperCountFromOther;
+ histogram[histogram.length - 1] += otherHistogram.getUpperOutlierCount();
+ histogram[0] += otherHistogram.getLowerOutlierCount();
+ count += upperCountFromOther + otherHistogram.getLowerOutlierCount() + otherHistogram.getUpperOutlierCount();
+ if (upperCountFromOther + otherHistogram.getUpperOutlierCount() > 0) {
+ max = upperLimit;
+ }
+ if (otherHistogram.getLowerOutlierCount() > 0) {
+ min = lowerLimit;
+ }
+ } else if (rangeStart > lowerLimit && rangeEnd < upperLimit) {
+ // I contain them
+
+ histogram[histogram.length - 1] += otherHistogram.getUpperOutlierCount();
+ histogram[0] += otherHistogram.getLowerOutlierCount();
+ count += otherHistogram.getUpperOutlierCount() + otherHistogram.getLowerOutlierCount();
+ if (otherHistogram.getUpperOutlierCount() > 0) {
+ max = Math.max(max, upperLimit);
+ } else if (otherHistogram.getCount() > 0) {
+ max = Math.max(max, otherHistogram.getMax());
+ }
+ if (otherHistogram.getLowerOutlierCount() > 0) {
+ min = Math.min(min, lowerLimit);
+ } else if (otherHistogram.getCount() > 0) {
+ min = Math.min(min, otherHistogram.getMin());
+ }
+ }
+ }
+
+ @Override
+ public void handleOutliersForCombineSameBuckets(FixedBucketsHistogram otherHistogram)
+ {
+ if (otherHistogram.getLowerOutlierCount() > 0) {
+ histogram[0] += otherHistogram.getLowerOutlierCount();
+ count += otherHistogram.getLowerOutlierCount();
+ min = Math.min(lowerLimit, min);
+ }
+ if (otherHistogram.getUpperOutlierCount() > 0) {
+ histogram[histogram.length - 1] += otherHistogram.getUpperOutlierCount();
+ count += otherHistogram.getUpperOutlierCount();
+ max = Math.max(upperLimit, max);
+ }
+ }
+
+ @Override
+ public void handleOutliersCombineDifferentBucketsAllUpper(FixedBucketsHistogram otherHistogram)
+ {
+ long otherCount = otherHistogram.getCount() + otherHistogram.getUpperOutlierCount();
+ histogram[histogram.length - 1] += otherCount;
+ count += otherCount;
+ if (otherCount > 0) {
+ max = Math.max(max, upperLimit);
+ }
+ }
+
+ @Override
+ public void handleOutliersCombineDifferentBucketsAllLower(FixedBucketsHistogram otherHistogram)
+ {
+ long otherCount = otherHistogram.getCount() + otherHistogram.getLowerOutlierCount();
+ histogram[0] += otherCount;
+ count += otherCount;
+ if (otherCount > 0) {
+ min = Math.min(min, lowerLimit);
+ }
+ }
+ }
+}
diff --git a/extensions-core/histogram/src/main/java/org/apache/druid/query/aggregation/histogram/FixedBucketsHistogramAggregator.java b/extensions-core/histogram/src/main/java/org/apache/druid/query/aggregation/histogram/FixedBucketsHistogramAggregator.java
new file mode 100644
index 0000000..3611b58
--- /dev/null
+++ b/extensions-core/histogram/src/main/java/org/apache/druid/query/aggregation/histogram/FixedBucketsHistogramAggregator.java
@@ -0,0 +1,127 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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.apache.druid.query.aggregation.histogram;
+
+import com.google.common.primitives.Longs;
+import org.apache.druid.common.config.NullHandling;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.logger.Logger;
+import org.apache.druid.query.aggregation.Aggregator;
+import org.apache.druid.segment.BaseObjectColumnValueSelector;
+
+import javax.annotation.Nullable;
+import java.util.Comparator;
+
+public class FixedBucketsHistogramAggregator implements Aggregator
+{
+ private static final Logger LOG = new Logger(FixedBucketsHistogramAggregator.class);
+
+ public static final String TYPE_NAME = "fixedBucketsHistogram";
+
+ public static final Comparator COMPARATOR = new Comparator()
+ {
+ @Override
+ public int compare(Object o, Object o1)
+ {
+ return Longs.compare(((FixedBucketsHistogram) o).getCount(), ((FixedBucketsHistogram) o1).getCount());
+ }
+ };
+
+ private final BaseObjectColumnValueSelector selector;
+
+ private FixedBucketsHistogram histogram;
+
+ public FixedBucketsHistogramAggregator(
+ BaseObjectColumnValueSelector selector,
+ double lowerLimit,
+ double upperLimit,
+ int numBuckets,
+ FixedBucketsHistogram.OutlierHandlingMode outlierHandlingMode
+ )
+ {
+ this.selector = selector;
+ this.histogram = new FixedBucketsHistogram(
+ lowerLimit,
+ upperLimit,
+ numBuckets,
+ outlierHandlingMode
+ );
+ }
+
+ @Override
+ public void aggregate()
+ {
+ Object val = selector.getObject();
+
+ if (val == null) {
+ if (NullHandling.replaceWithDefault()) {
+ histogram.add(NullHandling.defaultDoubleValue());
+ } else {
+ histogram.incrementMissing();
+ }
+ } else if (val instanceof String) {
+ LOG.info((String) val);
+ histogram.combineHistogram(FixedBucketsHistogram.fromBase64((String) val));
+ } else if (val instanceof FixedBucketsHistogram) {
+ histogram.combineHistogram((FixedBucketsHistogram) val);
+ } else if (val instanceof Number) {
+ histogram.add(((Number) val).doubleValue());
+ } else {
+ throw new ISE("Unknown class for object: " + val.getClass());
+ }
+ }
+
+ @Nullable
+ @Override
+ public Object get()
+ {
+ return histogram;
+ }
+
+ @Override
+ public float getFloat()
+ {
+ throw new UnsupportedOperationException("FixedBucketsHistogramAggregator does not support getFloat()");
+ }
+
+ @Override
+ public long getLong()
+ {
+ throw new UnsupportedOperationException("FixedBucketsHistogramAggregator does not support getLong()");
+ }
+
+ @Override
+ public double getDouble()
+ {
+ throw new UnsupportedOperationException("FixedBucketsHistogramAggregator does not support getDouble()");
+ }
+
+ @Override
+ public boolean isNull()
+ {
+ return false;
+ }
+
+ @Override
+ public void close()
+ {
+
+ }
+}
diff --git a/extensions-core/histogram/src/main/java/org/apache/druid/query/aggregation/histogram/FixedBucketsHistogramAggregatorFactory.java b/extensions-core/histogram/src/main/java/org/apache/druid/query/aggregation/histogram/FixedBucketsHistogramAggregatorFactory.java
new file mode 100644
index 0000000..0dca0c6
--- /dev/null
+++ b/extensions-core/histogram/src/main/java/org/apache/druid/query/aggregation/histogram/FixedBucketsHistogramAggregatorFactory.java
@@ -0,0 +1,334 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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.apache.druid.query.aggregation.histogram;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonTypeName;
+import org.apache.commons.codec.binary.Base64;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.query.aggregation.AggregateCombiner;
+import org.apache.druid.query.aggregation.Aggregator;
+import org.apache.druid.query.aggregation.AggregatorFactory;
+import org.apache.druid.query.aggregation.AggregatorFactoryNotMergeableException;
+import org.apache.druid.query.aggregation.AggregatorUtil;
+import org.apache.druid.query.aggregation.BufferAggregator;
+import org.apache.druid.query.aggregation.ObjectAggregateCombiner;
+import org.apache.druid.segment.ColumnSelectorFactory;
+import org.apache.druid.segment.ColumnValueSelector;
+
+import javax.annotation.Nullable;
+import java.nio.ByteBuffer;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Objects;
+
+@JsonTypeName(FixedBucketsHistogramAggregator.TYPE_NAME)
+public class FixedBucketsHistogramAggregatorFactory extends AggregatorFactory
+{
+ private static int DEFAULT_NUM_BUCKETS = 10;
+
+ private final String name;
+ private final String fieldName;
+
+ private double lowerLimit;
+ private double upperLimit;
+ private int numBuckets;
+
+ private FixedBucketsHistogram.OutlierHandlingMode outlierHandlingMode;
+
+ @JsonCreator
+ public FixedBucketsHistogramAggregatorFactory(
+ @JsonProperty("name") String name,
+ @JsonProperty("fieldName") String fieldName,
+ @Nullable @JsonProperty("numBuckets") Integer numBuckets,
+ @JsonProperty("lowerLimit") double lowerLimit,
+ @JsonProperty("upperLimit") double upperLimit,
+ @JsonProperty("outlierHandlingMode") FixedBucketsHistogram.OutlierHandlingMode outlierHandlingMode
+ )
+ {
+ this.name = name;
+ this.fieldName = fieldName;
+ this.numBuckets = numBuckets == null ? DEFAULT_NUM_BUCKETS : numBuckets;
+ this.lowerLimit = lowerLimit;
+ this.upperLimit = upperLimit;
+ this.outlierHandlingMode = outlierHandlingMode;
+ }
+
+ @Override
+ public Aggregator factorize(ColumnSelectorFactory metricFactory)
+ {
+ return new FixedBucketsHistogramAggregator(
+ metricFactory.makeColumnValueSelector(fieldName),
+ lowerLimit,
+ upperLimit,
+ numBuckets,
+ outlierHandlingMode
+ );
+ }
+
+ @Override
+ public BufferAggregator factorizeBuffered(ColumnSelectorFactory metricFactory)
+ {
+ return new FixedBucketsHistogramBufferAggregator(
+ metricFactory.makeColumnValueSelector(fieldName),
+ lowerLimit,
+ upperLimit,
+ numBuckets,
+ outlierHandlingMode
+ );
+ }
+
+ @Override
+ public Comparator getComparator()
+ {
+ return FixedBucketsHistogramAggregator.COMPARATOR;
+ }
+
+ @Nullable
+ @Override
+ public Object combine(@Nullable Object lhs, @Nullable Object rhs)
+ {
+ if (lhs == null) {
+ if (rhs == null) {
+ return null;
+ } else {
+ return rhs;
+ }
+ } else {
+ ((FixedBucketsHistogram) lhs).combineHistogram((FixedBucketsHistogram) rhs);
+ return lhs;
+ }
+ }
+
+ @Override
+ public AggregateCombiner makeAggregateCombiner()
+ {
+ return new ObjectAggregateCombiner()
+ {
+ private final FixedBucketsHistogram combined = new FixedBucketsHistogram(
+ lowerLimit,
+ upperLimit,
+ numBuckets,
+ outlierHandlingMode
+ );
+
+ @Override
+ public void reset(ColumnValueSelector selector)
+ {
+ FixedBucketsHistogram first = (FixedBucketsHistogram) selector.getObject();
+ combined.combineHistogram(first);
+ }
+
+ @Override
+ public void fold(ColumnValueSelector selector)
+ {
+ FixedBucketsHistogram other = (FixedBucketsHistogram) selector.getObject();
+ combined.combineHistogram(other);
+ }
+
+ @Override
+ public Class<FixedBucketsHistogram> classOfObject()
+ {
+ return FixedBucketsHistogram.class;
+ }
+
+ @Nullable
+ @Override
+ public FixedBucketsHistogram getObject()
+ {
+ return combined;
+ }
+ };
+ }
+
+ @Override
+ public AggregatorFactory getCombiningFactory()
+ {
+ return new FixedBucketsHistogramAggregatorFactory(
+ name,
+ name,
+ numBuckets,
+ lowerLimit,
+ upperLimit,
+ outlierHandlingMode
+ );
+ }
+
+ @Override
+ public AggregatorFactory getMergingFactory(AggregatorFactory other) throws AggregatorFactoryNotMergeableException
+ {
+ return new FixedBucketsHistogramAggregatorFactory(
+ name,
+ name,
+ numBuckets,
+ lowerLimit,
+ upperLimit,
+ outlierHandlingMode
+ );
+ }
+
+ @Override
+ public List<AggregatorFactory> getRequiredColumns()
+ {
+ return Collections.singletonList(
+ new FixedBucketsHistogramAggregatorFactory(
+ fieldName,
+ fieldName,
+ numBuckets,
+ lowerLimit,
+ upperLimit,
+ outlierHandlingMode
+ )
+ );
+ }
+
+ @Override
+ public Object deserialize(Object object)
+ {
+ if (object instanceof String) {
+ byte[] bytes = Base64.decodeBase64(StringUtils.toUtf8((String) object));
+ final FixedBucketsHistogram fbh = FixedBucketsHistogram.fromBytes(bytes);
+ return fbh;
+ } else {
+ return object;
+ }
+ }
+
+ @Nullable
+ @Override
+ public Object finalizeComputation(@Nullable Object object)
+ {
+ return object;
+ }
+
+ @JsonProperty
+ @Override
+ public String getName()
+ {
+ return name;
+ }
+
+ @Override
+ public List<String> requiredFields()
+ {
+ return Collections.singletonList(fieldName);
+ }
+
+ @Override
+ public String getTypeName()
+ {
+ return FixedBucketsHistogramAggregator.TYPE_NAME;
+ }
+
+ @Override
+ public int getMaxIntermediateSize()
+ {
+ return FixedBucketsHistogram.SERDE_HEADER_SIZE + FixedBucketsHistogram.getFullStorageSize(numBuckets);
+ }
+
+ @Override
+ public byte[] getCacheKey()
+ {
+ byte[] fieldNameBytes = StringUtils.toUtf8(fieldName);
+ return ByteBuffer.allocate(1 + fieldNameBytes.length + Integer.BYTES * 2 + Double.BYTES * 2)
+ .put(AggregatorUtil.FIXED_BUCKET_HIST_CACHE_TYPE_ID)
+ .put(fieldNameBytes)
+ .putInt(outlierHandlingMode.ordinal())
+ .putInt(numBuckets)
+ .putDouble(lowerLimit)
+ .putDouble(upperLimit).array();
+ }
+
+ @JsonProperty
+ public String getFieldName()
+ {
+ return fieldName;
+ }
+
+ @JsonProperty
+ public double getLowerLimit()
+ {
+ return lowerLimit;
+ }
+
+ @JsonProperty
+ public double getUpperLimit()
+ {
+ return upperLimit;
+ }
+
+ @JsonProperty
+ public int getNumBuckets()
+ {
+ return numBuckets;
+ }
+
+ @JsonProperty
+ public FixedBucketsHistogram.OutlierHandlingMode getOutlierHandlingMode()
+ {
+ return outlierHandlingMode;
+ }
+
+ @Override
+ public boolean equals(Object o)
+ {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ FixedBucketsHistogramAggregatorFactory that = (FixedBucketsHistogramAggregatorFactory) o;
+ return Double.compare(that.getLowerLimit(), getLowerLimit()) == 0 &&
+ Double.compare(that.getUpperLimit(), getUpperLimit()) == 0 &&
+ getNumBuckets() == that.getNumBuckets() &&
+ Objects.equals(getName(), that.getName()) &&
+ Objects.equals(getFieldName(), that.getFieldName()) &&
+ getOutlierHandlingMode() == that.getOutlierHandlingMode();
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return Objects.hash(
+ getName(),
+ getFieldName(),
+ getLowerLimit(),
+ getUpperLimit(),
+ getNumBuckets(),
+ getOutlierHandlingMode()
+ );
+ }
+
+ @Override
+ public String toString()
+ {
+ return "FixedBucketsHistogramAggregatorFactory{" +
+ "name='" + name + '\'' +
+ ", fieldName='" + fieldName + '\'' +
+ ", lowerLimit=" + lowerLimit +
+ ", upperLimit=" + upperLimit +
+ ", numBuckets=" + numBuckets +
+ ", outlierHandlingMode=" + outlierHandlingMode +
+ '}';
+ }
+}
diff --git a/extensions-core/histogram/src/main/java/org/apache/druid/query/aggregation/histogram/FixedBucketsHistogramBufferAggregator.java b/extensions-core/histogram/src/main/java/org/apache/druid/query/aggregation/histogram/FixedBucketsHistogramBufferAggregator.java
new file mode 100644
index 0000000..0e894d7
--- /dev/null
+++ b/extensions-core/histogram/src/main/java/org/apache/druid/query/aggregation/histogram/FixedBucketsHistogramBufferAggregator.java
@@ -0,0 +1,125 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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.apache.druid.query.aggregation.histogram;
+
+import org.apache.druid.common.config.NullHandling;
+import org.apache.druid.query.aggregation.BufferAggregator;
+import org.apache.druid.query.monomorphicprocessing.RuntimeShapeInspector;
+import org.apache.druid.segment.BaseObjectColumnValueSelector;
+
+import java.nio.ByteBuffer;
+
+public class FixedBucketsHistogramBufferAggregator implements BufferAggregator
+{
+ private final BaseObjectColumnValueSelector selector;
+
+ private FixedBucketsHistogram histogram;
+
+ public FixedBucketsHistogramBufferAggregator(
+ BaseObjectColumnValueSelector selector,
+ double lowerLimit,
+ double upperLimit,
+ int numBuckets,
+ FixedBucketsHistogram.OutlierHandlingMode outlierHandlingMode
+ )
+ {
+ this.selector = selector;
+ this.histogram = new FixedBucketsHistogram(
+ lowerLimit,
+ upperLimit,
+ numBuckets,
+ outlierHandlingMode
+ );
+ }
+
+ @Override
+ public void init(ByteBuffer buf, int position)
+ {
+ ByteBuffer mutationBuffer = buf.duplicate();
+ mutationBuffer.position(position);
+ mutationBuffer.put(histogram.toBytesFull(false));
+ }
+
+ @Override
+ public void aggregate(ByteBuffer buf, int position)
+ {
+ ByteBuffer mutationBuffer = buf.duplicate();
+ mutationBuffer.position(position);
+
+ FixedBucketsHistogram h0 = FixedBucketsHistogram.fromByteBufferFullNoSerdeHeader(mutationBuffer);
+
+ Object val = selector.getObject();
+ if (val == null) {
+ if (NullHandling.replaceWithDefault()) {
+ h0.incrementMissing();
+ } else {
+ h0.add(NullHandling.defaultDoubleValue());
+ }
+ } else if (val instanceof String) {
+ h0.combineHistogram(FixedBucketsHistogram.fromBase64((String) val));
+ } else if (val instanceof FixedBucketsHistogram) {
+ h0.combineHistogram((FixedBucketsHistogram) val);
+ } else {
+ Double x = ((Number) val).doubleValue();
+ h0.add(x);
+ }
+
+ mutationBuffer.position(position);
+ mutationBuffer.put(h0.toBytesFull(false));
+ }
+
+ @Override
+ public Object get(ByteBuffer buf, int position)
+ {
+ ByteBuffer mutationBuffer = buf.duplicate();
+ mutationBuffer.position(position);
+ return FixedBucketsHistogram.fromByteBufferFullNoSerdeHeader(mutationBuffer);
+ }
+
+ @Override
+ public float getFloat(ByteBuffer buf, int position)
+ {
+ throw new UnsupportedOperationException("FixedBucketsHistogramBufferAggregator does not support getFloat()");
+ }
+
+ @Override
+ public long getLong(ByteBuffer buf, int position)
+ {
+ throw new UnsupportedOperationException("FixedBucketsHistogramBufferAggregator does not support getLong()");
+ }
+
+ @Override
+ public double getDouble(ByteBuffer buf, int position)
+ {
+ throw new UnsupportedOperationException("FixedBucketsHistogramBufferAggregator does not support getDouble()");
+ }
+
+ @Override
+ public void close()
+ {
+ // no resources to cleanup
+ }
+
+ @Override
+ public void inspectRuntimeShape(RuntimeShapeInspector inspector)
+ {
+ inspector.visit("selector", selector);
+ }
+}
diff --git a/extensions-core/histogram/src/main/java/org/apache/druid/query/aggregation/histogram/FixedBucketsHistogramSerde.java b/extensions-core/histogram/src/main/java/org/apache/druid/query/aggregation/histogram/FixedBucketsHistogramSerde.java
new file mode 100644
index 0000000..0f7f743
--- /dev/null
+++ b/extensions-core/histogram/src/main/java/org/apache/druid/query/aggregation/histogram/FixedBucketsHistogramSerde.java
@@ -0,0 +1,189 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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.apache.druid.query.aggregation.histogram;
+
+import com.google.common.collect.Ordering;
+import org.apache.druid.common.config.NullHandling;
+import org.apache.druid.data.input.InputRow;
+import org.apache.druid.data.input.Rows;
+import org.apache.druid.java.util.common.logger.Logger;
+import org.apache.druid.java.util.common.parsers.ParseException;
+import org.apache.druid.query.aggregation.AggregatorFactory;
+import org.apache.druid.segment.GenericColumnSerializer;
+import org.apache.druid.segment.column.ColumnBuilder;
+import org.apache.druid.segment.data.GenericIndexed;
+import org.apache.druid.segment.data.ObjectStrategy;
+import org.apache.druid.segment.serde.ComplexColumnPartSupplier;
+import org.apache.druid.segment.serde.ComplexMetricExtractor;
+import org.apache.druid.segment.serde.ComplexMetricSerde;
+import org.apache.druid.segment.serde.LargeColumnSupportedComplexColumnSerializer;
+import org.apache.druid.segment.writeout.SegmentWriteOutMedium;
+
+import javax.annotation.Nullable;
+import java.nio.ByteBuffer;
+
+public class FixedBucketsHistogramSerde extends ComplexMetricSerde
+{
+ private static final Logger LOG = new Logger(FixedBucketsHistogramSerde.class);
+
+ private static Ordering<FixedBucketsHistogram> comparator = new Ordering<FixedBucketsHistogram>()
+ {
+ @Override
+ public int compare(
+ FixedBucketsHistogram arg1,
+ FixedBucketsHistogram arg2
+ )
+ {
+ return FixedBucketsHistogramAggregator.COMPARATOR.compare(arg1, arg2);
+ }
+ }.nullsFirst();
+
+ @Override
+ public String getTypeName()
+ {
+ return FixedBucketsHistogramAggregator.TYPE_NAME;
+ }
+
+ @Override
+ public ComplexMetricExtractor getExtractor()
+ {
+ return new ComplexMetricExtractor()
+ {
+ @Override
+ public Class<FixedBucketsHistogram> extractedClass()
+ {
+ return FixedBucketsHistogram.class;
+ }
+
+ @Nullable
+ @Override
+ public Object extractValue(InputRow inputRow, String metricName)
+ {
+ throw new UnsupportedOperationException("extractValue without an aggregator factory is not supported.");
+ }
+
+ @Override
+ public FixedBucketsHistogram extractValue(InputRow inputRow, String metricName, AggregatorFactory agg)
+ {
+ Object rawValue = inputRow.getRaw(metricName);
+
+ FixedBucketsHistogramAggregatorFactory aggregatorFactory = (FixedBucketsHistogramAggregatorFactory) agg;
+
+ if (rawValue == null) {
+ FixedBucketsHistogram fbh = new FixedBucketsHistogram(
+ aggregatorFactory.getLowerLimit(),
+ aggregatorFactory.getUpperLimit(),
+ aggregatorFactory.getNumBuckets(),
+ aggregatorFactory.getOutlierHandlingMode()
+ );
+ if (NullHandling.replaceWithDefault()) {
+ fbh.add(NullHandling.defaultDoubleValue());
+ } else {
+ fbh.incrementMissing();
+ }
+ return fbh;
+ } else if (rawValue instanceof Number) {
+ FixedBucketsHistogram fbh = new FixedBucketsHistogram(
+ aggregatorFactory.getLowerLimit(),
+ aggregatorFactory.getUpperLimit(),
+ aggregatorFactory.getNumBuckets(),
+ aggregatorFactory.getOutlierHandlingMode()
+ );
+ fbh.add(((Number) rawValue).doubleValue());
+ return fbh;
+ } else if (rawValue instanceof FixedBucketsHistogram) {
+ return (FixedBucketsHistogram) rawValue;
+ } else if (rawValue instanceof String) {
+ Number numberAttempt;
+ try {
+ numberAttempt = Rows.objectToNumber(metricName, rawValue);
+ FixedBucketsHistogram fbh = new FixedBucketsHistogram(
+ aggregatorFactory.getLowerLimit(),
+ aggregatorFactory.getUpperLimit(),
+ aggregatorFactory.getNumBuckets(),
+ aggregatorFactory.getOutlierHandlingMode()
+ );
+ fbh.add(numberAttempt.doubleValue());
+ return fbh;
+ }
+ catch (ParseException pe) {
+ FixedBucketsHistogram fbh = FixedBucketsHistogram.fromBase64((String) rawValue);
+ return fbh;
+ }
+ } else {
+ throw new UnsupportedOperationException("Unknown type: " + rawValue.getClass());
+ }
+ }
+ };
+ }
+
+ @Override
+ public void deserializeColumn(ByteBuffer buffer, ColumnBuilder builder)
+ {
+ final GenericIndexed column = GenericIndexed.read(buffer, getObjectStrategy(), builder.getFileMapper());
+ builder.setComplexColumnSupplier(new ComplexColumnPartSupplier(getTypeName(), column));
+ }
+
+ @Override
+ public ObjectStrategy getObjectStrategy()
+ {
+ return new ObjectStrategy<FixedBucketsHistogram>()
+ {
+ @Override
+ public Class<? extends FixedBucketsHistogram> getClazz()
+ {
+ return FixedBucketsHistogram.class;
+ }
+
+ @Override
+ public FixedBucketsHistogram fromByteBuffer(ByteBuffer buffer, int numBytes)
+ {
+ buffer.limit(buffer.position() + numBytes);
+ FixedBucketsHistogram fbh = FixedBucketsHistogram.fromByteBuffer(buffer);
+ return fbh;
+ }
+
+ @Override
+ public byte[] toBytes(FixedBucketsHistogram h)
+ {
+ if (h == null) {
+ return new byte[]{};
+ }
+
+ return h.toBytes();
+ }
+
+ @Override
+ public int compare(FixedBucketsHistogram o1, FixedBucketsHistogram o2)
+ {
+ return comparator.compare(o1, o2);
+ }
+ };
+ }
+
+ @Override
+ public GenericColumnSerializer getSerializer(
+ SegmentWriteOutMedium segmentWriteOutMedium,
+ String column
+ )
+ {
+ return LargeColumnSupportedComplexColumnSerializer.create(segmentWriteOutMedium, column, this.getObjectStrategy());
+ }
+}
diff --git a/extensions-core/histogram/src/main/java/org/apache/druid/query/aggregation/histogram/MaxPostAggregator.java b/extensions-core/histogram/src/main/java/org/apache/druid/query/aggregation/histogram/MaxPostAggregator.java
index 47ff828..23d59c9 100644
--- a/extensions-core/histogram/src/main/java/org/apache/druid/query/aggregation/histogram/MaxPostAggregator.java
+++ b/extensions-core/histogram/src/main/java/org/apache/druid/query/aggregation/histogram/MaxPostAggregator.java
@@ -68,8 +68,14 @@ public class MaxPostAggregator extends ApproximateHistogramPostAggregator
@Override
public Object compute(Map<String, Object> values)
{
- final ApproximateHistogram ah = (ApproximateHistogram) values.get(fieldName);
- return ah.getMax();
+ Object val = values.get(fieldName);
+ if (val instanceof ApproximateHistogram) {
+ final ApproximateHistogram ah = (ApproximateHistogram) val;
+ return ah.getMax();
+ } else {
+ final FixedBucketsHistogram fbh = (FixedBucketsHistogram) val;
+ return fbh.getMax();
+ }
}
@Override
diff --git a/extensions-core/histogram/src/main/java/org/apache/druid/query/aggregation/histogram/MinPostAggregator.java b/extensions-core/histogram/src/main/java/org/apache/druid/query/aggregation/histogram/MinPostAggregator.java
index a209b8e..3aa443d 100644
--- a/extensions-core/histogram/src/main/java/org/apache/druid/query/aggregation/histogram/MinPostAggregator.java
+++ b/extensions-core/histogram/src/main/java/org/apache/druid/query/aggregation/histogram/MinPostAggregator.java
@@ -23,6 +23,7 @@ import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.annotation.JsonTypeName;
import com.google.common.collect.Sets;
+import org.apache.druid.java.util.common.ISE;
import org.apache.druid.query.aggregation.AggregatorFactory;
import org.apache.druid.query.aggregation.PostAggregator;
import org.apache.druid.query.aggregation.post.PostAggregatorIds;
@@ -68,8 +69,15 @@ public class MinPostAggregator extends ApproximateHistogramPostAggregator
@Override
public Object compute(Map<String, Object> values)
{
- final ApproximateHistogram ah = (ApproximateHistogram) values.get(fieldName);
- return ah.getMin();
+ Object val = values.get(fieldName);
+ if (val instanceof ApproximateHistogram) {
+ final ApproximateHistogram ah = (ApproximateHistogram) val;
+ return ah.getMin();
+ } else if (val instanceof FixedBucketsHistogram) {
+ final FixedBucketsHistogram fbh = (FixedBucketsHistogram) val;
+ return fbh.getMin();
+ }
+ throw new ISE("Unknown value type: " + val.getClass());
}
@Override
diff --git a/extensions-core/histogram/src/main/java/org/apache/druid/query/aggregation/histogram/QuantilePostAggregator.java b/extensions-core/histogram/src/main/java/org/apache/druid/query/aggregation/histogram/QuantilePostAggregator.java
index 0cfb7b7..09fd1fa 100644
--- a/extensions-core/histogram/src/main/java/org/apache/druid/query/aggregation/histogram/QuantilePostAggregator.java
+++ b/extensions-core/histogram/src/main/java/org/apache/druid/query/aggregation/histogram/QuantilePostAggregator.java
@@ -79,8 +79,15 @@ public class QuantilePostAggregator extends ApproximateHistogramPostAggregator
@Override
public Object compute(Map<String, Object> values)
{
- final ApproximateHistogram ah = (ApproximateHistogram) values.get(fieldName);
- return ah.getQuantiles(new float[]{probability})[0];
+ Object value = values.get(fieldName);
+ if (value instanceof ApproximateHistogram) {
+ final ApproximateHistogram ah = (ApproximateHistogram) value;
+ return ah.getQuantiles(new float[]{probability})[0];
+ } else {
+ final FixedBucketsHistogram fbh = (FixedBucketsHistogram) value;
+ float x = fbh.percentilesFloat(new double[]{probability * 100.0})[0];
+ return x;
+ }
}
@Override
diff --git a/extensions-core/histogram/src/main/java/org/apache/druid/query/aggregation/histogram/QuantilesPostAggregator.java b/extensions-core/histogram/src/main/java/org/apache/druid/query/aggregation/histogram/QuantilesPostAggregator.java
index 434eb83..2ef4c50 100644
--- a/extensions-core/histogram/src/main/java/org/apache/druid/query/aggregation/histogram/QuantilesPostAggregator.java
+++ b/extensions-core/histogram/src/main/java/org/apache/druid/query/aggregation/histogram/QuantilesPostAggregator.java
@@ -24,6 +24,7 @@ import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.annotation.JsonTypeName;
import com.google.common.collect.Sets;
import org.apache.druid.java.util.common.IAE;
+import org.apache.druid.java.util.common.ISE;
import org.apache.druid.query.aggregation.AggregatorFactory;
import org.apache.druid.query.aggregation.PostAggregator;
import org.apache.druid.query.aggregation.post.PostAggregatorIds;
@@ -71,9 +72,24 @@ public class QuantilesPostAggregator extends ApproximateHistogramPostAggregator
@Override
public Object compute(Map<String, Object> values)
{
- final ApproximateHistogram ah = (ApproximateHistogram) values.get(fieldName);
-
- return new Quantiles(probabilities, ah.getQuantiles(probabilities), ah.getMin(), ah.getMax());
+ Object val = values.get(fieldName);
+ if (val instanceof ApproximateHistogram) {
+ final ApproximateHistogram ah = (ApproximateHistogram) val;
+ return new Quantiles(probabilities, ah.getQuantiles(probabilities), ah.getMin(), ah.getMax());
+ } else if (val instanceof FixedBucketsHistogram) {
+ final FixedBucketsHistogram fbh = (FixedBucketsHistogram) val;
+ double[] adjustedProbabilites = new double[probabilities.length];
+ for (int i = 0; i < probabilities.length; i++) {
+ adjustedProbabilites[i] = probabilities[i] * 100.0;
+ }
+ return new Quantiles(
+ probabilities,
+ fbh.percentilesFloat(adjustedProbabilites),
+ (float) fbh.getMin(),
+ (float) fbh.getMax()
+ );
+ }
+ throw new ISE("Unknown value type: " + val.getClass());
}
@Override
@@ -91,7 +107,7 @@ public class QuantilesPostAggregator extends ApproximateHistogramPostAggregator
@Override
public String toString()
{
- return "EqualBucketsPostAggregator{" +
+ return "QuantilesPostAggregator{" +
"name='" + name + '\'' +
", fieldName='" + fieldName + '\'' +
", probabilities=" + Arrays.toString(this.getProbabilities()) +
diff --git a/extensions-core/histogram/src/test/java/org/apache/druid/query/aggregation/histogram/FixedBucketsHistogramAggregationTest.java b/extensions-core/histogram/src/test/java/org/apache/druid/query/aggregation/histogram/FixedBucketsHistogramAggregationTest.java
new file mode 100644
index 0000000..838cae0
--- /dev/null
+++ b/extensions-core/histogram/src/test/java/org/apache/druid/query/aggregation/histogram/FixedBucketsHistogramAggregationTest.java
@@ -0,0 +1,162 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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.apache.druid.query.aggregation.histogram;
+
+import com.google.common.collect.Lists;
+import org.apache.druid.common.config.NullHandling;
+import org.apache.druid.data.input.MapBasedRow;
+import org.apache.druid.java.util.common.granularity.Granularities;
+import org.apache.druid.java.util.common.guava.Sequence;
+import org.apache.druid.query.aggregation.AggregationTestHelper;
+import org.apache.druid.query.groupby.GroupByQueryConfig;
+import org.apache.druid.query.groupby.GroupByQueryRunnerTest;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+
+/**
+ */
+@RunWith(Parameterized.class)
+public class FixedBucketsHistogramAggregationTest
+{
+ private AggregationTestHelper helper;
+
+ @Rule
+ public final TemporaryFolder tempFolder = new TemporaryFolder();
+
+ public FixedBucketsHistogramAggregationTest(final GroupByQueryConfig config)
+ {
+ ApproximateHistogramDruidModule module = new ApproximateHistogramDruidModule();
+ module.configure(null);
+ helper = AggregationTestHelper.createGroupByQueryAggregationTestHelper(
+ Lists.newArrayList(module.getJacksonModules()),
+ config,
+ tempFolder
+ );
+ }
+
+ @Parameterized.Parameters(name = "{0}")
+ public static Collection<?> constructorFeeder()
+ {
+ final List<Object[]> constructors = new ArrayList<>();
+ for (GroupByQueryConfig config : GroupByQueryRunnerTest.testConfigs()) {
+ constructors.add(new Object[]{config});
+ }
+ return constructors;
+ }
+
+ @After
+ public void teardown() throws IOException
+ {
+ helper.close();
+ }
+
+ @Test
+ public void testIngestWithNullsIgnoredAndQuery() throws Exception
+ {
+ MapBasedRow row = ingestAndQuery();
+ if (!NullHandling.replaceWithDefault()) {
+ Assert.assertEquals(92.782760, row.getMetric("index_min").floatValue(), 0.0001);
+ Assert.assertEquals(135.109191, row.getMetric("index_max").floatValue(), 0.0001);
+ Assert.assertEquals(135.9499969482422, row.getMetric("index_quantile").floatValue(), 0.0001);
+ } else {
+ Assert.assertEquals(0.0, row.getMetric("index_min"));
+ Assert.assertEquals(135.109191, row.getMetric("index_max").floatValue(), 0.0001);
+ Assert.assertEquals(135.8699951171875, row.getMetric("index_quantile").floatValue(), 0.0001);
+ }
+ }
+
+ private MapBasedRow ingestAndQuery() throws Exception
+ {
+ String ingestionAgg = FixedBucketsHistogramAggregator.TYPE_NAME;
+
+ String metricSpec = "[{"
+ + "\"type\": \"" + ingestionAgg + "\","
+ + "\"name\": \"index_fbh\","
+ + "\"numBuckets\": 200,"
+ + "\"lowerLimit\": 0,"
+ + "\"upperLimit\": 200,"
+ + "\"outlierHandlingMode\": \"overflow\","
+ + "\"fieldName\": \"index\""
+ + "}]";
+
+ String parseSpec = "{"
+ + "\"type\" : \"string\","
+ + "\"parseSpec\" : {"
+ + " \"format\" : \"tsv\","
+ + " \"timestampSpec\" : {"
+ + " \"column\" : \"timestamp\","
+ + " \"format\" : \"auto\""
+ + "},"
+ + " \"dimensionsSpec\" : {"
+ + " \"dimensions\": [],"
+ + " \"dimensionExclusions\" : [],"
+ + " \"spatialDimensions\" : []"
+ + " },"
+ + " \"columns\": [\"timestamp\", \"market\", \"quality\", \"placement\", \"placementish\", \"index\"]"
+ + " }"
+ + "}";
+
+ String query = "{"
+ + "\"queryType\": \"groupBy\","
+ + "\"dataSource\": \"test_datasource\","
+ + "\"granularity\": \"ALL\","
+ + "\"dimensions\": [],"
+ + "\"aggregations\": ["
+ + " {"
+ + " \"type\": \"fixedBucketsHistogram\","
+ + " \"name\": \"index_fbh\","
+ + " \"fieldName\": \"index_fbh\","
+ + " \"numBuckets\": 200,"
+ + " \"lowerLimit\": 0,"
+ + " \"upperLimit\": 200,"
+ + " \"outlierHandlingMode\": \"overflow\""
+ + " }"
+ + "],"
+ + "\"postAggregations\": ["
+ + " { \"type\": \"min\", \"name\": \"index_min\", \"fieldName\": \"index_fbh\"},"
+ + " { \"type\": \"max\", \"name\": \"index_max\", \"fieldName\": \"index_fbh\"},"
+ + " { \"type\": \"quantile\", \"name\": \"index_quantile\", \"fieldName\": \"index_fbh\", \"probability\" : 0.99 }"
+ + "],"
+ + "\"intervals\": [ \"1970/2050\" ]"
+ + "}";
+
+ Sequence seq = helper.createIndexAndRunQueryOnSegment(
+ this.getClass().getClassLoader().getResourceAsStream("sample.data.tsv"),
+ parseSpec,
+ metricSpec,
+ 0,
+ Granularities.NONE,
+ 50000,
+ query
+ );
+
+ return (MapBasedRow) seq.toList().get(0);
+ }
+}
diff --git a/extensions-core/histogram/src/test/java/org/apache/druid/query/aggregation/histogram/FixedBucketsHistogramBufferAggregatorTest.java b/extensions-core/histogram/src/test/java/org/apache/druid/query/aggregation/histogram/FixedBucketsHistogramBufferAggregatorTest.java
new file mode 100644
index 0000000..a84124f
--- /dev/null
+++ b/extensions-core/histogram/src/test/java/org/apache/druid/query/aggregation/histogram/FixedBucketsHistogramBufferAggregatorTest.java
@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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.apache.druid.query.aggregation.histogram;
+
+import org.apache.druid.query.aggregation.BufferAggregator;
+import org.apache.druid.query.aggregation.TestFloatColumnSelector;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.nio.ByteBuffer;
+
+public class FixedBucketsHistogramBufferAggregatorTest
+{
+ private void aggregateBuffer(TestFloatColumnSelector selector, BufferAggregator agg, ByteBuffer buf, int position)
+ {
+ agg.aggregate(buf, position);
+ selector.increment();
+ }
+
+ @Test
+ public void testBufferAggregate()
+ {
+ final float[] values = {23, 19, 10, 16, 36, 2, 9, 32, 30, 45};
+
+ final TestFloatColumnSelector selector = new TestFloatColumnSelector(values);
+
+ FixedBucketsHistogramAggregatorFactory factory = new FixedBucketsHistogramAggregatorFactory(
+ "billy",
+ "billy",
+ 5,
+ 0,
+ 50,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW
+ );
+
+ FixedBucketsHistogramBufferAggregator agg = new FixedBucketsHistogramBufferAggregator(
+ selector,
+ 0,
+ 50,
+ 5,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW
+ );
+
+ ByteBuffer buf = ByteBuffer.allocate(factory.getMaxIntermediateSizeWithNulls());
+ int position = 0;
+
+ agg.init(buf, position);
+ //noinspection ForLoopReplaceableByForEach
+ for (int i = 0; i < values.length; i++) {
+ aggregateBuffer(selector, agg, buf, position);
+ }
+
+ FixedBucketsHistogram h = ((FixedBucketsHistogram) agg.get(buf, position));
+
+ Assert.assertArrayEquals(
+ "final bin counts don't match expected counts",
+ new long[]{2, 3, 1, 3, 1}, h.getHistogram()
+ );
+
+ Assert.assertEquals("getMin value doesn't match expected getMin", 2, h.getMin(), 0);
+ Assert.assertEquals("getMax value doesn't match expected getMax", 45, h.getMax(), 0);
+
+ Assert.assertEquals("count doesn't match expected count", 10, h.getCount());
+ }
+}
diff --git a/extensions-core/histogram/src/test/java/org/apache/druid/query/aggregation/histogram/FixedBucketsHistogramGroupByQueryTest.java b/extensions-core/histogram/src/test/java/org/apache/druid/query/aggregation/histogram/FixedBucketsHistogramGroupByQueryTest.java
new file mode 100644
index 0000000..4ee5283
--- /dev/null
+++ b/extensions-core/histogram/src/test/java/org/apache/druid/query/aggregation/histogram/FixedBucketsHistogramGroupByQueryTest.java
@@ -0,0 +1,262 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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.apache.druid.query.aggregation.histogram;
+
+import com.google.common.collect.ImmutableList;
+import org.apache.druid.data.input.Row;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.io.Closer;
+import org.apache.druid.query.QueryRunner;
+import org.apache.druid.query.QueryRunnerTestHelper;
+import org.apache.druid.query.dimension.DefaultDimensionSpec;
+import org.apache.druid.query.groupby.GroupByQuery;
+import org.apache.druid.query.groupby.GroupByQueryConfig;
+import org.apache.druid.query.groupby.GroupByQueryRunnerFactory;
+import org.apache.druid.query.groupby.GroupByQueryRunnerTest;
+import org.apache.druid.query.groupby.GroupByQueryRunnerTestHelper;
+import org.apache.druid.query.groupby.orderby.DefaultLimitSpec;
+import org.apache.druid.query.groupby.orderby.OrderByColumnSpec;
+import org.apache.druid.query.groupby.strategy.GroupByStrategySelector;
+import org.apache.druid.segment.TestHelper;
+import org.junit.After;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ */
+@RunWith(Parameterized.class)
+public class FixedBucketsHistogramGroupByQueryTest
+{
+ private static final Closer resourceCloser = Closer.create();
+
+ private final QueryRunner<Row> runner;
+ private final GroupByQueryRunnerFactory factory;
+
+ @Parameterized.Parameters(name = "{0}")
+ public static Iterable<Object[]> constructorFeeder()
+ {
+ final GroupByQueryConfig v1Config = new GroupByQueryConfig()
+ {
+ @Override
+ public String getDefaultStrategy()
+ {
+ return GroupByStrategySelector.STRATEGY_V1;
+ }
+
+ @Override
+ public String toString()
+ {
+ return "v1";
+ }
+ };
+ final GroupByQueryConfig v1SingleThreadedConfig = new GroupByQueryConfig()
+ {
+ @Override
+ public boolean isSingleThreaded()
+ {
+ return true;
+ }
+
+ @Override
+ public String getDefaultStrategy()
+ {
+ return GroupByStrategySelector.STRATEGY_V1;
+ }
+
+ @Override
+ public String toString()
+ {
+ return "v1SingleThreaded";
+ }
+ };
+ final GroupByQueryConfig v2Config = new GroupByQueryConfig()
+ {
+ @Override
+ public String getDefaultStrategy()
+ {
+ return GroupByStrategySelector.STRATEGY_V2;
+ }
+
+ @Override
+ public String toString()
+ {
+ return "v2";
+ }
+ };
+
+ v1Config.setMaxIntermediateRows(10000);
+ v1SingleThreadedConfig.setMaxIntermediateRows(10000);
+
+ final List<Object[]> constructors = new ArrayList<>();
+ final List<GroupByQueryConfig> configs = ImmutableList.of(
+ v1Config,
+ v1SingleThreadedConfig,
+ v2Config
+ );
+
+ for (GroupByQueryConfig config : configs) {
+ final Pair<GroupByQueryRunnerFactory, Closer> factoryAndCloser = GroupByQueryRunnerTest.makeQueryRunnerFactory(
+ config
+ );
+ final GroupByQueryRunnerFactory factory = factoryAndCloser.lhs;
+ resourceCloser.register(factoryAndCloser.rhs);
+ for (QueryRunner<Row> runner : QueryRunnerTestHelper.makeQueryRunners(factory)) {
+ final String testName = StringUtils.format(
+ "config=%s, runner=%s",
+ config.toString(),
+ runner.toString()
+ );
+ constructors.add(new Object[]{testName, factory, runner});
+ }
+ }
+
+ return constructors;
+ }
+
+ public FixedBucketsHistogramGroupByQueryTest(
+ String testName,
+ GroupByQueryRunnerFactory factory,
+ QueryRunner runner
+ )
+ {
+ this.factory = factory;
+ this.runner = runner;
+
+ //Note: this is needed in order to properly register the serde for Histogram.
+ new ApproximateHistogramDruidModule().configure(null);
+ }
+
+ @After
+ public void teardown() throws IOException
+ {
+ resourceCloser.close();
+ }
+
+ @Test
+ public void testGroupByWithFixedHistogramAgg()
+ {
+ FixedBucketsHistogramAggregatorFactory aggFactory = new FixedBucketsHistogramAggregatorFactory(
+ "histo",
+ "index",
+ 10,
+ 0,
+ 2000,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW
+ );
+
+ GroupByQuery query = new GroupByQuery.Builder()
+ .setDataSource(QueryRunnerTestHelper.dataSource)
+ .setGranularity(QueryRunnerTestHelper.allGran).setDimensions(new DefaultDimensionSpec(
+ QueryRunnerTestHelper.marketDimension,
+ "marketalias"
+ ))
+ .setInterval(QueryRunnerTestHelper.fullOnInterval)
+ .setLimitSpec(
+ new DefaultLimitSpec(
+ Collections.singletonList(new OrderByColumnSpec("marketalias", OrderByColumnSpec.Direction.DESCENDING)),
+ 1
+ )
+ ).setAggregatorSpecs(QueryRunnerTestHelper.rowsCount, aggFactory)
+ .setPostAggregatorSpecs(
+ Collections.singletonList(
+ new QuantilePostAggregator("quantile", "histo", 0.5f)
+ )
+ )
+ .build();
+
+ List<Row> expectedResults = Collections.singletonList(
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "1970-01-01T00:00:00.000Z",
+ "marketalias", "upfront",
+ "rows", 186L,
+ "quantile", 969.6969604492188f,
+ "histo",
+ new FixedBucketsHistogram(
+ 0,
+ 2000,
+ 10,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW,
+ new long[]{0, 0, 4, 33, 66, 35, 25, 11, 10, 2},
+ 186,
+ 1870.061029,
+ 545.990623,
+ 0,
+ 0,
+ 0
+ )
+ )
+ );
+
+ Iterable<Row> results = GroupByQueryRunnerTestHelper.runQuery(factory, runner, query);
+ TestHelper.assertExpectedObjects(expectedResults, results, "fixed-histo");
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testGroupByWithSameNameComplexPostAgg()
+ {
+ FixedBucketsHistogramAggregatorFactory aggFactory = new FixedBucketsHistogramAggregatorFactory(
+ "histo",
+ "index",
+ 10,
+ 0,
+ 2000,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW
+ );
+
+ GroupByQuery query = new GroupByQuery.Builder()
+ .setDataSource(QueryRunnerTestHelper.dataSource)
+ .setGranularity(QueryRunnerTestHelper.allGran).setDimensions(new DefaultDimensionSpec(
+ QueryRunnerTestHelper.marketDimension,
+ "marketalias"
+ ))
+ .setInterval(QueryRunnerTestHelper.fullOnInterval)
+ .setLimitSpec(
+ new DefaultLimitSpec(
+ Collections.singletonList(new OrderByColumnSpec("marketalias", OrderByColumnSpec.Direction.DESCENDING)),
+ 1
+ )
+ ).setAggregatorSpecs(QueryRunnerTestHelper.rowsCount, aggFactory)
+ .setPostAggregatorSpecs(
+ Collections.singletonList(
+ new QuantilePostAggregator("quantile", "quantile", 0.5f)
+ )
+ )
+ .build();
+
+ List<Row> expectedResults = Collections.singletonList(
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "1970-01-01T00:00:00.000Z",
+ "marketalias", "upfront",
+ "rows", 186L,
+ "quantile", 969.6969604492188f
+ )
+ );
+
+ Iterable<Row> results = GroupByQueryRunnerTestHelper.runQuery(factory, runner, query);
+ TestHelper.assertExpectedObjects(expectedResults, results, "fixed-histo");
+ }
+}
diff --git a/extensions-core/histogram/src/test/java/org/apache/druid/query/aggregation/histogram/FixedBucketsHistogramTest.java b/extensions-core/histogram/src/test/java/org/apache/druid/query/aggregation/histogram/FixedBucketsHistogramTest.java
new file mode 100644
index 0000000..f89167d
--- /dev/null
+++ b/extensions-core/histogram/src/test/java/org/apache/druid/query/aggregation/histogram/FixedBucketsHistogramTest.java
@@ -0,0 +1,1453 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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.apache.druid.query.aggregation.histogram;
+
+import org.apache.commons.math3.distribution.NormalDistribution;
+import org.apache.commons.math3.random.JDKRandomGenerator;
+import org.apache.druid.java.util.common.logger.Logger;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Random;
+
+public class FixedBucketsHistogramTest
+{
+ private static final Logger log = new Logger(FixedBucketsHistogramTest.class);
+
+ static final float[] VALUES2 = {23, 19, 10, 16, 36, 2, 1, 9, 32, 30, 45, 46};
+
+ static final float[] VALUES3 = {
+ 20, 16, 19, 27, 17, 20, 18, 20, 28, 14, 17, 21, 20, 21, 10, 25, 23, 17, 21, 18,
+ 14, 20, 18, 12, 19, 20, 23, 25, 15, 22, 14, 17, 15, 23, 23, 15, 27, 20, 17, 15
+ };
+ static final float[] VALUES4 = {
+ 27.489f, 3.085f, 3.722f, 66.875f, 30.998f, -8.193f, 5.395f, 5.109f, 10.944f, 54.75f,
+ 14.092f, 15.604f, 52.856f, 66.034f, 22.004f, -14.682f, -50.985f, 2.872f, 61.013f,
+ -21.766f, 19.172f, 62.882f, 33.537f, 21.081f, 67.115f, 44.789f, 64.1f, 20.911f,
+ -6.553f, 2.178f
+ };
+ static final float[] VALUES5 = {1, 2, 3, 4, 5, 6, 7, 8, 9, 10};
+ static final float[] VALUES6 = {
+ 1f, 1.5f, 2f, 2.5f, 3f, 3.5f, 4f, 4.5f, 5f, 5.5f, 6f, 6.5f, 7f, 7.5f, 8f, 8.5f, 9f, 9.5f, 10f
+ };
+
+ static final float[] VALUES7 = {1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 12, 12, 12, 15, 20, 25, 25, 25};
+
+ protected FixedBucketsHistogram buildHistogram(
+ double lowerLimit,
+ double upperLimit,
+ int numBuckets,
+ FixedBucketsHistogram.OutlierHandlingMode outlierHandlingMode,
+ float[] values
+ )
+ {
+ FixedBucketsHistogram h = new FixedBucketsHistogram(
+ lowerLimit,
+ upperLimit,
+ numBuckets,
+ outlierHandlingMode
+ );
+
+ for (float v : values) {
+ h.add(v);
+ }
+ return h;
+ }
+
+ @Test
+ public void testOffer()
+ {
+ FixedBucketsHistogram h = buildHistogram(
+ 0,
+ 200,
+ 200,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW,
+ VALUES2
+ );
+
+ float[] quantiles = h.percentilesFloat(new double[]{12.5f, 50.0f, 98f});
+ double[] doubles = new double[VALUES2.length];
+
+ for (int i = 0; i < doubles.length; i++) {
+ doubles[i] = VALUES2[i];
+ }
+
+ Assert.assertArrayEquals(
+ new float[]{2.5f, 20.0f, 46.76f},
+ quantiles,
+ 0.01f
+ );
+ }
+
+ @Test
+ public void testOfferRandoms()
+ {
+ FixedBucketsHistogram h = buildHistogram(
+ 0,
+ 1000,
+ 50,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW,
+ new float[]{}
+ );
+
+ Random rng = new Random(1000);
+ double[] values = new double[100000];
+ for (int i = 0; i < 100000; i++) {
+ values[i] = (double) rng.nextInt(1000);
+ h.add(values[i]);
+ }
+
+ float[] quantiles = h.percentilesFloat(new double[]{12.5f, 25.0f, 50.0f, 98f});
+
+ Assert.assertArrayEquals(
+ new float[]{125.04082f, 248.84348f, 501.67166f, 979.7799f},
+ quantiles,
+ 0.01f
+ );
+ }
+
+ @Test
+ public void testNormalDistribution()
+ {
+ NormalDistribution normalDistribution = new NormalDistribution(
+ new JDKRandomGenerator(1000),
+ 50000,
+ 10000
+ );
+
+ FixedBucketsHistogram h = new FixedBucketsHistogram(
+ 0,
+ 100000,
+ 1000,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW
+ );
+
+ for (int i = 0; i < 100000; i++) {
+ double val = normalDistribution.sample();
+ h.add(val);
+ }
+ float[] quantiles = h.percentilesFloat(new double[]{12.5f, 25.0f, 50.0f, 98f});
+
+ Assert.assertArrayEquals(
+ new float[]{38565.324f, 43297.95f, 50091.902f, 70509.125f},
+ quantiles,
+ 0.01f
+ );
+ }
+
+ @Test
+ public void testOfferWithNegatives()
+ {
+ FixedBucketsHistogram h = buildHistogram(
+ -100,
+ 100,
+ 100,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW,
+ VALUES2
+ );
+
+ float[] quantiles = h.percentilesFloat(new double[]{12.5f, 50.0f, 98f});
+ double[] doubles = new double[VALUES2.length];
+
+ for (int i = 0; i < doubles.length; i++) {
+ doubles[i] = VALUES2[i];
+ }
+
+ Assert.assertArrayEquals(
+ new float[]{3.0f, 20.0f, 47.52f},
+ quantiles,
+ 0.01f
+ );
+ }
+
+ @Test
+ public void testOfferValues3()
+ {
+ FixedBucketsHistogram h = buildHistogram(
+ 0,
+ 200,
+ 100,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW,
+ VALUES3
+ );
+
+ float[] quantiles = h.percentilesFloat(new double[]{12.5f, 50.0f, 98f});
+ double[] doubles = new double[VALUES3.length];
+
+ for (int i = 0; i < doubles.length; i++) {
+ doubles[i] = VALUES3[i];
+ }
+
+ Assert.assertArrayEquals(
+ new float[]{14.857142f, 20.0f, 28.4f},
+ quantiles,
+ 0.01f
+ );
+ }
+
+ @Test
+ public void testOfferValues4()
+ {
+ FixedBucketsHistogram h = buildHistogram(
+ -100,
+ 100,
+ 100,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW,
+ VALUES4
+ );
+
+ float[] quantiles = h.percentilesFloat(new double[]{12.5f, 50.0f, 98f});
+ double[] doubles = new double[VALUES4.length];
+
+ for (int i = 0; i < doubles.length; i++) {
+ doubles[i] = VALUES4[i];
+ }
+
+ Assert.assertArrayEquals(
+ new float[]{-8.5f, 20.0f, 67.6f},
+ quantiles,
+ 0.01f
+ );
+ }
+
+ @Test
+ public void testOfferValues5()
+ {
+ FixedBucketsHistogram h = buildHistogram(
+ 0,
+ 10,
+ 10,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW,
+ VALUES5
+ );
+
+ float[] quantiles = h.percentilesFloat(new double[]{12.5f, 50.0f, 98f});
+ double[] doubles = new double[VALUES5.length];
+
+ for (int i = 0; i < doubles.length; i++) {
+ doubles[i] = VALUES5[i];
+ }
+
+ Assert.assertArrayEquals(
+ new float[]{2.125f, 5.5f, 9.82f},
+ quantiles,
+ 0.01f
+ );
+ }
+
+ @Test
+ public void testOfferValues6()
+ {
+ FixedBucketsHistogram h = buildHistogram(
+ 0,
+ 10,
+ 10,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW,
+ VALUES6
+ );
+
+ float[] quantiles = h.percentilesFloat(new double[]{12.5f, 50.0f, 98f});
+ double[] doubles = new double[VALUES6.length];
+
+ for (int i = 0; i < doubles.length; i++) {
+ doubles[i] = VALUES6[i];
+ }
+
+ Assert.assertArrayEquals(
+ new float[]{2.125f, 5.5f, 9.82f},
+ quantiles,
+ 0.01f
+ );
+ }
+
+ @Test
+ public void testOfferValues7()
+ {
+ FixedBucketsHistogram h = buildHistogram(
+ 0,
+ 50,
+ 50,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW,
+ VALUES7
+ );
+
+ float[] quantiles = h.percentilesFloat(new double[]{12.5f, 50.0f, 98f});
+ double[] doubles = new double[VALUES7.length];
+
+ for (int i = 0; i < doubles.length; i++) {
+ doubles[i] = VALUES7[i];
+ }
+
+ Assert.assertArrayEquals(
+ new float[]{3.25f, 10f, 25.88f},
+ quantiles,
+ 0.01f
+ );
+ }
+
+ @Test
+ public void testMergeSameBuckets()
+ {
+ FixedBucketsHistogram h = buildHistogram(
+ 0,
+ 20,
+ 5,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW,
+ new float[]{1, 2, 7, 12, 19, 25, -5}
+ );
+
+ FixedBucketsHistogram hClip = buildHistogram(
+ 0,
+ 20,
+ 5,
+ FixedBucketsHistogram.OutlierHandlingMode.CLIP,
+ new float[]{1, 2, 7, 12, 19, 25, -5}
+ );
+
+ FixedBucketsHistogram hIgnore = buildHistogram(
+ 0,
+ 20,
+ 5,
+ FixedBucketsHistogram.OutlierHandlingMode.IGNORE,
+ new float[]{1, 2, 7, 12, 19, 25, -5}
+ );
+
+ FixedBucketsHistogram h2 = buildHistogram(
+ 0,
+ 20,
+ 5,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW,
+ new float[]{3, 8, 9, 13, -99, 200}
+ );
+
+ h.combineHistogram(h2);
+ Assert.assertEquals(5, h.getNumBuckets());
+ Assert.assertEquals(4.0, h.getBucketSize(), 0.01);
+ Assert.assertEquals(0, h.getLowerLimit(), 0.01);
+ Assert.assertEquals(20, h.getUpperLimit(), 0.01);
+ Assert.assertEquals(FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW, h.getOutlierHandlingMode());
+ Assert.assertArrayEquals(new long[]{3, 1, 2, 2, 1}, h.getHistogram());
+ Assert.assertEquals(9, h.getCount());
+ Assert.assertEquals(1, h.getMin(), 0.01);
+ Assert.assertEquals(19, h.getMax(), 0.01);
+ Assert.assertEquals(0, h.getMissingValueCount());
+ Assert.assertEquals(2, h.getLowerOutlierCount());
+ Assert.assertEquals(2, h.getUpperOutlierCount());
+
+ hClip.combineHistogram(h2);
+ Assert.assertEquals(5, hClip.getNumBuckets());
+ Assert.assertEquals(4.0, hClip.getBucketSize(), 0.01);
+ Assert.assertEquals(0, hClip.getLowerLimit(), 0.01);
+ Assert.assertEquals(20, hClip.getUpperLimit(), 0.01);
+ Assert.assertEquals(FixedBucketsHistogram.OutlierHandlingMode.CLIP, hClip.getOutlierHandlingMode());
+ Assert.assertArrayEquals(new long[]{5, 1, 2, 2, 3}, hClip.getHistogram());
+ Assert.assertEquals(13, hClip.getCount());
+ Assert.assertEquals(0, hClip.getMin(), 0.01);
+ Assert.assertEquals(20, hClip.getMax(), 0.01);
+ Assert.assertEquals(0, hClip.getMissingValueCount());
+ Assert.assertEquals(0, hClip.getLowerOutlierCount());
+ Assert.assertEquals(0, hClip.getUpperOutlierCount());
+
+ hIgnore.combineHistogram(h2);
+ Assert.assertEquals(5, hIgnore.getNumBuckets());
+ Assert.assertEquals(4.0, hIgnore.getBucketSize(), 0.01);
+ Assert.assertEquals(0, hIgnore.getLowerLimit(), 0.01);
+ Assert.assertEquals(20, hIgnore.getUpperLimit(), 0.01);
+ Assert.assertEquals(FixedBucketsHistogram.OutlierHandlingMode.IGNORE, hIgnore.getOutlierHandlingMode());
+ Assert.assertArrayEquals(new long[]{3, 1, 2, 2, 1}, hIgnore.getHistogram());
+ Assert.assertEquals(9, hIgnore.getCount());
+ Assert.assertEquals(1, hIgnore.getMin(), 0.01);
+ Assert.assertEquals(19, hIgnore.getMax(), 0.01);
+ Assert.assertEquals(0, hIgnore.getMissingValueCount());
+ Assert.assertEquals(0, hIgnore.getLowerOutlierCount());
+ Assert.assertEquals(0, hIgnore.getUpperOutlierCount());
+ }
+
+ @Test
+ public void testMergeNoOverlapRight()
+ {
+ FixedBucketsHistogram h = buildHistogram(
+ 0,
+ 20,
+ 5,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW,
+ new float[]{1, 2, 7, 12, 19}
+ );
+
+ FixedBucketsHistogram hClip = buildHistogram(
+ 0,
+ 20,
+ 5,
+ FixedBucketsHistogram.OutlierHandlingMode.CLIP,
+ new float[]{1, 2, 7, 12, 19}
+ );
+
+ FixedBucketsHistogram hIgnore = buildHistogram(
+ 0,
+ 20,
+ 5,
+ FixedBucketsHistogram.OutlierHandlingMode.IGNORE,
+ new float[]{1, 2, 7, 12, 19}
+ );
+
+ FixedBucketsHistogram h2 = buildHistogram(
+ 50,
+ 100,
+ 5,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW,
+ new float[]{60, 70, 80, 90}
+ );
+
+ h.combineHistogram(h2);
+ Assert.assertEquals(5, h.getNumBuckets());
+ Assert.assertEquals(4.0, h.getBucketSize(), 0.01);
+ Assert.assertEquals(0, h.getLowerLimit(), 0.01);
+ Assert.assertEquals(20, h.getUpperLimit(), 0.01);
+ Assert.assertEquals(FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW, h.getOutlierHandlingMode());
+ Assert.assertArrayEquals(new long[]{2, 1, 0, 1, 1}, h.getHistogram());
+ Assert.assertEquals(5, h.getCount());
+ Assert.assertEquals(1, h.getMin(), 0.01);
+ Assert.assertEquals(19, h.getMax(), 0.01);
+ Assert.assertEquals(0, h.getMissingValueCount());
+ Assert.assertEquals(0, h.getLowerOutlierCount());
+ Assert.assertEquals(4, h.getUpperOutlierCount());
+
+ hClip.combineHistogram(h2);
+ Assert.assertEquals(5, hClip.getNumBuckets());
+ Assert.assertEquals(4.0, hClip.getBucketSize(), 0.01);
+ Assert.assertEquals(0, hClip.getLowerLimit(), 0.01);
+ Assert.assertEquals(20, hClip.getUpperLimit(), 0.01);
+ Assert.assertEquals(FixedBucketsHistogram.OutlierHandlingMode.CLIP, hClip.getOutlierHandlingMode());
+ Assert.assertArrayEquals(new long[]{2, 1, 0, 1, 5}, hClip.getHistogram());
+ Assert.assertEquals(9, hClip.getCount());
+ Assert.assertEquals(1, hClip.getMin(), 0.01);
+ Assert.assertEquals(20, hClip.getMax(), 0.01);
+ Assert.assertEquals(0, hClip.getMissingValueCount());
+ Assert.assertEquals(0, hClip.getLowerOutlierCount());
+ Assert.assertEquals(0, hClip.getUpperOutlierCount());
+
+ hIgnore.combineHistogram(h2);
+ Assert.assertEquals(5, hIgnore.getNumBuckets());
+ Assert.assertEquals(4.0, hIgnore.getBucketSize(), 0.01);
+ Assert.assertEquals(0, hIgnore.getLowerLimit(), 0.01);
+ Assert.assertEquals(20, hIgnore.getUpperLimit(), 0.01);
+ Assert.assertEquals(FixedBucketsHistogram.OutlierHandlingMode.IGNORE, hIgnore.getOutlierHandlingMode());
+ Assert.assertArrayEquals(new long[]{2, 1, 0, 1, 1}, hIgnore.getHistogram());
+ Assert.assertEquals(5, hIgnore.getCount());
+ Assert.assertEquals(1, hIgnore.getMin(), 0.01);
+ Assert.assertEquals(19, hIgnore.getMax(), 0.01);
+ Assert.assertEquals(0, hIgnore.getMissingValueCount());
+ Assert.assertEquals(0, hIgnore.getLowerOutlierCount());
+ Assert.assertEquals(0, hIgnore.getUpperOutlierCount());
+ }
+
+ @Test
+ public void testMergeNoOverlapLeft()
+ {
+ FixedBucketsHistogram h = buildHistogram(
+ 0,
+ 20,
+ 5,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW,
+ new float[]{1, 2, 7, 12, 19}
+ );
+
+ FixedBucketsHistogram hClip = buildHistogram(
+ 0,
+ 20,
+ 5,
+ FixedBucketsHistogram.OutlierHandlingMode.CLIP,
+ new float[]{1, 2, 7, 12, 19}
+ );
+
+ FixedBucketsHistogram hIgnore = buildHistogram(
+ 0,
+ 20,
+ 5,
+ FixedBucketsHistogram.OutlierHandlingMode.IGNORE,
+ new float[]{1, 2, 7, 12, 19}
+ );
+
+ FixedBucketsHistogram h2 = buildHistogram(
+ -100,
+ -50,
+ 5,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW,
+ new float[]{-60, -70, -80, -90}
+ );
+
+ h.combineHistogram(h2);
+ Assert.assertEquals(5, h.getNumBuckets());
+ Assert.assertEquals(4.0, h.getBucketSize(), 0.01);
+ Assert.assertEquals(0, h.getLowerLimit(), 0.01);
+ Assert.assertEquals(20, h.getUpperLimit(), 0.01);
+ Assert.assertEquals(FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW, h.getOutlierHandlingMode());
+ Assert.assertArrayEquals(new long[]{2, 1, 0, 1, 1}, h.getHistogram());
+ Assert.assertEquals(5, h.getCount());
+ Assert.assertEquals(1, h.getMin(), 0.01);
+ Assert.assertEquals(19, h.getMax(), 0.01);
+ Assert.assertEquals(0, h.getMissingValueCount());
+ Assert.assertEquals(4, h.getLowerOutlierCount());
+ Assert.assertEquals(0, h.getUpperOutlierCount());
+
+ hClip.combineHistogram(h2);
+ Assert.assertEquals(5, hClip.getNumBuckets());
+ Assert.assertEquals(4.0, hClip.getBucketSize(), 0.01);
+ Assert.assertEquals(0, hClip.getLowerLimit(), 0.01);
+ Assert.assertEquals(20, hClip.getUpperLimit(), 0.01);
+ Assert.assertEquals(FixedBucketsHistogram.OutlierHandlingMode.CLIP, hClip.getOutlierHandlingMode());
+ Assert.assertArrayEquals(new long[]{6, 1, 0, 1, 1}, hClip.getHistogram());
+ Assert.assertEquals(9, hClip.getCount());
+ Assert.assertEquals(0, hClip.getMin(), 0.01);
+ Assert.assertEquals(19, hClip.getMax(), 0.01);
+ Assert.assertEquals(0, hClip.getMissingValueCount());
+ Assert.assertEquals(0, hClip.getLowerOutlierCount());
+ Assert.assertEquals(0, hClip.getUpperOutlierCount());
+
+ hIgnore.combineHistogram(h2);
+ Assert.assertEquals(5, hIgnore.getNumBuckets());
+ Assert.assertEquals(4.0, hIgnore.getBucketSize(), 0.01);
+ Assert.assertEquals(0, hIgnore.getLowerLimit(), 0.01);
+ Assert.assertEquals(20, hIgnore.getUpperLimit(), 0.01);
+ Assert.assertEquals(FixedBucketsHistogram.OutlierHandlingMode.IGNORE, hIgnore.getOutlierHandlingMode());
+ Assert.assertArrayEquals(new long[]{2, 1, 0, 1, 1}, hIgnore.getHistogram());
+ Assert.assertEquals(5, hIgnore.getCount());
+ Assert.assertEquals(1, hIgnore.getMin(), 0.01);
+ Assert.assertEquals(19, hIgnore.getMax(), 0.01);
+ Assert.assertEquals(0, hIgnore.getMissingValueCount());
+ Assert.assertEquals(0, hIgnore.getLowerOutlierCount());
+ Assert.assertEquals(0, hIgnore.getUpperOutlierCount());
+ }
+
+
+ @Test
+ public void testMergeSameBucketsRightOverlap()
+ {
+ FixedBucketsHistogram h = buildHistogram(
+ 0,
+ 20,
+ 5,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW,
+ new float[]{1, 2, 7, 12, 19}
+ );
+
+ FixedBucketsHistogram hClip = buildHistogram(
+ 0,
+ 20,
+ 5,
+ FixedBucketsHistogram.OutlierHandlingMode.CLIP,
+ new float[]{1, 2, 7, 12, 19}
+ );
+
+ FixedBucketsHistogram hIgnore = buildHistogram(
+ 0,
+ 20,
+ 5,
+ FixedBucketsHistogram.OutlierHandlingMode.IGNORE,
+ new float[]{1, 2, 7, 12, 19}
+ );
+
+ FixedBucketsHistogram h2 = buildHistogram(
+ 12,
+ 32,
+ 5,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW,
+ new float[]{13, 18, 25, 29}
+ );
+
+ h.combineHistogram(h2);
+
+ Assert.assertEquals(5, h.getNumBuckets());
+ Assert.assertEquals(4.0, h.getBucketSize(), 0.01);
+ Assert.assertEquals(0, h.getLowerLimit(), 0.01);
+ Assert.assertEquals(20, h.getUpperLimit(), 0.01);
+ Assert.assertEquals(FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW, h.getOutlierHandlingMode());
+ Assert.assertArrayEquals(new long[]{2, 1, 0, 2, 2}, h.getHistogram());
+ Assert.assertEquals(7, h.getCount());
+ Assert.assertEquals(1, h.getMin(), 0.01);
+ Assert.assertEquals(19, h.getMax(), 0.01);
+ Assert.assertEquals(0, h.getMissingValueCount());
+ Assert.assertEquals(0, h.getLowerOutlierCount());
+ Assert.assertEquals(2, h.getUpperOutlierCount());
+
+ hClip.combineHistogram(h2);
+ Assert.assertEquals(5, hClip.getNumBuckets());
+ Assert.assertEquals(4.0, hClip.getBucketSize(), 0.01);
+ Assert.assertEquals(0, hClip.getLowerLimit(), 0.01);
+ Assert.assertEquals(20, hClip.getUpperLimit(), 0.01);
+ Assert.assertEquals(FixedBucketsHistogram.OutlierHandlingMode.CLIP, hClip.getOutlierHandlingMode());
+ Assert.assertArrayEquals(new long[]{2, 1, 0, 2, 4}, hClip.getHistogram());
+ Assert.assertEquals(9, hClip.getCount());
+ Assert.assertEquals(1, hClip.getMin(), 0.01);
+ Assert.assertEquals(20, hClip.getMax(), 0.01);
+ Assert.assertEquals(0, hClip.getMissingValueCount());
+ Assert.assertEquals(0, hClip.getLowerOutlierCount());
+ Assert.assertEquals(0, hClip.getUpperOutlierCount());
+
+ hIgnore.combineHistogram(h2);
+ Assert.assertEquals(5, hIgnore.getNumBuckets());
+ Assert.assertEquals(4.0, hIgnore.getBucketSize(), 0.01);
+ Assert.assertEquals(0, hIgnore.getLowerLimit(), 0.01);
+ Assert.assertEquals(20, hIgnore.getUpperLimit(), 0.01);
+ Assert.assertEquals(FixedBucketsHistogram.OutlierHandlingMode.IGNORE, hIgnore.getOutlierHandlingMode());
+ Assert.assertArrayEquals(new long[]{2, 1, 0, 2, 2}, hIgnore.getHistogram());
+ Assert.assertEquals(7, hIgnore.getCount());
+ Assert.assertEquals(1, hIgnore.getMin(), 0.01);
+ Assert.assertEquals(19, hIgnore.getMax(), 0.01);
+ Assert.assertEquals(0, hIgnore.getMissingValueCount());
+ Assert.assertEquals(0, hIgnore.getLowerOutlierCount());
+ Assert.assertEquals(0, hIgnore.getUpperOutlierCount());
+ }
+
+ @Test
+ public void testMergeSameBucketsLeftOverlap()
+ {
+ FixedBucketsHistogram h = buildHistogram(
+ 12,
+ 32,
+ 5,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW,
+ new float[]{13, 18, 25, 29}
+ );
+
+ FixedBucketsHistogram hClip = buildHistogram(
+ 12,
+ 32,
+ 5,
+ FixedBucketsHistogram.OutlierHandlingMode.CLIP,
+ new float[]{13, 18, 25, 29}
+ );
+
+ FixedBucketsHistogram hIgnore = buildHistogram(
+ 12,
+ 32,
+ 5,
+ FixedBucketsHistogram.OutlierHandlingMode.IGNORE,
+ new float[]{13, 18, 25, 29}
+ );
+
+ FixedBucketsHistogram h2 = buildHistogram(
+ 0,
+ 20,
+ 5,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW,
+ new float[]{1, 2, 7, 13, 19}
+ );
+
+ h.combineHistogram(h2);
+
+ Assert.assertEquals(5, h.getNumBuckets());
+ Assert.assertEquals(4.0, h.getBucketSize(), 0.01);
+ Assert.assertEquals(12, h.getLowerLimit(), 0.01);
+ Assert.assertEquals(32, h.getUpperLimit(), 0.01);
+ Assert.assertEquals(FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW, h.getOutlierHandlingMode());
+ Assert.assertArrayEquals(new long[]{2, 2, 0, 1, 1}, h.getHistogram());
+ Assert.assertEquals(6, h.getCount());
+ Assert.assertEquals(12, h.getMin(), 0.01);
+ Assert.assertEquals(29, h.getMax(), 0.01);
+ Assert.assertEquals(0, h.getMissingValueCount());
+ Assert.assertEquals(3, h.getLowerOutlierCount());
+ Assert.assertEquals(0, h.getUpperOutlierCount());
+
+ hClip.combineHistogram(h2);
+ Assert.assertEquals(5, hClip.getNumBuckets());
+ Assert.assertEquals(4.0, hClip.getBucketSize(), 0.01);
+ Assert.assertEquals(12, hClip.getLowerLimit(), 0.01);
+ Assert.assertEquals(32, hClip.getUpperLimit(), 0.01);
+ Assert.assertEquals(FixedBucketsHistogram.OutlierHandlingMode.CLIP, hClip.getOutlierHandlingMode());
+ Assert.assertArrayEquals(new long[]{5, 2, 0, 1, 1}, hClip.getHistogram());
+ Assert.assertEquals(9, hClip.getCount());
+ Assert.assertEquals(12, hClip.getMin(), 0.01);
+ Assert.assertEquals(29, hClip.getMax(), 0.01);
+ Assert.assertEquals(0, hClip.getMissingValueCount());
+ Assert.assertEquals(0, hClip.getLowerOutlierCount());
+ Assert.assertEquals(0, hClip.getUpperOutlierCount());
+
+ hIgnore.combineHistogram(h2);
+ Assert.assertEquals(5, hIgnore.getNumBuckets());
+ Assert.assertEquals(4.0, hIgnore.getBucketSize(), 0.01);
+ Assert.assertEquals(12, hIgnore.getLowerLimit(), 0.01);
+ Assert.assertEquals(32, hIgnore.getUpperLimit(), 0.01);
+ Assert.assertEquals(FixedBucketsHistogram.OutlierHandlingMode.IGNORE, hIgnore.getOutlierHandlingMode());
+ Assert.assertArrayEquals(new long[]{2, 2, 0, 1, 1}, hIgnore.getHistogram());
+ Assert.assertEquals(6, hIgnore.getCount());
+ Assert.assertEquals(12, hIgnore.getMin(), 0.01);
+ Assert.assertEquals(29, hIgnore.getMax(), 0.01);
+ Assert.assertEquals(0, hIgnore.getMissingValueCount());
+ Assert.assertEquals(0, hIgnore.getLowerOutlierCount());
+ Assert.assertEquals(0, hIgnore.getUpperOutlierCount());
+ }
+
+ @Test
+ public void testMergeSameBucketsContainsOther()
+ {
+ FixedBucketsHistogram h = buildHistogram(
+ 0,
+ 50,
+ 10,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW,
+ new float[]{2, 18, 34, 48}
+ );
+
+ FixedBucketsHistogram hClip = buildHistogram(
+ 0,
+ 50,
+ 10,
+ FixedBucketsHistogram.OutlierHandlingMode.CLIP,
+ new float[]{2, 18, 34, 48}
+ );
+
+ FixedBucketsHistogram hIgnore = buildHistogram(
+ 0,
+ 50,
+ 10,
+ FixedBucketsHistogram.OutlierHandlingMode.IGNORE,
+ new float[]{2, 18, 34, 48}
+ );
+
+ FixedBucketsHistogram h2 = buildHistogram(
+ 10,
+ 30,
+ 4,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW,
+ new float[]{11, 15, 21, 29, 99, -100}
+ );
+
+ h.combineHistogram(h2);
+ Assert.assertEquals(10, h.getNumBuckets());
+ Assert.assertEquals(5.0, h.getBucketSize(), 0.01);
+ Assert.assertEquals(0, h.getLowerLimit(), 0.01);
+ Assert.assertEquals(50, h.getUpperLimit(), 0.01);
+ Assert.assertEquals(FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW, h.getOutlierHandlingMode());
+ Assert.assertArrayEquals(new long[]{1, 0, 1, 2, 1, 1, 1, 0, 0, 1}, h.getHistogram());
+ Assert.assertEquals(8, h.getCount());
+ Assert.assertEquals(2, h.getMin(), 0.01);
+ Assert.assertEquals(48, h.getMax(), 0.01);
+ Assert.assertEquals(0, h.getMissingValueCount());
+ Assert.assertEquals(1, h.getLowerOutlierCount());
+ Assert.assertEquals(1, h.getUpperOutlierCount());
+
+ hClip.combineHistogram(h2);
+ Assert.assertEquals(10, hClip.getNumBuckets());
+ Assert.assertEquals(5.0, hClip.getBucketSize(), 0.01);
+ Assert.assertEquals(0, hClip.getLowerLimit(), 0.01);
+ Assert.assertEquals(50, hClip.getUpperLimit(), 0.01);
+ Assert.assertEquals(FixedBucketsHistogram.OutlierHandlingMode.CLIP, hClip.getOutlierHandlingMode());
+ Assert.assertArrayEquals(new long[]{2, 0, 1, 2, 1, 1, 1, 0, 0, 2}, hClip.getHistogram());
+ Assert.assertEquals(10, hClip.getCount());
+ Assert.assertEquals(0, hClip.getMin(), 0.01);
+ Assert.assertEquals(50, hClip.getMax(), 0.01);
+ Assert.assertEquals(0, hClip.getMissingValueCount());
+ Assert.assertEquals(0, hClip.getLowerOutlierCount());
+ Assert.assertEquals(0, hClip.getUpperOutlierCount());
+
+ hIgnore.combineHistogram(h2);
+ Assert.assertEquals(10, hIgnore.getNumBuckets());
+ Assert.assertEquals(5.0, hIgnore.getBucketSize(), 0.01);
+ Assert.assertEquals(0, hIgnore.getLowerLimit(), 0.01);
+ Assert.assertEquals(50, hIgnore.getUpperLimit(), 0.01);
+ Assert.assertEquals(FixedBucketsHistogram.OutlierHandlingMode.IGNORE, hIgnore.getOutlierHandlingMode());
+ Assert.assertArrayEquals(new long[]{1, 0, 1, 2, 1, 1, 1, 0, 0, 1}, hIgnore.getHistogram());
+ Assert.assertEquals(8, hIgnore.getCount());
+ Assert.assertEquals(2, hIgnore.getMin(), 0.01);
+ Assert.assertEquals(48, hIgnore.getMax(), 0.01);
+ Assert.assertEquals(0, hIgnore.getMissingValueCount());
+ Assert.assertEquals(0, hIgnore.getLowerOutlierCount());
+ Assert.assertEquals(0, hIgnore.getUpperOutlierCount());
+ }
+
+ @Test
+ public void testMergeSameBucketsContainedByOther()
+ {
+ FixedBucketsHistogram h = buildHistogram(
+ 10,
+ 30,
+ 4,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW,
+ new float[]{11, 15, 21, 29}
+ );
+
+ FixedBucketsHistogram hClip = buildHistogram(
+ 10,
+ 30,
+ 4,
+ FixedBucketsHistogram.OutlierHandlingMode.CLIP,
+ new float[]{11, 15, 21, 29}
+ );
+
+ FixedBucketsHistogram hIgnore = buildHistogram(
+ 10,
+ 30,
+ 4,
+ FixedBucketsHistogram.OutlierHandlingMode.IGNORE,
+ new float[]{11, 15, 21, 29}
+ );
+
+ FixedBucketsHistogram h2 = buildHistogram(
+ 0,
+ 50,
+ 10,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW,
+ new float[]{2, 18, 34, 48, 99, -100}
+ );
+
+ h.combineHistogram(h2);
+ Assert.assertEquals(4, h.getNumBuckets());
+ Assert.assertEquals(5.0, h.getBucketSize(), 0.01);
+ Assert.assertEquals(10, h.getLowerLimit(), 0.01);
+ Assert.assertEquals(30, h.getUpperLimit(), 0.01);
+ Assert.assertEquals(FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW, h.getOutlierHandlingMode());
+ Assert.assertArrayEquals(new long[]{1, 2, 1, 1}, h.getHistogram());
+ Assert.assertEquals(5, h.getCount());
+ Assert.assertEquals(11, h.getMin(), 0.01);
+ Assert.assertEquals(29, h.getMax(), 0.01);
+ Assert.assertEquals(0, h.getMissingValueCount());
+ Assert.assertEquals(2, h.getLowerOutlierCount());
+ Assert.assertEquals(3, h.getUpperOutlierCount());
+
+ hClip.combineHistogram(h2);
+ Assert.assertEquals(4, hClip.getNumBuckets());
+ Assert.assertEquals(5.0, hClip.getBucketSize(), 0.01);
+ Assert.assertEquals(10, hClip.getLowerLimit(), 0.01);
+ Assert.assertEquals(30, hClip.getUpperLimit(), 0.01);
+ Assert.assertEquals(FixedBucketsHistogram.OutlierHandlingMode.CLIP, hClip.getOutlierHandlingMode());
+ Assert.assertArrayEquals(new long[]{3, 2, 1, 4}, hClip.getHistogram());
+ Assert.assertEquals(10, hClip.getCount());
+ Assert.assertEquals(10, hClip.getMin(), 0.01);
+ Assert.assertEquals(30, hClip.getMax(), 0.01);
+ Assert.assertEquals(0, hClip.getMissingValueCount());
+ Assert.assertEquals(0, hClip.getLowerOutlierCount());
+ Assert.assertEquals(0, hClip.getUpperOutlierCount());
+
+ hIgnore.combineHistogram(h2);
+ Assert.assertEquals(4, hIgnore.getNumBuckets());
+ Assert.assertEquals(5.0, hIgnore.getBucketSize(), 0.01);
+ Assert.assertEquals(10, hIgnore.getLowerLimit(), 0.01);
+ Assert.assertEquals(30, hIgnore.getUpperLimit(), 0.01);
+ Assert.assertEquals(FixedBucketsHistogram.OutlierHandlingMode.IGNORE, hIgnore.getOutlierHandlingMode());
+ Assert.assertArrayEquals(new long[]{1, 2, 1, 1}, hIgnore.getHistogram());
+ Assert.assertEquals(5, hIgnore.getCount());
+ Assert.assertEquals(11, hIgnore.getMin(), 0.01);
+ Assert.assertEquals(29, hIgnore.getMax(), 0.01);
+ Assert.assertEquals(0, hIgnore.getMissingValueCount());
+ Assert.assertEquals(0, hIgnore.getLowerOutlierCount());
+ Assert.assertEquals(0, hIgnore.getUpperOutlierCount());
+ }
+
+ @Test
+ public void testMergeDifferentBuckets2()
+ {
+ FixedBucketsHistogram h = buildHistogram(
+ 0,
+ 6,
+ 6,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW,
+ new float[]{}
+ );
+
+ FixedBucketsHistogram h2 = buildHistogram(
+ 0,
+ 12,
+ 4,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW,
+ new float[]{1, 1, 1, 1, 5, 5, 5, 5}
+ );
+
+ h.combineHistogram(h2);
+ Assert.assertEquals(6, h.getNumBuckets());
+ Assert.assertEquals(1.0, h.getBucketSize(), 0.01);
+ Assert.assertEquals(0, h.getLowerLimit(), 0.01);
+ Assert.assertEquals(6, h.getUpperLimit(), 0.01);
+ Assert.assertEquals(FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW, h.getOutlierHandlingMode());
+ Assert.assertArrayEquals(new long[]{1, 1, 1, 1, 1, 1}, h.getHistogram());
+ Assert.assertEquals(6, h.getCount());
+ Assert.assertEquals(0, h.getMin(), 0.01);
+ Assert.assertEquals(5.25, h.getMax(), 0.01);
+ Assert.assertEquals(0, h.getMissingValueCount());
+ Assert.assertEquals(0, h.getLowerOutlierCount());
+ Assert.assertEquals(0, h.getUpperOutlierCount());
+ }
+
+
+ @Test
+ public void testMergeDifferentBuckets()
+ {
+ FixedBucketsHistogram h = buildHistogram(
+ 0,
+ 20,
+ 5,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW,
+ new float[]{1, 2, 7, 12, 18}
+ );
+
+ FixedBucketsHistogram hClip = buildHistogram(
+ 0,
+ 20,
+ 5,
+ FixedBucketsHistogram.OutlierHandlingMode.CLIP,
+ new float[]{1, 2, 7, 12, 18}
+ );
+
+ FixedBucketsHistogram hIgnore = buildHistogram(
+ 0,
+ 20,
+ 5,
+ FixedBucketsHistogram.OutlierHandlingMode.IGNORE,
+ new float[]{1, 2, 7, 12, 18}
+ );
+
+ FixedBucketsHistogram h2 = buildHistogram(
+ 0,
+ 20,
+ 7,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW,
+ new float[]{3, 8, 9, 19, 99, -50}
+ );
+
+ h.combineHistogram(h2);
+ Assert.assertEquals(5, h.getNumBuckets());
+ Assert.assertEquals(4.0, h.getBucketSize(), 0.01);
+ Assert.assertEquals(0, h.getLowerLimit(), 0.01);
+ Assert.assertEquals(20, h.getUpperLimit(), 0.01);
+ Assert.assertEquals(FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW, h.getOutlierHandlingMode());
+ Assert.assertArrayEquals(new long[]{2, 3, 1, 1, 2}, h.getHistogram());
+ Assert.assertEquals(9, h.getCount());
+ Assert.assertEquals(1, h.getMin(), 0.01);
+ Assert.assertEquals(18, h.getMax(), 0.01);
+ Assert.assertEquals(0, h.getMissingValueCount());
+ Assert.assertEquals(1, h.getLowerOutlierCount());
+ Assert.assertEquals(1, h.getUpperOutlierCount());
+
+ hClip.combineHistogram(h2);
+ Assert.assertEquals(5, hClip.getNumBuckets());
+ Assert.assertEquals(4.0, hClip.getBucketSize(), 0.01);
+ Assert.assertEquals(0, hClip.getLowerLimit(), 0.01);
+ Assert.assertEquals(20, hClip.getUpperLimit(), 0.01);
+ Assert.assertEquals(FixedBucketsHistogram.OutlierHandlingMode.CLIP, hClip.getOutlierHandlingMode());
+ Assert.assertArrayEquals(new long[]{3, 3, 1, 1, 3}, hClip.getHistogram());
+ Assert.assertEquals(11, hClip.getCount());
+ Assert.assertEquals(0, hClip.getMin(), 0.01);
+ Assert.assertEquals(20, hClip.getMax(), 0.01);
+ Assert.assertEquals(0, hClip.getMissingValueCount());
+ Assert.assertEquals(0, hClip.getLowerOutlierCount());
+ Assert.assertEquals(0, hClip.getUpperOutlierCount());
+
+ hIgnore.combineHistogram(h2);
+ Assert.assertEquals(5, hIgnore.getNumBuckets());
+ Assert.assertEquals(4.0, hIgnore.getBucketSize(), 0.01);
+ Assert.assertEquals(0, hIgnore.getLowerLimit(), 0.01);
+ Assert.assertEquals(20, hIgnore.getUpperLimit(), 0.01);
+ Assert.assertEquals(FixedBucketsHistogram.OutlierHandlingMode.IGNORE, hIgnore.getOutlierHandlingMode());
+ Assert.assertArrayEquals(new long[]{2, 3, 1, 1, 2}, hIgnore.getHistogram());
+ Assert.assertEquals(9, hIgnore.getCount());
+ Assert.assertEquals(1, hIgnore.getMin(), 0.01);
+ Assert.assertEquals(18, hIgnore.getMax(), 0.01);
+ Assert.assertEquals(0, hIgnore.getMissingValueCount());
+ Assert.assertEquals(0, hIgnore.getLowerOutlierCount());
+ Assert.assertEquals(0, hIgnore.getUpperOutlierCount());
+ }
+
+ @Test
+ public void testMergeDifferentBucketsRightOverlap()
+ {
+ FixedBucketsHistogram h = buildHistogram(
+ 0,
+ 20,
+ 5,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW,
+ new float[]{1, 2, 7, 12, 19}
+ );
+
+ FixedBucketsHistogram hClip = buildHistogram(
+ 0,
+ 20,
+ 5,
+ FixedBucketsHistogram.OutlierHandlingMode.CLIP,
+ new float[]{1, 2, 7, 12, 19}
+ );
+
+ FixedBucketsHistogram hIgnore = buildHistogram(
+ 0,
+ 20,
+ 5,
+ FixedBucketsHistogram.OutlierHandlingMode.IGNORE,
+ new float[]{1, 2, 7, 12, 19}
+ );
+
+ FixedBucketsHistogram h2 = buildHistogram(
+ 12,
+ 32,
+ 7,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW,
+ new float[]{13, 18, 25, 29, -10}
+ );
+
+ h.combineHistogram(h2);
+ Assert.assertEquals(5, h.getNumBuckets());
+ Assert.assertEquals(4.0, h.getBucketSize(), 0.01);
+ Assert.assertEquals(0, h.getLowerLimit(), 0.01);
+ Assert.assertEquals(20, h.getUpperLimit(), 0.01);
+ Assert.assertEquals(FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW, h.getOutlierHandlingMode());
+ Assert.assertEquals(7, h.getCount());
+ Assert.assertArrayEquals(new long[]{2, 1, 0, 2, 2}, h.getHistogram());
+ Assert.assertEquals(1, h.getMin(), 0.01);
+ Assert.assertEquals(19, h.getMax(), 0.01);
+ Assert.assertEquals(0, h.getMissingValueCount());
+ Assert.assertEquals(1, h.getLowerOutlierCount());
+ Assert.assertEquals(2, h.getUpperOutlierCount());
+
+ hClip.combineHistogram(h2);
+ Assert.assertEquals(5, hClip.getNumBuckets());
+ Assert.assertEquals(4.0, hClip.getBucketSize(), 0.01);
+ Assert.assertEquals(0, hClip.getLowerLimit(), 0.01);
+ Assert.assertEquals(20, hClip.getUpperLimit(), 0.01);
+ Assert.assertEquals(FixedBucketsHistogram.OutlierHandlingMode.CLIP, hClip.getOutlierHandlingMode());
+ Assert.assertEquals(10, hClip.getCount());
+ Assert.assertArrayEquals(new long[]{3, 1, 0, 2, 4}, hClip.getHistogram());
+ Assert.assertEquals(0, hClip.getMin(), 0.01);
+ Assert.assertEquals(20, hClip.getMax(), 0.01);
+ Assert.assertEquals(0, hClip.getMissingValueCount());
+ Assert.assertEquals(0, hClip.getLowerOutlierCount());
+ Assert.assertEquals(0, hClip.getUpperOutlierCount());
+
+ hIgnore.combineHistogram(h2);
+ Assert.assertEquals(5, hIgnore.getNumBuckets());
+ Assert.assertEquals(4.0, hIgnore.getBucketSize(), 0.01);
+ Assert.assertEquals(0, hIgnore.getLowerLimit(), 0.01);
+ Assert.assertEquals(20, hIgnore.getUpperLimit(), 0.01);
+ Assert.assertEquals(FixedBucketsHistogram.OutlierHandlingMode.IGNORE, hIgnore.getOutlierHandlingMode());
+ Assert.assertEquals(7, hIgnore.getCount());
+ Assert.assertArrayEquals(new long[]{2, 1, 0, 2, 2}, hIgnore.getHistogram());
+ Assert.assertEquals(1, hIgnore.getMin(), 0.01);
+ Assert.assertEquals(19, hIgnore.getMax(), 0.01);
+ Assert.assertEquals(0, hIgnore.getMissingValueCount());
+ Assert.assertEquals(0, hIgnore.getLowerOutlierCount());
+ Assert.assertEquals(0, hIgnore.getUpperOutlierCount());
+ }
+
+ @Test
+ public void testMergeDifferentBucketsLeftOverlap()
+ {
+ FixedBucketsHistogram h = buildHistogram(
+ 12,
+ 32,
+ 5,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW,
+ new float[]{13, 18, 25, 29}
+ );
+
+ FixedBucketsHistogram hClip = buildHistogram(
+ 12,
+ 32,
+ 5,
+ FixedBucketsHistogram.OutlierHandlingMode.CLIP,
+ new float[]{13, 18, 25, 29}
+ );
+
+ FixedBucketsHistogram hIgnore = buildHistogram(
+ 12,
+ 32,
+ 5,
+ FixedBucketsHistogram.OutlierHandlingMode.IGNORE,
+ new float[]{13, 18, 25, 29}
+ );
+
+ FixedBucketsHistogram h2 = buildHistogram(
+ 0,
+ 20,
+ 9,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW,
+ new float[]{1, 2, 7, 12, 19, -99, 100}
+ );
+
+ h.combineHistogram(h2);
+ Assert.assertEquals(5, h.getNumBuckets());
+ Assert.assertEquals(4.0, h.getBucketSize(), 0.01);
+ Assert.assertEquals(12, h.getLowerLimit(), 0.01);
+ Assert.assertEquals(32, h.getUpperLimit(), 0.01);
+ Assert.assertEquals(FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW, h.getOutlierHandlingMode());
+ Assert.assertArrayEquals(new long[]{2, 2, 0, 1, 1}, h.getHistogram());
+ Assert.assertEquals(6, h.getCount());
+ Assert.assertEquals(13, h.getMin(), 0.01);
+ Assert.assertEquals(29, h.getMax(), 0.01);
+ Assert.assertEquals(0, h.getMissingValueCount());
+ Assert.assertEquals(4, h.getLowerOutlierCount());
+ Assert.assertEquals(1, h.getUpperOutlierCount());
+
+ hClip.combineHistogram(h2);
+ Assert.assertEquals(5, hClip.getNumBuckets());
+ Assert.assertEquals(4.0, hClip.getBucketSize(), 0.01);
+ Assert.assertEquals(12, hClip.getLowerLimit(), 0.01);
+ Assert.assertEquals(32, hClip.getUpperLimit(), 0.01);
+ Assert.assertEquals(FixedBucketsHistogram.OutlierHandlingMode.CLIP, hClip.getOutlierHandlingMode());
+ Assert.assertArrayEquals(new long[]{6, 2, 0, 1, 2}, hClip.getHistogram());
+ Assert.assertEquals(11, hClip.getCount());
+ Assert.assertEquals(12, hClip.getMin(), 0.01);
+ Assert.assertEquals(32, hClip.getMax(), 0.01);
+ Assert.assertEquals(0, hClip.getMissingValueCount());
+ Assert.assertEquals(0, hClip.getLowerOutlierCount());
+ Assert.assertEquals(0, hClip.getUpperOutlierCount());
+
+ hIgnore.combineHistogram(h2);
+ Assert.assertEquals(5, hIgnore.getNumBuckets());
+ Assert.assertEquals(4.0, hIgnore.getBucketSize(), 0.01);
+ Assert.assertEquals(12, hIgnore.getLowerLimit(), 0.01);
+ Assert.assertEquals(32, hIgnore.getUpperLimit(), 0.01);
+ Assert.assertEquals(FixedBucketsHistogram.OutlierHandlingMode.IGNORE, hIgnore.getOutlierHandlingMode());
+ Assert.assertArrayEquals(new long[]{2, 2, 0, 1, 1}, hIgnore.getHistogram());
+ Assert.assertEquals(6, hIgnore.getCount());
+ Assert.assertEquals(13, hIgnore.getMin(), 0.01);
+ Assert.assertEquals(29, hIgnore.getMax(), 0.01);
+ Assert.assertEquals(0, hIgnore.getMissingValueCount());
+ Assert.assertEquals(0, hIgnore.getLowerOutlierCount());
+ Assert.assertEquals(0, hIgnore.getUpperOutlierCount());
+ }
+
+ @Test
+ public void testMergeDifferentBucketsContainsOther()
+ {
+ FixedBucketsHistogram h = buildHistogram(
+ 0,
+ 50,
+ 10,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW,
+ new float[]{2, 18, 34, 48}
+ );
+
+ FixedBucketsHistogram hClip = buildHistogram(
+ 0,
+ 50,
+ 10,
+ FixedBucketsHistogram.OutlierHandlingMode.CLIP,
+ new float[]{2, 18, 34, 48}
+ );
+
+ FixedBucketsHistogram hIgnore = buildHistogram(
+ 0,
+ 50,
+ 10,
+ FixedBucketsHistogram.OutlierHandlingMode.IGNORE,
+ new float[]{2, 18, 34, 48}
+ );
+
+
+ FixedBucketsHistogram h2 = buildHistogram(
+ 10,
+ 30,
+ 7,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW,
+ new float[]{11, 15, 21, 21, 29, -99, 100}
+ );
+
+ h.combineHistogram(h2);
+ Assert.assertEquals(10, h.getNumBuckets());
+ Assert.assertEquals(5.0, h.getBucketSize(), 0.01);
+ Assert.assertEquals(0, h.getLowerLimit(), 0.01);
+ Assert.assertEquals(50, h.getUpperLimit(), 0.01);
+ Assert.assertEquals(FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW, h.getOutlierHandlingMode());
+ Assert.assertArrayEquals(new long[]{1, 0, 2, 2, 1, 1, 1, 0, 0, 1}, h.getHistogram());
+ Assert.assertEquals(9, h.getCount());
+ Assert.assertEquals(2, h.getMin(), 0.01);
+ Assert.assertEquals(48, h.getMax(), 0.01);
+ Assert.assertEquals(0, h.getMissingValueCount());
+ Assert.assertEquals(1, h.getLowerOutlierCount());
+ Assert.assertEquals(1, h.getUpperOutlierCount());
+
+ hClip.combineHistogram(h2);
+ Assert.assertEquals(10, hClip.getNumBuckets());
+ Assert.assertEquals(5.0, hClip.getBucketSize(), 0.01);
+ Assert.assertEquals(0, hClip.getLowerLimit(), 0.01);
+ Assert.assertEquals(50, hClip.getUpperLimit(), 0.01);
+ Assert.assertEquals(FixedBucketsHistogram.OutlierHandlingMode.CLIP, hClip.getOutlierHandlingMode());
+ Assert.assertArrayEquals(new long[]{2, 0, 2, 2, 1, 1, 1, 0, 0, 2}, hClip.getHistogram());
+ Assert.assertEquals(11, hClip.getCount());
+ Assert.assertEquals(0, hClip.getMin(), 0.01);
+ Assert.assertEquals(50, hClip.getMax(), 0.01);
+ Assert.assertEquals(0, hClip.getMissingValueCount());
+ Assert.assertEquals(0, hClip.getLowerOutlierCount());
+ Assert.assertEquals(0, hClip.getUpperOutlierCount());
+
+ hIgnore.combineHistogram(h2);
+ Assert.assertEquals(10, hIgnore.getNumBuckets());
+ Assert.assertEquals(5.0, hIgnore.getBucketSize(), 0.01);
+ Assert.assertEquals(0, hIgnore.getLowerLimit(), 0.01);
+ Assert.assertEquals(50, hIgnore.getUpperLimit(), 0.01);
+ Assert.assertEquals(FixedBucketsHistogram.OutlierHandlingMode.IGNORE, hIgnore.getOutlierHandlingMode());
+ Assert.assertArrayEquals(new long[]{1, 0, 2, 2, 1, 1, 1, 0, 0, 1}, hIgnore.getHistogram());
+ Assert.assertEquals(9, hIgnore.getCount());
+ Assert.assertEquals(2, hIgnore.getMin(), 0.01);
+ Assert.assertEquals(48, hIgnore.getMax(), 0.01);
+ Assert.assertEquals(0, hIgnore.getMissingValueCount());
+ Assert.assertEquals(0, hIgnore.getLowerOutlierCount());
+ Assert.assertEquals(0, hIgnore.getUpperOutlierCount());
+ }
+
+ @Test
+ public void testMergeDifferentBucketsContainedByOther()
+ {
+ FixedBucketsHistogram h = buildHistogram(
+ 10,
+ 30,
+ 4,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW,
+ new float[]{11, 15, 21, 29}
+ );
+
+ FixedBucketsHistogram hClip = buildHistogram(
+ 10,
+ 30,
+ 4,
+ FixedBucketsHistogram.OutlierHandlingMode.CLIP,
+ new float[]{11, 15, 21, 29}
+ );
+
+ FixedBucketsHistogram hIgnore = buildHistogram(
+ 10,
+ 30,
+ 4,
+ FixedBucketsHistogram.OutlierHandlingMode.IGNORE,
+ new float[]{11, 15, 21, 29}
+ );
+
+ FixedBucketsHistogram h2 = buildHistogram(
+ 0,
+ 50,
+ 13,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW,
+ new float[]{2, 18, 34, 48}
+ );
+
+ h.combineHistogram(h2);
+ Assert.assertEquals(4, h.getNumBuckets());
+ Assert.assertEquals(5.0, h.getBucketSize(), 0.01);
+ Assert.assertEquals(10, h.getLowerLimit(), 0.01);
+ Assert.assertEquals(30, h.getUpperLimit(), 0.01);
+ Assert.assertEquals(FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW, h.getOutlierHandlingMode());
+ Assert.assertArrayEquals(new long[]{1, 2, 1, 1}, h.getHistogram());
+ Assert.assertEquals(5, h.getCount());
+ Assert.assertEquals(11, h.getMin(), 0.01);
+ Assert.assertEquals(29, h.getMax(), 0.01);
+ Assert.assertEquals(0, h.getMissingValueCount());
+ Assert.assertEquals(1, h.getLowerOutlierCount());
+ Assert.assertEquals(2, h.getUpperOutlierCount());
+
+ hClip.combineHistogram(h2);
+ Assert.assertEquals(4, hClip.getNumBuckets());
+ Assert.assertEquals(5.0, hClip.getBucketSize(), 0.01);
+ Assert.assertEquals(10, hClip.getLowerLimit(), 0.01);
+ Assert.assertEquals(30, hClip.getUpperLimit(), 0.01);
+ Assert.assertEquals(FixedBucketsHistogram.OutlierHandlingMode.CLIP, hClip.getOutlierHandlingMode());
+ Assert.assertArrayEquals(new long[]{2, 2, 1, 3}, hClip.getHistogram());
+ Assert.assertEquals(8, hClip.getCount());
+ Assert.assertEquals(10, hClip.getMin(), 0.01);
+ Assert.assertEquals(30, hClip.getMax(), 0.01);
+ Assert.assertEquals(0, hClip.getMissingValueCount());
+ Assert.assertEquals(0, hClip.getLowerOutlierCount());
+ Assert.assertEquals(0, hClip.getUpperOutlierCount());
+
+ hIgnore.combineHistogram(h2);
+ Assert.assertEquals(4, hIgnore.getNumBuckets());
+ Assert.assertEquals(5.0, hIgnore.getBucketSize(), 0.01);
+ Assert.assertEquals(10, hIgnore.getLowerLimit(), 0.01);
+ Assert.assertEquals(30, hIgnore.getUpperLimit(), 0.01);
+ Assert.assertEquals(FixedBucketsHistogram.OutlierHandlingMode.IGNORE, hIgnore.getOutlierHandlingMode());
+ Assert.assertArrayEquals(new long[]{1, 2, 1, 1}, hIgnore.getHistogram());
+ Assert.assertEquals(5, hIgnore.getCount());
+ Assert.assertEquals(11, hIgnore.getMin(), 0.01);
+ Assert.assertEquals(29, hIgnore.getMax(), 0.01);
+ Assert.assertEquals(0, hIgnore.getMissingValueCount());
+ Assert.assertEquals(0, hIgnore.getLowerOutlierCount());
+ Assert.assertEquals(0, hIgnore.getUpperOutlierCount());
+ }
+
+ @Test
+ public void testMissing()
+ {
+ FixedBucketsHistogram h = new FixedBucketsHistogram(
+ 0,
+ 200,
+ 200,
+ FixedBucketsHistogram.OutlierHandlingMode.IGNORE
+ );
+
+ h.incrementMissing();
+ h.incrementMissing();
+ Assert.assertEquals(2, h.getMissingValueCount());
+
+ FixedBucketsHistogram h2 = new FixedBucketsHistogram(
+ 0,
+ 200,
+ 200,
+ FixedBucketsHistogram.OutlierHandlingMode.IGNORE
+ );
+
+ h2.incrementMissing();
+ h2.incrementMissing();
+ h2.incrementMissing();
+ h.combineHistogram(h2);
+ Assert.assertEquals(5, h.getMissingValueCount());
+ }
+
+
+ @Test
+ public void testOutlierIgnore()
+ {
+ FixedBucketsHistogram h = new FixedBucketsHistogram(
+ 0,
+ 200,
+ 200,
+ FixedBucketsHistogram.OutlierHandlingMode.IGNORE
+ );
+
+ h.add(900);
+ h.add(300);
+ h.add(-275);
+ h.add(500);
+ h.add(-1000);
+ h.add(10);
+ h.add(199);
+
+ Assert.assertEquals(0, h.getUpperOutlierCount());
+ Assert.assertEquals(0, h.getLowerOutlierCount());
+ Assert.assertEquals(2, h.getCount());
+ Assert.assertEquals(10, h.getMin(), 0.01);
+ Assert.assertEquals(199, h.getMax(), 0.01);
+ }
+
+ @Test
+ public void testOutlierOverflow()
+ {
+ FixedBucketsHistogram h = new FixedBucketsHistogram(
+ 0,
+ 200,
+ 200,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW
+ );
+
+ h.add(900);
+ h.add(300);
+ h.add(-275);
+ h.add(500);
+ h.add(-1000);
+ h.add(10);
+ h.add(199);
+
+ Assert.assertEquals(3, h.getUpperOutlierCount());
+ Assert.assertEquals(2, h.getLowerOutlierCount());
+ Assert.assertEquals(2, h.getCount());
+ Assert.assertEquals(10, h.getMin(), 0.01);
+ Assert.assertEquals(199, h.getMax(), 0.01);
+ }
+
+ @Test
+ public void testOutlierClip()
+ {
+ FixedBucketsHistogram h = new FixedBucketsHistogram(
+ 0,
+ 200,
+ 200,
+ FixedBucketsHistogram.OutlierHandlingMode.CLIP
+ );
+
+ h.add(900);
+ h.add(300);
+ h.add(-275);
+ h.add(500);
+ h.add(-1000);
+ h.add(10);
+ h.add(199);
+
+ Assert.assertEquals(0, h.getUpperOutlierCount());
+ Assert.assertEquals(0, h.getLowerOutlierCount());
+ Assert.assertEquals(7, h.getCount());
+ Assert.assertEquals(0, h.getMin(), 0.01);
+ Assert.assertEquals(200, h.getMax(), 0.01);
+ }
+
+ @Test
+ public void testSerdeFullHistogram()
+ {
+ FixedBucketsHistogram hFull = new FixedBucketsHistogram(
+ -100,
+ 100,
+ 200,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW
+ );
+
+ for (int i = -100; i < 100; i++) {
+ hFull.add(i);
+ }
+
+ hFull.incrementMissing();
+
+ byte[] fullWithHeader = hFull.toBytesFull(true);
+ byte[] sparse = hFull.toBytesSparse(hFull.getNonEmptyBucketCount());
+ String asBase64Full = hFull.toBase64();
+
+ byte[] asBytesAuto = hFull.toBytes();
+
+ Assert.assertArrayEquals(fullWithHeader, asBytesAuto);
+
+ FixedBucketsHistogram fromFullWithHeader = FixedBucketsHistogram.fromBytes(fullWithHeader);
+ Assert.assertEquals(hFull, fromFullWithHeader);
+
+ FixedBucketsHistogram fromSparse = FixedBucketsHistogram.fromBytes(sparse);
+ Assert.assertEquals(hFull, fromSparse);
+
+ FixedBucketsHistogram fromBase64 = FixedBucketsHistogram.fromBase64(asBase64Full);
+ Assert.assertEquals(hFull, fromBase64);
+ }
+
+ @Test
+ public void testSerdeSparseHistogram()
+ {
+ FixedBucketsHistogram hSparse = buildHistogram(
+ -10,
+ 200,
+ 100,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW,
+ VALUES3
+ );
+
+ hSparse.add(300);
+ hSparse.add(400);
+ hSparse.add(500);
+ hSparse.add(-300);
+ hSparse.add(-700);
+ hSparse.incrementMissing();
+
+ byte[] fullWithHeader = hSparse.toBytesFull(true);
+ byte[] sparse = hSparse.toBytesSparse(hSparse.getNonEmptyBucketCount());
+ String asBase64Full = hSparse.toBase64();
+
+ byte[] asBytesAuto = hSparse.toBytes();
+
+ Assert.assertArrayEquals(sparse, asBytesAuto);
+
+ FixedBucketsHistogram fromFullWithHeader = FixedBucketsHistogram.fromBytes(fullWithHeader);
+ Assert.assertEquals(hSparse, fromFullWithHeader);
+
+ FixedBucketsHistogram fromSparse = FixedBucketsHistogram.fromBytes(sparse);
+ Assert.assertEquals(hSparse, fromSparse);
+
+ FixedBucketsHistogram fromBase64 = FixedBucketsHistogram.fromBase64(asBase64Full);
+ Assert.assertEquals(hSparse, fromBase64);
+ }
+}
diff --git a/extensions-core/histogram/src/test/java/org/apache/druid/query/aggregation/histogram/FixedBucketsHistogramTopNQueryTest.java b/extensions-core/histogram/src/test/java/org/apache/druid/query/aggregation/histogram/FixedBucketsHistogramTopNQueryTest.java
new file mode 100644
index 0000000..859b052
--- /dev/null
+++ b/extensions-core/histogram/src/test/java/org/apache/druid/query/aggregation/histogram/FixedBucketsHistogramTopNQueryTest.java
@@ -0,0 +1,247 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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.apache.druid.query.aggregation.histogram;
+
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
+import org.apache.druid.collections.CloseableStupidPool;
+import org.apache.druid.java.util.common.DateTimes;
+import org.apache.druid.java.util.common.io.Closer;
+import org.apache.druid.query.QueryPlus;
+import org.apache.druid.query.QueryRunner;
+import org.apache.druid.query.QueryRunnerTestHelper;
+import org.apache.druid.query.Result;
+import org.apache.druid.query.TestQueryRunners;
+import org.apache.druid.query.aggregation.DoubleMaxAggregatorFactory;
+import org.apache.druid.query.aggregation.DoubleMinAggregatorFactory;
+import org.apache.druid.query.topn.TopNQuery;
+import org.apache.druid.query.topn.TopNQueryBuilder;
+import org.apache.druid.query.topn.TopNQueryConfig;
+import org.apache.druid.query.topn.TopNQueryQueryToolChest;
+import org.apache.druid.query.topn.TopNQueryRunnerFactory;
+import org.apache.druid.query.topn.TopNResultValue;
+import org.apache.druid.segment.TestHelper;
+import org.junit.AfterClass;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+@RunWith(Parameterized.class)
+public class FixedBucketsHistogramTopNQueryTest
+{
+ private static final Closer resourceCloser = Closer.create();
+
+ @AfterClass
+ public static void teardown() throws IOException
+ {
+ resourceCloser.close();
+ }
+
+ @Parameterized.Parameters(name = "{0}")
+ public static Iterable<Object[]> constructorFeeder()
+ {
+ final CloseableStupidPool<ByteBuffer> defaultPool = TestQueryRunners.createDefaultNonBlockingPool();
+ final CloseableStupidPool<ByteBuffer> customPool = new CloseableStupidPool<>(
+ "TopNQueryRunnerFactory-bufferPool",
+ () -> ByteBuffer.allocate(2000)
+ );
+ resourceCloser.register(defaultPool);
+ resourceCloser.register(customPool);
+
+ return QueryRunnerTestHelper.transformToConstructionFeeder(
+ Iterables.concat(
+ QueryRunnerTestHelper.makeQueryRunners(
+ new TopNQueryRunnerFactory(
+ defaultPool,
+ new TopNQueryQueryToolChest(
+ new TopNQueryConfig(),
+ QueryRunnerTestHelper.NoopIntervalChunkingQueryRunnerDecorator()
+ ),
+ QueryRunnerTestHelper.NOOP_QUERYWATCHER
+ )
+ ),
+ QueryRunnerTestHelper.makeQueryRunners(
+ new TopNQueryRunnerFactory(
+ customPool,
+ new TopNQueryQueryToolChest(
+ new TopNQueryConfig(),
+ QueryRunnerTestHelper.NoopIntervalChunkingQueryRunnerDecorator()
+ ),
+ QueryRunnerTestHelper.NOOP_QUERYWATCHER
+ )
+ )
+ )
+ );
+ }
+
+ private final QueryRunner runner;
+
+ public FixedBucketsHistogramTopNQueryTest(
+ QueryRunner runner
+ )
+ {
+ this.runner = runner;
+ }
+
+ @Test
+ public void testTopNWithFixedHistogramAgg()
+ {
+ FixedBucketsHistogramAggregatorFactory factory = new FixedBucketsHistogramAggregatorFactory(
+ "histo",
+ "index",
+ 10,
+ 0,
+ 2000,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW
+ );
+
+ TopNQuery query = new TopNQueryBuilder()
+ .dataSource(QueryRunnerTestHelper.dataSource)
+ .granularity(QueryRunnerTestHelper.allGran)
+ .dimension(QueryRunnerTestHelper.marketDimension)
+ .metric(QueryRunnerTestHelper.dependentPostAggMetric)
+ .threshold(4)
+ .intervals(QueryRunnerTestHelper.fullOnInterval)
+ .aggregators(
+ Lists.newArrayList(
+ Iterables.concat(
+ QueryRunnerTestHelper.commonDoubleAggregators,
+ Lists.newArrayList(
+ new DoubleMaxAggregatorFactory("maxIndex", "index"),
+ new DoubleMinAggregatorFactory("minIndex", "index"),
+ factory
+ )
+ )
+ )
+ )
+ .postAggregators(
+ Arrays.asList(
+ QueryRunnerTestHelper.addRowsIndexConstant,
+ QueryRunnerTestHelper.dependentPostAgg,
+ new QuantilePostAggregator("quantile", "histo", 0.5f)
+ )
+ )
+ .build();
+
+ List<Result<TopNResultValue>> expectedResults = Collections.singletonList(
+ new Result<TopNResultValue>(
+ DateTimes.of("2011-01-12T00:00:00.000Z"),
+ new TopNResultValue(
+ Arrays.<Map<String, Object>>asList(
+ ImmutableMap.<String, Object>builder()
+ .put(QueryRunnerTestHelper.marketDimension, "total_market")
+ .put("rows", 186L)
+ .put("index", 215679.82879638672D)
+ .put("addRowsIndexConstant", 215866.82879638672D)
+ .put(QueryRunnerTestHelper.dependentPostAggMetric, 216053.82879638672D)
+ .put("uniques", QueryRunnerTestHelper.UNIQUES_2)
+ .put("maxIndex", 1743.9217529296875D)
+ .put("minIndex", 792.3260498046875D)
+ .put("quantile", 1135.238f)
+ .put(
+ "histo",
+ new FixedBucketsHistogram(
+ 0,
+ 2000,
+ 10,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW,
+ new long[]{0, 0, 0, 1, 21, 105, 42, 12, 5, 0},
+ 186,
+ 1743.92175,
+ 792.326066,
+ 0,
+ 0,
+ 0
+ )
+ )
+ .build(),
+ ImmutableMap.<String, Object>builder()
+ .put(QueryRunnerTestHelper.marketDimension, "upfront")
+ .put("rows", 186L)
+ .put("index", 192046.1060180664D)
+ .put("addRowsIndexConstant", 192233.1060180664D)
+ .put(QueryRunnerTestHelper.dependentPostAggMetric, 192420.1060180664D)
+ .put("uniques", QueryRunnerTestHelper.UNIQUES_2)
+ .put("maxIndex", 1870.06103515625D)
+ .put("minIndex", 545.9906005859375D)
+ .put("quantile", 969.69696f)
+ .put(
+ "histo",
+ new FixedBucketsHistogram(
+ 0,
+ 2000,
+ 10,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW,
+ new long[]{0, 0, 4, 33, 66, 35, 25, 11, 10, 2},
+ 186,
+ 1870.061029,
+ 545.990623,
+ 0,
+ 0,
+ 0
+ )
+ )
+ .build(),
+ ImmutableMap.<String, Object>builder()
+ .put(QueryRunnerTestHelper.marketDimension, "spot")
+ .put("rows", 837L)
+ .put("index", 95606.57232284546D)
+ .put("addRowsIndexConstant", 96444.57232284546D)
+ .put(QueryRunnerTestHelper.dependentPostAggMetric, 97282.57232284546D)
+ .put("uniques", QueryRunnerTestHelper.UNIQUES_9)
+ .put("maxIndex", 277.2735290527344D)
+ .put("minIndex", 59.02102279663086D)
+ .put("quantile", 100.23952f)
+ .put(
+ "histo",
+ new FixedBucketsHistogram(
+ 0,
+ 2000,
+ 10,
+ FixedBucketsHistogram.OutlierHandlingMode.OVERFLOW,
+ new long[]{835, 2, 0, 0, 0, 0, 0, 0, 0, 0},
+ 837,
+ 277.273533,
+ 59.021022,
+ 0,
+ 0,
+ 0
+ )
+ )
+ .build()
+ )
+ )
+ )
+ );
+ HashMap<String, Object> context = new HashMap<String, Object>();
+
+ List<Result<TopNResultValue>> results = runner.run(QueryPlus.wrap(query), context).toList();
+ TestHelper.assertExpectedResults(expectedResults, results);
+ }
+}
diff --git a/processing/src/main/java/org/apache/druid/query/aggregation/AggregatorUtil.java b/processing/src/main/java/org/apache/druid/query/aggregation/AggregatorUtil.java
index 1d291b7..32003c6 100644
--- a/processing/src/main/java/org/apache/druid/query/aggregation/AggregatorUtil.java
+++ b/processing/src/main/java/org/apache/druid/query/aggregation/AggregatorUtil.java
@@ -110,6 +110,9 @@ public class AggregatorUtil
public static final byte HLL_SKETCH_TO_STRING_CACHE_TYPE_ID = 0x31;
public static final byte HLL_SKETCH_TO_ESTIMATE_AND_BOUNDS_CACHE_TYPE_ID = 0x32;
+ // Fixed buckets histogram aggregator
+ public static final byte FIXED_BUCKET_HIST_CACHE_TYPE_ID = 0x33;
+
/**
* returns the list of dependent postAggregators that should be calculated in order to calculate given postAgg
*
diff --git a/processing/src/main/java/org/apache/druid/segment/incremental/IncrementalIndex.java b/processing/src/main/java/org/apache/druid/segment/incremental/IncrementalIndex.java
index e7700cf..f20484e 100644
--- a/processing/src/main/java/org/apache/druid/segment/incremental/IncrementalIndex.java
+++ b/processing/src/main/java/org/apache/druid/segment/incremental/IncrementalIndex.java
@@ -194,7 +194,7 @@ public abstract class IncrementalIndex<AggregatorType> extends AbstractIndex imp
@Override
public Object getObject()
{
- return extractor.extractValue(in.get(), column);
+ return extractor.extractValue(in.get(), column, agg);
}
@Override
diff --git a/processing/src/main/java/org/apache/druid/segment/serde/ComplexMetricExtractor.java b/processing/src/main/java/org/apache/druid/segment/serde/ComplexMetricExtractor.java
index 6e237f3..9eca971 100644
--- a/processing/src/main/java/org/apache/druid/segment/serde/ComplexMetricExtractor.java
+++ b/processing/src/main/java/org/apache/druid/segment/serde/ComplexMetricExtractor.java
@@ -21,6 +21,7 @@ package org.apache.druid.segment.serde;
import org.apache.druid.data.input.InputRow;
import org.apache.druid.guice.annotations.ExtensionPoint;
+import org.apache.druid.query.aggregation.AggregatorFactory;
import javax.annotation.Nullable;
@@ -33,4 +34,10 @@ public interface ComplexMetricExtractor<T>
@Nullable
T extractValue(InputRow inputRow, String metricName);
+
+ @Nullable
+ default T extractValue(InputRow inputRow, String metricName, AggregatorFactory agg)
+ {
+ return extractValue(inputRow, metricName);
+ }
}
diff --git a/processing/src/main/java/org/apache/druid/segment/serde/LargeColumnSupportedComplexColumnSerializer.java b/processing/src/main/java/org/apache/druid/segment/serde/LargeColumnSupportedComplexColumnSerializer.java
index 8e4f2c2..bd68aef 100644
--- a/processing/src/main/java/org/apache/druid/segment/serde/LargeColumnSupportedComplexColumnSerializer.java
+++ b/processing/src/main/java/org/apache/druid/segment/serde/LargeColumnSupportedComplexColumnSerializer.java
@@ -116,5 +116,4 @@ public class LargeColumnSupportedComplexColumnSerializer<T> implements GenericCo
{
writer.writeTo(channel, smoosher);
}
-
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org