googleapis / googleapis/google-cloud-java

[java-bigquerystorage] Datetime converter problem in Storage Write API

Ouverte
#11,966 1 commentaire 0 réactions 0 personnes assignées Voir sur GitHub
api: bigquerystorage priority: p3 type: bug
Langage dominant
Java
Étoiles
2.1k
Forks
1.2k
Merge moyen
1 j 23 h
PR mergées (30 j)
154

Description

#### Environment details

1. Specify the API at the beginning of the title. For example, "BigQuery: ...").
General, Core, and Other are also allowed as types : BigQuery Storage Write API
2. OS type and version: MacOs Sonoma 14.4.1
3. Java version: Java17 and Java20
4. version(s):3.16.2 and 3.5.1

#### Steps to reproduce

We are currently using the BigQuery Storage Write API and have encountered an error regarding datetime formatting. However, upon further inspection, we clearly see that this problem is caused by the integration of protobuf and the BigQuery Storage Write API in Java. This is because if we use a time zone, we have a problem.

We see that using Storage Write API in BigQuery:

```"BQ_ERROR_MESSAGE":"Field root.ModifiedDateTime failed to convert to DATETIME. Error: Text '2025-08-14T12:24:03.484Z' could not be parsed, unparsed text found at index 23"```

Also we that this problem

```
com.google.cloud.bigquery.storage.v1.Exceptions$AppendSerializationError: INVALID_ARGUMENT: Append serialization failed for writer: projects/hb-datalake-test/datasets/test/tables/deneme_table/_default
at com.google.cloud.bigquery.storage.v1.SchemaAwareStreamWriter.appendWithUniqueId(SchemaAwareStreamWriter.java:251) ~[google-cloud-bigquerystorage-3.16.2.jar:3.16.2]
at com.google.cloud.bigquery.storage.v1.SchemaAwareStreamWriter.append(SchemaAwareStreamWriter.java:140) ~[google-cloud-bigquerystorage-3.16.2.jar:3.16.2]
at com.google.cloud.bigquery.storage.v1.JsonStreamWriter.append(JsonStreamWriter.java:68) ~[google-cloud-bigquerystorage-3.16.2.jar:3.16.2]
at com.hepsiburada.lightrabbitconnector.service.SWADefaultStream$DataWriter.append(SWADefaultStream.java:190) ~[classes/:na]
at com.hepsiburada.lightrabbitconnector.service.SWADefaultStream.writeToDefaultStream(SWADefaultStream.java:77) ~[classes/:na]
at com.hepsiburada.lightrabbitconnector.service.RabbitMessageProcessor.listen(RabbitMessageProcessor.java:100) ~[classes/:na]
at java.base/jdk.internal.reflect.DirectMethodHandleAccessor.invoke(DirectMethodHandleAccessor.java:104) ~[na:na]
at java.base/java.lang.reflect.Method.invoke(Method.java:578) ~[na:na]
at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.doInvoke(InvocableHandlerMethod.java:169) ~[spring-messaging-6.0.13.jar:6.0.13]
at org.springframework.amqp.rabbit.listener.adapter.KotlinAwareInvocableHandlerMethod.doInvoke(KotlinAwareInvocableHandlerMethod.java:45) ~[spring-rabbit-3.0.10.jar:3.0.10]
at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.invoke(InvocableHandlerMethod.java:119) ~[spring-messaging-6.0.13.jar:6.0.13]
at org.springframework.amqp.rabbit.listener.adapter.HandlerAdapter.invoke(HandlerAdapter.java:75) ~[spring-rabbit-3.0.10.jar:3.0.10]
at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:274) ~[spring-rabbit-3.0.10.jar:3.0.10]
at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandlerAndProcessResult(MessagingMessageListenerAdapter.java:224) ~[spring-rabbit-3.0.10.jar:3.0.10]
at org.springframework.amqp.rabbit.listener.adapter.BatchMessagingMessageListenerAdapter.onMessageBatch(BatchMessagingMessageListenerAdapter.java:80) ~[spring-rabbit-3.0.10.jar:3.0.10]
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:1659) ~[spring-rabbit-3.0.10.jar:3.0.10]
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.actualInvokeListener(AbstractMessageListenerContainer.java:1582) ~[spring-rabbit-3.0.10.jar:3.0.10]
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:1570) ~[spring-rabbit-3.0.10.jar:3.0.10]
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:1565) ~[spring-rabbit-3.0.10.jar:3.0.10]
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListenerAndHandleException(AbstractMessageListenerContainer.java:1506) ~[spring-rabbit-3.0.10.jar:3.0.10]
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1487) ~[spring-rabbit-3.0.10.jar:3.0.10]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.executeWithList(SimpleMessageListenerContainer.java:1057) ~[spring-rabbit-3.0.10.jar:3.0.10]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:1046) ~[spring-rabbit-3.0.10.jar:3.0.10]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:941) ~[spring-rabbit-3.0.10.jar:3.0.10]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.mainLoop(SimpleMessageListenerContainer.java:1323) ~[spring-rabbit-3.0.10.jar:3.0.10]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1225) ~[spring-rabbit-3.0.10.jar:3.0.10]
at java.base/java.lang.Thread.run(Thread.java:1623) ~[na:na]
```

When I encounter an error, I investigate it, and I think I have found the source of the problem. This code snippet is BQTableSchemaToProtoDescriptor in BigQuery Storage Write API.

#### Code example

We see that in this code in BQTableSchemaToProtoDescriptor

```
public class BQTableSchemaToProtoDescriptor {
private static ImmutableMap
BQTableSchemaModeMap =
ImmutableMap.of(
TableFieldSchema.Mode.NULLABLE, FieldDescriptorProto.Label.LABEL_OPTIONAL,
TableFieldSchema.Mode.REPEATED, FieldDescriptorProto.Label.LABEL_REPEATED,
TableFieldSchema.Mode.REQUIRED, FieldDescriptorProto.Label.LABEL_REQUIRED);

private static ImmutableMap
BQTableSchemaTypeMap =
new ImmutableMap.Builder()
.put(TableFieldSchema.Type.BOOL, FieldDescriptorProto.Type.TYPE_BOOL)
.put(TableFieldSchema.Type.BYTES, FieldDescriptorProto.Type.TYPE_BYTES)
.put(TableFieldSchema.Type.DATE, FieldDescriptorProto.Type.TYPE_INT32)
.put(TableFieldSchema.Type.DATETIME, FieldDescriptorProto.Type.TYPE_INT64)
.put(TableFieldSchema.Type.DOUBLE, FieldDescriptorProto.Type.TYPE_DOUBLE)
.put(TableFieldSchema.Type.GEOGRAPHY, FieldDescriptorProto.Type.TYPE_STRING)
.put(TableFieldSchema.Type.INT64, FieldDescriptorProto.Type.TYPE_INT64)
.put(TableFieldSchema.Type.NUMERIC, FieldDescriptorProto.Type.TYPE_BYTES)
.put(TableFieldSchema.Type.BIGNUMERIC, FieldDescriptorProto.Type.TYPE_BYTES)
.put(TableFieldSchema.Type.STRING, FieldDescriptorProto.Type.TYPE_STRING)
.put(TableFieldSchema.Type.STRUCT, FieldDescriptorProto.Type.TYPE_MESSAGE)
.put(TableFieldSchema.Type.TIME, FieldDescriptorProto.Type.TYPE_INT64)
.put(TableFieldSchema.Type.TIMESTAMP, FieldDescriptorProto.Type.TYPE_INT64)
.put(TableFieldSchema.Type.JSON, FieldDescriptorProto.Type.TYPE_STRING)
.put(TableFieldSchema.Type.INTERVAL, FieldDescriptorProto.Type.TYPE_STRING)
.put(TableFieldSchema.Type.RANGE, FieldDescriptorProto.Type.TYPE_MESSAGE)
.build();
```

Howewer we see that converter like that

```
private static ImmutableMap BQTableSchemaModeMap =
ImmutableMap.of(
Field.Mode.NULLABLE, TableFieldSchema.Mode.NULLABLE,
Field.Mode.REPEATED, TableFieldSchema.Mode.REPEATED,
Field.Mode.REQUIRED, TableFieldSchema.Mode.REQUIRED);

private static ImmutableMap BQTableSchemaTypeMap =
new ImmutableMap.Builder()
.put(StandardSQLTypeName.BOOL, TableFieldSchema.Type.BOOL)
.put(StandardSQLTypeName.BYTES, TableFieldSchema.Type.BYTES)
.put(StandardSQLTypeName.DATE, TableFieldSchema.Type.DATE)
.put(StandardSQLTypeName.DATETIME, TableFieldSchema.Type.DATETIME)
.put(StandardSQLTypeName.FLOAT64, TableFieldSchema.Type.DOUBLE)
.put(StandardSQLTypeName.GEOGRAPHY, TableFieldSchema.Type.GEOGRAPHY)
.put(StandardSQLTypeName.INT64, TableFieldSchema.Type.INT64)
.put(StandardSQLTypeName.NUMERIC, TableFieldSchema.Type.NUMERIC)
.put(StandardSQLTypeName.STRING, TableFieldSchema.Type.STRING)
.put(StandardSQLTypeName.STRUCT, TableFieldSchema.Type.STRUCT)
.put(StandardSQLTypeName.TIME, TableFieldSchema.Type.TIME)
.put(StandardSQLTypeName.TIMESTAMP, TableFieldSchema.Type.TIMESTAMP)
.build();
```

As you can see, the datetime format is not converted with the time zone, such as +03, for the INT64 reason. That's why we get an error. We could write a converter, but we would have a huge latency problem because we use 2 PBs of data per day with this Write API. Therefore, we cannot convert every single datetime in memory.

Guide de contribution

Ouvrir le guide de contribution

Évaluation

Cette issue n'a pas encore été évaluée.

Recevez les nouvelles issues par e-mail

Un résumé court des issues GitHub adaptées aux débutants.