[BEAM-13990] support date and timestamp fields

buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy

@@ -631,6 +631,7 @@ class BeamModulePlugin implements Plugin<Project> {
         jackson_dataformat_xml                      : "com.fasterxml.jackson.dataformat:jackson-dataformat-xml:$jackson_version",
         jackson_dataformat_yaml                     : "com.fasterxml.jackson.dataformat:jackson-dataformat-yaml:$jackson_version",
         jackson_datatype_joda                       : "com.fasterxml.jackson.datatype:jackson-datatype-joda:$jackson_version",
+        jackson_datatype_jsr310                     : "com.fasterxml.jackson.datatype:jackson-datatype-jsr310:$jackson_version",
         jackson_module_scala_2_11                   : "com.fasterxml.jackson.module:jackson-module-scala_2.11:$jackson_version",
         jackson_module_scala_2_12                   : "com.fasterxml.jackson.module:jackson-module-scala_2.12:$jackson_version",
         // Swap to use the officially published version of 0.4.x once available

sdks/java/io/google-cloud-platform/build.gradle

@@ -116,6 +116,8 @@ dependencies {
   implementation library.java.http_core
   implementation library.java.jackson_core
   implementation library.java.jackson_databind
+  implementation library.java.jackson_datatype_joda
+  implementation library.java.jackson_datatype_jsr310
   implementation library.java.joda_time
   implementation library.java.junit
   implementation library.java.netty_handler

sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java

@@ -3085,6 +3085,8 @@ static void clearStaticCaches() throws ExecutionException, InterruptedException
     CreateTables.clearCreatedTables();
     TwoLevelMessageConverterCache.clear();
     StorageApiDynamicDestinationsTableRow.clearSchemaCache();
+    StorageApiWriteUnshardedRecords.clearCache();
+    StorageApiWritesShardedRecords.clearCache();
   }
 
   /////////////////////////////////////////////////////////////////////////////

sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiDynamicDestinationsTableRow.java

@@ -70,6 +70,7 @@ public MessageConverter<T> getMessageConverter(
       DestinationT destination, DatasetService datasetService) throws Exception {
     return new MessageConverter<T>() {
       TableSchema tableSchema;
+      TableRowToStorageApiProto.SchemaInformation schemaInformation;
       Descriptor descriptor;
       long descriptorHash;
 
@@ -103,7 +104,8 @@ public MessageConverter<T> getMessageConverter(
               MoreObjects.firstNonNull(
                   SCHEMA_CACHE.putSchemaIfAbsent(tableReference, tableSchema), tableSchema);
         }
-
+        schemaInformation =
+            TableRowToStorageApiProto.SchemaInformation.fromTableSchema(tableSchema);
         descriptor = TableRowToStorageApiProto.getDescriptorFromTableSchema(tableSchema);
         descriptorHash = BigQueryUtils.hashSchemaDescriptorDeterministic(descriptor);
       }
@@ -138,6 +140,8 @@ public void refreshSchemaInternal() throws Exception {
         }
         synchronized (this) {
           tableSchema = newSchema;
+          schemaInformation =
+              TableRowToStorageApiProto.SchemaInformation.fromTableSchema(tableSchema);
           descriptor = TableRowToStorageApiProto.getDescriptorFromTableSchema(tableSchema);
           long newHash = BigQueryUtils.hashSchemaDescriptorDeterministic(descriptor);
           if (descriptorHash != newHash) {
@@ -154,16 +158,21 @@ public void refreshSchemaInternal() throws Exception {
       public StorageApiWritePayload toMessage(T element) throws Exception {
         int attempt = 0;
         do {
+          TableRowToStorageApiProto.SchemaInformation localSchemaInformation;
           Descriptor localDescriptor;
           long localDescriptorHash;
           synchronized (this) {
+            localSchemaInformation = schemaInformation;
             localDescriptor = descriptor;
             localDescriptorHash = descriptorHash;
           }
           try {
             Message msg =
                 TableRowToStorageApiProto.messageFromTableRow(
-                    localDescriptor, formatFunction.apply(element), ignoreUnknownValues);
+                    localSchemaInformation,
+                    localDescriptor,
+                    formatFunction.apply(element),
+                    ignoreUnknownValues);
             return new AutoValue_StorageApiWritePayload(msg.toByteArray(), localDescriptorHash);
           } catch (SchemaTooNarrowException e) {
             if (attempt > schemaUpdateRetries) {

sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiWriteUnshardedRecords.java

@@ -87,6 +87,10 @@ public class StorageApiWriteUnshardedRecords<DestinationT, ElementT>
               })
           .build();
 
+  static void clearCache() {
+    APPEND_CLIENTS.invalidateAll();
+  }
+
   // Run a closure asynchronously, ignoring failures.
   private interface ThrowingRunnable {
     void run() throws Exception;

sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiWritesShardedRecords.java

@@ -119,6 +119,10 @@ public class StorageApiWritesShardedRecords<DestinationT, ElementT>
               })
           .build();
 
+  static void clearCache() {
+    APPEND_CLIENTS.invalidateAll();
+  }
+
   // Run a closure asynchronously, ignoring failures.
   private interface ThrowingRunnable {
     void run() throws Exception;

sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/TableRowJsonCoder.java

@@ -19,6 +19,8 @@
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.SerializationFeature;
+import com.fasterxml.jackson.datatype.joda.JodaModule;
+import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
 import com.google.api.services.bigquery.model.TableRow;
 import java.io.IOException;
 import java.io.InputStream;
@@ -68,7 +70,11 @@ public long getEncodedElementByteSize(TableRow value) throws Exception {
   // 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);
+      new ObjectMapper()
+          .registerModule(new JavaTimeModule())
+          .registerModule(new JodaModule())
+          .disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS)
+          .disable(SerializationFeature.FAIL_ON_EMPTY_BEANS);
 
   private static final TableRowJsonCoder INSTANCE = new TableRowJsonCoder();
   private static final TypeDescriptor<TableRow> TYPE_DESCRIPTOR = new TypeDescriptor<TableRow>() {};