Fix decay for quantile digests which are more frequently read than modified

Author: tdcmeehan

stats/src/main/java/com/facebook/airlift/stats/QuantileDigest.java

@@ -32,6 +32,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.LongBinaryOperator;
 
 import static com.facebook.airlift.stats.QuantileDigest.MiddleFunction.DEFAULT;
 import static com.google.common.base.Preconditions.checkArgument;
@@ -328,6 +329,14 @@ public void scale(double scaleFactor)
      */
     public List<Long> getQuantilesLowerBound(List<Double> quantiles)
     {
+        if (alpha > 0.0) {
+            long nowInSeconds = TimeUnit.NANOSECONDS.toSeconds(ticker.read());
+            if (nowInSeconds - landmarkInSeconds >= RESCALE_THRESHOLD_SECONDS) {
+                rescale(nowInSeconds);
+                compress(); // rescale affects weights globally, so force compression
+            }
+        }
+
         checkArgument(Ordering.natural().isOrdered(quantiles), "quantiles must be sorted in increasing order");
         for (double quantile : quantiles) {
             checkArgument(quantile >= 0 && quantile <= 1, "quantile must be between [0,1]");
@@ -379,6 +388,14 @@ public boolean process(int node)
      */
     public List<Long> getQuantilesUpperBound(List<Double> quantiles)
     {
+        if (alpha > 0.0) {
+            long nowInSeconds = TimeUnit.NANOSECONDS.toSeconds(ticker.read());
+            if (nowInSeconds - landmarkInSeconds >= RESCALE_THRESHOLD_SECONDS) {
+                rescale(nowInSeconds);
+                compress(); // rescale affects weights globally, so force compression
+            }
+        }
+
         checkArgument(Ordering.natural().isOrdered(quantiles), "quantiles must be sorted in increasing order");
         for (double quantile : quantiles) {
             checkArgument(quantile >= 0 && quantile <= 1, "quantile must be between [0,1]");
@@ -411,9 +428,12 @@ public boolean process(int node)
 
         // we finished the traversal without consuming all quantiles. This means the remaining quantiles
         // correspond to the max known value
-        while (iterator.hasNext()) {
-            builder.add(max);
-            iterator.next();
+        if (iterator.hasNext()) {
+            long max = getMax();
+            while (iterator.hasNext()) {
+                builder.add(max);
+                iterator.next();
+            }
         }
 
         return builder.build();
@@ -624,7 +644,13 @@ void compress()
     {
         double bound = Math.floor(weightedCount / calculateCompressionFactor());
 
+        AtomicLong max = new AtomicLong(Integer.MIN_VALUE);
+        AtomicLong min = new AtomicLong(Integer.MAX_VALUE);
         postOrderTraversal(root, node -> {
+            if (counts[node] >= ZERO_WEIGHT_THRESHOLD) {
+                max.accumulateAndGet(upperBound(node), Math::max);
+                min.accumulateAndGet(lowerBound(node), Math::min);
+            }
             // if children's weights are 0 remove them and shift the weight to their parent
             int left = lefts[node];
             int right = rights[node];
@@ -655,6 +681,15 @@ void compress()
         // root's count may have decayed to ~0
         if (root != -1 && counts[root] < ZERO_WEIGHT_THRESHOLD) {
             root = tryRemove(root);
+            if (root < 0) {
+                this.min = Long.MAX_VALUE;
+                this.max = Long.MIN_VALUE;
+            }
+            // If decay is being used, the min and max values may have decayed as well
+            else if (alpha > 0) {
+                this.min = min.get();
+                this.max = max.get();
+            }
         }
     }
 

stats/src/test/java/com/facebook/airlift/stats/TestQuantileDigest.java

@@ -433,6 +433,31 @@ public void testDecayedCountsWithClockIncrementSmallerThanRescaleThreshold()
         assertEquals(digest.getCount(), 15.0);
     }
 
+    @Test
+    public void testDecayedQuantilesWithNoMergeOrAdd()
+            throws Exception
+    {
+        TestingTicker ticker = new TestingTicker();
+        QuantileDigest digest = new QuantileDigest(0.01, ExponentialDecay.computeAlpha(0.5, 60), ticker);
+
+        addAll(digest, asList(0, 1, 2, 3, 4, 5, 6, 7, 8, 9));
+
+        assertEquals(digest.getConfidenceFactor(), 0.0);
+        assertEquals(digest.getQuantile(0.5), 5);
+
+        // Decay the digest
+        ticker.increment(60, TimeUnit.SECONDS);
+        assertEquals(digest.getQuantile(0.5), 5);
+
+        // Allow further decay
+        ticker.increment(6, TimeUnit.MINUTES);
+        assertEquals(digest.getQuantile(0.5), 5);
+
+        // Values have decayed to 0
+        ticker.increment(60, TimeUnit.MINUTES);
+        assertEquals(digest.getQuantile(0.5), Long.MIN_VALUE); // Default value for empty digests
+    }
+
     @Test
     public void testMinMax()
             throws Exception