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

reduce overhead for group by result #1117

Merged
merged 1 commit into from
Mar 8, 2024
Merged

reduce overhead for group by result #1117

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
62 changes: 36 additions & 26 deletions spectator-reg-atlas/src/main/java/com/netflix/spectator/atlas/impl/DataExpr.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
/*
* Copyright 2014-2023 Netflix, Inc.
* Copyright 2014-2024 Netflix, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
Expand Down Expand Up @@ -116,6 +116,19 @@ interface Aggregator {
Iterable<TagsValuePair> result();
}

/** Base type for simple aggregators that have a single result value. */
interface SimpleAggregator extends Aggregator {
/** Compute the result for the aggregation. Can be used to avoid the list allocation with result. */
TagsValuePair resultPair();

@Override default Iterable<TagsValuePair> result() {
TagsValuePair pair = resultPair();
return pair == null
? Collections.emptyList()
: Collections.singletonList(pair);
}
}

/**
* Includes all datapoints that match the query expression.
*/
Expand Down Expand Up @@ -217,7 +230,7 @@ final class Sum implements AggregateFunction {
}

@Override public Aggregator aggregator(Map<String, String> tags, boolean shouldCheckQuery) {
return new Aggregator() {
return new SimpleAggregator() {
private double aggr = 0.0;
private int count = 0;

Expand All @@ -228,10 +241,8 @@ final class Sum implements AggregateFunction {
}
}

@Override public Iterable<TagsValuePair> result() {
return (count > 0)
? Collections.singletonList(new TagsValuePair(tags, aggr))
: Collections.emptyList();
@Override public TagsValuePair resultPair() {
return (count > 0) ? new TagsValuePair(tags, aggr) : null;
}
};
}
Expand Down Expand Up @@ -284,7 +295,7 @@ final class Min implements AggregateFunction {
}

@Override public Aggregator aggregator(Map<String, String> tags, boolean shouldCheckQuery) {
return new Aggregator() {
return new SimpleAggregator() {
private double aggr = Double.MAX_VALUE;
private int count = 0;

Expand All @@ -295,10 +306,8 @@ final class Min implements AggregateFunction {
}
}

@Override public Iterable<TagsValuePair> result() {
return (count > 0)
? Collections.singletonList(new TagsValuePair(tags, aggr))
: Collections.emptyList();
@Override public TagsValuePair resultPair() {
return (count > 0) ? new TagsValuePair(tags, aggr) : null;
}
};
}
Expand Down Expand Up @@ -351,7 +360,7 @@ final class Max implements AggregateFunction {
}

@Override public Aggregator aggregator(Map<String, String> tags, boolean shouldCheckQuery) {
return new Aggregator() {
return new SimpleAggregator() {
private double aggr = -Double.MAX_VALUE;
private int count = 0;

Expand All @@ -362,10 +371,8 @@ final class Max implements AggregateFunction {
}
}

@Override public Iterable<TagsValuePair> result() {
return (count > 0)
? Collections.singletonList(new TagsValuePair(tags, aggr))
: Collections.emptyList();
@Override public TagsValuePair resultPair() {
return (count > 0) ? new TagsValuePair(tags, aggr) : null;
}
};
}
Expand Down Expand Up @@ -422,7 +429,7 @@ final class Count implements AggregateFunction {
}

@Override public Aggregator aggregator(Map<String, String> tags, boolean shouldCheckQuery) {
return new Aggregator() {
return new SimpleAggregator() {
private int aggr = 0;

@Override public void update(TagsValuePair p) {
Expand All @@ -431,10 +438,8 @@ final class Count implements AggregateFunction {
}
}

@Override public Iterable<TagsValuePair> result() {
return (aggr > 0)
? Collections.singletonList(new TagsValuePair(tags, aggr))
: Collections.emptyList();
@Override public TagsValuePair resultPair() {
return (aggr > 0) ? new TagsValuePair(tags, aggr) : null;
}
};
}
Expand Down Expand Up @@ -513,22 +518,27 @@ private Map<String, String> keyTags(Map<String, String> tags) {

@Override public Aggregator aggregator(Map<String, String> ignored, boolean shouldCheckQuery) {
return new Aggregator() {
private final Map<Map<String, String>, Aggregator> aggrs = new HashMap<>();
private final Map<Map<String, String>, SimpleAggregator> aggrs = new HashMap<>();

@Override public void update(TagsValuePair p) {
Map<String, String> tags = p.tags();
if (!shouldCheckQuery || af.query().matches(tags)) {
Map<String, String> k = resultTags(tags);
if (k != null) {
aggrs.computeIfAbsent(k, ks -> af.aggregator(ks, false)).update(p);
aggrs.computeIfAbsent(k, ks -> (SimpleAggregator) af.aggregator(ks, false)).update(p);
}
}
}

@Override public Iterable<TagsValuePair> result() {
return aggrs.values().stream()
.flatMap(a -> StreamSupport.stream(a.result().spliterator(), false))
.collect(Collectors.toList());
List<TagsValuePair> pairs = new ArrayList<>(aggrs.size());
for (SimpleAggregator aggr : aggrs.values()) {
TagsValuePair pair = aggr.resultPair();
if (pair != null) {
pairs.add(pair);
}
}
return pairs;
}
};
}
Expand Down
Loading