beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From rober...@apache.org
Subject [2/3] beam git commit: Try PTransform-based coder inference before using fallback coder.
Date Wed, 03 May 2017 21:36:26 GMT
Try PTransform-based coder inference before using fallback coder.

This is particularly important for fallback coders that claim
to provide a coder for Object (or equivalently an unconstrained
type parameter).  See BEAM-1642.


Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/289d2dec
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/289d2dec
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/289d2dec

Branch: refs/heads/master
Commit: 289d2decba57ff990f44567d1a16c189cfbd1cc8
Parents: 9b6b906
Author: Robert Bradshaw <robertwb@gmail.com>
Authored: Fri Apr 28 15:53:42 2017 -0700
Committer: Robert Bradshaw <robertwb@gmail.com>
Committed: Wed May 3 14:36:03 2017 -0700

----------------------------------------------------------------------
 .../beam/runners/direct/DirectRunnerTest.java   | 10 ++++++++++
 .../org/apache/beam/sdk/values/TypedPValue.java | 20 ++++++++++----------
 2 files changed, 20 insertions(+), 10 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/beam/blob/289d2dec/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectRunnerTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectRunnerTest.java
b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectRunnerTest.java
index 6f9adc4..83881fc 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectRunnerTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectRunnerTest.java
@@ -578,4 +578,14 @@ public class DirectRunnerTest implements Serializable {
       return underlying.getDefaultOutputCoder();
     }
   }
+
+  @Test
+  public void fallbackCoderProviderAllowsInference() {
+    // See https://issues.apache.org/jira/browse/BEAM-1642
+    Pipeline p = getPipeline();
+    p.getCoderRegistry().setFallbackCoderProvider(
+        org.apache.beam.sdk.coders.AvroCoder.PROVIDER);
+    p.apply(Create.of(Arrays.asList(100, 200))).apply(Count.<Integer>globally());
+    p.run().waitUntilFinish();
+  }
 }

http://git-wip-us.apache.org/repos/asf/beam/blob/289d2dec/sdks/java/core/src/main/java/org/apache/beam/sdk/values/TypedPValue.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/TypedPValue.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/TypedPValue.java
index 54af747..f473776 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/TypedPValue.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/TypedPValue.java
@@ -140,7 +140,16 @@ public abstract class TypedPValue<T> extends PValueBase implements
PValue {
       return coderOrFailure;
     }
 
-    // Second option for a coder: Look in the coder registry.
+    // Second option for a coder: use the default Coder from the producing PTransform.
+    CannotProvideCoderException inputCoderException;
+    try {
+      return new CoderOrFailure<>(
+          ((PTransform) transform).getDefaultOutputCoder(input, this), null);
+    } catch (CannotProvideCoderException exc) {
+      inputCoderException = exc;
+    }
+
+    // Third option for a coder: Look in the coder registry.
     TypeDescriptor<T> token = getTypeDescriptor();
     CannotProvideCoderException inferFromTokenException = null;
     if (token != null) {
@@ -162,15 +171,6 @@ public abstract class TypedPValue<T> extends PValueBase implements
PValue {
       }
     }
 
-    // Third option for a coder: use the default Coder from the producing PTransform.
-    CannotProvideCoderException inputCoderException;
-    try {
-      return new CoderOrFailure<>(
-          ((PTransform) transform).getDefaultOutputCoder(input, this), null);
-    } catch (CannotProvideCoderException exc) {
-      inputCoderException = exc;
-    }
-
     // Build up the error message and list of causes.
     StringBuilder messageBuilder = new StringBuilder()
         .append("Unable to return a default Coder for ").append(this)


Mime
View raw message