You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by "rohangarg (via GitHub)" <gi...@apache.org> on 2023/08/02 13:25:12 UTC
Re: [PR] Minimize PostAggregator computations (druid)
rohangarg commented on code in PR #14708:
URL: https://github.com/apache/druid/pull/14708#discussion_r1281830258
##########
processing/src/main/java/org/apache/druid/query/topn/TopNQueryQueryToolChest.java:
##########
@@ -151,128 +185,80 @@ public Function<Result<TopNResultValue>, Result<TopNResultValue>> makePreCompute
final MetricManipulationFn fn
)
{
+ //noinspection ObjectEquality
+ if (MetricManipulatorFns.deserializing() != fn) {
+ throw DruidException.defensive("This method can only be used to deserialize.");
+ }
+
return new Function<Result<TopNResultValue>, Result<TopNResultValue>>()
{
- private String dimension = query.getDimensionSpec().getOutputName();
- private final List<PostAggregator> prunedAggs = prunePostAggregators(query);
private final AggregatorFactory[] aggregatorFactories = query.getAggregatorSpecs()
.toArray(new AggregatorFactory[0]);
private final String[] aggFactoryNames = extractFactoryName(query.getAggregatorSpecs());
@Override
public Result<TopNResultValue> apply(Result<TopNResultValue> result)
{
- List<Map<String, Object>> serializedValues = Lists.newArrayList(
- Iterables.transform(
- result.getValue(),
- new Function<DimensionAndMetricValueExtractor, Map<String, Object>>()
- {
- @Override
- public Map<String, Object> apply(DimensionAndMetricValueExtractor input)
- {
- final Map<String, Object> values = Maps.newHashMapWithExpectedSize(
- aggregatorFactories.length
- + prunedAggs.size()
- + 1
- );
-
- for (int i = 0; i < aggregatorFactories.length; ++i) {
- final String aggName = aggFactoryNames[i];
- values.put(aggName, fn.manipulate(aggregatorFactories[i], input.getMetric(aggName)));
- }
+ final List<DimensionAndMetricValueExtractor> values = result.getValue().getValue();
+ final List<DimensionAndMetricValueExtractor> newValues = new ArrayList<>(values.size());
- for (PostAggregator postAgg : prunedAggs) {
- final String name = postAgg.getName();
- Object calculatedPostAgg = input.getMetric(name);
- if (calculatedPostAgg != null) {
- values.put(name, calculatedPostAgg);
- } else {
- values.put(name, postAgg.compute(values));
- }
- }
- values.put(dimension, input.getDimensionValue(dimension));
+ for (DimensionAndMetricValueExtractor input : values) {
+ final Map<String, Object> map = new LinkedHashMap<>(input.getBaseObject());
- return values;
- }
- }
- )
- );
+ for (int i = 0; i < aggregatorFactories.length; ++i) {
+ final String aggName = aggFactoryNames[i];
+ map.put(aggName, aggregatorFactories[i].deserialize(map.get(aggName)));
Review Comment:
should we call `fn.apply` here instead?
##########
processing/src/main/java/org/apache/druid/query/topn/TopNQueryQueryToolChest.java:
##########
@@ -151,128 +185,80 @@ public Function<Result<TopNResultValue>, Result<TopNResultValue>> makePreCompute
final MetricManipulationFn fn
)
{
+ //noinspection ObjectEquality
+ if (MetricManipulatorFns.deserializing() != fn) {
+ throw DruidException.defensive("This method can only be used to deserialize.");
+ }
+
return new Function<Result<TopNResultValue>, Result<TopNResultValue>>()
{
- private String dimension = query.getDimensionSpec().getOutputName();
- private final List<PostAggregator> prunedAggs = prunePostAggregators(query);
private final AggregatorFactory[] aggregatorFactories = query.getAggregatorSpecs()
.toArray(new AggregatorFactory[0]);
private final String[] aggFactoryNames = extractFactoryName(query.getAggregatorSpecs());
@Override
public Result<TopNResultValue> apply(Result<TopNResultValue> result)
{
- List<Map<String, Object>> serializedValues = Lists.newArrayList(
- Iterables.transform(
- result.getValue(),
- new Function<DimensionAndMetricValueExtractor, Map<String, Object>>()
- {
- @Override
- public Map<String, Object> apply(DimensionAndMetricValueExtractor input)
- {
- final Map<String, Object> values = Maps.newHashMapWithExpectedSize(
- aggregatorFactories.length
- + prunedAggs.size()
- + 1
- );
-
- for (int i = 0; i < aggregatorFactories.length; ++i) {
- final String aggName = aggFactoryNames[i];
- values.put(aggName, fn.manipulate(aggregatorFactories[i], input.getMetric(aggName)));
- }
+ final List<DimensionAndMetricValueExtractor> values = result.getValue().getValue();
+ final List<DimensionAndMetricValueExtractor> newValues = new ArrayList<>(values.size());
- for (PostAggregator postAgg : prunedAggs) {
- final String name = postAgg.getName();
- Object calculatedPostAgg = input.getMetric(name);
- if (calculatedPostAgg != null) {
- values.put(name, calculatedPostAgg);
- } else {
- values.put(name, postAgg.compute(values));
- }
- }
- values.put(dimension, input.getDimensionValue(dimension));
+ for (DimensionAndMetricValueExtractor input : values) {
+ final Map<String, Object> map = new LinkedHashMap<>(input.getBaseObject());
- return values;
- }
- }
- )
- );
+ for (int i = 0; i < aggregatorFactories.length; ++i) {
+ final String aggName = aggFactoryNames[i];
+ map.put(aggName, aggregatorFactories[i].deserialize(map.get(aggName)));
+ }
- return new Result<TopNResultValue>(
- result.getTimestamp(),
- new TopNResultValue(serializedValues)
- );
+ newValues.add(new DimensionAndMetricValueExtractor(map));
+ }
+
+ return new Result<>(result.getTimestamp(), new TopNResultValue(newValues));
}
};
}
+ @SuppressWarnings("ObjectEquality")
@Override
public Function<Result<TopNResultValue>, Result<TopNResultValue>> makePostComputeManipulatorFn(
- final TopNQuery query,
- final MetricManipulationFn fn
+ TopNQuery query,
+ MetricManipulationFn fn
)
{
- return new Function<Result<TopNResultValue>, Result<TopNResultValue>>()
- {
- private String dimension = query.getDimensionSpec().getOutputName();
- private final AggregatorFactory[] aggregatorFactories = query.getAggregatorSpecs()
- .toArray(new AggregatorFactory[0]);
- private final String[] aggFactoryNames = extractFactoryName(query.getAggregatorSpecs());
- private final PostAggregator[] postAggregators = query.getPostAggregatorSpecs().toArray(new PostAggregator[0]);
+ if (MetricManipulatorFns.identity() == fn) {
+ return result -> result;
+ }
- @Override
- public Result<TopNResultValue> apply(Result<TopNResultValue> result)
+ if (MetricManipulatorFns.finalizing() == fn) {
+ return new Function<Result<TopNResultValue>, Result<TopNResultValue>>()
{
- List<Map<String, Object>> serializedValues = Lists.newArrayList(
- Iterables.transform(
- result.getValue(),
- new Function<DimensionAndMetricValueExtractor, Map<String, Object>>()
- {
- @Override
- public Map<String, Object> apply(DimensionAndMetricValueExtractor input)
- {
- final Map<String, Object> values = Maps.newHashMapWithExpectedSize(
- aggregatorFactories.length
- + query.getPostAggregatorSpecs().size()
- + 1
- );
-
- // Put non-finalized aggregators before post-aggregators.
- for (final String name : aggFactoryNames) {
- values.put(name, input.getMetric(name));
- }
+ private final AggregatorFactory[] aggregatorFactories = query.getAggregatorSpecs()
+ .toArray(new AggregatorFactory[0]);
+ private final String[] aggFactoryNames = extractFactoryName(query.getAggregatorSpecs());
- // Put dimension, post-aggregators might depend on it.
- values.put(dimension, input.getDimensionValue(dimension));
-
- // Put post-aggregators.
- for (PostAggregator postAgg : postAggregators) {
- Object calculatedPostAgg = input.getMetric(postAgg.getName());
- if (calculatedPostAgg != null) {
- values.put(postAgg.getName(), calculatedPostAgg);
- } else {
- values.put(postAgg.getName(), postAgg.compute(values));
- }
- }
+ @Override
+ public Result<TopNResultValue> apply(Result<TopNResultValue> result)
+ {
+ final List<DimensionAndMetricValueExtractor> values = result.getValue().getValue();
+ final List<DimensionAndMetricValueExtractor> newValues = new ArrayList<>(values.size());
- // Put finalized aggregators now that post-aggregators are done.
- for (int i = 0; i < aggFactoryNames.length; ++i) {
- final String name = aggFactoryNames[i];
- values.put(name, fn.manipulate(aggregatorFactories[i], input.getMetric(name)));
- }
+ for (DimensionAndMetricValueExtractor input : values) {
+ final Map<String, Object> map = new LinkedHashMap<>(input.getBaseObject());
- return values;
- }
- }
- )
- );
+ for (int i = 0; i < aggregatorFactories.length; ++i) {
+ final String aggName = aggFactoryNames[i];
+ map.put(aggName, aggregatorFactories[i].finalizeComputation(map.get(aggName)));
Review Comment:
same here - `fn.apply` would be better I think
##########
processing/src/main/java/org/apache/druid/query/aggregation/AggregatorUtil.java:
##########
@@ -166,6 +166,10 @@ public class AggregatorUtil
*/
public static List<PostAggregator> pruneDependentPostAgg(List<PostAggregator> postAggregatorList, String postAggName)
Review Comment:
the documentation and var naming in this function is a bit confusing - since it can be used to prune postAggs not needed for a topn metric. Do we want to think that even topn metrics are sort of postAggs?
##########
processing/src/main/java/org/apache/druid/query/topn/TopNQueryQueryToolChest.java:
##########
@@ -151,128 +185,80 @@ public Function<Result<TopNResultValue>, Result<TopNResultValue>> makePreCompute
final MetricManipulationFn fn
)
{
+ //noinspection ObjectEquality
+ if (MetricManipulatorFns.deserializing() != fn) {
+ throw DruidException.defensive("This method can only be used to deserialize.");
+ }
+
return new Function<Result<TopNResultValue>, Result<TopNResultValue>>()
{
- private String dimension = query.getDimensionSpec().getOutputName();
- private final List<PostAggregator> prunedAggs = prunePostAggregators(query);
private final AggregatorFactory[] aggregatorFactories = query.getAggregatorSpecs()
.toArray(new AggregatorFactory[0]);
private final String[] aggFactoryNames = extractFactoryName(query.getAggregatorSpecs());
@Override
public Result<TopNResultValue> apply(Result<TopNResultValue> result)
{
- List<Map<String, Object>> serializedValues = Lists.newArrayList(
- Iterables.transform(
- result.getValue(),
- new Function<DimensionAndMetricValueExtractor, Map<String, Object>>()
- {
- @Override
- public Map<String, Object> apply(DimensionAndMetricValueExtractor input)
- {
- final Map<String, Object> values = Maps.newHashMapWithExpectedSize(
- aggregatorFactories.length
- + prunedAggs.size()
- + 1
- );
-
- for (int i = 0; i < aggregatorFactories.length; ++i) {
- final String aggName = aggFactoryNames[i];
- values.put(aggName, fn.manipulate(aggregatorFactories[i], input.getMetric(aggName)));
- }
+ final List<DimensionAndMetricValueExtractor> values = result.getValue().getValue();
+ final List<DimensionAndMetricValueExtractor> newValues = new ArrayList<>(values.size());
- for (PostAggregator postAgg : prunedAggs) {
- final String name = postAgg.getName();
- Object calculatedPostAgg = input.getMetric(name);
- if (calculatedPostAgg != null) {
- values.put(name, calculatedPostAgg);
- } else {
- values.put(name, postAgg.compute(values));
- }
- }
- values.put(dimension, input.getDimensionValue(dimension));
+ for (DimensionAndMetricValueExtractor input : values) {
+ final Map<String, Object> map = new LinkedHashMap<>(input.getBaseObject());
- return values;
- }
- }
- )
- );
+ for (int i = 0; i < aggregatorFactories.length; ++i) {
+ final String aggName = aggFactoryNames[i];
+ map.put(aggName, aggregatorFactories[i].deserialize(map.get(aggName)));
+ }
- return new Result<TopNResultValue>(
- result.getTimestamp(),
- new TopNResultValue(serializedValues)
- );
+ newValues.add(new DimensionAndMetricValueExtractor(map));
+ }
+
+ return new Result<>(result.getTimestamp(), new TopNResultValue(newValues));
}
};
}
+ @SuppressWarnings("ObjectEquality")
@Override
public Function<Result<TopNResultValue>, Result<TopNResultValue>> makePostComputeManipulatorFn(
- final TopNQuery query,
- final MetricManipulationFn fn
+ TopNQuery query,
+ MetricManipulationFn fn
)
{
- return new Function<Result<TopNResultValue>, Result<TopNResultValue>>()
- {
- private String dimension = query.getDimensionSpec().getOutputName();
- private final AggregatorFactory[] aggregatorFactories = query.getAggregatorSpecs()
- .toArray(new AggregatorFactory[0]);
- private final String[] aggFactoryNames = extractFactoryName(query.getAggregatorSpecs());
- private final PostAggregator[] postAggregators = query.getPostAggregatorSpecs().toArray(new PostAggregator[0]);
+ if (MetricManipulatorFns.identity() == fn) {
+ return result -> result;
+ }
- @Override
- public Result<TopNResultValue> apply(Result<TopNResultValue> result)
+ if (MetricManipulatorFns.finalizing() == fn) {
+ return new Function<Result<TopNResultValue>, Result<TopNResultValue>>()
{
- List<Map<String, Object>> serializedValues = Lists.newArrayList(
- Iterables.transform(
- result.getValue(),
- new Function<DimensionAndMetricValueExtractor, Map<String, Object>>()
- {
- @Override
- public Map<String, Object> apply(DimensionAndMetricValueExtractor input)
- {
- final Map<String, Object> values = Maps.newHashMapWithExpectedSize(
- aggregatorFactories.length
- + query.getPostAggregatorSpecs().size()
- + 1
- );
-
- // Put non-finalized aggregators before post-aggregators.
- for (final String name : aggFactoryNames) {
- values.put(name, input.getMetric(name));
- }
+ private final AggregatorFactory[] aggregatorFactories = query.getAggregatorSpecs()
+ .toArray(new AggregatorFactory[0]);
+ private final String[] aggFactoryNames = extractFactoryName(query.getAggregatorSpecs());
- // Put dimension, post-aggregators might depend on it.
- values.put(dimension, input.getDimensionValue(dimension));
-
- // Put post-aggregators.
- for (PostAggregator postAgg : postAggregators) {
- Object calculatedPostAgg = input.getMetric(postAgg.getName());
- if (calculatedPostAgg != null) {
- values.put(postAgg.getName(), calculatedPostAgg);
- } else {
- values.put(postAgg.getName(), postAgg.compute(values));
- }
- }
+ @Override
+ public Result<TopNResultValue> apply(Result<TopNResultValue> result)
+ {
+ final List<DimensionAndMetricValueExtractor> values = result.getValue().getValue();
+ final List<DimensionAndMetricValueExtractor> newValues = new ArrayList<>(values.size());
- // Put finalized aggregators now that post-aggregators are done.
- for (int i = 0; i < aggFactoryNames.length; ++i) {
- final String name = aggFactoryNames[i];
- values.put(name, fn.manipulate(aggregatorFactories[i], input.getMetric(name)));
- }
+ for (DimensionAndMetricValueExtractor input : values) {
+ final Map<String, Object> map = new LinkedHashMap<>(input.getBaseObject());
- return values;
- }
- }
- )
- );
+ for (int i = 0; i < aggregatorFactories.length; ++i) {
+ final String aggName = aggFactoryNames[i];
+ map.put(aggName, aggregatorFactories[i].finalizeComputation(map.get(aggName)));
+ }
- return new Result<>(
- result.getTimestamp(),
- new TopNResultValue(serializedValues)
- );
- }
- };
+ newValues.add(new DimensionAndMetricValueExtractor(map));
+ }
+
+ return new Result<>(result.getTimestamp(), new TopNResultValue(newValues));
+ }
+ };
+ }
+
+ throw DruidException.defensive("This method can only be used to finalize.");
Review Comment:
could do an early return with this exception in the method
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org