You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by dh...@apache.org on 2016/07/12 01:03:07 UTC
[1/2] incubator-beam git commit: [BEAM-389] Fix the uses of
DelegateCoder in Combine
Repository: incubator-beam
Updated Branches:
refs/heads/master d93f564eb -> a59ddab21
[BEAM-389] Fix the uses of DelegateCoder in Combine
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/5316f20a
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/5316f20a
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/5316f20a
Branch: refs/heads/master
Commit: 5316f20a8bdf30f5eabba3763544e39608d53d12
Parents: d93f564
Author: Pei He <pe...@google.com>
Authored: Mon Jul 11 13:01:24 2016 -0700
Committer: Dan Halperin <dh...@google.com>
Committed: Mon Jul 11 18:02:29 2016 -0700
----------------------------------------------------------------------
.../org/apache/beam/sdk/transforms/Combine.java | 167 +++++++++++++------
.../org/apache/beam/sdk/transforms/SumTest.java | 33 ++++
2 files changed, 152 insertions(+), 48 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/5316f20a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Combine.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Combine.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Combine.java
index 9a87b36..96c03eb 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Combine.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Combine.java
@@ -723,19 +723,7 @@ public class Combine {
@Override
public Coder<int[]> getAccumulatorCoder(CoderRegistry registry, Coder<Integer> inputCoder) {
return DelegateCoder.of(
- inputCoder,
- new DelegateCoder.CodingFunction<int[], Integer>() {
- @Override
- public Integer apply(int[] accumulator) {
- return accumulator[0];
- }
- },
- new DelegateCoder.CodingFunction<Integer, int[]>() {
- @Override
- public int[] apply(Integer value) {
- return wrap(value);
- }
- });
+ inputCoder, new ToIntegerCodingFunction(), new FromIntegerCodingFunction());
}
@Override
@@ -744,12 +732,48 @@ public class Combine {
return inputCoder;
}
- private int[] wrap(int value) {
+ private static int[] wrap(int value) {
return new int[] { value };
}
- public Counter<Integer> getCounter(String name) {
- throw new UnsupportedOperationException("BinaryCombineDoubleFn does not support getCounter");
+ public Counter<Integer> getCounter(@SuppressWarnings("unused") String name) {
+ throw new UnsupportedOperationException("BinaryCombineIntegerFn does not support getCounter");
+ }
+
+ private static final class ToIntegerCodingFunction
+ implements DelegateCoder.CodingFunction<int[], Integer> {
+ @Override
+ public Integer apply(int[] accumulator) {
+ return accumulator[0];
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ return o instanceof ToIntegerCodingFunction;
+ }
+
+ @Override
+ public int hashCode() {
+ return this.getClass().hashCode();
+ }
+ }
+
+ private static final class FromIntegerCodingFunction
+ implements DelegateCoder.CodingFunction<Integer, int[]> {
+ @Override
+ public int[] apply(Integer value) {
+ return wrap(value);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ return o instanceof FromIntegerCodingFunction;
+ }
+
+ @Override
+ public int hashCode() {
+ return this.getClass().hashCode();
+ }
}
}
@@ -803,20 +827,7 @@ public class Combine {
@Override
public Coder<long[]> getAccumulatorCoder(CoderRegistry registry, Coder<Long> inputCoder) {
- return DelegateCoder.of(
- inputCoder,
- new DelegateCoder.CodingFunction<long[], Long>() {
- @Override
- public Long apply(long[] accumulator) {
- return accumulator[0];
- }
- },
- new DelegateCoder.CodingFunction<Long, long[]>() {
- @Override
- public long[] apply(Long value) {
- return wrap(value);
- }
- });
+ return DelegateCoder.of(inputCoder, new ToLongCodingFunction(), new FromLongCodingFunction());
}
@Override
@@ -824,12 +835,48 @@ public class Combine {
return inputCoder;
}
- private long[] wrap(long value) {
+ private static long[] wrap(long value) {
return new long[] { value };
}
- public Counter<Long> getCounter(String name) {
- throw new UnsupportedOperationException("BinaryCombineDoubleFn does not support getCounter");
+ public Counter<Long> getCounter(@SuppressWarnings("unused") String name) {
+ throw new UnsupportedOperationException("BinaryCombineLongFn does not support getCounter");
+ }
+
+ private static final class ToLongCodingFunction
+ implements DelegateCoder.CodingFunction<long[], Long> {
+ @Override
+ public Long apply(long[] accumulator) {
+ return accumulator[0];
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ return o instanceof ToLongCodingFunction;
+ }
+
+ @Override
+ public int hashCode() {
+ return this.getClass().hashCode();
+ }
+ }
+
+ private static final class FromLongCodingFunction
+ implements DelegateCoder.CodingFunction<Long, long[]> {
+ @Override
+ public long[] apply(Long value) {
+ return wrap(value);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ return o instanceof FromLongCodingFunction;
+ }
+
+ @Override
+ public int hashCode() {
+ return this.getClass().hashCode();
+ }
}
}
@@ -885,19 +932,7 @@ public class Combine {
@Override
public Coder<double[]> getAccumulatorCoder(CoderRegistry registry, Coder<Double> inputCoder) {
return DelegateCoder.of(
- inputCoder,
- new DelegateCoder.CodingFunction<double[], Double>() {
- @Override
- public Double apply(double[] accumulator) {
- return accumulator[0];
- }
- },
- new DelegateCoder.CodingFunction<Double, double[]>() {
- @Override
- public double[] apply(Double value) {
- return wrap(value);
- }
- });
+ inputCoder, new ToDoubleCodingFunction(), new FromDoubleCodingFunction());
}
@Override
@@ -905,13 +940,49 @@ public class Combine {
return inputCoder;
}
- private double[] wrap(double value) {
+ private static double[] wrap(double value) {
return new double[] { value };
}
- public Counter<Double> getCounter(String name) {
+ public Counter<Double> getCounter(@SuppressWarnings("unused") String name) {
throw new UnsupportedOperationException("BinaryCombineDoubleFn does not support getCounter");
}
+
+ private static final class ToDoubleCodingFunction
+ implements DelegateCoder.CodingFunction<double[], Double> {
+ @Override
+ public Double apply(double[] accumulator) {
+ return accumulator[0];
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ return o instanceof ToDoubleCodingFunction;
+ }
+
+ @Override
+ public int hashCode() {
+ return this.getClass().hashCode();
+ }
+ }
+
+ private static final class FromDoubleCodingFunction
+ implements DelegateCoder.CodingFunction<Double, double[]> {
+ @Override
+ public double[] apply(Double value) {
+ return wrap(value);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ return o instanceof FromDoubleCodingFunction;
+ }
+
+ @Override
+ public int hashCode() {
+ return this.getClass().hashCode();
+ }
+ }
}
/////////////////////////////////////////////////////////////////////////////
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/5316f20a/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 4e98055..4af6db1 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
@@ -20,6 +20,14 @@ package org.apache.beam.sdk.transforms;
import static org.apache.beam.sdk.TestUtils.checkCombineFn;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
+
+import org.apache.beam.sdk.coders.BigEndianIntegerCoder;
+import org.apache.beam.sdk.coders.BigEndianLongCoder;
+import org.apache.beam.sdk.coders.CoderRegistry;
+import org.apache.beam.sdk.coders.DoubleCoder;
+import org.apache.beam.sdk.coders.VarIntCoder;
+import org.apache.beam.sdk.coders.VarLongCoder;
import com.google.common.collect.Lists;
@@ -32,6 +40,7 @@ import org.junit.runners.JUnit4;
*/
@RunWith(JUnit4.class)
public class SumTest {
+ private static final CoderRegistry STANDARD_REGISTRY = new CoderRegistry();
@Test
public void testSumGetNames() {
@@ -66,4 +75,28 @@ public class SumTest {
Lists.newArrayList(1.0, 2.0, 3.0, 4.0),
10.0);
}
+
+ @Test
+ public void testGetAccumulatorCoderEquals() {
+ Sum.SumIntegerFn sumIntegerFn = new Sum.SumIntegerFn();
+ assertEquals(
+ sumIntegerFn.getAccumulatorCoder(STANDARD_REGISTRY, VarIntCoder.of()),
+ sumIntegerFn.getAccumulatorCoder(STANDARD_REGISTRY, VarIntCoder.of()));
+ assertNotEquals(
+ sumIntegerFn.getAccumulatorCoder(STANDARD_REGISTRY, VarIntCoder.of()),
+ sumIntegerFn.getAccumulatorCoder(STANDARD_REGISTRY, BigEndianIntegerCoder.of()));
+
+ Sum.SumLongFn sumLongFn = new Sum.SumLongFn();
+ assertEquals(
+ sumLongFn.getAccumulatorCoder(STANDARD_REGISTRY, VarLongCoder.of()),
+ sumLongFn.getAccumulatorCoder(STANDARD_REGISTRY, VarLongCoder.of()));
+ assertNotEquals(
+ sumLongFn.getAccumulatorCoder(STANDARD_REGISTRY, VarLongCoder.of()),
+ sumLongFn.getAccumulatorCoder(STANDARD_REGISTRY, BigEndianLongCoder.of()));
+
+ Sum.SumDoubleFn sumDoubleFn = new Sum.SumDoubleFn();
+ assertEquals(
+ sumDoubleFn.getAccumulatorCoder(STANDARD_REGISTRY, DoubleCoder.of()),
+ sumDoubleFn.getAccumulatorCoder(STANDARD_REGISTRY, DoubleCoder.of()));
+ }
}
[2/2] incubator-beam git commit: Closes #626
Posted by dh...@apache.org.
Closes #626
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/a59ddab2
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/a59ddab2
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/a59ddab2
Branch: refs/heads/master
Commit: a59ddab2164b7c5c25ddde1f88e62ea91a710581
Parents: d93f564 5316f20
Author: Dan Halperin <dh...@google.com>
Authored: Mon Jul 11 18:02:30 2016 -0700
Committer: Dan Halperin <dh...@google.com>
Committed: Mon Jul 11 18:02:30 2016 -0700
----------------------------------------------------------------------
.../org/apache/beam/sdk/transforms/Combine.java | 167 +++++++++++++------
.../org/apache/beam/sdk/transforms/SumTest.java | 33 ++++
2 files changed, 152 insertions(+), 48 deletions(-)
----------------------------------------------------------------------