Implement Counter#merge

Merge is referenced in isCompatibleWith but is not implemented via the
Counter interface.

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

Author: tgroh

sdk/src/main/java/com/google/cloud/dataflow/sdk/util/common/Counter.java

@@ -19,6 +19,7 @@
 import static com.google.cloud.dataflow.sdk.util.common.Counter.AggregationKind.AND;
 import static com.google.cloud.dataflow.sdk.util.common.Counter.AggregationKind.MEAN;
 import static com.google.cloud.dataflow.sdk.util.common.Counter.AggregationKind.OR;
+import static com.google.common.base.Preconditions.checkArgument;
 
 import com.google.cloud.dataflow.sdk.values.TypeDescriptor;
 import com.google.common.util.concurrent.AtomicDouble;
@@ -328,6 +329,13 @@ public boolean isCompatibleWith(Counter<?> that) {
         && this.getClass().equals(that.getClass());
   }
 
+  /**
+   * Merges this counter with the provided counter, returning this counter with the combined value
+   * of both counters. This may reset the delta of this counter.
+   *
+   * @throws IllegalArgumentException if the provided Counter is not compatible with this Counter
+   */
+  public abstract Counter<T> merge(Counter<T> that);
 
   //////////////////////////////////////////////////////////////////////////////
 
@@ -494,6 +502,25 @@ public CounterMean<Long> getMean() {
       return mean.get();
     }
 
+    @Override
+    public Counter<Long> merge(Counter<Long> that) {
+      checkArgument(this.isCompatibleWith(that), "Counters %s and %s are incompatible", this, that);
+      switch (kind) {
+        case SUM:
+        case MIN:
+        case MAX:
+          return addValue(that.getAggregate());
+        case MEAN:
+          CounterMean<Long> thisCounterMean = this.getMean();
+          CounterMean<Long> thatCounterMean = that.getMean();
+          return resetMeanToValue(
+              thisCounterMean.getCount() + thatCounterMean.getCount(),
+              thisCounterMean.getAggregate() + thatCounterMean.getAggregate());
+        default:
+          throw illegalArgumentException();
+      }
+    }
+
     private static class LongCounterMean implements CounterMean<Long> {
       private final long aggregate;
       private final long count;
@@ -670,6 +697,25 @@ public CounterMean<Double> getMean() {
       return mean.get();
     }
 
+    @Override
+    public Counter<Double> merge(Counter<Double> that) {
+      checkArgument(this.isCompatibleWith(that), "Counters %s and %s are incompatible", this, that);
+      switch (kind) {
+        case SUM:
+        case MIN:
+        case MAX:
+          return addValue(that.getAggregate());
+        case MEAN:
+          CounterMean<Double> thisCounterMean = this.getMean();
+          CounterMean<Double> thatCounterMean = that.getMean();
+          return resetMeanToValue(
+              thisCounterMean.getCount() + thatCounterMean.getCount(),
+              thisCounterMean.getAggregate() + thatCounterMean.getAggregate());
+        default:
+          throw illegalArgumentException();
+      }
+    }
+
     private static class DoubleCounterMean implements CounterMean<Double> {
       private final double aggregate;
       private final long count;
@@ -763,13 +809,18 @@ public Boolean getAggregate() {
     public CounterMean<Boolean> getMean() {
       throw illegalArgumentException();
     }
+
+    @Override
+    public Counter<Boolean> merge(Counter<Boolean> that) {
+      checkArgument(this.isCompatibleWith(that), "Counters %s and %s are incompatible", this, that);
+      return addValue(that.getAggregate());
+    }
   }
 
   /**
    * Implements a {@link Counter} for {@link String} values.
    */
   private static class StringCounter extends Counter<String> {
-
     /** Initializes a new {@link Counter} for {@link String} values. */
     private StringCounter(String name, AggregationKind kind) {
       super(name, kind);
@@ -833,6 +884,15 @@ public CounterMean<String> getMean() {
           throw illegalArgumentException();
       }
     }
+
+    @Override
+    public Counter<String> merge(Counter<String> that) {
+      checkArgument(this.isCompatibleWith(that), "Counters %s and %s are incompatible", this, that);
+      switch (kind) {
+        default:
+          throw illegalArgumentException();
+      }
+    }
   }
 
   /**
@@ -985,6 +1045,25 @@ public CounterMean<Integer> getMean() {
       return mean.get();
     }
 
+    @Override
+    public Counter<Integer> merge(Counter<Integer> that) {
+      checkArgument(this.isCompatibleWith(that), "Counters %s and %s are incompatible", this, that);
+      switch (kind) {
+        case SUM:
+        case MIN:
+        case MAX:
+          return addValue(that.getAggregate());
+        case MEAN:
+          CounterMean<Integer> thisCounterMean = this.getMean();
+          CounterMean<Integer> thatCounterMean = that.getMean();
+          return resetMeanToValue(
+              thisCounterMean.getCount() + thatCounterMean.getCount(),
+              thisCounterMean.getAggregate() + thatCounterMean.getAggregate());
+        default:
+          throw illegalArgumentException();
+      }
+    }
+
     private static class IntegerCounterMean implements CounterMean<Integer> {
       private final int aggregate;
       private final long count;

sdk/src/test/java/com/google/cloud/dataflow/sdk/util/common/CounterTest.java

@@ -33,12 +33,15 @@
 import com.google.cloud.dataflow.sdk.util.CloudCounterUtils;
 import com.google.cloud.dataflow.sdk.util.common.Counter.CounterMean;
 
+import org.junit.Rule;
 import org.junit.Test;
+import org.junit.rules.ExpectedException;
 import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;
 
 import java.util.Arrays;
 import java.util.HashSet;
+import java.util.List;
 import java.util.Set;
 
 /**
@@ -47,6 +50,9 @@
 @RunWith(JUnit4.class)
 public class CounterTest {
 
+  @Rule
+  public ExpectedException thrown = ExpectedException.none();
+
   private static MetricUpdate flush(Counter<?> c) {
     // TODO: Move this out into a separate Counter test.
     return CounterTestUtils.extractCounterUpdate(c, true);
@@ -211,6 +217,13 @@ public void testSumLong() {
     c.resetToValue(100L).addValue(17L).addValue(49L);
     expectedTotal = expectedDelta = 166;
     assertOK(expectedTotal, expectedDelta, c);
+
+    Counter<Long> other = Counter.longs("sum-long", SUM);
+    other.addValue(12L);
+    expectedDelta = 12L;
+    expectedTotal += 12L;
+    c.merge(other);
+    assertOK(expectedTotal, expectedDelta, c);
   }
 
   @Test
@@ -241,6 +254,13 @@ public void testSumDouble() {
     c.resetToValue(Math.sqrt(17)).addValue(17.0).addValue(49.0);
     expectedTotal = expectedDelta = Math.sqrt(17.0) + 17.0 + 49.0;
     assertOK(expectedTotal, expectedDelta, c);
+
+    Counter<Double> other = Counter.doubles("sum-double", SUM);
+    other.addValue(12 * Math.PI);
+    expectedDelta = 12 * Math.PI;
+    expectedTotal += 12 * Math.PI;
+    c.merge(other);
+    assertOK(expectedTotal, expectedDelta, c);
   }
 
 
@@ -272,6 +292,12 @@ public void testMaxLong() {
     c.resetToValue(100L).addValue(171L).addValue(49L);
     expectedTotal = expectedDelta = 171;
     assertOK(expectedTotal, expectedDelta, c);
+
+    Counter<Long> other = Counter.longs("max-long", MAX);
+    other.addValue(12L);
+    expectedDelta = 12L;
+    c.merge(other);
+    assertOK(expectedTotal, expectedDelta, c);
   }
 
   @Test
@@ -300,6 +326,12 @@ public void testMaxDouble() {
     c.resetToValue(Math.sqrt(17)).addValue(171.0).addValue(49.0);
     expectedTotal = expectedDelta = 171.0;
     assertOK(expectedTotal, expectedDelta, c);
+
+    Counter<Double> other = Counter.doubles("max-double", MAX);
+    other.addValue(12 * Math.PI);
+    expectedDelta = 12 * Math.PI;
+    c.merge(other);
+    assertOK(expectedTotal, expectedDelta, c);
   }
 
 
@@ -331,6 +363,12 @@ public void testMinLong() {
     c.resetToValue(100L).addValue(171L).addValue(49L);
     expectedTotal = expectedDelta = 49;
     assertOK(expectedTotal, expectedDelta, c);
+
+    Counter<Long> other = Counter.longs("min-long", MIN);
+    other.addValue(42L);
+    expectedTotal = expectedDelta = 42L;
+    c.merge(other);
+    assertOK(expectedTotal, expectedDelta, c);
   }
 
   @Test
@@ -359,6 +397,12 @@ public void testMinDouble() {
     c.resetToValue(Math.sqrt(17)).addValue(171.0).addValue(0.0);
     expectedTotal = expectedDelta = 0.0;
     assertOK(expectedTotal, expectedDelta, c);
+
+    Counter<Double> other = Counter.doubles("min-double", MIN);
+    other.addValue(42 * Math.E);
+    expectedDelta = 42 * Math.E;
+    c.merge(other);
+    assertOK(expectedTotal, expectedDelta, c);
   }
 
 
@@ -419,6 +463,15 @@ public void testMeanLong() {
     expTotal = expDelta = 166;
     expCountTotal = expCountDelta = 5;
     assertMean(expTotal, expDelta, expCountTotal, expCountDelta, c);
+
+    Counter<Long> other = Counter.longs("mean-long", MEAN);
+    other.addValue(12L).addValue(44L).addValue(-5L);
+    expTotal += 12L + 44L - 5L;
+    expDelta += 12L + 44L - 5L;
+    expCountTotal += 3;
+    expCountDelta += 3;
+    c.merge(other);
+    assertMean(expTotal, expDelta, expCountTotal, expCountDelta, c);
   }
 
   @Test
@@ -458,6 +511,15 @@ public void testMeanDouble() {
     expTotal = expDelta = Math.sqrt(17.0) + 17.0 + 49.0;
     expCountTotal = expCountDelta = 5;
     assertMean(expTotal, expDelta, expCountTotal, expCountDelta, c);
+
+    Counter<Double> other = Counter.doubles("mean-double", MEAN);
+    other.addValue(3 * Math.PI).addValue(12 * Math.E);
+    expTotal += 3 * Math.PI + 12 * Math.E;
+    expDelta += 3 * Math.PI + 12 * Math.E;
+    expCountTotal += 2;
+    expCountDelta += 2;
+    c.merge(other);
+    assertMean(expTotal, expDelta, expCountTotal, expCountDelta, c);
   }
 
 
@@ -590,4 +652,40 @@ public void testExtraction() {
     assertEquals(cloudCountersFromSet, cloudCountersFromArray);
     assertEquals(2, cloudCountersFromSet.size());
   }
+
+  @Test
+  public void testMergeIncompatibleCounters() {
+    Counter<Long> longSums = Counter.longs("longsums", SUM);
+    Counter<Long> longMean = Counter.longs("longmean", MEAN);
+    Counter<Long> longMin = Counter.longs("longmin", MIN);
+
+    Counter<Long> otherLongSums = Counter.longs("othersums", SUM);
+    Counter<Long> otherLongMean = Counter.longs("otherlongmean", MEAN);
+
+    Counter<Double> doubleSums = Counter.doubles("doublesums", SUM);
+    Counter<Double> doubleMean = Counter.doubles("doublemean", MEAN);
+
+    Counter<Boolean> boolAnd = Counter.booleans("and", AND);
+    Counter<Boolean> boolOr = Counter.booleans("or", OR);
+
+    List<Counter<Long>> longCounters =
+        Arrays.asList(longSums, longMean, longMin, otherLongSums, otherLongMean);
+    for (Counter<Long> left : longCounters) {
+      for (Counter<Long> right : longCounters) {
+        if (left != right) {
+          assertIncompatibleMerge(left, right);
+        }
+      }
+    }
+
+    assertIncompatibleMerge(doubleSums, doubleMean);
+    assertIncompatibleMerge(boolAnd, boolOr);
+  }
+
+  private <T> void assertIncompatibleMerge(Counter<T> left, Counter<T> right) {
+    thrown.expect(IllegalArgumentException.class);
+    thrown.expectMessage("Counters");
+    thrown.expectMessage("are incompatible");
+    left.merge(right);
+  }
 }