Skip to content

Commit

Permalink
support date and timestamp fields
Browse files Browse the repository at this point in the history
  • Loading branch information
reuvenlax committed Apr 29, 2022
1 parent e6aae70 commit 2774b8a
Show file tree
Hide file tree
Showing 12 changed files with 707 additions and 136 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
2 changes: 2 additions & 0 deletions sdks/java/io/google-cloud-platform/build.gradle
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -3085,6 +3085,8 @@ static void clearStaticCaches() throws ExecutionException, InterruptedException
CreateTables.clearCreatedTables();
TwoLevelMessageConverterCache.clear();
StorageApiDynamicDestinationsTableRow.clearSchemaCache();
StorageApiWriteUnshardedRecords.clearCache();
StorageApiWritesShardedRecords.clearCache();
}

/////////////////////////////////////////////////////////////////////////////
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down Expand Up @@ -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);
}
Expand Down Expand Up @@ -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) {
Expand All @@ -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) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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>() {};
Expand Down

0 comments on commit 2774b8a

Please sign in to comment.