Finalize fields in postaggs (apache#3957)

* initial commits for finalizeFieldAccess apache#2433

* fix some bugs to run a query

* change name of method Queries.verifyAggregations to Queries.prepareAggregations

* add Uts

* fix Ut failures

* rebased to master

* address comments and add a Ut for arithmetic post aggregators

* rebased to the master

* address the comment of injection within arithmetic post aggregator

* address comments and introduce decorate() in the PostAggregator interface.

* Address comments. 1. Implements getComparator in FinalizingFieldAccessPostAggregator and add Uts for it 2. Some minor changes like renaming a method name.

* Fix a code style mismatch.

* Rebased to the master

extensions-core/datasketches/src/main/java/io/druid/query/aggregation/datasketches/theta/SketchEstimatePostAggregator.java

@@ -25,6 +25,7 @@
 import com.google.common.collect.Ordering;
 import com.google.common.collect.Sets;
 import com.google.common.primitives.Doubles;
+import io.druid.query.aggregation.AggregatorFactory;
 import io.druid.query.aggregation.PostAggregator;
 import io.druid.query.aggregation.post.PostAggregatorIds;
 import io.druid.query.cache.CacheKeyBuilder;
@@ -97,6 +98,12 @@ public String getName()
     return name;
   }
 
+  @Override
+  public PostAggregator decorate(Map<String, AggregatorFactory> aggregators)
+  {
+    return this;
+  }
+
   @JsonProperty
   public PostAggregator getField()
   {

extensions-core/datasketches/src/main/java/io/druid/query/aggregation/datasketches/theta/SketchSetPostAggregator.java

@@ -24,6 +24,7 @@
 import com.google.common.collect.Sets;
 import com.yahoo.sketches.Util;
 import io.druid.java.util.common.IAE;
+import io.druid.query.aggregation.AggregatorFactory;
 import io.druid.query.aggregation.PostAggregator;
 import io.druid.query.aggregation.post.PostAggregatorIds;
 import io.druid.query.cache.CacheKeyBuilder;
@@ -93,6 +94,12 @@ public String getName()
     return name;
   }
 
+  @Override
+  public PostAggregator decorate(Map<String, AggregatorFactory> aggregators)
+  {
+    return this;
+  }
+
   @JsonProperty
   public String getFunc()
   {

extensions-core/histogram/src/main/java/io/druid/query/aggregation/histogram/BucketsPostAggregator.java

@@ -24,6 +24,8 @@
 import com.fasterxml.jackson.annotation.JsonTypeName;
 import com.google.common.collect.Sets;
 import io.druid.java.util.common.IAE;
+import io.druid.query.aggregation.AggregatorFactory;
+import io.druid.query.aggregation.PostAggregator;
 import io.druid.query.aggregation.post.PostAggregatorIds;
 import io.druid.query.cache.CacheKeyBuilder;
 
@@ -65,6 +67,12 @@ public Object compute(Map<String, Object> values)
     return ah.toHistogram(bucketSize, offset);
   }
 
+  @Override
+  public PostAggregator decorate(Map<String, AggregatorFactory> aggregators)
+  {
+    return this;
+  }
+
   @JsonProperty
   public float getBucketSize()
   {

extensions-core/histogram/src/main/java/io/druid/query/aggregation/histogram/CustomBucketsPostAggregator.java

@@ -23,6 +23,8 @@
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.annotation.JsonTypeName;
 import com.google.common.collect.Sets;
+import io.druid.query.aggregation.AggregatorFactory;
+import io.druid.query.aggregation.PostAggregator;
 import io.druid.query.aggregation.post.PostAggregatorIds;
 import io.druid.query.cache.CacheKeyBuilder;
 
@@ -61,6 +63,12 @@ public Object compute(Map<String, Object> values)
     return ah.toHistogram(breaks);
   }
 
+  @Override
+  public PostAggregator decorate(Map<String, AggregatorFactory> aggregators)
+  {
+    return this;
+  }
+
   @JsonProperty
   public float[] getBreaks()
   {

extensions-core/histogram/src/main/java/io/druid/query/aggregation/histogram/EqualBucketsPostAggregator.java

@@ -24,6 +24,8 @@
 import com.fasterxml.jackson.annotation.JsonTypeName;
 import com.google.common.collect.Sets;
 import io.druid.java.util.common.IAE;
+import io.druid.query.aggregation.AggregatorFactory;
+import io.druid.query.aggregation.PostAggregator;
 import io.druid.query.aggregation.post.PostAggregatorIds;
 import io.druid.query.cache.CacheKeyBuilder;
 
@@ -62,6 +64,12 @@ public Object compute(Map<String, Object> values)
     return ah.toHistogram(numBuckets);
   }
 
+  @Override
+  public PostAggregator decorate(Map<String, AggregatorFactory> aggregators)
+  {
+    return this;
+  }
+
   @JsonProperty
   public int getNumBuckets()
   {

extensions-core/histogram/src/main/java/io/druid/query/aggregation/histogram/MaxPostAggregator.java

@@ -23,6 +23,8 @@
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.annotation.JsonTypeName;
 import com.google.common.collect.Sets;
+import io.druid.query.aggregation.AggregatorFactory;
+import io.druid.query.aggregation.PostAggregator;
 import io.druid.query.aggregation.post.PostAggregatorIds;
 import io.druid.query.cache.CacheKeyBuilder;
 
@@ -70,6 +72,12 @@ public Object compute(Map<String, Object> values)
     return ah.getMax();
   }
 
+  @Override
+  public PostAggregator decorate(Map<String, AggregatorFactory> aggregators)
+  {
+    return this;
+  }
+
   @Override
   public String toString()
   {

extensions-core/histogram/src/main/java/io/druid/query/aggregation/histogram/MinPostAggregator.java

@@ -23,6 +23,8 @@
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.annotation.JsonTypeName;
 import com.google.common.collect.Sets;
+import io.druid.query.aggregation.AggregatorFactory;
+import io.druid.query.aggregation.PostAggregator;
 import io.druid.query.aggregation.post.PostAggregatorIds;
 import io.druid.query.cache.CacheKeyBuilder;
 
@@ -70,6 +72,12 @@ public Object compute(Map<String, Object> values)
     return ah.getMin();
   }
 
+  @Override
+  public PostAggregator decorate(Map<String, AggregatorFactory> aggregators)
+  {
+    return this;
+  }
+
   @Override
   public String toString()
   {

extensions-core/histogram/src/main/java/io/druid/query/aggregation/histogram/QuantilePostAggregator.java

@@ -24,6 +24,8 @@
 import com.fasterxml.jackson.annotation.JsonTypeName;
 import com.google.common.collect.Sets;
 import io.druid.java.util.common.IAE;
+import io.druid.query.aggregation.AggregatorFactory;
+import io.druid.query.aggregation.PostAggregator;
 import io.druid.query.aggregation.post.PostAggregatorIds;
 import io.druid.query.cache.CacheKeyBuilder;
 
@@ -81,6 +83,12 @@ public Object compute(Map<String, Object> values)
     return ah.getQuantiles(new float[]{probability})[0];
   }
 
+  @Override
+  public PostAggregator decorate(Map<String, AggregatorFactory> aggregators)
+  {
+    return this;
+  }
+
   @JsonProperty
   public float getProbability()
   {

extensions-core/histogram/src/main/java/io/druid/query/aggregation/histogram/QuantilesPostAggregator.java

@@ -24,6 +24,8 @@
 import com.fasterxml.jackson.annotation.JsonTypeName;
 import com.google.common.collect.Sets;
 import io.druid.java.util.common.IAE;
+import io.druid.query.aggregation.AggregatorFactory;
+import io.druid.query.aggregation.PostAggregator;
 import io.druid.query.aggregation.post.PostAggregatorIds;
 import io.druid.query.cache.CacheKeyBuilder;
 
@@ -74,6 +76,12 @@ public Object compute(Map<String, Object> values)
     return new Quantiles(probabilities, ah.getQuantiles(probabilities), ah.getMin(), ah.getMax());
   }
 
+  @Override
+  public PostAggregator decorate(Map<String, AggregatorFactory> aggregators)
+  {
+    return this;
+  }
+
   @JsonProperty
   public float[] getProbabilities()
   {

extensions-core/stats/src/main/java/io/druid/query/aggregation/variance/StandardDeviationPostAggregator.java

@@ -24,6 +24,7 @@
 import com.fasterxml.jackson.annotation.JsonTypeName;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.Sets;
+import io.druid.query.aggregation.AggregatorFactory;
 import io.druid.query.aggregation.PostAggregator;
 import io.druid.query.aggregation.post.ArithmeticPostAggregator;
 import io.druid.query.aggregation.post.PostAggregatorIds;
@@ -82,6 +83,12 @@ public String getName()
     return name;
   }
 
+  @Override
+  public PostAggregator decorate(Map<String, AggregatorFactory> aggregators)
+  {
+    return this;
+  }
+
   @JsonProperty("fieldName")
   public String getFieldName()
   {

processing/src/main/java/io/druid/jackson/AggregatorsModule.java

@@ -50,6 +50,7 @@
 import io.druid.query.aggregation.post.DoubleLeastPostAggregator;
 import io.druid.query.aggregation.post.ExpressionPostAggregator;
 import io.druid.query.aggregation.post.FieldAccessPostAggregator;
+import io.druid.query.aggregation.post.FinalizingFieldAccessPostAggregator;
 import io.druid.query.aggregation.post.JavaScriptPostAggregator;
 import io.druid.query.aggregation.post.LongGreatestPostAggregator;
 import io.druid.query.aggregation.post.LongLeastPostAggregator;
@@ -103,6 +104,7 @@ public static interface AggregatorFactoryMixin
       @JsonSubTypes.Type(name = "expression", value = ExpressionPostAggregator.class),
       @JsonSubTypes.Type(name = "arithmetic", value = ArithmeticPostAggregator.class),
       @JsonSubTypes.Type(name = "fieldAccess", value = FieldAccessPostAggregator.class),
+      @JsonSubTypes.Type(name = "finalizingFieldAccess", value = FinalizingFieldAccessPostAggregator.class),
       @JsonSubTypes.Type(name = "constant", value = ConstantPostAggregator.class),
       @JsonSubTypes.Type(name = "javascript", value = JavaScriptPostAggregator.class),
       @JsonSubTypes.Type(name = "hyperUniqueCardinality", value = HyperUniqueFinalizingPostAggregator.class),

processing/src/main/java/io/druid/query/Queries.java

@@ -20,42 +20,64 @@
 package io.druid.query;
 
 import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
 import com.google.common.collect.Sets;
 import io.druid.query.aggregation.AggregatorFactory;
 import io.druid.query.aggregation.PostAggregator;
 
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 
 /**
  */
 public class Queries
 {
-  public static void verifyAggregations(
+  public static List<PostAggregator> decoratePostAggregators(List<PostAggregator> postAggs,
+                                                             Map<String, AggregatorFactory> aggFactories)
+  {
+    List<PostAggregator> decorated = Lists.newArrayListWithExpectedSize(postAggs.size());
+    for (PostAggregator aggregator : postAggs) {
+      decorated.add(aggregator.decorate(aggFactories));
+    }
+    return decorated;
+  }
+
+  public static List<PostAggregator> prepareAggregations(
       List<AggregatorFactory> aggFactories,
       List<PostAggregator> postAggs
   )
   {
     Preconditions.checkNotNull(aggFactories, "aggregations cannot be null");
 
-    final Set<String> aggNames = Sets.newHashSet();
+    final Map<String, AggregatorFactory> aggsFactoryMap = Maps.newHashMap();
     for (AggregatorFactory aggFactory : aggFactories) {
-      Preconditions.checkArgument(aggNames.add(aggFactory.getName()), "[%s] already defined", aggFactory.getName());
+      Preconditions.checkArgument(!aggsFactoryMap.containsKey(aggFactory.getName()),
+                                  "[%s] already defined", aggFactory.getName());
+      aggsFactoryMap.put(aggFactory.getName(), aggFactory);
     }
 
     if (postAggs != null && !postAggs.isEmpty()) {
-      final Set<String> combinedAggNames = Sets.newHashSet(aggNames);
+      final Set<String> combinedAggNames = Sets.newHashSet(aggsFactoryMap.keySet());
 
-      for (PostAggregator postAgg : postAggs) {
+      List<PostAggregator> decorated = Lists.newArrayListWithExpectedSize(postAggs.size());
+      for (final PostAggregator postAgg : postAggs) {
         final Set<String> dependencies = postAgg.getDependentFields();
         final Set<String> missing = Sets.difference(dependencies, combinedAggNames);
 
         Preconditions.checkArgument(
             missing.isEmpty(),
             "Missing fields [%s] for postAggregator [%s]", missing, postAgg.getName()
         );
-        Preconditions.checkArgument(combinedAggNames.add(postAgg.getName()), "[%s] already defined", postAgg.getName());
+        Preconditions.checkArgument(combinedAggNames.add(postAgg.getName()),
+                                    "[%s] already defined", postAgg.getName());
+
+        decorated.add(postAgg.decorate(aggsFactoryMap));
       }
+      return decorated;
     }
+
+    return postAggs;
   }
 }

processing/src/main/java/io/druid/query/aggregation/PostAggregator.java

@@ -37,4 +37,13 @@ public interface PostAggregator extends Cacheable
   Object compute(Map<String, Object> combinedAggregators);
 
   String getName();
+
+  /**
+   * Returns a richer post aggregator which are built from the given aggregators with their names and some accessible
+   * environmental variables such as ones in the object scope.
+   *
+   * @param aggregators A map of aggregator factories with their names.
+   *
+   */
+  PostAggregator decorate(Map<String, AggregatorFactory> aggregators);
 }

processing/src/main/java/io/druid/query/aggregation/hyperloglog/HyperUniqueFinalizingPostAggregator.java

@@ -24,6 +24,7 @@
 import com.google.common.base.Preconditions;
 import com.google.common.collect.Ordering;
 import com.google.common.collect.Sets;
+import io.druid.query.aggregation.AggregatorFactory;
 import io.druid.query.aggregation.PostAggregator;
 import io.druid.query.aggregation.post.PostAggregatorIds;
 import io.druid.query.cache.CacheKeyBuilder;
@@ -86,6 +87,12 @@ public String getName()
     return name;
   }
 
+  @Override
+  public HyperUniqueFinalizingPostAggregator decorate(Map<String, AggregatorFactory> aggregators)
+  {
+    return this;
+  }
+
   @JsonProperty("fieldName")
   public String getFieldName()
   {

processing/src/main/java/io/druid/query/aggregation/post/ArithmeticPostAggregator.java

@@ -25,6 +25,8 @@
 import com.google.common.collect.Maps;
 import com.google.common.collect.Sets;
 import io.druid.java.util.common.IAE;
+import io.druid.query.Queries;
+import io.druid.query.aggregation.AggregatorFactory;
 import io.druid.query.aggregation.PostAggregator;
 import io.druid.query.cache.CacheKeyBuilder;
 
@@ -53,6 +55,7 @@ public int compare(Object o, Object o1)
   private final Ops op;
   private final Comparator comparator;
   private final String ordering;
+  private Map<String, AggregatorFactory> aggFactoryMap;
 
   public ArithmeticPostAggregator(
       String name,
@@ -124,6 +127,12 @@ public String getName()
     return name;
   }
 
+  @Override
+  public ArithmeticPostAggregator decorate(Map<String, AggregatorFactory> aggregators)
+  {
+    return new ArithmeticPostAggregator(name, fnName, Queries.decoratePostAggregators(fields, aggregators), ordering);
+  }
+
   @Override
   public byte[] getCacheKey()
   {

processing/src/main/java/io/druid/query/aggregation/post/ConstantPostAggregator.java

@@ -23,6 +23,7 @@
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.Sets;
+import io.druid.query.aggregation.AggregatorFactory;
 import io.druid.query.aggregation.PostAggregator;
 import io.druid.query.cache.CacheKeyBuilder;
 
@@ -80,6 +81,12 @@ public String getName()
     return name;
   }
 
+  @Override
+  public ConstantPostAggregator decorate(Map<String, AggregatorFactory> aggregators)
+  {
+    return this;
+  }
+
   @JsonProperty("value")
   public Number getConstantValue()
   {