Optimize mergeAccumulators by reusing an existing accumulator

Also, added units tests for Sum, Min, Max, Mean.

----Release Notes----
[]
-------------
Created by MOE: https://github.com/google/moe
MOE_MIGRATED_REVID=112723885

Author: peihe

sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/Combine.java

@@ -534,17 +534,23 @@ public Holder<V> addInput(Holder<V> accumulator, V input) {
 
     @Override
     public Holder<V> mergeAccumulators(Iterable<Holder<V>> accumulators) {
-      Holder<V> running = new Holder<>();
-      for (Holder<V> accumulator : accumulators) {
-        if (accumulator.present) {
-          if (running.present) {
-            running.set(apply(running.value, accumulator.value));
-          } else {
-            running.set(accumulator.value);
+      Iterator<Holder<V>> iter = accumulators.iterator();
+      if (!iter.hasNext()) {
+        return createAccumulator();
+      } else {
+        Holder<V> running = iter.next();
+        while (iter.hasNext()) {
+          Holder<V> accum = iter.next();
+          if (accum.present) {
+            if (running.present) {
+              running.set(apply(running.value, accum.value));
+            } else {
+              running.set(accum.value);
+            }
           }
         }
+        return running;
       }
-      return running;
     }
 
     @Override
@@ -632,7 +638,9 @@ public void verifyDeterministic() throws NonDeterministicException {
 
   /**
    * An abstract subclass of {@link CombineFn} for implementing combiners that are more
-   * easily and efficiently expressed as binary operations on <code>int</code>s.
+   * easily and efficiently expressed as binary operations on <code>int</code>s
+   *
+   * <p> It uses {@code int[0]} as the mutable accumulator.
    */
   public abstract static class BinaryCombineIntegerFn extends CombineFn<Integer, int[], Integer> {
 
@@ -664,11 +672,11 @@ public int[] mergeAccumulators(Iterable<int[]> accumulators) {
       if (!iter.hasNext()) {
         return createAccumulator();
       } else {
-        int running = iter.next()[0];
+        int[] running = iter.next();
         while (iter.hasNext()) {
-          running = apply(running, iter.next()[0]);
+          running[0] = apply(running[0], iter.next()[0]);
         }
-        return wrap(running);
+        return running;
       }
     }
 
@@ -713,6 +721,8 @@ public Counter<Integer> getCounter(String name) {
   /**
    * An abstract subclass of {@link CombineFn} for implementing combiners that are more
    * easily and efficiently expressed as binary operations on <code>long</code>s.
+   *
+   * <p> It uses {@code long[0]} as the mutable accumulator.
    */
   public abstract static class BinaryCombineLongFn extends CombineFn<Long, long[], Long> {
     /**
@@ -743,11 +753,11 @@ public long[] mergeAccumulators(Iterable<long[]> accumulators) {
       if (!iter.hasNext()) {
         return createAccumulator();
       } else {
-        long running = iter.next()[0];
+        long[] running = iter.next();
         while (iter.hasNext()) {
-          running = apply(running, iter.next()[0]);
+          running[0] = apply(running[0], iter.next()[0]);
         }
-        return wrap(running);
+        return running;
       }
     }
 
@@ -791,6 +801,8 @@ public Counter<Long> getCounter(String name) {
   /**
    * An abstract subclass of {@link CombineFn} for implementing combiners that are more
    * easily and efficiently expressed as binary operations on <code>double</code>s.
+   *
+   * <p> It uses {@code double[0]} as the mutable accumulator.
    */
   public abstract static class BinaryCombineDoubleFn extends CombineFn<Double, double[], Double> {
 
@@ -822,11 +834,11 @@ public double[] mergeAccumulators(Iterable<double[]> accumulators) {
       if (!iter.hasNext()) {
         return createAccumulator();
       } else {
-        double running = iter.next()[0];
+        double[] running = iter.next();
         while (iter.hasNext()) {
-          running = apply(running, iter.next()[0]);
+          running[0] = apply(running[0], iter.next()[0]);
         }
-        return wrap(running);
+        return running;
       }
     }
 

sdk/src/test/java/com/google/cloud/dataflow/sdk/transforms/MaxTest.java

@@ -16,8 +16,11 @@
 
 package com.google.cloud.dataflow.sdk.transforms;
 
+import static com.google.cloud.dataflow.sdk.TestUtils.checkCombineFn;
 import static org.junit.Assert.assertEquals;
 
+import com.google.common.collect.Lists;
+
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;
@@ -36,4 +39,28 @@ public void testMeanGetNames() {
     assertEquals("Max.PerKey", Max.doublesPerKey().getName());
     assertEquals("Max.PerKey", Max.longsPerKey().getName());
   }
+
+  @Test
+  public void testMaxIntegerFn() {
+    checkCombineFn(
+        new Max.MaxIntegerFn(),
+        Lists.newArrayList(1, 2, 3, 4),
+        4);
+  }
+
+  @Test
+  public void testMaxLongFn() {
+    checkCombineFn(
+        new Max.MaxLongFn(),
+        Lists.newArrayList(1L, 2L, 3L, 4L),
+        4L);
+  }
+
+  @Test
+  public void testMaxDoubleFn() {
+    checkCombineFn(
+        new Max.MaxDoubleFn(),
+        Lists.newArrayList(1.0, 2.0, 3.0, 4.0),
+        4.0);
+  }
 }

sdk/src/test/java/com/google/cloud/dataflow/sdk/transforms/MeanTest.java

@@ -16,12 +16,14 @@
 
 package com.google.cloud.dataflow.sdk.transforms;
 
+import static com.google.cloud.dataflow.sdk.TestUtils.checkCombineFn;
 import static org.junit.Assert.assertEquals;
 
 import com.google.cloud.dataflow.sdk.coders.Coder;
 import com.google.cloud.dataflow.sdk.testing.CoderProperties;
 import com.google.cloud.dataflow.sdk.transforms.Mean.CountSum;
 import com.google.cloud.dataflow.sdk.transforms.Mean.CountSumCoder;
+import com.google.common.collect.Lists;
 
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -59,4 +61,12 @@ public void testCountSumCoderEncodeDecode() throws Exception {
   public void testCountSumCoderSerializable() throws Exception {
     CoderProperties.coderSerializable(TEST_CODER);
   }
+
+  @Test
+  public void testMeanFn() throws Exception {
+    checkCombineFn(
+        new Mean.MeanFn<Integer>(),
+        Lists.newArrayList(1, 2, 3, 4),
+        2.5);
+  }
 }

sdk/src/test/java/com/google/cloud/dataflow/sdk/transforms/MinTest.java

@@ -16,8 +16,11 @@
 
 package com.google.cloud.dataflow.sdk.transforms;
 
+import static com.google.cloud.dataflow.sdk.TestUtils.checkCombineFn;
 import static org.junit.Assert.assertEquals;
 
+import com.google.common.collect.Lists;
+
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;
@@ -36,4 +39,28 @@ public void testMeanGetNames() {
     assertEquals("Min.PerKey", Min.doublesPerKey().getName());
     assertEquals("Min.PerKey", Min.longsPerKey().getName());
   }
+
+  @Test
+  public void testMinIntegerFn() {
+    checkCombineFn(
+        new Min.MinIntegerFn(),
+        Lists.newArrayList(1, 2, 3, 4),
+        1);
+  }
+
+  @Test
+  public void testMinLongFn() {
+    checkCombineFn(
+        new Min.MinLongFn(),
+        Lists.newArrayList(1L, 2L, 3L, 4L),
+        1L);
+  }
+
+  @Test
+  public void testMinDoubleFn() {
+    checkCombineFn(
+        new Min.MinDoubleFn(),
+        Lists.newArrayList(1.0, 2.0, 3.0, 4.0),
+        1.0);
+  }
 }

sdk/src/test/java/com/google/cloud/dataflow/sdk/transforms/SumTest.java

@@ -13,11 +13,13 @@
  * License for the specific language governing permissions and limitations under
  * the License.
  */
-
 package com.google.cloud.dataflow.sdk.transforms;
 
+import static com.google.cloud.dataflow.sdk.TestUtils.checkCombineFn;
 import static org.junit.Assert.assertEquals;
 
+import com.google.common.collect.Lists;
+
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;
@@ -37,4 +39,28 @@ public void testSumGetNames() {
     assertEquals("Sum.PerKey", Sum.doublesPerKey().getName());
     assertEquals("Sum.PerKey", Sum.longsPerKey().getName());
   }
+
+  @Test
+  public void testSumIntegerFn() {
+    checkCombineFn(
+        new Sum.SumIntegerFn(),
+        Lists.newArrayList(1, 2, 3, 4),
+        10);
+  }
+
+  @Test
+  public void testSumLongFn() {
+    checkCombineFn(
+        new Sum.SumLongFn(),
+        Lists.newArrayList(1L, 2L, 3L, 4L),
+        10L);
+  }
+
+  @Test
+  public void testSumDoubleFn() {
+    checkCombineFn(
+        new Sum.SumDoubleFn(),
+        Lists.newArrayList(1.0, 2.0, 3.0, 4.0),
+        10.0);
+  }
 }