Fix cache bug in stats module (#5650)

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

@@ -107,7 +107,8 @@ public class StandardDeviationPostAggregator implements PostAggregator
     return "StandardDeviationPostAggregator{" +
            "name='" + name + '\'' +
            ", fieldName='" + fieldName + '\'' +
-           ", isVariancePop='" + isVariancePop + '\'' +
+           ", estimator='" + estimator + '\'' +
+           ", isVariancePop=" + isVariancePop +
            '}';
   }
 
@@ -116,6 +117,7 @@ public class StandardDeviationPostAggregator implements PostAggregator
   {
     return new CacheKeyBuilder(PostAggregatorIds.VARIANCE_STANDARD_DEVIATION)
         .appendString(fieldName)
+        .appendString(estimator)
         .appendBoolean(isVariancePop)
         .build();
   }

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

@@ -59,20 +59,15 @@ public class VarianceAggregatorCollector
     return new VarianceAggregatorCollector(buffer.getLong(), buffer.getDouble(), buffer.getDouble());
   }
 
-  public static final Comparator<VarianceAggregatorCollector> COMPARATOR = new Comparator<VarianceAggregatorCollector>()
+  public static final Comparator<VarianceAggregatorCollector> COMPARATOR = (o1, o2) -> {
-  {
+    int compare = Longs.compare(o1.count, o2.count);
-    @Override
+    if (compare == 0) {
-    public int compare(VarianceAggregatorCollector o1, VarianceAggregatorCollector o2)
+      compare = Doubles.compare(o1.sum, o2.sum);
-    {
-      int compare = Longs.compare(o1.count, o2.count);
       if (compare == 0) {
-        compare = Doubles.compare(o1.sum, o2.sum);
+        compare = Doubles.compare(o1.nvariance, o2.nvariance);
-        if (compare == 0) {
-          compare = Doubles.compare(o1.nvariance, o2.nvariance);
-        }
       }
-      return compare;
     }
+    return compare;
   };
 
   void fold(@Nullable VarianceAggregatorCollector other)
@@ -114,13 +109,6 @@ public class VarianceAggregatorCollector
     this(0, 0, 0);
   }
 
-  public void reset()
-  {
-    count = 0;
-    sum = 0;
-    nvariance = 0;
-  }
-
   void copyFrom(VarianceAggregatorCollector other)
   {
     this.count = other.count;

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

@@ -34,13 +34,13 @@ import io.druid.query.aggregation.BufferAggregator;
 import io.druid.query.aggregation.NoopAggregator;
 import io.druid.query.aggregation.NoopBufferAggregator;
 import io.druid.query.aggregation.ObjectAggregateCombiner;
+import io.druid.query.cache.CacheKeyBuilder;
 import io.druid.segment.ColumnSelectorFactory;
 import io.druid.segment.ColumnValueSelector;
 import io.druid.segment.NilColumnValueSelector;
 import org.apache.commons.codec.binary.Base64;
 
 import java.nio.ByteBuffer;
-import java.util.Arrays;
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.List;
@@ -184,7 +184,7 @@ public class VarianceAggregatorFactory extends AggregatorFactory
   @Override
   public List<AggregatorFactory> getRequiredColumns()
   {
-    return Arrays.<AggregatorFactory>asList(new VarianceAggregatorFactory(fieldName, fieldName, estimator, inputType));
+    return Collections.singletonList(new VarianceAggregatorFactory(fieldName, fieldName, estimator, inputType));
   }
 
   @Override
@@ -258,25 +258,23 @@ public class VarianceAggregatorFactory extends AggregatorFactory
   @Override
   public byte[] getCacheKey()
   {
-    byte[] fieldNameBytes = StringUtils.toUtf8(fieldName);
+    return new CacheKeyBuilder(AggregatorUtil.VARIANCE_CACHE_TYPE_ID)
-    byte[] inputTypeBytes = StringUtils.toUtf8(inputType);
+        .appendString(fieldName)
-    return ByteBuffer.allocate(2 + fieldNameBytes.length + 1 + inputTypeBytes.length)
+        .appendString(inputType)
-                     .put(AggregatorUtil.VARIANCE_CACHE_TYPE_ID)
+        .appendBoolean(isVariancePop)
-                     .put(isVariancePop ? (byte) 1 : 0)
+        .appendString(estimator)
-                     .put(fieldNameBytes)
+        .build();
-                     .put((byte) 0xFF)
-                     .put(inputTypeBytes)
-                     .array();
   }
 
   @Override
   public String toString()
   {
-    return getClass().getSimpleName() + "{" +
+    return "VarianceAggregatorFactory{" +
            "fieldName='" + fieldName + '\'' +
            ", name='" + name + '\'' +
-           ", isVariancePop='" + isVariancePop + '\'' +
+           ", estimator='" + estimator + '\'' +
            ", inputType='" + inputType + '\'' +
+           ", isVariancePop=" + isVariancePop +
            '}';
   }
 
@@ -289,29 +287,18 @@ public class VarianceAggregatorFactory extends AggregatorFactory
     if (o == null || getClass() != o.getClass()) {
       return false;
     }
-
     VarianceAggregatorFactory that = (VarianceAggregatorFactory) o;
-
+    return isVariancePop == that.isVariancePop &&
-    if (!Objects.equals(name, that.name)) {
+           Objects.equals(fieldName, that.fieldName) &&
-      return false;
+           Objects.equals(name, that.name) &&
-    }
+           Objects.equals(estimator, that.estimator) &&
-    if (!Objects.equals(isVariancePop, that.isVariancePop)) {
+           Objects.equals(inputType, that.inputType);
-      return false;
-    }
-    if (!Objects.equals(inputType, that.inputType)) {
-      return false;
-    }
-
-    return true;
   }
 
   @Override
   public int hashCode()
   {
-    int result = fieldName.hashCode();
+
-    result = 31 * result + Objects.hashCode(name);
+    return Objects.hash(fieldName, name, estimator, inputType, isVariancePop);
-    result = 31 * result + Objects.hashCode(isVariancePop);
-    result = 31 * result + Objects.hashCode(inputType);
-    return result;
   }
 }