Make TDigestState configurable (elastic#96794)

* Initial import for TDigest forking. * Fix MedianTest. More work needed for TDigestPercentile*Tests and the TDigestTest (and the rest of the tests) in the tdigest lib to pass. * Fix Dist. * Fix AVLTreeDigest.quantile to match Dist for uniform centroids. * Update docs/changelog/96086.yaml * Fix `MergingDigest.quantile` to match `Dist` on uniform distribution. * Add merging to TDigestState.hashCode and .equals. Remove wrong asserts from tests and MergingDigest. * Fix style violations for tdigest library. * Fix typo. * Fix more style violations. * Fix more style violations. * Fix remaining style violations in tdigest library. * Update results in docs based on the forked tdigest. * Fix YAML tests in aggs module. * Fix YAML tests in x-pack/plugin. * Skip failing V7 compat tests in modules/aggregations. * Fix TDigest library unittests. Remove redundant serializing interfaces from the library. * Remove YAML test versions for older releases. These tests don't address compatibility issues in mixed cluster tests as the latter contain a mix of older and newer nodes, so the output depends on which node is picked as a data node since the forked TDigest library is not backwards compatible (produces slightly different results). * Fix test failures in docs and mixed cluster. * Reduce buffer sizes in MergingDigest to avoid oom. * Exclude more failing V7 compatibility tests. * Update results for JdbcCsvSpecIT tests. * Update results for JdbcDocCsvSpecIT tests. * Revert unrelated change. * More test fixes. * Use version skips instead of blacklisting in mixed cluster tests. * Switch TDigestState back to AVLTreeDigest. * Update docs and tests with AVLTreeDigest output. * Update flaky test. * Remove dead code, esp around tracking of incoming data. * Update docs/changelog/96086.yaml * Delete docs/changelog/96086.yaml * Remove explicit compression calls. This was added to prevent concurrency tests from failing, but it leads to reduces precision. Submit this to see if the concurrency tests are still failing. * Revert "Remove explicit compression calls." This reverts commit 5352c96. * Remove explicit compression calls to MedianAbsoluteDeviation input. * Add unittests for AVL and merging digest accuracy. * Fix spotless violations. * Delete redundant tests and benchmarks. * Fix spotless violation. * Use the old implementation of AVLTreeDigest. The latest library version is 50% slower and less accurate, as verified by ComparisonTests. * Update docs with latest percentile results. * Update docs with latest percentile results. * Remove repeated compression calls. * Update more percentile results. * Use approximate percentile values in integration tests. This helps with mixed cluster tests, where some of the tests where blocked. * Fix expected percentile value in test. * Revert in-place node updates in AVL tree. Update quantile calculations between centroids and min/max values to match v.3.2. * Add SortingDigest and HybridDigest. The SortingDigest tracks all samples in an ArrayList that gets sorted for quantile calculations. This approach provides perfectly accurate results and is the most efficient implementation for up to millions of samples, at the cost of bloated memory footprint. The HybridDigest uses a SortingDigest for small sample populations, then switches to a MergingDigest. This approach combines to the best performance and results for small sample counts with very good performance and acceptable accuracy for effectively unbounded sample counts. * Remove deps to the 3.2 library. * Remove unused licenses for tdigest. * Revert changes for SortingDigest and HybridDigest. These will be submitted in a follow-up PR for enabling MergingDigest. * Remove unused Histogram classes and unit tests. Delete dead and commented out code, make the remaining tests run reasonably fast. Remove unused annotations, esp. SuppressWarnings. * Remove Comparison class, not used. * Revert "Revert changes for SortingDigest and HybridDigest." This reverts commit 2336b11. * Use HybridDigest as default tdigest implementation Add SortingDigest as a simple structure for percentile calculations that tracks all data points in a sorted array. This is a fast and perfectly accurate solution that leads to bloated memory allocation. Add HybridDigest that uses SortingDigest for small sample counts, then switches to MergingDigest. This approach delivers extreme performance and accuracy for small populations while scaling indefinitely and maintaining acceptable performance and accuracy with constant memory allocation (15kB by default). Provide knobs to switch back to AVLTreeDigest, either per query or through ClusterSettings. * Small fixes. * Add javadoc and tests. * Add javadoc and tests. * Remove special logic for singletons in the boundaries. While this helps with the case where the digest contains only singletons (perfect accuracy), it has a major issue problem (non-monotonic quantile function) when the first singleton is followed by a non-singleton centroid. It's preferable to revert to the old version from 3.2; inaccuracies in a singleton-only digest should be mitigated by using a sorted array for small sample counts. * Revert changes to expected values in tests. This is due to restoring quantile functions to match head. * Revert changes to expected values in tests. This is due to restoring quantile functions to match head. * Tentatively restore percentile rank expected results. * Use cdf version from 3.2 Update Dist.cdf to use interpolation, use the same cdf version in AVLTreeDigest and MergingDigest. * Revert "Tentatively restore percentile rank expected results." This reverts commit 7718dbb. * Revert remaining changes compared to main. * Revert excluded V7 compat tests. * Exclude V7 compat tests still failing. * Exclude V7 compat tests still failing. * Remove ClusterSettings tentatively. * Initial import for TDigest forking. * Fix MedianTest. More work needed for TDigestPercentile*Tests and the TDigestTest (and the rest of the tests) in the tdigest lib to pass. * Fix Dist. * Fix AVLTreeDigest.quantile to match Dist for uniform centroids. * Update docs/changelog/96086.yaml * Fix `MergingDigest.quantile` to match `Dist` on uniform distribution. * Add merging to TDigestState.hashCode and .equals. Remove wrong asserts from tests and MergingDigest. * Fix style violations for tdigest library. * Fix typo. * Fix more style violations. * Fix more style violations. * Fix remaining style violations in tdigest library. * Update results in docs based on the forked tdigest. * Fix YAML tests in aggs module. * Fix YAML tests in x-pack/plugin. * Skip failing V7 compat tests in modules/aggregations. * Fix TDigest library unittests. Remove redundant serializing interfaces from the library. * Remove YAML test versions for older releases. These tests don't address compatibility issues in mixed cluster tests as the latter contain a mix of older and newer nodes, so the output depends on which node is picked as a data node since the forked TDigest library is not backwards compatible (produces slightly different results). * Fix test failures in docs and mixed cluster. * Reduce buffer sizes in MergingDigest to avoid oom. * Exclude more failing V7 compatibility tests. * Update results for JdbcCsvSpecIT tests. * Update results for JdbcDocCsvSpecIT tests. * Revert unrelated change. * More test fixes. * Use version skips instead of blacklisting in mixed cluster tests. * Switch TDigestState back to AVLTreeDigest. * Update docs and tests with AVLTreeDigest output. * Update flaky test. * Remove dead code, esp around tracking of incoming data. * Remove explicit compression calls. This was added to prevent concurrency tests from failing, but it leads to reduces precision. Submit this to see if the concurrency tests are still failing. * Update docs/changelog/96086.yaml * Delete docs/changelog/96086.yaml * Revert "Remove explicit compression calls." This reverts commit 5352c96. * Remove explicit compression calls to MedianAbsoluteDeviation input. * Add unittests for AVL and merging digest accuracy. * Fix spotless violations. * Delete redundant tests and benchmarks. * Fix spotless violation. * Use the old implementation of AVLTreeDigest. The latest library version is 50% slower and less accurate, as verified by ComparisonTests. * Update docs with latest percentile results. * Update docs with latest percentile results. * Remove repeated compression calls. * Update more percentile results. * Use approximate percentile values in integration tests. This helps with mixed cluster tests, where some of the tests where blocked. * Fix expected percentile value in test. * Revert in-place node updates in AVL tree. Update quantile calculations between centroids and min/max values to match v.3.2. * Add SortingDigest and HybridDigest. The SortingDigest tracks all samples in an ArrayList that gets sorted for quantile calculations. This approach provides perfectly accurate results and is the most efficient implementation for up to millions of samples, at the cost of bloated memory footprint. The HybridDigest uses a SortingDigest for small sample populations, then switches to a MergingDigest. This approach combines to the best performance and results for small sample counts with very good performance and acceptable accuracy for effectively unbounded sample counts. * Remove deps to the 3.2 library. * Remove unused licenses for tdigest. * Revert changes for SortingDigest and HybridDigest. These will be submitted in a follow-up PR for enabling MergingDigest. * Remove unused Histogram classes and unit tests. Delete dead and commented out code, make the remaining tests run reasonably fast. Remove unused annotations, esp. SuppressWarnings. * Remove Comparison class, not used. * Revert "Revert changes for SortingDigest and HybridDigest." This reverts commit 2336b11. * Use HybridDigest as default tdigest implementation Add SortingDigest as a simple structure for percentile calculations that tracks all data points in a sorted array. This is a fast and perfectly accurate solution that leads to bloated memory allocation. Add HybridDigest that uses SortingDigest for small sample counts, then switches to MergingDigest. This approach delivers extreme performance and accuracy for small populations while scaling indefinitely and maintaining acceptable performance and accuracy with constant memory allocation (15kB by default). Provide knobs to switch back to AVLTreeDigest, either per query or through ClusterSettings. * Add javadoc and tests. * Remove ClusterSettings tentatively. * Restore bySize function in TDigest and subclasses. * Update Dist.cdf to match the rest. Update tests. * Revert outdated test changes. * Revert outdated changes. * Small fixes. * Update docs/changelog/96794.yaml * Make HybridDigest the default implementation. * Update boxplot documentation. * Restore AVLTreeDigest as the default in TDigestState. TDigest.createHybridDigest nw returns the right type. The switch in TDigestState will happen in a separate PR as it requires many test updates. * Use execution_hint in tdigest spec. * Fix Dist.cdf for empty digest. * Bump up TransportVersion. * Bump up TransportVersion for real. * HybridDigest uses its final implementation during deserialization. * Restore the right TransportVersion in TDigestState.read * Use TDigestExecutionHint instead of strings. * Add link to TDigest javadoc. * Spotless fix. * Small fixes. * Bump up TransportVersion. * Bump up the TransportVersion, again.
salvatore-campagna · Jun 19, 2023 · 32655b6 · 32655b6
1 parent a994a92
commit 32655b6
Show file tree

Hide file tree

Showing 58 changed files with 1,330 additions and 198 deletions.
diff --git a/docs/changelog/96794.yaml b/docs/changelog/96794.yaml
@@ -0,0 +1,5 @@
+pr: 96794
+summary: Make TDigestState configurable
+area: Aggregations
+type: enhancement
+issues: []
diff --git a/docs/reference/aggregations/metrics/boxplot-aggregation.asciidoc b/docs/reference/aggregations/metrics/boxplot-aggregation.asciidoc
@@ -166,6 +166,35 @@ GET latency/_search
 
 include::percentile-aggregation.asciidoc[tags=t-digest]
 
+==== Execution hint
+
+The default implementation of TDigest is optimized for performance, scaling to millions or even
+billions of sample values while maintaining acceptable accuracy levels (close to 1% relative error
+for millions of samples in some cases). There's an option to use an implementation optimized
+for accuracy by setting parameter `execution_hint` to value `high_accuracy`:
+
+[source,console]
+--------------------------------------------------
+GET latency/_search
+{
+  "size": 0,
+  "aggs": {
+    "load_time_boxplot": {
+      "boxplot": {
+        "field": "load_time",
+        "execution_hint": "high_accuracy"    <1>
+      }
+    }
+  }
+}
+--------------------------------------------------
+// TEST[setup:latency]
+
+<1> Optimize TDigest for accuracy, at the expense of performance
+
+This option can lead to improved accuracy (relative error close to 0.01% for millions of samples in some
+cases) but then percentile queries take 2x-10x longer to complete.
+
 ==== Missing value
 
 The `missing` parameter defines how documents that are missing a value should be treated.

diff --git a/docs/reference/aggregations/metrics/percentile-aggregation.asciidoc b/docs/reference/aggregations/metrics/percentile-aggregation.asciidoc
@@ -306,6 +306,38 @@ TDigest roughly 64KB in size. In practice data tends to be more random and
 the TDigest will use less memory.
 // end::t-digest[]
 
+[[search-aggregations-metrics-percentile-aggregation-execution-hint]]
+==== Execution hint
+
+The default implementation of TDigest is optimized for performance, scaling to millions or even
+billions of sample values while maintaining acceptable accuracy levels (close to 1% relative error
+for millions of samples in some cases). There's an option to use an implementation optimized
+for accuracy by setting parameter `execution_hint` to value `high_accuracy`:
+
+[source,console]
+--------------------------------------------------
+GET latency/_search
+{
+  "size": 0,
+  "aggs": {
+    "load_time_outlier": {
+      "percentiles": {
+        "field": "load_time",
+        "tdigest": {
+          "execution_hint": "high_accuracy"    <1>
+        }
+      }
+    }
+  }
+}
+--------------------------------------------------
+// TEST[setup:latency]
+
+<1> Optimize TDigest for accuracy, at the expense of performance
+
+This option can lead to improved accuracy (relative error close to 0.01% for millions of samples in some
+cases) but then percentile queries take 2x-10x longer to complete.
+
 ==== HDR Histogram
 
 https://github.com/HdrHistogram/HdrHistogram[HDR Histogram] (High Dynamic Range Histogram) is an alternative implementation

diff --git a/docs/reference/aggregations/metrics/percentile-rank-aggregation.asciidoc b/docs/reference/aggregations/metrics/percentile-rank-aggregation.asciidoc
@@ -10,9 +10,10 @@ extracted from specific numeric or <<histogram,histogram fields>> in the documen
 
 [NOTE]
 ==================================================
-Please see <<search-aggregations-metrics-percentile-aggregation-approximation>>
-and <<search-aggregations-metrics-percentile-aggregation-compression>> for advice
-regarding approximation and memory use of the percentile ranks aggregation
+Please see <<search-aggregations-metrics-percentile-aggregation-approximation>>,
+<<search-aggregations-metrics-percentile-aggregation-compression>> and
+<<search-aggregations-metrics-percentile-aggregation-execution-hint>> for advice
+regarding approximation, performance and memory use of the percentile ranks aggregation
 ==================================================
 
 Percentile rank show the percentage of observed values which are below certain

diff --git a/libs/tdigest/src/main/java/org/elasticsearch/tdigest/Dist.java b/libs/tdigest/src/main/java/org/elasticsearch/tdigest/Dist.java
@@ -30,30 +30,68 @@
 public class Dist {
 
     private static double cdf(final double x, final int length, Function<Integer, Double> elementGetter) {
+        if (length == 0) {
+            // no data to examine
+            return Double.NaN;
+        }
+        if (length == 1) {
+            double value = elementGetter.apply(0);
+            if (x < value) return 0;
+            if (x > value) return 1;
+            return 0.5;
+        }
+
         if (Double.compare(x, elementGetter.apply(0)) < 0) {
             return 0;
         }
 
-        double n1 = 0.5;
-        int n2 = 0;
-        for (int i = 1; i < length; i++) {
-            double value = elementGetter.apply(i);
-            int compareResult = Double.compare(value, x);
-            if (compareResult > 0) {
-                if (Double.compare(n2, 0) > 0) {
-                    return (n1 + 0.5 * n2) / length;
-                }
-                double previousValue = elementGetter.apply(i - 1);
-                double factor = (x - previousValue) / (value - previousValue);
-                return (n1 + factor) / length;
+        if (Double.compare(x, elementGetter.apply(0)) == 0) {
+            // we have one or more centroids == x, treat them as one
+            // dw will accumulate the weight of all of the centroids at x
+            double dw = 0;
+            for (int i = 0; i < length && Double.compare(elementGetter.apply(i), x) == 0; i++) {
+                dw += 1;
+            }
+            return dw / 2.0 / length;
+        }
+
+        if (x > elementGetter.apply(length - 1)) {
+            return 1;
+        }
+        if (x == elementGetter.apply(length - 1)) {
+            double dw = 0;
+            for (int i = length - 1; i >= 0 && Double.compare(elementGetter.apply(i), x) == 0; i--) {
+                dw += 1;
             }
-            if (compareResult < 0) {
-                n1++;
-            } else {
-                n2++;
+            return (length - dw / 2.0) / length;
+        }
+
+        // initially, we set left width equal to right width
+        double left = (elementGetter.apply(1) - elementGetter.apply(0)) / 2;
+        double weightSoFar = 0;
+
+        for (int i = 0; i < length - 1; i++) {
+            double right = (elementGetter.apply(i + 1) - elementGetter.apply(i)) / 2;
+            if (x < elementGetter.apply(i) + right) {
+                double value = (weightSoFar + AbstractTDigest.interpolate(x, elementGetter.apply(i) - left, elementGetter.apply(i) + right))
+                    / length;
+                return Math.max(value, 0.0);
             }
+            weightSoFar += 1;
+            left = right;
+        }
+
+        // for the last element, assume right width is same as left
+        int lastOffset = length - 1;
+        double right = (elementGetter.apply(lastOffset) - elementGetter.apply(lastOffset - 1)) / 2;
+        if (x < elementGetter.apply(lastOffset) + right) {
+            return (weightSoFar + AbstractTDigest.interpolate(
+                x,
+                elementGetter.apply(lastOffset) - right,
+                elementGetter.apply(lastOffset) + right
+            )) / length;
         }
-        return (length - 0.5 * n2) / length;
+        return 1;
     }
 
     public static double cdf(final double x, double[] data) {

diff --git a/libs/tdigest/src/main/java/org/elasticsearch/tdigest/HybridDigest.java b/libs/tdigest/src/main/java/org/elasticsearch/tdigest/HybridDigest.java
@@ -0,0 +1,193 @@
+/*
+ * Licensed to Elasticsearch B.V. under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.tdigest;
+
+import java.util.Collection;
+import java.util.List;
+
+/**
+ * Uses a {@link SortingDigest} implementation under the covers for small sample populations, then switches to {@link MergingDigest}.
+ * The {@link SortingDigest} is perfectly accurate and the fastest implementation for up to millions of samples, at the cost of increased
+ * memory footprint as it tracks all samples. Conversely, the {@link MergingDigest} pre-allocates its memory (tens of KBs) and provides
+ * better performance for hundreds of millions of samples and more, while accuracy stays bounded to 0.1-1% for most cases.
+ *
+ * This hybrid  approach provides the best of both worlds, i.e. speedy and accurate percentile calculations for small populations with
+ * bounded memory allocation and acceptable speed and accuracy for larger ones.
+ */
+public class HybridDigest extends AbstractTDigest {
+
+    // See MergingDigest's compression param.
+    private final double compression;
+
+    // Indicates the sample size over which it switches from SortingDigest to MergingDigest.
+    private final long maxSortingSize;
+
+    // This is set to null when the implementation switches to MergingDigest.
+    private SortingDigest sortingDigest = new SortingDigest();
+
+    // This gets initialized when the implementation switches to MergingDigest.
+    private MergingDigest mergingDigest;
+
+    /**
+     * Creates a hybrid digest that uses a {@link SortingDigest} for up to {@param maxSortingSize} samples,
+     * then switches to a {@link MergingDigest}.
+     *
+     * @param compression The compression factor for the MergingDigest
+     * @param maxSortingSize The sample size limit for switching from a {@link SortingDigest} to a {@link MergingDigest} implementation
+     */
+    HybridDigest(double compression, long maxSortingSize) {
+        this.compression = compression;
+        this.maxSortingSize = maxSortingSize;
+    }
+
+    /**
+     * Similar to the constructor above. The limit for switching from a {@link SortingDigest} to a {@link MergingDigest} implementation
+     * is calculated based on the passed compression factor.
+     *
+     * @param compression The compression factor for the MergingDigest
+     */
+    HybridDigest(double compression) {
+        // The default maxSortingSize is calculated so that the SortingDigest will have comparable size with the MergingDigest
+        // at the point where implementations switch, e.g. for default compression 100 SortingDigest allocates ~16kB and MergingDigest
+        // allocates ~15kB.
+        this(compression, Math.round(compression) * 20);
+    }
+
+    @Override
+    public void add(double x, int w) {
+        reserve(w);
+        if (mergingDigest != null) {
+            mergingDigest.add(x, w);
+        } else {
+            sortingDigest.add(x, w);
+        }
+    }
+
+    @Override
+    public void reserve(long size) {
+        if (mergingDigest != null) {
+            mergingDigest.reserve(size);
+            return;
+        }
+        // Check if we need to switch implementations.
+        assert sortingDigest != null;
+        if (sortingDigest.size() + size >= maxSortingSize) {
+            mergingDigest = new MergingDigest(compression);
+            for (double value : sortingDigest.values) {
+                mergingDigest.add(value);
+            }
+            mergingDigest.reserve(size);
+            // Release the allocated SortingDigest.
+            sortingDigest = null;
+        } else {
+            sortingDigest.reserve(size);
+        }
+    }
+
+    @Override
+    public void add(List<? extends TDigest> others) {
+        if (mergingDigest != null) {
+            mergingDigest.add(others);
+        } else {
+            sortingDigest.add(others);
+        }
+    }
+
+    @Override
+    public void compress() {
+        if (mergingDigest != null) {
+            mergingDigest.compress();
+        } else {
+            sortingDigest.compress();
+        }
+    }
+
+    @Override
+    public long size() {
+        if (mergingDigest != null) {
+            return mergingDigest.size();
+        }
+        return sortingDigest.size();
+    }
+
+    @Override
+    public double cdf(double x) {
+        if (mergingDigest != null) {
+            return mergingDigest.cdf(x);
+        }
+        return sortingDigest.cdf(x);
+    }
+
+    @Override
+    public double quantile(double q) {
+        if (mergingDigest != null) {
+            return mergingDigest.quantile(q);
+        }
+        return sortingDigest.quantile(q);
+    }
+
+    @Override
+    public Collection<Centroid> centroids() {
+        if (mergingDigest != null) {
+            return mergingDigest.centroids();
+        }
+        return sortingDigest.centroids();
+    }
+
+    @Override
+    public double compression() {
+        if (mergingDigest != null) {
+            return mergingDigest.compression();
+        }
+        return sortingDigest.compression();
+    }
+
+    @Override
+    public int centroidCount() {
+        if (mergingDigest != null) {
+            return mergingDigest.centroidCount();
+        }
+        return sortingDigest.centroidCount();
+    }
+
+    @Override
+    public double getMin() {
+        if (mergingDigest != null) {
+            return mergingDigest.getMin();
+        }
+        return sortingDigest.getMin();
+    }
+
+    @Override
+    public double getMax() {
+        if (mergingDigest != null) {
+            return mergingDigest.getMax();
+        }
+        return sortingDigest.getMax();
+    }
+
+    @Override
+    public int byteSize() {
+        if (mergingDigest != null) {
+            return mergingDigest.byteSize();
+        }
+        return sortingDigest.byteSize();
+    }
+}