Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Fix decay for quantile digests which are more frequently read than mo… #37

Open
wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Fix decay for quantile digests which are more frequently read than mo… #37

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 32 additions & 0 deletions stats/src/main/java/com/facebook/airlift/stats/QuantileDigest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -328,6 +328,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]");
Expand Down Expand Up @@ -379,6 +387,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]");
Expand Down Expand Up @@ -624,7 +640,14 @@ 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 decay is enabled and this node has a non-zero weight, determine if the value is the new max or min
if (counts[node] >= ZERO_WEIGHT_THRESHOLD && alpha > 0.0) {
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];
Expand Down Expand Up @@ -655,6 +678,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();
}
}
}

Expand Down
25 changes: 25 additions & 0 deletions stats/src/test/java/com/facebook/airlift/stats/TestQuantileDigest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down