beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From lc...@apache.org
Subject [1/2] beam git commit: [BEAM-1786, BEAM-1871] Add the ability to register coder factories for classes allowing TableRowJsonCoder to move to sdks/java/io/google-cloud-platform
Date Thu, 20 Apr 2017 21:19:39 GMT
Repository: beam
Updated Branches:
  refs/heads/master b47fd52cb -> 36a12d023


[BEAM-1786, BEAM-1871] Add the ability to register coder factories for classes allowing TableRowJsonCoder to move to sdks/java/io/google-cloud-platform


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

Branch: refs/heads/master
Commit: 3c811f5eabc823c5895a7b3c62379370bff8b22c
Parents: b47fd52
Author: Luke Cwik <lcwik@google.com>
Authored: Wed Apr 19 17:59:15 2017 -0700
Committer: Luke Cwik <lcwik@google.com>
Committed: Thu Apr 20 14:19:03 2017 -0700

----------------------------------------------------------------------
 .../org/apache/beam/runners/core/StateTags.java |   6 +-
 .../beam/runners/core/ReduceFnTester.java       |  12 +--
 .../apache/beam/runners/core/StateTagTest.java  |   3 +-
 .../spark/translation/SparkRuntimeContext.java  |   3 +-
 sdks/java/core/pom.xml                          |   5 -
 .../main/java/org/apache/beam/sdk/Pipeline.java |   3 +-
 .../apache/beam/sdk/coders/CoderFactories.java  |  22 +++-
 .../apache/beam/sdk/coders/CoderRegistrar.java  |  45 ++++++++
 .../apache/beam/sdk/coders/CoderRegistry.java   | 108 ++++++++++++++-----
 .../apache/beam/sdk/coders/StringUtf8Coder.java |   2 +-
 .../beam/sdk/coders/TableRowJsonCoder.java      |  88 ---------------
 .../apache/beam/sdk/util/state/StateSpecs.java  |   6 +-
 .../org/apache/beam/SdkCoreApiSurfaceTest.java  |   1 -
 .../beam/sdk/coders/CoderRegistryTest.java      |  77 +++++++------
 .../beam/sdk/coders/DefaultCoderTest.java       |   8 +-
 .../beam/sdk/coders/TableRowJsonCoderTest.java  |  94 ----------------
 .../beam/sdk/transforms/LatestFnTest.java       |   2 +-
 .../org/apache/beam/sdk/transforms/SumTest.java |   2 +-
 .../transforms/reflect/DoFnInvokersTest.java    |   4 +-
 sdks/java/io/google-cloud-platform/pom.xml      |   5 +
 .../io/gcp/bigquery/BigQueryCoderRegistrar.java |  39 +++++++
 .../beam/sdk/io/gcp/bigquery/BigQueryIO.java    |   1 -
 .../sdk/io/gcp/bigquery/BigQuerySourceBase.java |   1 -
 .../io/gcp/bigquery/StreamingWriteTables.java   |   1 -
 .../sdk/io/gcp/bigquery/TableRowInfoCoder.java  |   1 -
 .../sdk/io/gcp/bigquery/TableRowJsonCoder.java  |  91 ++++++++++++++++
 .../sdk/io/gcp/bigquery/TableRowWriter.java     |   1 -
 .../bigquery/BigQueryCoderRegistrarTest.java    |  40 +++++++
 .../sdk/io/gcp/bigquery/BigQueryIOTest.java     |   1 -
 .../io/gcp/bigquery/FakeBigQueryServices.java   |   1 -
 .../sdk/io/gcp/bigquery/FakeJobService.java     |   1 -
 .../io/gcp/bigquery/TableRowJsonCoderTest.java  |  95 ++++++++++++++++
 32 files changed, 479 insertions(+), 290 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/beam/blob/3c811f5e/runners/core-java/src/main/java/org/apache/beam/runners/core/StateTags.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/StateTags.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/StateTags.java
index 77ae8f5..3a45569 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/StateTags.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/StateTags.java
@@ -47,11 +47,7 @@ import org.apache.beam.sdk.util.state.WatermarkHoldState;
 @Experimental(Kind.STATE)
 public class StateTags {
 
-  private static final CoderRegistry STANDARD_REGISTRY = new CoderRegistry();
-
-  static {
-    STANDARD_REGISTRY.registerStandardCoders();
-  }
+  private static final CoderRegistry STANDARD_REGISTRY = CoderRegistry.createDefault();
 
   /** @deprecated for migration purposes only */
   @Deprecated

http://git-wip-us.apache.org/repos/asf/beam/blob/3c811f5e/runners/core-java/src/test/java/org/apache/beam/runners/core/ReduceFnTester.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/test/java/org/apache/beam/runners/core/ReduceFnTester.java b/runners/core-java/src/test/java/org/apache/beam/runners/core/ReduceFnTester.java
index 914550e..549fd8a 100644
--- a/runners/core-java/src/test/java/org/apache/beam/runners/core/ReduceFnTester.java
+++ b/runners/core-java/src/test/java/org/apache/beam/runners/core/ReduceFnTester.java
@@ -180,8 +180,7 @@ public class ReduceFnTester<InputT, OutputT, W extends BoundedWindow> {
           Coder<OutputT> outputCoder)
           throws Exception {
 
-    CoderRegistry registry = new CoderRegistry();
-    registry.registerStandardCoders();
+    CoderRegistry registry = CoderRegistry.createDefault();
     AppliedCombineFn<String, Integer, AccumT, OutputT> fn =
         AppliedCombineFn.<String, Integer, AccumT, OutputT>withInputCoder(
             combineFn, registry, KvCoder.of(StringUtf8Coder.of(), VarIntCoder.of()));
@@ -207,8 +206,7 @@ public class ReduceFnTester<InputT, OutputT, W extends BoundedWindow> {
       Coder<OutputT> outputCoder)
       throws Exception {
 
-    CoderRegistry registry = new CoderRegistry();
-    registry.registerStandardCoders();
+    CoderRegistry registry = CoderRegistry.createDefault();
     AppliedCombineFn<String, Integer, AccumT, OutputT> fn =
         AppliedCombineFn.<String, Integer, AccumT, OutputT>withInputCoder(
             combineFn, registry, KvCoder.of(StringUtf8Coder.of(), VarIntCoder.of()));
@@ -230,8 +228,7 @@ public class ReduceFnTester<InputT, OutputT, W extends BoundedWindow> {
           PipelineOptions options,
           SideInputReader sideInputReader)
           throws Exception {
-    CoderRegistry registry = new CoderRegistry();
-    registry.registerStandardCoders();
+    CoderRegistry registry = CoderRegistry.createDefault();
     AppliedCombineFn<String, Integer, AccumT, OutputT> fn =
         AppliedCombineFn.<String, Integer, AccumT, OutputT>withInputCoder(
             combineFn, registry, KvCoder.of(StringUtf8Coder.of(), VarIntCoder.of()));
@@ -254,8 +251,7 @@ public class ReduceFnTester<InputT, OutputT, W extends BoundedWindow> {
       PipelineOptions options,
       SideInputReader sideInputReader)
       throws Exception {
-    CoderRegistry registry = new CoderRegistry();
-    registry.registerStandardCoders();
+    CoderRegistry registry = CoderRegistry.createDefault();
     AppliedCombineFn<String, Integer, AccumT, OutputT> fn =
         AppliedCombineFn.<String, Integer, AccumT, OutputT>withInputCoder(
             combineFn, registry, KvCoder.of(StringUtf8Coder.of(), VarIntCoder.of()));

http://git-wip-us.apache.org/repos/asf/beam/blob/3c811f5e/runners/core-java/src/test/java/org/apache/beam/runners/core/StateTagTest.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/test/java/org/apache/beam/runners/core/StateTagTest.java b/runners/core-java/src/test/java/org/apache/beam/runners/core/StateTagTest.java
index 0584643..5f5d92d 100644
--- a/runners/core-java/src/test/java/org/apache/beam/runners/core/StateTagTest.java
+++ b/runners/core-java/src/test/java/org/apache/beam/runners/core/StateTagTest.java
@@ -158,8 +158,7 @@ public class StateTagTest {
   @SuppressWarnings({"unchecked", "rawtypes"})
   @Test
   public void testCombiningValueWithContextEquality() {
-    CoderRegistry registry = new CoderRegistry();
-    registry.registerStandardCoders();
+    CoderRegistry registry = CoderRegistry.createDefault();
 
     Combine.BinaryCombineIntegerFn maxFn = Max.ofIntegers();
     Combine.BinaryCombineIntegerFn minFn = Min.ofIntegers();

http://git-wip-us.apache.org/repos/asf/beam/blob/3c811f5e/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkRuntimeContext.java
----------------------------------------------------------------------
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkRuntimeContext.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkRuntimeContext.java
index 9d0f576..6abab17 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkRuntimeContext.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkRuntimeContext.java
@@ -112,8 +112,7 @@ public class SparkRuntimeContext implements Serializable {
 
   public CoderRegistry getCoderRegistry() {
     if (coderRegistry == null) {
-      coderRegistry = new CoderRegistry();
-      coderRegistry.registerStandardCoders();
+      coderRegistry = CoderRegistry.createDefault();
     }
     return coderRegistry;
   }

http://git-wip-us.apache.org/repos/asf/beam/blob/3c811f5e/sdks/java/core/pom.xml
----------------------------------------------------------------------
diff --git a/sdks/java/core/pom.xml b/sdks/java/core/pom.xml
index 2860be2..930632d 100644
--- a/sdks/java/core/pom.xml
+++ b/sdks/java/core/pom.xml
@@ -141,11 +141,6 @@
 
     <dependency>
       <groupId>com.google.apis</groupId>
-      <artifactId>google-api-services-bigquery</artifactId>
-    </dependency>
-
-    <dependency>
-      <groupId>com.google.apis</groupId>
       <artifactId>google-api-services-storage</artifactId>
     </dependency>
 

http://git-wip-us.apache.org/repos/asf/beam/blob/3c811f5e/sdks/java/core/src/main/java/org/apache/beam/sdk/Pipeline.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/Pipeline.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/Pipeline.java
index e3b9309..0a1dc13 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/Pipeline.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/Pipeline.java
@@ -292,8 +292,7 @@ public class Pipeline {
    */
   public CoderRegistry getCoderRegistry() {
     if (coderRegistry == null) {
-      coderRegistry = new CoderRegistry();
-      coderRegistry.registerStandardCoders();
+      coderRegistry = CoderRegistry.createDefault();
     }
     return coderRegistry;
   }

http://git-wip-us.apache.org/repos/asf/beam/blob/3c811f5e/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/CoderFactories.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/CoderFactories.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/CoderFactories.java
index e1a202a..0031698 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/CoderFactories.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/CoderFactories.java
@@ -17,6 +17,7 @@
  */
 package org.apache.beam.sdk.coders;
 
+import com.google.common.base.MoreObjects;
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 import java.lang.reflect.Modifier;
@@ -129,13 +130,13 @@ public final class CoderFactories {
     // Method to create a coder given component coders
     // For a Coder class of kind * -> * -> ... n times ... -> *
     // this has type Coder<?> -> Coder<?> -> ... n times ... -> Coder<T>
-    private Method factoryMethod;
+    private final Method factoryMethod;
 
     // Method to decompose a value of type T into its parts.
     // For a Coder class of kind * -> * -> ... n times ... -> *
     // this has type T -> List<Object>
     // where the list has n elements.
-    private Method getComponentsMethod;
+    private final Method getComponentsMethod;
 
     /**
      * Returns a CoderFactory that invokes the given static factory method
@@ -248,6 +249,14 @@ public final class CoderFactories {
           "cannot build CoderFactory from class " + coderType
           + ": does not implement Coder<T> for any T.");
     }
+
+    @Override
+    public String toString() {
+      return MoreObjects.toStringHelper(getClass())
+          .add("factoryMethod", factoryMethod)
+          .add("getComponentsMethod", getComponentsMethod)
+          .toString();
+    }
   }
 
   /**
@@ -255,7 +264,7 @@ public final class CoderFactories {
    * {@link CoderFactory}.
    */
   private static class CoderFactoryForCoder<T> implements CoderFactory {
-    private Coder<T> coder;
+    private final Coder<T> coder;
 
     public CoderFactoryForCoder(Coder<T> coder) {
       this.coder = coder;
@@ -270,5 +279,12 @@ public final class CoderFactories {
     public List<Object> getInstanceComponents(Object value) {
       return Collections.emptyList();
     }
+
+    @Override
+    public String toString() {
+      return MoreObjects.toStringHelper(getClass())
+          .add("coder", coder)
+          .toString();
+    }
   }
 }

http://git-wip-us.apache.org/repos/asf/beam/blob/3c811f5e/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/CoderRegistrar.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/CoderRegistrar.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/CoderRegistrar.java
new file mode 100644
index 0000000..fced976
--- /dev/null
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/CoderRegistrar.java
@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.coders;
+
+import com.google.auto.service.AutoService;
+import java.util.Map;
+import java.util.ServiceLoader;
+import org.apache.beam.sdk.annotations.Experimental;
+
+/**
+ * {@link Coder} creators have the ability to automatically have their
+ * {@link Coder coders} registered with this SDK by creating a {@link ServiceLoader} entry
+ * and a concrete implementation of this interface.
+ *
+ * <p>It is optional but recommended to use one of the many build time tools such as
+ * {@link AutoService} to generate the necessary META-INF files automatically.
+ */
+@Experimental
+public interface CoderRegistrar {
+  /**
+   * Returns a mapping of {@link Class classes} to {@link CoderFactory coder factories} which
+   * will be registered by default within each {@link CoderRegistry coder registry} instance.
+   *
+   * <p>See {@link CoderFactories} for convenience methods to construct a {@link CoderFactory}.
+   *
+   * <p>Note that a warning is logged if multiple {@link CoderRegistrar coder registrars} provide
+   * mappings for the same {@link Class}.
+   */
+  Map<Class<?>, CoderFactory> getCoderFactoriesToUseForClasses();
+}

http://git-wip-us.apache.org/repos/asf/beam/blob/3c811f5e/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/CoderRegistry.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/CoderRegistry.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/CoderRegistry.java
index 65f4209..6b909d4 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/CoderRegistry.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/CoderRegistry.java
@@ -19,10 +19,14 @@ package org.apache.beam.sdk.coders;
 
 import static com.google.common.base.Preconditions.checkArgument;
 
-import com.google.api.services.bigquery.model.TableRow;
 import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Predicate;
+import com.google.common.collect.HashMultimap;
 import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
+import com.google.common.collect.Multimap;
+import com.google.common.collect.Sets;
 import com.google.protobuf.ByteString;
 import java.lang.reflect.ParameterizedType;
 import java.lang.reflect.Type;
@@ -35,13 +39,17 @@ import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.ServiceLoader;
 import java.util.Set;
+import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
 import org.apache.beam.sdk.coders.CannotProvideCoderException.ReasonCode;
 import org.apache.beam.sdk.coders.protobuf.ProtoCoder;
 import org.apache.beam.sdk.transforms.SerializableFunction;
 import org.apache.beam.sdk.transforms.windowing.IntervalWindow;
 import org.apache.beam.sdk.util.CoderUtils;
+import org.apache.beam.sdk.util.common.ReflectHelpers;
+import org.apache.beam.sdk.util.common.ReflectHelpers.ObjectsClassComparator;
 import org.apache.beam.sdk.values.KV;
 import org.apache.beam.sdk.values.TimestampedValue;
 import org.apache.beam.sdk.values.TypeDescriptor;
@@ -63,8 +71,7 @@ import org.slf4j.LoggerFactory;
  *       <li>A {@link Coder} class with the static methods to satisfy
  *           {@link CoderFactories#fromStaticMethods} can be registered via
  *           {@link #registerCoder(Class, Class)}.</li>
- *       <li>Built-in types are registered via
- *           {@link #registerStandardCoders()}.</li>
+ *       <li>Types can be automatically registered via {@link CoderRegistrar coder registrars}.</li>
  *     </ul>
  *   <li>Annotations: {@link DefaultCoder} can be used to annotate a type with
  *       the default {@code Coder} type. The {@link Coder} class must satisfy the requirements
@@ -82,33 +89,86 @@ import org.slf4j.LoggerFactory;
 public class CoderRegistry implements CoderProvider {
 
   private static final Logger LOG = LoggerFactory.getLogger(CoderRegistry.class);
+  private static final Map<Class<?>, CoderFactory> REGISTERED_CODER_FACTORIES_PER_CLASS;
+
+  static {
+    // Register the standard coders first so they are choosen as the default
+    Multimap<Class<?>, CoderFactory> codersToRegister = HashMultimap.create();
+    codersToRegister.put(Byte.class, CoderFactories.fromStaticMethods(ByteCoder.class));
+    codersToRegister.put(ByteString.class, CoderFactories.fromStaticMethods(ByteStringCoder.class));
+    codersToRegister.put(Double.class, CoderFactories.fromStaticMethods(DoubleCoder.class));
+    codersToRegister.put(Instant.class, CoderFactories.fromStaticMethods(InstantCoder.class));
+    codersToRegister.put(Integer.class, CoderFactories.fromStaticMethods(VarIntCoder.class));
+    codersToRegister.put(Iterable.class, CoderFactories.fromStaticMethods(IterableCoder.class));
+    codersToRegister.put(KV.class, CoderFactories.fromStaticMethods(KvCoder.class));
+    codersToRegister.put(List.class, CoderFactories.fromStaticMethods(ListCoder.class));
+    codersToRegister.put(Long.class, CoderFactories.fromStaticMethods(VarLongCoder.class));
+    codersToRegister.put(Map.class, CoderFactories.fromStaticMethods(MapCoder.class));
+    codersToRegister.put(Set.class, CoderFactories.fromStaticMethods(SetCoder.class));
+    codersToRegister.put(String.class, CoderFactories.fromStaticMethods(StringUtf8Coder.class));
+    codersToRegister.put(TimestampedValue.class,
+        CoderFactories.fromStaticMethods(TimestampedValue.TimestampedValueCoder.class));
+    codersToRegister.put(Void.class, CoderFactories.fromStaticMethods(VoidCoder.class));
+    codersToRegister.put(byte[].class, CoderFactories.fromStaticMethods(ByteArrayCoder.class));
+    codersToRegister.put(IntervalWindow.class, CoderFactories.forCoder(IntervalWindow.getCoder()));
+
+    // Enumerate all the CoderRegistrars in a deterministic order, adding all coders to register
+    Set<CoderRegistrar> registrars = Sets.newTreeSet(ObjectsClassComparator.INSTANCE);
+    registrars.addAll(Lists.newArrayList(
+        ServiceLoader.load(CoderRegistrar.class, ReflectHelpers.findClassLoader())));
+    for (CoderRegistrar registrar : registrars) {
+      for (Map.Entry<Class<?>, CoderFactory> entry
+          : registrar.getCoderFactoriesToUseForClasses().entrySet()) {
+        codersToRegister.put(entry.getKey(), entry.getValue());
+      }
+    }
+
+    // Warn the user if multiple coders want to be registered for the same class
+    Map<Class<?>, Collection<CoderFactory>> multipleRegistrations =
+        Maps.filterValues(codersToRegister.asMap(), new Predicate<Collection<CoderFactory>>() {
+      @Override
+      public boolean apply(@Nonnull Collection<CoderFactory> input) {
+        return input.size() > 1;
+      }
+    });
+    for (Map.Entry<Class<?>, Collection<CoderFactory>> entry : multipleRegistrations.entrySet()) {
+      LOG.warn("Multiple CoderFactory registrations {} found for class {}, using {}.",
+          entry.getKey(), entry.getValue(), entry.getValue().iterator().next());
+    }
+
+    // Build a map choosing the first coder within the multimap as the default
+    ImmutableMap.Builder<Class<?>, CoderFactory> registeredCoderFactoriesPerClassBuilder =
+        ImmutableMap.builder();
+    for (Map.Entry<Class<?>, Collection<CoderFactory>> entry
+        : codersToRegister.asMap().entrySet()) {
+      registeredCoderFactoriesPerClassBuilder.put(
+          entry.getKey(), entry.getValue().iterator().next());
+    }
+    REGISTERED_CODER_FACTORIES_PER_CLASS = registeredCoderFactoriesPerClassBuilder.build();
+  }
+
+  /**
+   * Creates a CoderRegistry containing registrations for all standard coders part of the core Java
+   * Apache Beam SDK and also any registrations provided by {@link CoderRegistrar coder registrars}.
+   *
+   * <p>Multiple registrations for the same class result in the (in order of precedence):
+   * <ul>
+   *   <li>Standard coder part of the core Apache Beam Java SDK being used.</li>
+   *   <li>The coder from the {@link CoderRegistrar} with the lexicographically smallest
+   *   {@link Class#getName() class name} being used.</li>
+   * </ul>
+   */
+  public static CoderRegistry createDefault() {
+    return new CoderRegistry();
+  }
 
   public CoderRegistry() {
+    coderFactoryMap = new HashMap<>(REGISTERED_CODER_FACTORIES_PER_CLASS);
     setFallbackCoderProvider(
         CoderProviders.firstOf(ProtoCoder.coderProvider(), SerializableCoder.PROVIDER));
   }
 
-  /**
-   * Registers standard Coders with this CoderRegistry.
-   */
   public void registerStandardCoders() {
-    registerCoder(Byte.class, ByteCoder.class);
-    registerCoder(ByteString.class, ByteStringCoder.class);
-    registerCoder(Double.class, DoubleCoder.class);
-    registerCoder(Instant.class, InstantCoder.class);
-    registerCoder(Integer.class, VarIntCoder.class);
-    registerCoder(Iterable.class, IterableCoder.class);
-    registerCoder(KV.class, KvCoder.class);
-    registerCoder(List.class, ListCoder.class);
-    registerCoder(Long.class, VarLongCoder.class);
-    registerCoder(Map.class, MapCoder.class);
-    registerCoder(Set.class, SetCoder.class);
-    registerCoder(String.class, StringUtf8Coder.class);
-    registerCoder(TableRow.class, TableRowJsonCoder.class);
-    registerCoder(TimestampedValue.class, TimestampedValue.TimestampedValueCoder.class);
-    registerCoder(Void.class, VoidCoder.class);
-    registerCoder(byte[].class, ByteArrayCoder.class);
-    registerCoder(IntervalWindow.class, IntervalWindow.getCoder());
   }
 
   /**
@@ -642,7 +702,7 @@ public class CoderRegistry implements CoderProvider {
    * The map of classes to the CoderFactories to use to create their
    * default Coders.
    */
-  private Map<Class<?>, CoderFactory> coderFactoryMap = new HashMap<>();
+  private Map<Class<?>, CoderFactory> coderFactoryMap;
 
   /**
    * A provider of coders for types where no coder is registered.

http://git-wip-us.apache.org/repos/asf/beam/blob/3c811f5e/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/StringUtf8Coder.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/StringUtf8Coder.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/StringUtf8Coder.java
index ca7912c..cd124ef 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/StringUtf8Coder.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/StringUtf8Coder.java
@@ -127,7 +127,7 @@ public class StringUtf8Coder extends AtomicCoder<String> {
    * the byte size of the encoding plus the encoded length prefix.
    */
   @Override
-  protected long getEncodedElementByteSize(String value, Context context)
+  public long getEncodedElementByteSize(String value, Context context)
       throws Exception {
     if (value == null) {
       throw new CoderException("cannot encode a null String");

http://git-wip-us.apache.org/repos/asf/beam/blob/3c811f5e/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/TableRowJsonCoder.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/TableRowJsonCoder.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/TableRowJsonCoder.java
deleted file mode 100644
index 5c0929c..0000000
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/TableRowJsonCoder.java
+++ /dev/null
@@ -1,88 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.beam.sdk.coders;
-
-import com.fasterxml.jackson.annotation.JsonCreator;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.fasterxml.jackson.databind.SerializationFeature;
-import com.google.api.services.bigquery.model.TableRow;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import org.apache.beam.sdk.values.TypeDescriptor;
-
-/**
- * A {@link Coder} that encodes BigQuery {@link TableRow} objects in their native JSON format.
- */
-public class TableRowJsonCoder extends AtomicCoder<TableRow> {
-
-  @JsonCreator
-  public static TableRowJsonCoder of() {
-    return INSTANCE;
-  }
-
-  @Override
-  public void encode(TableRow value, OutputStream outStream, Context context)
-      throws IOException {
-    String strValue = MAPPER.writeValueAsString(value);
-    StringUtf8Coder.of().encode(strValue, outStream, context);
-  }
-
-  @Override
-  public TableRow decode(InputStream inStream, Context context)
-      throws IOException {
-    String strValue = StringUtf8Coder.of().decode(inStream, context);
-    return MAPPER.readValue(strValue, TableRow.class);
-  }
-
-  @Override
-  protected long getEncodedElementByteSize(TableRow value, Context context)
-      throws Exception {
-    String strValue = MAPPER.writeValueAsString(value);
-    return StringUtf8Coder.of().getEncodedElementByteSize(strValue, context);
-  }
-
-  /////////////////////////////////////////////////////////////////////////////
-
-  // FAIL_ON_EMPTY_BEANS is disabled in order to handle null values in
-  // TableRow.
-  private static final ObjectMapper MAPPER =
-      new ObjectMapper().disable(SerializationFeature.FAIL_ON_EMPTY_BEANS);
-
-  private static final TableRowJsonCoder INSTANCE = new TableRowJsonCoder();
-  private static final TypeDescriptor<TableRow> TYPE_DESCRIPTOR = new TypeDescriptor<TableRow>() {};
-
-  private TableRowJsonCoder() { }
-
-  /**
-   * {@inheritDoc}
-   *
-   * @throws NonDeterministicException always. A {@link TableRow} can hold arbitrary
-   *         {@link Object} instances, which makes the encoding non-deterministic.
-   */
-  @Override
-  public void verifyDeterministic() throws NonDeterministicException {
-    throw new NonDeterministicException(this,
-        "TableCell can hold arbitrary instances, which may be non-deterministic.");
-  }
-
-  @Override
-  public TypeDescriptor<TableRow> getEncodedTypeDescriptor() {
-    return TYPE_DESCRIPTOR;
-  }
-}

http://git-wip-us.apache.org/repos/asf/beam/blob/3c811f5e/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/StateSpecs.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/StateSpecs.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/StateSpecs.java
index 30a7a6d..dc647da 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/StateSpecs.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/StateSpecs.java
@@ -38,11 +38,7 @@ import org.apache.beam.sdk.transforms.windowing.OutputTimeFn;
 @Experimental(Kind.STATE)
 public class StateSpecs {
 
-  private static final CoderRegistry STANDARD_REGISTRY = new CoderRegistry();
-
-  static {
-    STANDARD_REGISTRY.registerStandardCoders();
-  }
+  private static final CoderRegistry STANDARD_REGISTRY = CoderRegistry.createDefault();
 
   private StateSpecs() {}
 

http://git-wip-us.apache.org/repos/asf/beam/blob/3c811f5e/sdks/java/core/src/test/java/org/apache/beam/SdkCoreApiSurfaceTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/SdkCoreApiSurfaceTest.java b/sdks/java/core/src/test/java/org/apache/beam/SdkCoreApiSurfaceTest.java
index eed4457..b6e9205 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/SdkCoreApiSurfaceTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/SdkCoreApiSurfaceTest.java
@@ -39,7 +39,6 @@ public class SdkCoreApiSurfaceTest {
         ImmutableSet.of(
             "org.apache.beam",
             "com.google.api.client",
-            "com.google.api.services.bigquery",
             "com.google.api.services.storage",
             "com.google.protobuf",
             "com.fasterxml.jackson.annotation",

http://git-wip-us.apache.org/repos/asf/beam/blob/3c811f5e/sdks/java/core/src/test/java/org/apache/beam/sdk/coders/CoderRegistryTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/coders/CoderRegistryTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/coders/CoderRegistryTest.java
index 10177e7..774ca9d 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/coders/CoderRegistryTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/coders/CoderRegistryTest.java
@@ -21,8 +21,10 @@ import static org.hamcrest.Matchers.allOf;
 import static org.hamcrest.Matchers.containsString;
 import static org.junit.Assert.assertEquals;
 
+import com.google.auto.service.AutoService;
 import com.google.cloud.dataflow.sdk.coders.Proto2CoderTestMessages.MessageA;
 import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
 import com.google.protobuf.Duration;
 import java.io.IOException;
 import java.io.InputStream;
@@ -37,6 +39,7 @@ import java.util.Map;
 import java.util.Set;
 import org.apache.beam.sdk.coders.CoderRegistry.IncompatibleCoderException;
 import org.apache.beam.sdk.coders.protobuf.ProtoCoder;
+import org.apache.beam.sdk.testing.ExpectedLogs;
 import org.apache.beam.sdk.testing.NeedsRunner;
 import org.apache.beam.sdk.testing.TestPipeline;
 import org.apache.beam.sdk.transforms.Create;
@@ -67,11 +70,8 @@ public class CoderRegistryTest {
   @Rule
   public ExpectedException thrown = ExpectedException.none();
 
-  public static CoderRegistry getStandardRegistry() {
-    CoderRegistry registry = new CoderRegistry();
-    registry.registerStandardCoders();
-    return registry;
-  }
+  @Rule
+  public ExpectedLogs expectedLogs = ExpectedLogs.none(CoderRegistry.class);
 
   private static class SerializableClass implements Serializable {
   }
@@ -80,7 +80,7 @@ public class CoderRegistryTest {
 
   @Test
   public void testSerializableFallbackCoderProvider() throws Exception {
-    CoderRegistry registry = getStandardRegistry();
+    CoderRegistry registry = CoderRegistry.createDefault();
     registry.setFallbackCoderProvider(SerializableCoder.PROVIDER);
     Coder<?> serializableCoder = registry.getDefaultCoder(SerializableClass.class);
 
@@ -89,7 +89,7 @@ public class CoderRegistryTest {
 
   @Test
   public void testProtoCoderFallbackCoderProvider() throws Exception {
-    CoderRegistry registry = getStandardRegistry();
+    CoderRegistry registry = CoderRegistry.createDefault();
 
     // MessageA is a Protocol Buffers test message with syntax 2
     assertEquals(registry.getDefaultCoder(MessageA.class), ProtoCoder.of(MessageA.class));
@@ -100,7 +100,7 @@ public class CoderRegistryTest {
 
   @Test
   public void testAvroFallbackCoderProvider() throws Exception {
-    CoderRegistry registry = getStandardRegistry();
+    CoderRegistry registry = CoderRegistry.createDefault();
     registry.setFallbackCoderProvider(AvroCoder.PROVIDER);
     Coder<?> avroCoder = registry.getDefaultCoder(NotSerializableClass.class);
 
@@ -109,7 +109,7 @@ public class CoderRegistryTest {
 
   @Test
   public void testRegisterInstantiatedCoder() throws Exception {
-    CoderRegistry registry = new CoderRegistry();
+    CoderRegistry registry = CoderRegistry.createDefault();
     registry.registerCoder(MyValue.class, MyValueCoder.of());
     assertEquals(registry.getDefaultCoder(MyValue.class), MyValueCoder.of());
   }
@@ -137,19 +137,19 @@ public class CoderRegistryTest {
   public void testRegisterInstantiatedCoderInvalidRawtype() throws Exception {
     thrown.expect(IllegalArgumentException.class);
     thrown.expectMessage("may not be used with unspecialized generic classes");
-    CoderRegistry registry = new CoderRegistry();
+    CoderRegistry registry = CoderRegistry.createDefault();
     registry.registerCoder(List.class, new MyListCoder());
   }
 
   @Test
   public void testSimpleDefaultCoder() throws Exception {
-    CoderRegistry registry = getStandardRegistry();
+    CoderRegistry registry = CoderRegistry.createDefault();
     assertEquals(StringUtf8Coder.of(), registry.getDefaultCoder(String.class));
   }
 
   @Test
   public void testSimpleUnknownDefaultCoder() throws Exception {
-    CoderRegistry registry = getStandardRegistry();
+    CoderRegistry registry = CoderRegistry.createDefault();
     thrown.expect(CannotProvideCoderException.class);
     thrown.expectMessage(allOf(
         containsString(UnknownType.class.getCanonicalName()),
@@ -161,7 +161,7 @@ public class CoderRegistryTest {
 
   @Test
   public void testParameterizedDefaultListCoder() throws Exception {
-    CoderRegistry registry = getStandardRegistry();
+    CoderRegistry registry = CoderRegistry.createDefault();
     TypeDescriptor<List<Integer>> listToken = new TypeDescriptor<List<Integer>>() {};
     assertEquals(ListCoder.of(VarIntCoder.of()),
                  registry.getDefaultCoder(listToken));
@@ -177,7 +177,7 @@ public class CoderRegistryTest {
 
   @Test
   public void testParameterizedDefaultMapCoder() throws Exception {
-    CoderRegistry registry = getStandardRegistry();
+    CoderRegistry registry = CoderRegistry.createDefault();
     TypeDescriptor<Map<Integer, String>> mapToken = new TypeDescriptor<Map<Integer, String>>() {};
     assertEquals(MapCoder.of(VarIntCoder.of(), StringUtf8Coder.of()),
                  registry.getDefaultCoder(mapToken));
@@ -185,7 +185,7 @@ public class CoderRegistryTest {
 
   @Test
   public void testParameterizedDefaultNestedMapCoder() throws Exception {
-    CoderRegistry registry = getStandardRegistry();
+    CoderRegistry registry = CoderRegistry.createDefault();
     TypeDescriptor<Map<Integer, Map<String, Double>>> mapToken =
         new TypeDescriptor<Map<Integer, Map<String, Double>>>() {};
     assertEquals(
@@ -195,21 +195,21 @@ public class CoderRegistryTest {
 
   @Test
   public void testParameterizedDefaultSetCoder() throws Exception {
-    CoderRegistry registry = getStandardRegistry();
+    CoderRegistry registry = CoderRegistry.createDefault();
     TypeDescriptor<Set<Integer>> setToken = new TypeDescriptor<Set<Integer>>() {};
     assertEquals(SetCoder.of(VarIntCoder.of()), registry.getDefaultCoder(setToken));
   }
 
   @Test
   public void testParameterizedDefaultNestedSetCoder() throws Exception {
-    CoderRegistry registry = getStandardRegistry();
+    CoderRegistry registry = CoderRegistry.createDefault();
     TypeDescriptor<Set<Set<Integer>>> setToken = new TypeDescriptor<Set<Set<Integer>>>() {};
     assertEquals(SetCoder.of(SetCoder.of(VarIntCoder.of())), registry.getDefaultCoder(setToken));
   }
 
   @Test
   public void testParameterizedDefaultCoderUnknown() throws Exception {
-    CoderRegistry registry = getStandardRegistry();
+    CoderRegistry registry = CoderRegistry.createDefault();
     TypeDescriptor<List<UnknownType>> listUnknownToken = new TypeDescriptor<List<UnknownType>>() {};
 
     thrown.expect(CannotProvideCoderException.class);
@@ -223,7 +223,7 @@ public class CoderRegistryTest {
 
   @Test
   public void testTypeParameterInferenceForward() throws Exception {
-    CoderRegistry registry = getStandardRegistry();
+    CoderRegistry registry = CoderRegistry.createDefault();
     MyGenericClass<MyValue, List<MyValue>> instance =
         new MyGenericClass<MyValue, List<MyValue>>() {};
 
@@ -239,7 +239,7 @@ public class CoderRegistryTest {
 
   @Test
   public void testTypeParameterInferenceBackward() throws Exception {
-    CoderRegistry registry = getStandardRegistry();
+    CoderRegistry registry = CoderRegistry.createDefault();
     MyGenericClass<MyValue, List<MyValue>> instance =
         new MyGenericClass<MyValue, List<MyValue>>() {};
 
@@ -256,7 +256,7 @@ public class CoderRegistryTest {
 
   @Test
   public void testGetDefaultCoderFromIntegerValue() throws Exception {
-    CoderRegistry registry = getStandardRegistry();
+    CoderRegistry registry = CoderRegistry.createDefault();
     Integer i = 13;
     Coder<Integer> coder = registry.getDefaultCoder(i);
     assertEquals(VarIntCoder.of(), coder);
@@ -264,13 +264,13 @@ public class CoderRegistryTest {
 
   @Test
   public void testGetDefaultCoderFromNullValue() throws Exception {
-    CoderRegistry registry = getStandardRegistry();
+    CoderRegistry registry = CoderRegistry.createDefault();
     assertEquals(VoidCoder.of(), registry.getDefaultCoder((Void) null));
   }
 
   @Test
   public void testGetDefaultCoderFromKvValue() throws Exception {
-    CoderRegistry registry = getStandardRegistry();
+    CoderRegistry registry = CoderRegistry.createDefault();
     KV<Integer, String> kv = KV.of(13, "hello");
     Coder<KV<Integer, String>> coder = registry.getDefaultCoder(kv);
     assertEquals(KvCoder.of(VarIntCoder.of(), StringUtf8Coder.of()),
@@ -279,7 +279,7 @@ public class CoderRegistryTest {
 
   @Test
   public void testGetDefaultCoderFromKvNullValue() throws Exception {
-    CoderRegistry registry = getStandardRegistry();
+    CoderRegistry registry = CoderRegistry.createDefault();
     KV<Void, Void> kv = KV.of((Void) null, (Void) null);
     assertEquals(KvCoder.of(VoidCoder.of(), VoidCoder.of()),
         registry.getDefaultCoder(kv));
@@ -287,7 +287,7 @@ public class CoderRegistryTest {
 
   @Test
   public void testGetDefaultCoderFromNestedKvValue() throws Exception {
-    CoderRegistry registry = getStandardRegistry();
+    CoderRegistry registry = CoderRegistry.createDefault();
     KV<Integer, KV<Long, KV<String, String>>> kv = KV.of(13, KV.of(17L, KV.of("hello", "goodbye")));
     Coder<KV<Integer, KV<Long, KV<String, String>>>> coder = registry.getDefaultCoder(kv);
     assertEquals(
@@ -346,8 +346,7 @@ public class CoderRegistryTest {
 
   @Test
   public void testDefaultCoderAnnotationGenericRawtype() throws Exception {
-    CoderRegistry registry = new CoderRegistry();
-    registry.registerStandardCoders();
+    CoderRegistry registry = CoderRegistry.createDefault();
     assertEquals(
         registry.getDefaultCoder(MySerializableGeneric.class),
         SerializableCoder.of(MySerializableGeneric.class));
@@ -355,8 +354,7 @@ public class CoderRegistryTest {
 
   @Test
   public void testDefaultCoderAnnotationGeneric() throws Exception {
-    CoderRegistry registry = new CoderRegistry();
-    registry.registerStandardCoders();
+    CoderRegistry registry = CoderRegistry.createDefault();
     assertEquals(
         registry.getDefaultCoder(new TypeDescriptor<MySerializableGeneric<String>>() {}),
         SerializableCoder.of(MySerializableGeneric.class));
@@ -383,7 +381,7 @@ public class CoderRegistryTest {
    */
   @Test
   public void testTypeVariableErrorMessage() throws Exception {
-    CoderRegistry registry = new CoderRegistry();
+    CoderRegistry registry = CoderRegistry.createDefault();
 
     thrown.expect(CannotProvideCoderException.class);
     thrown.expectMessage(allOf(
@@ -399,7 +397,7 @@ public class CoderRegistryTest {
   @Test
   @SuppressWarnings("rawtypes")
   public void testSerializableTypeVariableDefaultCoder() throws Exception {
-    CoderRegistry registry = new CoderRegistry();
+    CoderRegistry registry = CoderRegistry.createDefault();
 
     TypeDescriptor type = TypeDescriptor.of(
         TestSerializableGenericClass.class.getTypeParameters()[0]);
@@ -540,4 +538,21 @@ public class CoderRegistryTest {
     @SuppressWarnings("unused")
     private T foo;
   }
+
+  @Test
+  public void testAutomaticRegistrationOfCoders() throws Exception {
+    assertEquals(CoderRegistry.createDefault().getDefaultCoder(MyValue.class), MyValueCoder.of());
+  }
+
+  /**
+   * A {@link CoderRegistrar} to demonstrate default {@link Coder} registration.
+   */
+  @AutoService(CoderRegistrar.class)
+  public static class RegisteredTestCoderRegistrar implements CoderRegistrar {
+    @Override
+    public Map<Class<?>, CoderFactory> getCoderFactoriesToUseForClasses() {
+      return ImmutableMap.<Class<?>, CoderFactory>of(
+          MyValue.class, CoderFactories.forCoder(MyValueCoder.of()));
+    }
+  }
 }

http://git-wip-us.apache.org/repos/asf/beam/blob/3c811f5e/sdks/java/core/src/test/java/org/apache/beam/sdk/coders/DefaultCoderTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/coders/DefaultCoderTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/coders/DefaultCoderTest.java
index 59749ae..d335b18 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/coders/DefaultCoderTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/coders/DefaultCoderTest.java
@@ -25,7 +25,6 @@ import static org.junit.Assert.assertThat;
 import java.io.Serializable;
 import java.util.List;
 import org.apache.beam.sdk.values.TypeDescriptor;
-import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.ExpectedException;
@@ -41,12 +40,7 @@ public class DefaultCoderTest {
   @Rule
   public ExpectedException thrown = ExpectedException.none();
 
-  public CoderRegistry registry = new CoderRegistry();
-
-  @Before
-  public void registerStandardCoders() {
-    registry.registerStandardCoders();
-  }
+  public CoderRegistry registry = CoderRegistry.createDefault();
 
   @DefaultCoder(AvroCoder.class)
   private static class AvroRecord {

http://git-wip-us.apache.org/repos/asf/beam/blob/3c811f5e/sdks/java/core/src/test/java/org/apache/beam/sdk/coders/TableRowJsonCoderTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/coders/TableRowJsonCoderTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/coders/TableRowJsonCoderTest.java
deleted file mode 100644
index 5253ce7..0000000
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/coders/TableRowJsonCoderTest.java
+++ /dev/null
@@ -1,94 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.beam.sdk.coders;
-
-import static org.hamcrest.Matchers.equalTo;
-import static org.junit.Assert.assertThat;
-
-import com.google.api.services.bigquery.model.TableRow;
-import java.util.Arrays;
-import java.util.List;
-import org.apache.beam.sdk.testing.CoderProperties;
-import org.apache.beam.sdk.values.TypeDescriptor;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.junit.runners.JUnit4;
-
-/**
- * Test case for {@link TableRowJsonCoder}.
- */
-@RunWith(JUnit4.class)
-public class TableRowJsonCoderTest {
-
-  private static class TableRowBuilder {
-    private TableRow row;
-    public TableRowBuilder() {
-      row = new TableRow();
-    }
-    public TableRowBuilder set(String fieldName, Object value) {
-      row.set(fieldName, value);
-      return this;
-    }
-    public TableRow build() {
-      return row;
-    }
-  }
-
-  private static final Coder<TableRow> TEST_CODER = TableRowJsonCoder.of();
-
-  private static final List<TableRow> TEST_VALUES = Arrays.asList(
-      new TableRowBuilder().build(),
-      new TableRowBuilder().set("a", "1").build(),
-      new TableRowBuilder().set("b", 3.14).build(),
-      new TableRowBuilder().set("a", "1").set("b", true).set("c", "hi").build());
-
-  @Test
-  public void testDecodeEncodeEqual() throws Exception {
-    for (TableRow value : TEST_VALUES) {
-      CoderProperties.coderDecodeEncodeEqual(TEST_CODER, value);
-    }
-  }
-
-  // This identifier should only change if the JSON format of results from the BigQuery API changes.
-  private static final String EXPECTED_ENCODING_ID = "";
-
-  @Test
-  public void testEncodingId() throws Exception {
-    CoderProperties.coderHasEncodingId(TEST_CODER, EXPECTED_ENCODING_ID);
-  }
-
-  /**
-   * Generated data to check that the wire format has not changed. To regenerate, see
-   * {@link org.apache.beam.sdk.coders.PrintBase64Encodings}.
-   */
-  private static final List<String> TEST_ENCODINGS = Arrays.asList(
-      "e30",
-      "eyJhIjoiMSJ9",
-      "eyJiIjozLjE0fQ",
-      "eyJhIjoiMSIsImIiOnRydWUsImMiOiJoaSJ9");
-
-  @Test
-  public void testWireFormatEncode() throws Exception {
-    CoderProperties.coderEncodesBase64(TEST_CODER, TEST_VALUES, TEST_ENCODINGS);
-  }
-
-  @Test
-  public void testEncodedTypeDescriptor() throws Exception {
-    assertThat(TEST_CODER.getEncodedTypeDescriptor(), equalTo(TypeDescriptor.of(TableRow.class)));
-  }
-}

http://git-wip-us.apache.org/repos/asf/beam/blob/3c811f5e/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/LatestFnTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/LatestFnTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/LatestFnTest.java
index 31acb08..f49c765 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/LatestFnTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/LatestFnTest.java
@@ -196,7 +196,7 @@ public class LatestFnTest {
   public void testDefaultCoderHandlesNull() throws CannotProvideCoderException {
     Latest.LatestFn<Long> fn = new Latest.LatestFn<>();
 
-    CoderRegistry registry = new CoderRegistry();
+    CoderRegistry registry = CoderRegistry.createDefault();
     TimestampedValue.TimestampedValueCoder<Long> inputCoder =
         TimestampedValue.TimestampedValueCoder.of(VarLongCoder.of());
 

http://git-wip-us.apache.org/repos/asf/beam/blob/3c811f5e/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 b2f8aa8..9d2c6f6 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
@@ -37,7 +37,7 @@ import org.junit.runners.JUnit4;
  */
 @RunWith(JUnit4.class)
 public class SumTest {
-  private static final CoderRegistry STANDARD_REGISTRY = new CoderRegistry();
+  private static final CoderRegistry STANDARD_REGISTRY = CoderRegistry.createDefault();
 
   @Test
   public void testSumGetNames() {

http://git-wip-us.apache.org/repos/asf/beam/blob/3c811f5e/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/DoFnInvokersTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/DoFnInvokersTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/DoFnInvokersTest.java
index 8b4df4c..5732438 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/DoFnInvokersTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/DoFnInvokersTest.java
@@ -335,7 +335,7 @@ public class DoFnInvokersTest {
     when(fn.newTracker(restriction)).thenReturn(tracker);
     fn.processElement(mockProcessContext, tracker);
 
-    assertEquals(coder, invoker.invokeGetRestrictionCoder(new CoderRegistry()));
+    assertEquals(coder, invoker.invokeGetRestrictionCoder(CoderRegistry.createDefault()));
     assertEquals(restriction, invoker.invokeGetInitialRestriction("blah"));
     final List<SomeRestriction> outputs = new ArrayList<>();
     invoker.invokeSplitRestriction(
@@ -415,7 +415,7 @@ public class DoFnInvokersTest {
     MockFn fn = mock(MockFn.class);
     DoFnInvoker<String, String> invoker = DoFnInvokers.invokerFor(fn);
 
-    CoderRegistry coderRegistry = new CoderRegistry();
+    CoderRegistry coderRegistry = CoderRegistry.createDefault();
     coderRegistry.registerCoder(RestrictionWithDefaultTracker.class, CoderForDefaultTracker.class);
     assertThat(
         invoker.<RestrictionWithDefaultTracker>invokeGetRestrictionCoder(coderRegistry),

http://git-wip-us.apache.org/repos/asf/beam/blob/3c811f5e/sdks/java/io/google-cloud-platform/pom.xml
----------------------------------------------------------------------
diff --git a/sdks/java/io/google-cloud-platform/pom.xml b/sdks/java/io/google-cloud-platform/pom.xml
index 4cd0337..3778a63 100644
--- a/sdks/java/io/google-cloud-platform/pom.xml
+++ b/sdks/java/io/google-cloud-platform/pom.xml
@@ -77,6 +77,11 @@
     </dependency>
 
     <dependency>
+      <groupId>com.fasterxml.jackson.core</groupId>
+      <artifactId>jackson-databind</artifactId>
+    </dependency>
+
+    <dependency>
       <groupId>com.google.apis</groupId>
       <artifactId>google-api-services-bigquery</artifactId>
     </dependency>

http://git-wip-us.apache.org/repos/asf/beam/blob/3c811f5e/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryCoderRegistrar.java
----------------------------------------------------------------------
diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryCoderRegistrar.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryCoderRegistrar.java
new file mode 100644
index 0000000..847c7b5
--- /dev/null
+++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryCoderRegistrar.java
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.gcp.bigquery;
+
+import com.google.api.services.bigquery.model.TableRow;
+import com.google.auto.service.AutoService;
+import com.google.common.collect.ImmutableMap;
+import java.util.Map;
+import org.apache.beam.sdk.coders.CoderFactories;
+import org.apache.beam.sdk.coders.CoderFactory;
+import org.apache.beam.sdk.coders.CoderRegistrar;
+
+/**
+ * A {@link CoderRegistrar} for standard types used with {@link BigQueryIO}.
+ */
+@AutoService(CoderRegistrar.class)
+public class BigQueryCoderRegistrar implements CoderRegistrar {
+  @Override
+  public Map<Class<?>, CoderFactory> getCoderFactoriesToUseForClasses() {
+    return ImmutableMap.of(
+        TableRow.class, CoderFactories.forCoder(TableRowJsonCoder.of()),
+        TableRowInfo.class, CoderFactories.forCoder(TableRowInfoCoder.of()));
+  }
+}

http://git-wip-us.apache.org/repos/asf/beam/blob/3c811f5e/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 3f5947e..a13d61d 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
@@ -41,7 +41,6 @@ import javax.annotation.Nullable;
 
 import org.apache.beam.sdk.coders.Coder;
 import org.apache.beam.sdk.coders.KvCoder;
-import org.apache.beam.sdk.coders.TableRowJsonCoder;
 import org.apache.beam.sdk.coders.VoidCoder;
 import org.apache.beam.sdk.io.BoundedSource;
 import org.apache.beam.sdk.io.gcp.bigquery.BigQueryHelpers.BeamJobUuidToBigQueryJobUuid;

http://git-wip-us.apache.org/repos/asf/beam/blob/3c811f5e/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQuerySourceBase.java
----------------------------------------------------------------------
diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQuerySourceBase.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQuerySourceBase.java
index 4142da9..53d395b 100644
--- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQuerySourceBase.java
+++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQuerySourceBase.java
@@ -33,7 +33,6 @@ import java.util.List;
 import java.util.NoSuchElementException;
 import org.apache.avro.generic.GenericRecord;
 import org.apache.beam.sdk.coders.Coder;
-import org.apache.beam.sdk.coders.TableRowJsonCoder;
 import org.apache.beam.sdk.io.AvroSource;
 import org.apache.beam.sdk.io.BoundedSource;
 import org.apache.beam.sdk.io.gcp.bigquery.BigQueryHelpers.Status;

http://git-wip-us.apache.org/repos/asf/beam/blob/3c811f5e/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StreamingWriteTables.java
----------------------------------------------------------------------
diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StreamingWriteTables.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StreamingWriteTables.java
index 4ddc1df..4d130b6 100644
--- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StreamingWriteTables.java
+++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StreamingWriteTables.java
@@ -20,7 +20,6 @@ package org.apache.beam.sdk.io.gcp.bigquery;
 import com.google.api.services.bigquery.model.TableRow;
 import org.apache.beam.sdk.coders.KvCoder;
 import org.apache.beam.sdk.coders.StringUtf8Coder;
-import org.apache.beam.sdk.coders.TableRowJsonCoder;
 import org.apache.beam.sdk.transforms.PTransform;
 import org.apache.beam.sdk.transforms.ParDo;
 import org.apache.beam.sdk.util.Reshuffle;

http://git-wip-us.apache.org/repos/asf/beam/blob/3c811f5e/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/TableRowInfoCoder.java
----------------------------------------------------------------------
diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/TableRowInfoCoder.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/TableRowInfoCoder.java
index 5e8fa29..9ef947e 100644
--- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/TableRowInfoCoder.java
+++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/TableRowInfoCoder.java
@@ -26,7 +26,6 @@ import java.io.OutputStream;
 import org.apache.beam.sdk.coders.AtomicCoder;
 import org.apache.beam.sdk.coders.CoderException;
 import org.apache.beam.sdk.coders.StringUtf8Coder;
-import org.apache.beam.sdk.coders.TableRowJsonCoder;
 
 /**
  * Defines a coder for {@link TableRowInfo} objects.

http://git-wip-us.apache.org/repos/asf/beam/blob/3c811f5e/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/TableRowJsonCoder.java
----------------------------------------------------------------------
diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/TableRowJsonCoder.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/TableRowJsonCoder.java
new file mode 100644
index 0000000..ce4b669
--- /dev/null
+++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/TableRowJsonCoder.java
@@ -0,0 +1,91 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.gcp.bigquery;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializationFeature;
+import com.google.api.services.bigquery.model.TableRow;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import org.apache.beam.sdk.coders.AtomicCoder;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.values.TypeDescriptor;
+
+/**
+ * A {@link Coder} that encodes BigQuery {@link TableRow} objects in their native JSON format.
+ */
+public class TableRowJsonCoder extends AtomicCoder<TableRow> {
+
+  @JsonCreator
+  public static TableRowJsonCoder of() {
+    return INSTANCE;
+  }
+
+  @Override
+  public void encode(TableRow value, OutputStream outStream, Context context)
+      throws IOException {
+    String strValue = MAPPER.writeValueAsString(value);
+    StringUtf8Coder.of().encode(strValue, outStream, context);
+  }
+
+  @Override
+  public TableRow decode(InputStream inStream, Context context)
+      throws IOException {
+    String strValue = StringUtf8Coder.of().decode(inStream, context);
+    return MAPPER.readValue(strValue, TableRow.class);
+  }
+
+  @Override
+  protected long getEncodedElementByteSize(TableRow value, Context context)
+      throws Exception {
+    String strValue = MAPPER.writeValueAsString(value);
+    return StringUtf8Coder.of().getEncodedElementByteSize(strValue, context);
+  }
+
+  /////////////////////////////////////////////////////////////////////////////
+
+  // FAIL_ON_EMPTY_BEANS is disabled in order to handle null values in
+  // TableRow.
+  private static final ObjectMapper MAPPER =
+      new ObjectMapper().disable(SerializationFeature.FAIL_ON_EMPTY_BEANS);
+
+  private static final TableRowJsonCoder INSTANCE = new TableRowJsonCoder();
+  private static final TypeDescriptor<TableRow> TYPE_DESCRIPTOR = new TypeDescriptor<TableRow>() {};
+
+  private TableRowJsonCoder() { }
+
+  /**
+   * {@inheritDoc}
+   *
+   * @throws NonDeterministicException always. A {@link TableRow} can hold arbitrary
+   *         {@link Object} instances, which makes the encoding non-deterministic.
+   */
+  @Override
+  public void verifyDeterministic() throws NonDeterministicException {
+    throw new NonDeterministicException(this,
+        "TableCell can hold arbitrary instances, which may be non-deterministic.");
+  }
+
+  @Override
+  public TypeDescriptor<TableRow> getEncodedTypeDescriptor() {
+    return TYPE_DESCRIPTOR;
+  }
+}

http://git-wip-us.apache.org/repos/asf/beam/blob/3c811f5e/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/TableRowWriter.java
----------------------------------------------------------------------
diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/TableRowWriter.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/TableRowWriter.java
index 91ef404..cb51158 100644
--- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/TableRowWriter.java
+++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/TableRowWriter.java
@@ -26,7 +26,6 @@ import java.nio.channels.WritableByteChannel;
 import java.nio.charset.StandardCharsets;
 import org.apache.beam.sdk.coders.Coder;
 import org.apache.beam.sdk.coders.Coder.Context;
-import org.apache.beam.sdk.coders.TableRowJsonCoder;
 import org.apache.beam.sdk.util.IOChannelUtils;
 import org.apache.beam.sdk.util.MimeTypes;
 import org.slf4j.Logger;

http://git-wip-us.apache.org/repos/asf/beam/blob/3c811f5e/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryCoderRegistrarTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryCoderRegistrarTest.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryCoderRegistrarTest.java
new file mode 100644
index 0000000..e7e9fe1
--- /dev/null
+++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryCoderRegistrarTest.java
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.gcp.bigquery;
+
+import com.google.api.services.bigquery.model.TableRow;
+import org.apache.beam.sdk.coders.CoderRegistry;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/**
+ * Tests for {@link BigQueryCoderRegistrar}.
+ */
+@RunWith(JUnit4.class)
+public class BigQueryCoderRegistrarTest {
+  @Test
+  public void testTableRowCoderIsRegistered() throws Exception {
+    CoderRegistry.createDefault().getDefaultCoder(TableRow.class);
+  }
+
+  @Test
+  public void testTableRowInfoCoderIsRegistered() throws Exception {
+    CoderRegistry.createDefault().getDefaultCoder(TableRowInfo.class);
+  }
+}

http://git-wip-us.apache.org/repos/asf/beam/blob/3c811f5e/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOTest.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOTest.java
index 62c5b5f..8e1632f 100644
--- a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOTest.java
+++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOTest.java
@@ -73,7 +73,6 @@ import org.apache.beam.sdk.coders.CoderException;
 import org.apache.beam.sdk.coders.IterableCoder;
 import org.apache.beam.sdk.coders.KvCoder;
 import org.apache.beam.sdk.coders.StringUtf8Coder;
-import org.apache.beam.sdk.coders.TableRowJsonCoder;
 import org.apache.beam.sdk.coders.VarIntCoder;
 import org.apache.beam.sdk.io.BoundedSource;
 import org.apache.beam.sdk.io.CountingInput;

http://git-wip-us.apache.org/repos/asf/beam/blob/3c811f5e/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/FakeBigQueryServices.java
----------------------------------------------------------------------
diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/FakeBigQueryServices.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/FakeBigQueryServices.java
index 6dfd9d7..43ad238 100644
--- a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/FakeBigQueryServices.java
+++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/FakeBigQueryServices.java
@@ -33,7 +33,6 @@ import java.util.NoSuchElementException;
 
 import org.apache.beam.sdk.coders.Coder.Context;
 import org.apache.beam.sdk.coders.ListCoder;
-import org.apache.beam.sdk.coders.TableRowJsonCoder;
 import org.apache.beam.sdk.options.BigQueryOptions;
 
 

http://git-wip-us.apache.org/repos/asf/beam/blob/3c811f5e/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/FakeJobService.java
----------------------------------------------------------------------
diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/FakeJobService.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/FakeJobService.java
index cffd873..bef9a26 100644
--- a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/FakeJobService.java
+++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/FakeJobService.java
@@ -61,7 +61,6 @@ import org.apache.avro.generic.GenericRecord;
 import org.apache.avro.generic.GenericRecordBuilder;
 import org.apache.beam.sdk.coders.Coder;
 import org.apache.beam.sdk.coders.Coder.Context;
-import org.apache.beam.sdk.coders.TableRowJsonCoder;
 import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.CreateDisposition;
 import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.WriteDisposition;
 import org.apache.beam.sdk.io.gcp.bigquery.BigQueryServices.JobService;

http://git-wip-us.apache.org/repos/asf/beam/blob/3c811f5e/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/TableRowJsonCoderTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/TableRowJsonCoderTest.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/TableRowJsonCoderTest.java
new file mode 100644
index 0000000..f6e02dc
--- /dev/null
+++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/TableRowJsonCoderTest.java
@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.gcp.bigquery;
+
+import static org.hamcrest.Matchers.equalTo;
+import static org.junit.Assert.assertThat;
+
+import com.google.api.services.bigquery.model.TableRow;
+import java.util.Arrays;
+import java.util.List;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.testing.CoderProperties;
+import org.apache.beam.sdk.values.TypeDescriptor;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/**
+ * Test case for {@link TableRowJsonCoder}.
+ */
+@RunWith(JUnit4.class)
+public class TableRowJsonCoderTest {
+
+  private static class TableRowBuilder {
+    private TableRow row;
+    public TableRowBuilder() {
+      row = new TableRow();
+    }
+    public TableRowBuilder set(String fieldName, Object value) {
+      row.set(fieldName, value);
+      return this;
+    }
+    public TableRow build() {
+      return row;
+    }
+  }
+
+  private static final Coder<TableRow> TEST_CODER = TableRowJsonCoder.of();
+
+  private static final List<TableRow> TEST_VALUES = Arrays.asList(
+      new TableRowBuilder().build(),
+      new TableRowBuilder().set("a", "1").build(),
+      new TableRowBuilder().set("b", 3.14).build(),
+      new TableRowBuilder().set("a", "1").set("b", true).set("c", "hi").build());
+
+  @Test
+  public void testDecodeEncodeEqual() throws Exception {
+    for (TableRow value : TEST_VALUES) {
+      CoderProperties.coderDecodeEncodeEqual(TEST_CODER, value);
+    }
+  }
+
+  // This identifier should only change if the JSON format of results from the BigQuery API changes.
+  private static final String EXPECTED_ENCODING_ID = "";
+
+  @Test
+  public void testEncodingId() throws Exception {
+    CoderProperties.coderHasEncodingId(TEST_CODER, EXPECTED_ENCODING_ID);
+  }
+
+  /**
+   * Generated data to check that the wire format has not changed. To regenerate, see
+   * {@link org.apache.beam.sdk.coders.PrintBase64Encodings}.
+   */
+  private static final List<String> TEST_ENCODINGS = Arrays.asList(
+      "e30",
+      "eyJhIjoiMSJ9",
+      "eyJiIjozLjE0fQ",
+      "eyJhIjoiMSIsImIiOnRydWUsImMiOiJoaSJ9");
+
+  @Test
+  public void testWireFormatEncode() throws Exception {
+    CoderProperties.coderEncodesBase64(TEST_CODER, TEST_VALUES, TEST_ENCODINGS);
+  }
+
+  @Test
+  public void testEncodedTypeDescriptor() throws Exception {
+    assertThat(TEST_CODER.getEncodedTypeDescriptor(), equalTo(TypeDescriptor.of(TableRow.class)));
+  }
+}


Mime
View raw message