[Backport]fix MAX_INTERMEDIATE_SIZE for DoubleMeanHolder (#9568)

processing/src/main/java/org/apache/druid/query/aggregation/mean/DoubleMeanAggregatorFactory.java

@@ -148,7 +148,9 @@ public List<AggregatorFactory> getRequiredColumns()
   @Override
   public Object deserialize(Object object)
   {
-    if (object instanceof String) {
+    if (object instanceof byte[]) {
+      return DoubleMeanHolder.fromBytes((byte[]) object);
+    } else if (object instanceof String) {
       return DoubleMeanHolder.fromBytes(StringUtils.decodeBase64(StringUtils.toUtf8((String) object)));
     } else if (object instanceof DoubleMeanHolder) {
       return object;
@@ -161,7 +163,9 @@ public Object deserialize(Object object)
   @Override
   public Object finalizeComputation(@Nullable Object object)
   {
-    if (object instanceof DoubleMeanHolder) {
+    if (object instanceof byte[]) {
+      return DoubleMeanHolder.fromBytes((byte[]) object).mean();
+    } else if (object instanceof DoubleMeanHolder) {
       return ((DoubleMeanHolder) object).mean();
     } else if (object == null) {
       return null;

processing/src/main/java/org/apache/druid/query/aggregation/mean/DoubleMeanHolder.java

@@ -27,10 +27,11 @@
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.Comparator;
+import java.util.Objects;
 
 public class DoubleMeanHolder
 {
-  public static final int MAX_INTERMEDIATE_SIZE = Long.SIZE + Double.SIZE;
+  public static final int MAX_INTERMEDIATE_SIZE = Long.BYTES + Double.BYTES;
   public static final Comparator<DoubleMeanHolder> COMPARATOR = (o1, o2) -> Doubles.compare(o1.mean(), o2.mean());
 
   private double sum;
@@ -62,16 +63,36 @@ public double mean()
 
   public byte[] toBytes()
   {
-    ByteBuffer buf = ByteBuffer.allocate(Double.SIZE + Long.SIZE);
+    ByteBuffer buf = ByteBuffer.allocate(Double.BYTES + Long.BYTES);
     buf.putDouble(0, sum);
-    buf.putLong(Double.SIZE, count);
+    buf.putLong(Double.BYTES, count);
     return buf.array();
   }
 
+  @Override
+  public boolean equals(Object o)
+  {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    DoubleMeanHolder that = (DoubleMeanHolder) o;
+    return Double.compare(that.sum, sum) == 0 &&
+           count == that.count;
+  }
+
+  @Override
+  public int hashCode()
+  {
+    return Objects.hash(sum, count);
+  }
+
   public static DoubleMeanHolder fromBytes(byte[] data)
   {
     ByteBuffer buf = ByteBuffer.wrap(data);
-    return new DoubleMeanHolder(buf.getDouble(0), buf.getLong(Double.SIZE));
+    return new DoubleMeanHolder(buf.getDouble(0), buf.getLong(Double.BYTES));
   }
 
   public static void init(ByteBuffer buf, int position)
@@ -109,12 +130,12 @@ private static double getSum(ByteBuffer buf, int position)
 
   private static void writeCount(ByteBuffer buf, int position, long count)
   {
-    buf.putLong(position + Double.SIZE, count);
+    buf.putLong(position + Double.BYTES, count);
   }
 
   private static long getCount(ByteBuffer buf, int position)
   {
-    return buf.getLong(position + Double.SIZE);
+    return buf.getLong(position + Double.BYTES);
   }
 
   public static class Serializer extends JsonSerializer<DoubleMeanHolder>

processing/src/test/java/org/apache/druid/query/aggregation/mean/DoubleMeanAggregatorFactoryTest.java

@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.query.aggregation.mean;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+public class DoubleMeanAggregatorFactoryTest
+{
+  @Test
+  public void testMaxIntermediateSize()
+  {
+    DoubleMeanAggregatorFactory factory = new DoubleMeanAggregatorFactory("name", "fieldName");
+    Assert.assertEquals(Double.BYTES + Long.BYTES, factory.getMaxIntermediateSize());
+    Assert.assertEquals(Double.BYTES + Long.BYTES, factory.getMaxIntermediateSizeWithNulls());
+  }
+
+  @Test
+  public void testDeserialyze()
+  {
+    DoubleMeanAggregatorFactory factory = new DoubleMeanAggregatorFactory("name", "fieldName");
+    DoubleMeanHolder expectedHolder = new DoubleMeanHolder(50.0, 10L);
+
+    DoubleMeanHolder actualHolder = (DoubleMeanHolder) factory.deserialize(expectedHolder);
+    Assert.assertEquals(expectedHolder, actualHolder);
+
+    actualHolder = (DoubleMeanHolder) factory.deserialize(expectedHolder.toBytes());
+    Assert.assertEquals(expectedHolder, actualHolder);
+  }
+
+  @Test
+  public void testFinalizeComputation()
+  {
+    DoubleMeanAggregatorFactory factory = new DoubleMeanAggregatorFactory("name", "fieldName");
+    double sum = 50.0;
+    long count = 10L;
+    double expecterMean = sum / count;
+    DoubleMeanHolder holder = new DoubleMeanHolder(sum, count);
+
+    double actualMean = (Double) factory.finalizeComputation(holder);
+    Assert.assertEquals("", expecterMean, actualMean, 1e-6);
+
+    actualMean = (Double) factory.finalizeComputation(holder.toBytes());
+    Assert.assertEquals("", expecterMean, actualMean, 1e-6);
+
+    Assert.assertNull(factory.finalizeComputation(null));
+  }
+}