Repository: beam Updated Branches: refs/heads/master e794f1486 -> d86db15ba
http://git-wip-us.apache.org/repos/asf/beam/blob/78a360ea/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/DoFnTest.java ---------------------------------------------------------------------- diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/DoFnTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/DoFnTest.java index 19b7c51..4e80154 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/DoFnTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/DoFnTest.java @@ -28,7 +28,6 @@ import org.apache.beam.sdk.Pipeline.PipelineExecutionException; import org.apache.beam.sdk.testing.NeedsRunner; import org.apache.beam.sdk.testing.TestPipeline; import org.apache.beam.sdk.transforms.Combine.CombineFn; -import org.apache.beam.sdk.transforms.Max.MaxIntegerFn; import org.apache.beam.sdk.transforms.display.DisplayData; import org.junit.Rule; import org.junit.Test; @@ -59,7 +58,7 @@ public class DoFnTest implements Serializable { @Test public void testCreateAggregatorWithCombinerSucceeds() { String name = "testAggregator"; - Sum.SumLongFn combiner = new Sum.SumLongFn(); + Combine.BinaryCombineLongFn combiner = Sum.ofLongs(); DoFn<Void, Void> doFn = new NoOpDoFn(); @@ -76,7 +75,7 @@ public class DoFnTest implements Serializable { DoFn<Void, Void> doFn = new NoOpDoFn(); - doFn.createAggregator(null, new Sum.SumLongFn()); + doFn.createAggregator(null, Sum.ofLongs()); } @Test @@ -106,7 +105,7 @@ public class DoFnTest implements Serializable { @Test public void testCreateAggregatorWithSameNameThrowsException() { String name = "testAggregator"; - CombineFn<Double, ?, Double> combiner = new Max.MaxDoubleFn(); + CombineFn<Double, ?, Double> combiner = Max.ofDoubles(); DoFn<Void, Void> doFn = new NoOpDoFn(); @@ -124,7 +123,7 @@ public class DoFnTest implements Serializable { public void testCreateAggregatorsWithDifferentNamesSucceeds() { String nameOne = "testAggregator"; String nameTwo = "aggregatorPrime"; - CombineFn<Double, ?, Double> combiner = new Max.MaxDoubleFn(); + CombineFn<Double, ?, Double> combiner = Max.ofDoubles(); DoFn<Void, Void> doFn = new NoOpDoFn(); @@ -151,7 +150,7 @@ public class DoFnTest implements Serializable { TestPipeline p = createTestPipeline(new DoFn<String, String>() { @StartBundle public void startBundle(Context c) { - createAggregator("anyAggregate", new MaxIntegerFn()); + createAggregator("anyAggregate", Max.ofIntegers()); } @ProcessElement @@ -170,7 +169,7 @@ public class DoFnTest implements Serializable { TestPipeline p = createTestPipeline(new DoFn<String, String>() { @ProcessElement public void processElement(ProcessContext c) { - createAggregator("anyAggregate", new MaxIntegerFn()); + createAggregator("anyAggregate", Max.ofIntegers()); } }); @@ -186,7 +185,7 @@ public class DoFnTest implements Serializable { TestPipeline p = createTestPipeline(new DoFn<String, String>() { @FinishBundle public void finishBundle(Context c) { - createAggregator("anyAggregate", new MaxIntegerFn()); + createAggregator("anyAggregate", Max.ofIntegers()); } @ProcessElement http://git-wip-us.apache.org/repos/asf/beam/blob/78a360ea/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/DoFnTesterTest.java ---------------------------------------------------------------------- diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/DoFnTesterTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/DoFnTesterTest.java index 3859c9f..699687f 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/DoFnTesterTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/DoFnTesterTest.java @@ -404,11 +404,11 @@ public class DoFnTesterTest { * {@link DoFn.ProcessElement @ProcessElement}. */ private static class CounterDoFn extends DoFn<Long, String> { - Aggregator<Long, Long> agg = createAggregator("ctr", new Sum.SumLongFn()); + Aggregator<Long, Long> agg = createAggregator("ctr", Sum.ofLongs()); Aggregator<Long, Long> startBundleCalls = - createAggregator("startBundleCalls", new Sum.SumLongFn()); + createAggregator("startBundleCalls", Sum.ofLongs()); Aggregator<Long, Long> finishBundleCalls = - createAggregator("finishBundleCalls", new Sum.SumLongFn()); + createAggregator("finishBundleCalls", Sum.ofLongs()); private enum LifecycleState { UNINITIALIZED, http://git-wip-us.apache.org/repos/asf/beam/blob/78a360ea/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/MaxTest.java ---------------------------------------------------------------------- diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/MaxTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/MaxTest.java index 4aa39a3..2b43560 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/MaxTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/MaxTest.java @@ -46,7 +46,7 @@ public class MaxTest { @Test public void testMaxIntegerFn() { checkCombineFn( - new Max.MaxIntegerFn(), + Max.ofIntegers(), Lists.newArrayList(1, 2, 3, 4), 4); } @@ -54,7 +54,7 @@ public class MaxTest { @Test public void testMaxLongFn() { checkCombineFn( - new Max.MaxLongFn(), + Max.ofLongs(), Lists.newArrayList(1L, 2L, 3L, 4L), 4L); } @@ -62,7 +62,7 @@ public class MaxTest { @Test public void testMaxDoubleFn() { checkCombineFn( - new Max.MaxDoubleFn(), + Max.ofDoubles(), Lists.newArrayList(1.0, 2.0, 3.0, 4.0), 4.0); } http://git-wip-us.apache.org/repos/asf/beam/blob/78a360ea/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/MeanTest.java ---------------------------------------------------------------------- diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/MeanTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/MeanTest.java index 84741ee..79ebc25 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/MeanTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/MeanTest.java @@ -65,7 +65,7 @@ public class MeanTest { @Test public void testMeanFn() throws Exception { checkCombineFn( - new Mean.MeanFn<Integer>(), + Mean.<Integer>of(), Lists.newArrayList(1, 2, 3, 4), 2.5); } http://git-wip-us.apache.org/repos/asf/beam/blob/78a360ea/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/MinTest.java ---------------------------------------------------------------------- diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/MinTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/MinTest.java index 4334ed9..e89b223 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/MinTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/MinTest.java @@ -46,7 +46,7 @@ public class MinTest { @Test public void testMinIntegerFn() { checkCombineFn( - new Min.MinIntegerFn(), + Min.ofIntegers(), Lists.newArrayList(1, 2, 3, 4), 1); } @@ -54,7 +54,7 @@ public class MinTest { @Test public void testMinLongFn() { checkCombineFn( - new Min.MinLongFn(), + Min.ofLongs(), Lists.newArrayList(1L, 2L, 3L, 4L), 1L); } @@ -62,7 +62,7 @@ public class MinTest { @Test public void testMinDoubleFn() { checkCombineFn( - new Min.MinDoubleFn(), + Min.ofDoubles(), Lists.newArrayList(1.0, 2.0, 3.0, 4.0), 1.0); } http://git-wip-us.apache.org/repos/asf/beam/blob/78a360ea/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/OldDoFnContextTest.java ---------------------------------------------------------------------- diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/OldDoFnContextTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/OldDoFnContextTest.java index b2d4aed..b5cb286 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/OldDoFnContextTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/OldDoFnContextTest.java @@ -55,7 +55,7 @@ public class OldDoFnContextTest { @Test public void testSetupDelegateAggregatorsCreatesAndLinksDelegateAggregators() { - Sum.SumLongFn combiner = new Sum.SumLongFn(); + Combine.BinaryCombineLongFn combiner = Sum.ofLongs(); Aggregator<Long, Long> delegateAggregator = fn.createAggregator("test", combiner); http://git-wip-us.apache.org/repos/asf/beam/blob/78a360ea/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/OldDoFnTest.java ---------------------------------------------------------------------- diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/OldDoFnTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/OldDoFnTest.java index cc84252..1c767b1 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/OldDoFnTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/OldDoFnTest.java @@ -26,7 +26,6 @@ import static org.junit.Assert.assertThat; import java.io.Serializable; import org.apache.beam.sdk.options.PipelineOptions; import org.apache.beam.sdk.transforms.Combine.CombineFn; -import org.apache.beam.sdk.transforms.Max.MaxIntegerFn; import org.apache.beam.sdk.transforms.display.DisplayData; import org.apache.beam.sdk.values.TupleTag; import org.joda.time.Instant; @@ -48,7 +47,7 @@ public class OldDoFnTest implements Serializable { @Test public void testCreateAggregatorWithCombinerSucceeds() { String name = "testAggregator"; - Sum.SumLongFn combiner = new Sum.SumLongFn(); + Combine.BinaryCombineLongFn combiner = Sum.ofLongs(); OldDoFn<Void, Void> doFn = new NoOpOldDoFn<>(); @@ -65,7 +64,7 @@ public class OldDoFnTest implements Serializable { OldDoFn<Void, Void> doFn = new NoOpOldDoFn<>(); - doFn.createAggregator(null, new Sum.SumLongFn()); + doFn.createAggregator(null, Sum.ofLongs()); } @Test @@ -95,7 +94,7 @@ public class OldDoFnTest implements Serializable { @Test public void testCreateAggregatorWithSameNameThrowsException() { String name = "testAggregator"; - CombineFn<Double, ?, Double> combiner = new Max.MaxDoubleFn(); + CombineFn<Double, ?, Double> combiner = Max.ofDoubles(); OldDoFn<Void, Void> doFn = new NoOpOldDoFn<>(); @@ -113,7 +112,7 @@ public class OldDoFnTest implements Serializable { public void testCreateAggregatorsWithDifferentNamesSucceeds() { String nameOne = "testAggregator"; String nameTwo = "aggregatorPrime"; - CombineFn<Double, ?, Double> combiner = new Max.MaxDoubleFn(); + CombineFn<Double, ?, Double> combiner = Max.ofDoubles(); OldDoFn<Void, Void> doFn = new NoOpOldDoFn<>(); @@ -135,7 +134,7 @@ public class OldDoFnTest implements Serializable { context.setupDelegateAggregators(); thrown.expect(isA(IllegalStateException.class)); - fn.createAggregator("anyAggregate", new MaxIntegerFn()); + fn.createAggregator("anyAggregate", Max.ofIntegers()); } private OldDoFn<String, String>.Context createContext(OldDoFn<String, String> fn) { http://git-wip-us.apache.org/repos/asf/beam/blob/78a360ea/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/SimpleStatsFnsTest.java ---------------------------------------------------------------------- diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/SimpleStatsFnsTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/SimpleStatsFnsTest.java index a782ecc..2fd86c6 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/SimpleStatsFnsTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/SimpleStatsFnsTest.java @@ -78,31 +78,31 @@ public class SimpleStatsFnsTest { @Test public void testInstantStats() { - assertEquals(new Instant(1000), Min.MinFn.<Instant>naturalOrder().apply( + assertEquals(new Instant(1000), Min.<Instant>naturalOrder().apply( Arrays.asList(new Instant(1000), new Instant(2000)))); - assertEquals(null, Min.MinFn.<Instant>naturalOrder().apply( + assertEquals(null, Min.<Instant>naturalOrder().apply( Collections.<Instant>emptyList())); - assertEquals(new Instant(5000), Min.MinFn.<Instant>naturalOrder(new Instant(5000)).apply( + assertEquals(new Instant(5000), Min.<Instant>naturalOrder(new Instant(5000)).apply( Collections.<Instant>emptyList())); - assertEquals(new Instant(2000), Max.MaxFn.<Instant>naturalOrder().apply( + assertEquals(new Instant(2000), Max.<Instant>naturalOrder().apply( Arrays.asList(new Instant(1000), new Instant(2000)))); - assertEquals(null, Max.MaxFn.<Instant>naturalOrder().apply( + assertEquals(null, Max.<Instant>naturalOrder().apply( Collections.<Instant>emptyList())); - assertEquals(new Instant(5000), Max.MaxFn.<Instant>naturalOrder(new Instant(5000)).apply( + assertEquals(new Instant(5000), Max.<Instant>naturalOrder(new Instant(5000)).apply( Collections.<Instant>emptyList())); } @Test public void testDoubleStats() { for (TestCase<Double> t : DOUBLE_CASES) { - assertEquals(t.sum, new Sum.SumDoubleFn().apply(t.data), + assertEquals(t.sum, Sum.ofDoubles().apply(t.data), DOUBLE_COMPARISON_ACCURACY); - assertEquals(t.min, new Min.MinDoubleFn().apply(t.data), + assertEquals(t.min, Min.ofDoubles().apply(t.data), DOUBLE_COMPARISON_ACCURACY); - assertEquals(t.max, new Max.MaxDoubleFn().apply(t.data), + assertEquals(t.max, Max.ofDoubles().apply(t.data), DOUBLE_COMPARISON_ACCURACY); - assertEquals(t.mean, new Mean.MeanFn<Double>().apply(t.data), + assertEquals(t.mean, Mean.<Double>of().apply(t.data), DOUBLE_COMPARISON_ACCURACY); } } @@ -110,20 +110,20 @@ public class SimpleStatsFnsTest { @Test public void testIntegerStats() { for (TestCase<Integer> t : INTEGER_CASES) { - assertEquals(t.sum, new Sum.SumIntegerFn().apply(t.data)); - assertEquals(t.min, new Min.MinIntegerFn().apply(t.data)); - assertEquals(t.max, new Max.MaxIntegerFn().apply(t.data)); - assertEquals(t.mean, new Mean.MeanFn<Integer>().apply(t.data)); + assertEquals(t.sum, Sum.ofIntegers().apply(t.data)); + assertEquals(t.min, Min.ofIntegers().apply(t.data)); + assertEquals(t.max, Max.ofIntegers().apply(t.data)); + assertEquals(t.mean, Mean.<Integer>of().apply(t.data)); } } @Test public void testLongStats() { for (TestCase<Long> t : LONG_CASES) { - assertEquals(t.sum, new Sum.SumLongFn().apply(t.data)); - assertEquals(t.min, new Min.MinLongFn().apply(t.data)); - assertEquals(t.max, new Max.MaxLongFn().apply(t.data)); - assertEquals(t.mean, new Mean.MeanFn<Long>().apply(t.data)); + assertEquals(t.sum, Sum.ofLongs().apply(t.data)); + assertEquals(t.min, Min.ofLongs().apply(t.data)); + assertEquals(t.max, Max.ofLongs().apply(t.data)); + assertEquals(t.mean, Mean.<Long>of().apply(t.data)); } } } http://git-wip-us.apache.org/repos/asf/beam/blob/78a360ea/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/SumTest.java ---------------------------------------------------------------------- diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/SumTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/SumTest.java index 04c0186..b2f8aa8 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/SumTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/SumTest.java @@ -52,7 +52,7 @@ public class SumTest { @Test public void testSumIntegerFn() { checkCombineFn( - new Sum.SumIntegerFn(), + Sum.ofIntegers(), Lists.newArrayList(1, 2, 3, 4), 10); } @@ -60,7 +60,7 @@ public class SumTest { @Test public void testSumLongFn() { checkCombineFn( - new Sum.SumLongFn(), + Sum.ofLongs(), Lists.newArrayList(1L, 2L, 3L, 4L), 10L); } @@ -68,14 +68,14 @@ public class SumTest { @Test public void testSumDoubleFn() { checkCombineFn( - new Sum.SumDoubleFn(), + Sum.ofDoubles(), Lists.newArrayList(1.0, 2.0, 3.0, 4.0), 10.0); } @Test public void testGetAccumulatorCoderEquals() { - Sum.SumIntegerFn sumIntegerFn = new Sum.SumIntegerFn(); + Combine.BinaryCombineIntegerFn sumIntegerFn = Sum.ofIntegers(); assertEquals( sumIntegerFn.getAccumulatorCoder(STANDARD_REGISTRY, VarIntCoder.of()), sumIntegerFn.getAccumulatorCoder(STANDARD_REGISTRY, VarIntCoder.of())); @@ -83,7 +83,7 @@ public class SumTest { sumIntegerFn.getAccumulatorCoder(STANDARD_REGISTRY, VarIntCoder.of()), sumIntegerFn.getAccumulatorCoder(STANDARD_REGISTRY, BigEndianIntegerCoder.of())); - Sum.SumLongFn sumLongFn = new Sum.SumLongFn(); + Combine.BinaryCombineLongFn sumLongFn = Sum.ofLongs(); assertEquals( sumLongFn.getAccumulatorCoder(STANDARD_REGISTRY, VarLongCoder.of()), sumLongFn.getAccumulatorCoder(STANDARD_REGISTRY, VarLongCoder.of())); @@ -91,7 +91,7 @@ public class SumTest { sumLongFn.getAccumulatorCoder(STANDARD_REGISTRY, VarLongCoder.of()), sumLongFn.getAccumulatorCoder(STANDARD_REGISTRY, BigEndianLongCoder.of())); - Sum.SumDoubleFn sumDoubleFn = new Sum.SumDoubleFn(); + Combine.BinaryCombineDoubleFn sumDoubleFn = Sum.ofDoubles(); assertEquals( sumDoubleFn.getAccumulatorCoder(STANDARD_REGISTRY, DoubleCoder.of()), sumDoubleFn.getAccumulatorCoder(STANDARD_REGISTRY, DoubleCoder.of())); http://git-wip-us.apache.org/repos/asf/beam/blob/78a360ea/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ViewTest.java ---------------------------------------------------------------------- diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ViewTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ViewTest.java index 1d8b32c..0249ac9 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ViewTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ViewTest.java @@ -1120,7 +1120,7 @@ public class ViewTest implements Serializable { final PCollectionView<Map<String, Integer>> view = pipeline.apply("CreateSideInput", Create.of(KV.of("a", 1), KV.of("a", 20), KV.of("b", 3))) - .apply("SumIntegers", Combine.perKey(new Sum.SumIntegerFn().<String>asKeyedFn())) + .apply("SumIntegers", Combine.perKey(Sum.ofIntegers().<String>asKeyedFn())) .apply(View.<String, Integer>asMap()); PCollection<KV<String, Integer>> output = http://git-wip-us.apache.org/repos/asf/beam/blob/78a360ea/sdks/java/core/src/test/java/org/apache/beam/sdk/util/CombineFnUtilTest.java ---------------------------------------------------------------------- diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/CombineFnUtilTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/CombineFnUtilTest.java index fe81275..36a90e9 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/CombineFnUtilTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/CombineFnUtilTest.java @@ -70,18 +70,18 @@ public class CombineFnUtilTest { @Test public void testToFnWithContextIdempotent() throws Exception { CombineFnWithContext<Integer, int[], Integer> fnWithContext = - CombineFnUtil.toFnWithContext(new Sum.SumIntegerFn()); + CombineFnUtil.toFnWithContext(Sum.ofIntegers()); assertTrue(fnWithContext == CombineFnUtil.toFnWithContext(fnWithContext)); KeyedCombineFnWithContext<Object, Integer, int[], Integer> keyedFnWithContext = - CombineFnUtil.toFnWithContext(new Sum.SumIntegerFn().asKeyedFn()); + CombineFnUtil.toFnWithContext(Sum.ofIntegers().asKeyedFn()); assertTrue(keyedFnWithContext == CombineFnUtil.toFnWithContext(keyedFnWithContext)); } @Test public void testToFnWithContext() throws Exception { CombineFnWithContext<Integer, int[], Integer> fnWithContext = - CombineFnUtil.toFnWithContext(new Sum.SumIntegerFn()); + CombineFnUtil.toFnWithContext(Sum.ofIntegers()); List<Integer> inputs = ImmutableList.of(1, 2, 3, 4); Context nullContext = CombineContextFactory.nullContext(); int[] accum = fnWithContext.createAccumulator(nullContext); @@ -91,7 +91,7 @@ public class CombineFnUtilTest { assertEquals(10, fnWithContext.extractOutput(accum, nullContext).intValue()); KeyedCombineFnWithContext<String, Integer, int[], Integer> keyedFnWithContext = - CombineFnUtil.toFnWithContext(new Sum.SumIntegerFn().<String>asKeyedFn()); + CombineFnUtil.toFnWithContext(Sum.ofIntegers().<String>asKeyedFn()); String key = "key"; accum = keyedFnWithContext.createAccumulator(key, nullContext); for (Integer i : inputs) { http://git-wip-us.apache.org/repos/asf/beam/blob/78a360ea/sdks/java/core/src/test/java/org/apache/beam/sdk/util/state/InMemoryStateInternalsTest.java ---------------------------------------------------------------------- diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/state/InMemoryStateInternalsTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/state/InMemoryStateInternalsTest.java index 08a6bc1..e43ad36 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/state/InMemoryStateInternalsTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/state/InMemoryStateInternalsTest.java @@ -48,7 +48,7 @@ public class InMemoryStateInternalsTest { StateTags.value("stringValue", StringUtf8Coder.of()); private static final StateTag<Object, AccumulatorCombiningState<Integer, int[], Integer>> SUM_INTEGER_ADDR = StateTags.combiningValueFromInputInternal( - "sumInteger", VarIntCoder.of(), new Sum.SumIntegerFn()); + "sumInteger", VarIntCoder.of(), Sum.ofIntegers()); private static final StateTag<Object, BagState<String>> STRING_BAG_ADDR = StateTags.bag("stringBag", StringUtf8Coder.of()); private static final StateTag<Object, WatermarkHoldState<BoundedWindow>> http://git-wip-us.apache.org/repos/asf/beam/blob/78a360ea/sdks/java/core/src/test/java/org/apache/beam/sdk/util/state/StateTagTest.java ---------------------------------------------------------------------- diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/state/StateTagTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/state/StateTagTest.java index 2c8c9cc..7c06dbd 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/state/StateTagTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/state/StateTagTest.java @@ -24,10 +24,9 @@ import org.apache.beam.sdk.coders.BigEndianIntegerCoder; import org.apache.beam.sdk.coders.Coder; import org.apache.beam.sdk.coders.CoderRegistry; import org.apache.beam.sdk.coders.VarIntCoder; +import org.apache.beam.sdk.transforms.Combine; import org.apache.beam.sdk.transforms.Max; -import org.apache.beam.sdk.transforms.Max.MaxIntegerFn; import org.apache.beam.sdk.transforms.Min; -import org.apache.beam.sdk.transforms.Min.MinIntegerFn; import org.apache.beam.sdk.transforms.windowing.OutputTimeFns; import org.apache.beam.sdk.util.CombineFnUtil; import org.junit.Test; @@ -86,10 +85,10 @@ public class StateTagTest { @SuppressWarnings({"unchecked", "rawtypes"}) @Test public void testCombiningValueEquality() { - MaxIntegerFn maxFn = new Max.MaxIntegerFn(); + Combine.BinaryCombineIntegerFn maxFn = Max.ofIntegers(); Coder<Integer> input1 = VarIntCoder.of(); Coder<Integer> input2 = BigEndianIntegerCoder.of(); - MinIntegerFn minFn = new Min.MinIntegerFn(); + Combine.BinaryCombineIntegerFn minFn = Min.ofIntegers(); StateTag<?, ?> fooCoder1Max1 = StateTags.combiningValueFromInputInternal("foo", input1, maxFn); StateTag<?, ?> fooCoder1Max2 = StateTags.combiningValueFromInputInternal("foo", input1, maxFn); @@ -129,8 +128,8 @@ public class StateTagTest { CoderRegistry registry = new CoderRegistry(); registry.registerStandardCoders(); - MaxIntegerFn maxFn = new Max.MaxIntegerFn(); - MinIntegerFn minFn = new Min.MinIntegerFn(); + Combine.BinaryCombineIntegerFn maxFn = Max.ofIntegers(); + Combine.BinaryCombineIntegerFn minFn = Min.ofIntegers(); Coder<int[]> accum1 = maxFn.getAccumulatorCoder(registry, VarIntCoder.of()); Coder<int[]> accum2 = minFn.getAccumulatorCoder(registry, BigEndianIntegerCoder.of()); http://git-wip-us.apache.org/repos/asf/beam/blob/78a360ea/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java ---------------------------------------------------------------------- diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java index 3a36add..7efa115 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java +++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java @@ -2577,7 +2577,7 @@ public class BigQueryIO { /** Tracks bytes written, exposed as "ByteCount" Counter. */ private Aggregator<Long, Long> byteCountAggregator = - createAggregator("ByteCount", new Sum.SumLongFn()); + createAggregator("ByteCount", Sum.ofLongs()); /** Constructor. */ StreamingWriteFn(ValueProvider<TableSchema> schema, BigQueryServices bqServices) {
