From df8cf750e62846531b0b0260e4c84d3bb6b8d2c7 Mon Sep 17 00:00:00 2001
From: Reuven Lax <relax@google.com>
Date: Mon, 8 May 2017 09:06:55 -0700
Subject: [PATCH 1/2] TableDescription is allowed to be null.

---
 .../sdk/io/gcp/bigquery/TableDestination.java   | 10 ++++++++--
 .../io/gcp/bigquery/TableDestinationCoder.java  | 17 +++++++++++------
 2 files changed, 19 insertions(+), 8 deletions(-)

diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/TableDestination.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/TableDestination.java
index 7a82c54e277a..59d34757b8c3 100644
--- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/TableDestination.java
+++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/TableDestination.java
@@ -21,6 +21,7 @@
 import com.google.api.services.bigquery.model.TableReference;
 import java.io.Serializable;
 import java.util.Objects;
+import javax.annotation.Nullable;
 
 /**
  * Encapsulates a BigQuery table destination.
@@ -28,10 +29,11 @@
 public class TableDestination implements Serializable {
   private static final long serialVersionUID = 1L;
   private final String tableSpec;
+  @Nullable
   private final String tableDescription;
 
 
-  public TableDestination(String tableSpec, String tableDescription) {
+  public TableDestination(String tableSpec, @Nullable String tableDescription) {
     this.tableSpec = tableSpec;
     this.tableDescription = tableDescription;
   }
@@ -55,7 +57,11 @@ public String getTableDescription() {
 
   @Override
   public String toString() {
-    return "tableSpec: " + tableSpec + " tableDescription: " + tableDescription;
+    String toString = "tableSpec: " + tableSpec;
+    if (tableDescription != null) {
+      toString += " tableDescription: " + tableDescription;
+    }
+    return toString;
   }
 
   @Override
diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/TableDestinationCoder.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/TableDestinationCoder.java
index 3059e2aecde2..426c54333262 100644
--- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/TableDestinationCoder.java
+++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/TableDestinationCoder.java
@@ -22,13 +22,18 @@
 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.CoderException;
+import org.apache.beam.sdk.coders.NullableCoder;
 import org.apache.beam.sdk.coders.StringUtf8Coder;
 
 /** A coder for {@link TableDestination} objects. */
 public class TableDestinationCoder extends AtomicCoder<TableDestination> {
   private static final TableDestinationCoder INSTANCE = new TableDestinationCoder();
-  private static final StringUtf8Coder stringCoder = StringUtf8Coder.of();
+  private static final Coder<String> tableSpecCoder = StringUtf8Coder.of();
+  private static final Coder<String> tableDescriptionCoder =
+      NullableCoder.of(StringUtf8Coder.of());
+
 
   public static TableDestinationCoder of() {
     return INSTANCE;
@@ -40,15 +45,15 @@ public void encode(TableDestination value, OutputStream outStream, Context conte
     if (value == null) {
       throw new CoderException("cannot encode a null value");
     }
-    stringCoder.encode(value.getTableSpec(), outStream, context.nested());
-    stringCoder.encode(value.getTableDescription(), outStream, context.nested());
+    tableSpecCoder.encode(value.getTableSpec(), outStream, context.nested());
+    tableDescriptionCoder.encode(value.getTableDescription(), outStream, context);
   }
 
   @Override
   public TableDestination decode(InputStream inStream, Context context) throws IOException {
-    return new TableDestination(
-        stringCoder.decode(inStream, context.nested()),
-        stringCoder.decode(inStream, context.nested()));
+    String tableSpec = tableSpecCoder.decode(inStream, context.nested());
+    String tableDescription = tableDescriptionCoder.decode(inStream, context.nested());
+    return new TableDestination(tableSpec, tableDescription);
   }
 
   @Override

From 9e70c836a1e97e1924eea91b1c6b76ae8a5807c7 Mon Sep 17 00:00:00 2001
From: Reuven Lax <relax@google.com>
Date: Mon, 8 May 2017 10:05:20 -0700
Subject: [PATCH 2/2] Add unit-test coverage for null table descriptions.

---
 .../java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOTest.java | 1 +
 1 file changed, 1 insertion(+)

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 6c15f8736377..e8005de8d0f9 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
@@ -595,6 +595,7 @@ public void testWrite() throws Exception {
         new TableRow().set("name", "c").set("number", 3))
         .withCoder(TableRowJsonCoder.of()))
     .apply(BigQueryIO.writeTableRows().to("dataset-id.table-id")
+        .withTableDescription(null)
         .withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED)
         .withSchema(new TableSchema().setFields(
             ImmutableList.of(