Feature/replace avl digest with merging digest

docs/changelog/35182.yaml

@@ -0,0 +1,5 @@
+pr: 35182
+summary: Feature/replace avl digest with merging digest
+area: Aggregations
+type: feature
+issues: []

gradle/verification-metadata.xml

@@ -919,6 +919,11 @@
             <sha256 value="03db291a8887b474f90db67bfb1f92d084e990150037e231babbb374ee11d7c3" origin="Generated by Gradle"/>
          </artifact>
       </component>
+      <component group="com.tdunning" name="t-digest" version="3.3">
+         <artifact name="t-digest-3.3.jar">
+            <sha256 value="dc8be5228e0733e12fbe35eeffb6f720dff576bb2eb5dee69c1c211adcd7aeb9" origin="Generated by Gradle"/>
+         </artifact>
+      </component>
       <component group="com.thoughtworks.paranamer" name="paranamer" version="2.3">
          <artifact name="paranamer-2.3.jar">
             <sha256 value="e93f50ae4d0de11080677f44ab268691266fed2b3ff7bc6fd97636febae7d8fe" origin="Generated by Gradle"/>

server/build.gradle

@@ -57,7 +57,7 @@ dependencies {
   implementation 'com.carrotsearch:hppc:0.8.1'
 
   // percentiles aggregation
-  api 'com.tdunning:t-digest:3.2'
+  api 'com.tdunning:t-digest:3.3'
   // precentil ranks aggregation
   api 'org.hdrhistogram:HdrHistogram:2.1.9'
 

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

@@ -33,11 +33,6 @@ public void add(List<? extends TDigest> others) {
         throw new UnsupportedOperationException("Immutable Empty TDigest");
     }
 
-    @Override
-    public void add(double x, int w, List<Double> data) {
-        throw new UnsupportedOperationException("Immutable Empty TDigest");
-    }
-
     @Override
     public void compress() {
         throw new UnsupportedOperationException("Immutable Empty TDigest");

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

@@ -7,8 +7,8 @@
  */
 package org.elasticsearch.search.aggregations.metrics;
 
-import com.tdunning.math.stats.AVLTreeDigest;
 import com.tdunning.math.stats.Centroid;
+import com.tdunning.math.stats.MergingDigest;
 
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
@@ -19,7 +19,7 @@
 /**
  * Extension of {@link com.tdunning.math.stats.TDigest} with custom serialization.
  */
-public class TDigestState extends AVLTreeDigest {
+public class TDigestState extends MergingDigest {
 
     private final double compression;
 
@@ -34,6 +34,7 @@ public double compression() {
     }
 
     public static void write(TDigestState state, StreamOutput out) throws IOException {
+        state.compress();
         out.writeDouble(state.compression);
         out.writeVInt(state.centroidCount());
         for (Centroid centroid : state.centroids()) {

server/src/test/java/org/elasticsearch/search/aggregations/metrics/EmptyTDigestStateTests.java

@@ -36,12 +36,6 @@ public void testCompress() {
         expectThrows(UnsupportedOperationException.class, singleton::compress);
     }
 
-    public void testTestAddList() {
-        expectThrows(
-            UnsupportedOperationException.class,
-            () -> singleton.add(randomDouble(), randomInt(10), List.of(randomDouble(), randomDouble()))
-        );
-    }
 
     public void testTestAddListTDigest() {
         expectThrows(UnsupportedOperationException.class, () -> singleton.add(List.of(new EmptyTDigestState(), new EmptyTDigestState())));

server/src/test/java/org/elasticsearch/search/aggregations/metrics/InternalTDigestPercentilesRanksTests.java

@@ -34,7 +34,7 @@ protected InternalTDigestPercentileRanks createTestInstance(
         final TDigestState state = new TDigestState(100);
         Arrays.stream(values).forEach(state::add);
 
-        assertEquals(state.centroidCount(), values.length);
+        assertEquals(state.size(), values.length);
         return new InternalTDigestPercentileRanks(name, percents, state, keyed, format, metadata);
     }
 

server/src/test/java/org/elasticsearch/search/aggregations/metrics/InternalTDigestPercentilesTests.java

@@ -35,7 +35,7 @@ protected InternalTDigestPercentiles createTestInstance(
         final TDigestState state = new TDigestState(100);
         Arrays.stream(values).forEach(state::add);
 
-        assertEquals(state.centroidCount(), values.length);
+        assertEquals(state.size(), values.length);
         return new InternalTDigestPercentiles(name, percents, state, keyed, format, metadata);
     }
 

server/src/test/java/org/elasticsearch/search/aggregations/metrics/TDigestPercentileRanksAggregatorTests.java

@@ -74,18 +74,14 @@ public void testSimple() throws IOException {
                 Iterator<Percentile> rankIterator = ranks.iterator();
                 Percentile rank = rankIterator.next();
                 assertEquals(0.1, rank.getValue(), 0d);
-                // TODO: Fix T-Digest: this assertion should pass but we currently get ~15
-                // https://github.com/elastic/elasticsearch/issues/14851
-                // assertThat(rank.getPercent(), Matchers.equalTo(0d));
+                assertThat(rank.getPercent(), Matchers.equalTo(0d));
                 rank = rankIterator.next();
                 assertEquals(0.5, rank.getValue(), 0d);
                 assertThat(rank.getPercent(), Matchers.greaterThan(0d));
                 assertThat(rank.getPercent(), Matchers.lessThan(100d));
                 rank = rankIterator.next();
                 assertEquals(12, rank.getValue(), 0d);
-                // TODO: Fix T-Digest: this assertion should pass but we currently get ~59
-                // https://github.com/elastic/elasticsearch/issues/14851
-                // assertThat(rank.getPercent(), Matchers.equalTo(100d));
+                assertThat(rank.getPercent(), Matchers.equalTo(100d));
                 assertFalse(rankIterator.hasNext());
                 assertTrue(AggregationInspectionHelper.hasValue(((InternalTDigestPercentileRanks) ranks)));
             }

server/src/test/java/org/elasticsearch/search/aggregations/metrics/TDigestPercentilesAggregatorTests.java

@@ -75,13 +75,13 @@ public void testSomeMatchesSortedNumericDocValues() throws IOException {
             iw.addDocument(singleton(new SortedNumericDocValuesField("number", 0)));
         }, tdigest -> {
             assertEquals(7L, tdigest.state.size());
-            assertEquals(7L, tdigest.state.centroidCount());
             assertEquals(4.5d, tdigest.percentile(75), 0.0d);
             assertEquals("4.5", tdigest.percentileAsString(75));
             assertEquals(2.0d, tdigest.percentile(50), 0.0d);
             assertEquals("2.0", tdigest.percentileAsString(50));
             assertEquals(1.0d, tdigest.percentile(22), 0.0d);
             assertEquals("1.0", tdigest.percentileAsString(22));
+            assertEquals(7L, tdigest.state.centroidCount());
             assertTrue(AggregationInspectionHelper.hasValue(tdigest));
         });
     }
@@ -97,7 +97,6 @@ public void testSomeMatchesNumericDocValues() throws IOException {
             iw.addDocument(singleton(new NumericDocValuesField("number", 0)));
         }, tdigest -> {
             assertEquals(tdigest.state.size(), 7L);
-            assertEquals(tdigest.state.centroidCount(), 7L);
             assertEquals(8.0d, tdigest.percentile(100), 0.0d);
             assertEquals("8.0", tdigest.percentileAsString(100));
             assertEquals(6.98d, tdigest.percentile(88), 0.0d);
@@ -108,6 +107,7 @@ public void testSomeMatchesNumericDocValues() throws IOException {
             assertEquals("1.0", tdigest.percentileAsString(25));
             assertEquals(0.0d, tdigest.percentile(1), 0.0d);
             assertEquals("0.0", tdigest.percentileAsString(1));
+            assertEquals(tdigest.state.centroidCount(), 7L);
             assertTrue(AggregationInspectionHelper.hasValue(tdigest));
         });
     }
@@ -125,10 +125,10 @@ public void testQueryFiltering() throws IOException {
 
         testCase(LongPoint.newRangeQuery("row", 1, 4), docs, tdigest -> {
             assertEquals(4L, tdigest.state.size());
-            assertEquals(4L, tdigest.state.centroidCount());
             assertEquals(2.0d, tdigest.percentile(100), 0.0d);
             assertEquals(1.0d, tdigest.percentile(50), 0.0d);
             assertEquals(0.5d, tdigest.percentile(25), 0.0d);
+            assertEquals(4L, tdigest.state.centroidCount());
             assertTrue(AggregationInspectionHelper.hasValue(tdigest));
         });
 

server/src/test/java/org/elasticsearch/search/aggregations/metrics/TDigestStateTests.java

@@ -130,4 +130,26 @@ public void testTDigestStateHash() {
         assertTrue(set.stream().anyMatch(digest -> digest.equals(empty1)));
         assertTrue(set.stream().anyMatch(digest -> digest.equals(empty2)));
     }
+
+    public void testBasicAssumptions() {
+        int size = 7;
+        TDigestState[] states = new TDigestState[size];
+        for (int counter = 0; counter < size; counter++) {
+            states[counter] = new TDigestState(50);
+            final int numberDocs = randomIntBetween(5, 50);
+            for (int i = 0; i < numberDocs; i++) {
+                double value = randomDoubleBetween(-1000, 1000, true);
+                states[counter].add(value);
+            }
+        }
+
+        int fromIndex = 0;
+        int toIndex = 6;
+        TDigestState result = new TDigestState(states[0].compression());
+        for (int i = fromIndex; i < toIndex; i++) {
+            result.add(states[i]);
+        }
+        assertNotEquals(Double.NEGATIVE_INFINITY, result.getMax());
+        assertTrue(result.getMax() > result.getMin());
+    }
 }