From 1726c85ee7d3b1be58e6616de199d6cc2c01959f Mon Sep 17 00:00:00 2001 From: Gaole Meng Date: Wed, 29 Mar 2023 18:54:08 -0700 Subject: [PATCH 1/9] feat: add public api to stream writer to set the maximum wait time --- .../bigquery/storage/v1/ConnectionWorker.java | 2 +- .../bigquery/storage/v1/StreamWriter.java | 10 ++++++ .../bigquery/storage/v1/StreamWriterTest.java | 31 +++++++++++++++++++ 3 files changed, 42 insertions(+), 1 deletion(-) diff --git a/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/ConnectionWorker.java b/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/ConnectionWorker.java index 1aeb911943..12afbf13e0 100644 --- a/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/ConnectionWorker.java +++ b/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/ConnectionWorker.java @@ -75,7 +75,7 @@ class ConnectionWorker implements AutoCloseable { * We will constantly checking how much time we have been waiting for the next request callback * if we wait too much time we will start shutting down the connections and clean up the queues. */ - private static Duration MAXIMUM_REQUEST_CALLBACK_WAIT_TIME = Duration.ofMinutes(15); + static Duration MAXIMUM_REQUEST_CALLBACK_WAIT_TIME = Duration.ofMinutes(15); private Lock lock; private Condition hasMessageInWaitingQueue; diff --git a/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/StreamWriter.java b/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/StreamWriter.java index b21a52a63d..bfa30c6141 100644 --- a/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/StreamWriter.java +++ b/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/StreamWriter.java @@ -518,6 +518,16 @@ public synchronized TableSchema getUpdatedSchema() { : null; } + /** + * Sets the maximum time a request is allowed to be waiting in request waiting queue. Under very + * low chance, it's possible for append request to be waiting indefintely for request callback + * when Google networking SDK does not detect the networking breakage. The default timeout is 15 + * minutes. We are investigating the root cause for callback not triggered by networking SDK. + */ + public static void setMaxRequestCallbackWaitTime(Duration waitTime) { + ConnectionWorker.MAXIMUM_REQUEST_CALLBACK_WAIT_TIME = waitTime; + } + long getCreationTimestamp() { return creationTimestamp; } diff --git a/google-cloud-bigquerystorage/src/test/java/com/google/cloud/bigquery/storage/v1/StreamWriterTest.java b/google-cloud-bigquerystorage/src/test/java/com/google/cloud/bigquery/storage/v1/StreamWriterTest.java index af36273102..bc6dd71690 100644 --- a/google-cloud-bigquerystorage/src/test/java/com/google/cloud/bigquery/storage/v1/StreamWriterTest.java +++ b/google-cloud-bigquerystorage/src/test/java/com/google/cloud/bigquery/storage/v1/StreamWriterTest.java @@ -15,6 +15,7 @@ */ package com.google.cloud.bigquery.storage.v1; +import static com.google.common.truth.Truth.assertThat; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertThrows; @@ -113,6 +114,7 @@ public StreamWriterTest() throws DescriptorValidationException {} @Before public void setUp() throws Exception { testBigQueryWrite = new FakeBigQueryWrite(); + StreamWriter.setMaxRequestCallbackWaitTime(java.time.Duration.ofSeconds(10000)); ConnectionWorker.setMaxInflightQueueWaitTime(300000); serviceHelper = new MockServiceHelper( @@ -947,6 +949,35 @@ public void testMessageTooLarge() throws Exception { writer.close(); } + @Test + public void testThrowExceptionWhileWithinAppendLoop_MaxWaitTimeExceed() throws Exception { + ProtoSchema schema1 = createProtoSchema("foo"); + StreamWriter.setMaxRequestCallbackWaitTime(java.time.Duration.ofSeconds(1)); + StreamWriter writer = + StreamWriter.newBuilder(TEST_STREAM_1, client).setWriterSchema(schema1).build(); + testBigQueryWrite.setResponseSleep(org.threeten.bp.Duration.ofSeconds(3)); + + long appendCount = 10; + for (int i = 0; i < appendCount; i++) { + testBigQueryWrite.addResponse(createAppendResponse(i)); + } + + // In total insert 5 requests, + List> futures = new ArrayList<>(); + for (int i = 0; i < appendCount; i++) { + futures.add(writer.append(createProtoRows(new String[] {String.valueOf(i)}), i)); + } + + for (int i = 0; i < appendCount; i++) { + int finalI = i; + ExecutionException ex = + assertThrows( + ExecutionException.class, + () -> futures.get(finalI).get().getAppendResult().getOffset().getValue()); + assertThat(ex.getCause()).hasMessageThat().contains("Request has waited in inflight queue"); + } + } + @Test public void testAppendWithResetSuccess() throws Exception { try (StreamWriter writer = getTestStreamWriter()) { From 3f0c5ebe7f400cb49aedbe329165725eda5f5329 Mon Sep 17 00:00:00 2001 From: Owl Bot Date: Thu, 30 Mar 2023 20:47:37 +0000 Subject: [PATCH 2/9] =?UTF-8?q?=F0=9F=A6=89=20Updates=20from=20OwlBot=20po?= =?UTF-8?q?st-processor?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md --- README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/README.md b/README.md index 145a863499..76e80afa7f 100644 --- a/README.md +++ b/README.md @@ -219,7 +219,7 @@ Java is a registered trademark of Oracle and/or its affiliates. [kokoro-badge-link-5]: http://storage.googleapis.com/cloud-devrel-public/java/badges/java-bigquerystorage/java11.html [stability-image]: https://img.shields.io/badge/stability-stable-green [maven-version-image]: https://img.shields.io/maven-central/v/com.google.cloud/google-cloud-bigquerystorage.svg -[maven-version-link]: https://central.sonatype.com/artifact/com.google.cloud/google-cloud-bigquerystorage/2.34.1 +[maven-version-link]: https://central.sonatype.com/artifact/com.google.cloud/google-cloud-bigquerystorage/2.34.1: [authentication]: https://github.com/googleapis/google-cloud-java#authentication [auth-scopes]: https://developers.google.com/identity/protocols/oauth2/scopes [predefined-iam-roles]: https://cloud.google.com/iam/docs/understanding-roles#predefined_roles From a448c8b205ea41796b77e5f644ce527f0db53501 Mon Sep 17 00:00:00 2001 From: Gaole Meng Date: Thu, 30 Mar 2023 14:51:35 -0700 Subject: [PATCH 3/9] modify back the readme change from owl post processor --- README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/README.md b/README.md index 76e80afa7f..145a863499 100644 --- a/README.md +++ b/README.md @@ -219,7 +219,7 @@ Java is a registered trademark of Oracle and/or its affiliates. [kokoro-badge-link-5]: http://storage.googleapis.com/cloud-devrel-public/java/badges/java-bigquerystorage/java11.html [stability-image]: https://img.shields.io/badge/stability-stable-green [maven-version-image]: https://img.shields.io/maven-central/v/com.google.cloud/google-cloud-bigquerystorage.svg -[maven-version-link]: https://central.sonatype.com/artifact/com.google.cloud/google-cloud-bigquerystorage/2.34.1: +[maven-version-link]: https://central.sonatype.com/artifact/com.google.cloud/google-cloud-bigquerystorage/2.34.1 [authentication]: https://github.com/googleapis/google-cloud-java#authentication [auth-scopes]: https://developers.google.com/identity/protocols/oauth2/scopes [predefined-iam-roles]: https://cloud.google.com/iam/docs/understanding-roles#predefined_roles From b15b140d703fe510861cda925aece31ed7c3b234 Mon Sep 17 00:00:00 2001 From: Owl Bot Date: Fri, 31 Mar 2023 20:18:30 +0000 Subject: [PATCH 4/9] =?UTF-8?q?=F0=9F=A6=89=20Updates=20from=20OwlBot=20po?= =?UTF-8?q?st-processor?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md --- README.md | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index 145a863499..5eaba81994 100644 --- a/README.md +++ b/README.md @@ -56,13 +56,13 @@ implementation 'com.google.cloud:google-cloud-bigquerystorage' If you are using Gradle without BOM, add this to your dependencies: ```Groovy -implementation 'com.google.cloud:google-cloud-bigquerystorage:2.34.1' +implementation 'com.google.cloud:google-cloud-bigquerystorage:2.34.2' ``` If you are using SBT, add this to your dependencies: ```Scala -libraryDependencies += "com.google.cloud" % "google-cloud-bigquerystorage" % "2.34.1" +libraryDependencies += "com.google.cloud" % "google-cloud-bigquerystorage" % "2.34.2" ``` @@ -219,7 +219,7 @@ Java is a registered trademark of Oracle and/or its affiliates. [kokoro-badge-link-5]: http://storage.googleapis.com/cloud-devrel-public/java/badges/java-bigquerystorage/java11.html [stability-image]: https://img.shields.io/badge/stability-stable-green [maven-version-image]: https://img.shields.io/maven-central/v/com.google.cloud/google-cloud-bigquerystorage.svg -[maven-version-link]: https://central.sonatype.com/artifact/com.google.cloud/google-cloud-bigquerystorage/2.34.1 +[maven-version-link]: https://central.sonatype.com/artifact/com.google.cloud/google-cloud-bigquerystorage/2.34.2 [authentication]: https://github.com/googleapis/google-cloud-java#authentication [auth-scopes]: https://developers.google.com/identity/protocols/oauth2/scopes [predefined-iam-roles]: https://cloud.google.com/iam/docs/understanding-roles#predefined_roles From affa11ac873b38d27b6f36d5a8501e70105ba68b Mon Sep 17 00:00:00 2001 From: Gaole Meng Date: Fri, 21 Apr 2023 17:49:54 -0700 Subject: [PATCH 5/9] fix: Reduce the timeout to 5 minutes for the requests wait time in queue. Since in write api server side we have total timeout of 2 minutes, it does not make sense to wait 15 minutes to determine whether we have met dead connection, let's reduce the timeout here --- .../google/cloud/bigquery/storage/v1/ConnectionWorker.java | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/ConnectionWorker.java b/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/ConnectionWorker.java index 12afbf13e0..64abf82bb9 100644 --- a/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/ConnectionWorker.java +++ b/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/ConnectionWorker.java @@ -75,7 +75,7 @@ class ConnectionWorker implements AutoCloseable { * We will constantly checking how much time we have been waiting for the next request callback * if we wait too much time we will start shutting down the connections and clean up the queues. */ - static Duration MAXIMUM_REQUEST_CALLBACK_WAIT_TIME = Duration.ofMinutes(15); + static Duration MAXIMUM_REQUEST_CALLBACK_WAIT_TIME = Duration.ofMinutes(5); private Lock lock; private Condition hasMessageInWaitingQueue; @@ -321,7 +321,7 @@ public void run() { } private void resetConnection() { - log.info("Reconnecting for stream:" + streamName + " id: " + writerId); + log.info("Start connecting stream: " + streamName + " id: " + writerId); if (this.streamConnection != null) { // It's safe to directly close the previous connection as the in flight messages // will be picked up by the next connection. @@ -344,7 +344,7 @@ public void run(Throwable finalStatus) { doneCallback(finalStatus); } }); - log.info("Reconnect done for stream:" + streamName + " id: " + writerId); + log.info("Finish connecting stream: " + streamName + " id: " + writerId); } /** Schedules the writing of rows at given offset. */ From 60869f74065fc100f34f80ba00ee816343f7ba01 Mon Sep 17 00:00:00 2001 From: Owl Bot Date: Tue, 25 Apr 2023 00:18:28 +0000 Subject: [PATCH 6/9] =?UTF-8?q?=F0=9F=A6=89=20Updates=20from=20OwlBot=20po?= =?UTF-8?q?st-processor?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md --- README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/README.md b/README.md index e453f63b9a..d4392b52df 100644 --- a/README.md +++ b/README.md @@ -50,7 +50,7 @@ If you are using Maven without the BOM, add this to your dependencies: If you are using Gradle 5.x or later, add this to your dependencies: ```Groovy -implementation platform('com.google.cloud:libraries-bom:26.12.0') +implementation platform('com.google.cloud:libraries-bom:26.13.0') implementation 'com.google.cloud:google-cloud-bigquerystorage' ``` From 687ef3abc42bb7620309313a588fd9cffcdf2938 Mon Sep 17 00:00:00 2001 From: Gaole Meng Date: Wed, 24 May 2023 17:35:30 -0700 Subject: [PATCH 7/9] fix: 1.disable refresh of stream writer when the table schema is explicitly provided 2. fix location string matching for multiplexing --- .../bigquery/storage/v1/ConnectionWorker.java | 4 +- .../storage/v1/SchemaAwareStreamWriter.java | 16 +- .../storage/v1/JsonStreamWriterTest.java | 155 ++++++++++++++++-- .../it/ITBigQueryWriteManualClientTest.java | 4 +- 4 files changed, 160 insertions(+), 19 deletions(-) diff --git a/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/ConnectionWorker.java b/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/ConnectionWorker.java index 64abf82bb9..7e86da4d81 100644 --- a/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/ConnectionWorker.java +++ b/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/ConnectionWorker.java @@ -349,7 +349,7 @@ public void run(Throwable finalStatus) { /** Schedules the writing of rows at given offset. */ ApiFuture append(StreamWriter streamWriter, ProtoRows rows, long offset) { - if (this.location != null && this.location != streamWriter.getLocation()) { + if (this.location != null && !this.location.equals(streamWriter.getLocation())) { throw new StatusRuntimeException( Status.fromCode(Code.INVALID_ARGUMENT) .withDescription( @@ -357,7 +357,7 @@ ApiFuture append(StreamWriter streamWriter, ProtoRows rows, + streamWriter.getLocation() + " is scheduled to use a connection with location " + this.location)); - } else if (this.location == null && streamWriter.getStreamName() != this.streamName) { + } else if (this.location == null && !streamWriter.getStreamName().equals(this.streamName)) { // Location is null implies this is non-multiplexed connection. throw new StatusRuntimeException( Status.fromCode(Code.INVALID_ARGUMENT) diff --git a/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/SchemaAwareStreamWriter.java b/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/SchemaAwareStreamWriter.java index 0e8f931914..10fceeee68 100644 --- a/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/SchemaAwareStreamWriter.java +++ b/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/SchemaAwareStreamWriter.java @@ -54,6 +54,10 @@ public class SchemaAwareStreamWriter implements AutoCloseable { private TableSchema tableSchema; private ProtoSchema protoSchema; + // During some sitaution we want to skip stream writer refresh for updated schema. e.g. when + // the user provides the table schema, we should always use that schema. + private final boolean skipRefreshStreamWriter; + /** * Constructs the SchemaAwareStreamWriter * @@ -87,6 +91,7 @@ private SchemaAwareStreamWriter(Builder builder) this.tableSchema = builder.tableSchema; this.toProtoConverter = builder.toProtoConverter; this.ignoreUnknownFields = builder.ignoreUnknownFields; + this.skipRefreshStreamWriter = builder.skipRefreshStreamWriter; } /** @@ -125,6 +130,10 @@ private Message buildMessage(T item) return this.toProtoConverter.convertToProtoMessage( this.descriptor, this.tableSchema, item, ignoreUnknownFields); } catch (Exceptions.DataHasUnknownFieldException ex) { + // Directly return error when stream writer refresh is disabled. + if (this.skipRefreshStreamWriter) { + throw ex; + } LOG.warning( "Saw unknown field " + ex.getFieldName() @@ -157,7 +166,7 @@ public ApiFuture append(Iterable items, long offset) // Handle schema updates in a Thread-safe way by locking down the operation synchronized (this) { // Create a new stream writer internally if a new updated schema is reported from backend. - if (this.streamWriter.getUpdatedSchema() != null) { + if (!this.skipRefreshStreamWriter && this.streamWriter.getUpdatedSchema() != null) { refreshWriter(this.streamWriter.getUpdatedSchema()); } @@ -404,6 +413,8 @@ public static final class Builder { private final BigQueryWriteClient client; private final TableSchema tableSchema; + private final boolean skipRefreshStreamWriter; + private final ToProtoConverter toProtoConverter; private TransportChannelProvider channelProvider; private CredentialsProvider credentialsProvider; @@ -459,11 +470,12 @@ private Builder( .build(); WriteStream writeStream = this.client.getWriteStream(writeStreamRequest); - this.tableSchema = writeStream.getTableSchema(); this.location = writeStream.getLocation(); + this.skipRefreshStreamWriter = false; } else { this.tableSchema = tableSchema; + this.skipRefreshStreamWriter = true; } this.toProtoConverter = toProtoConverter; } diff --git a/google-cloud-bigquerystorage/src/test/java/com/google/cloud/bigquery/storage/v1/JsonStreamWriterTest.java b/google-cloud-bigquerystorage/src/test/java/com/google/cloud/bigquery/storage/v1/JsonStreamWriterTest.java index 1bd91925e4..eed96886a4 100644 --- a/google-cloud-bigquerystorage/src/test/java/com/google/cloud/bigquery/storage/v1/JsonStreamWriterTest.java +++ b/google-cloud-bigquerystorage/src/test/java/com/google/cloud/bigquery/storage/v1/JsonStreamWriterTest.java @@ -160,6 +160,13 @@ private JsonStreamWriter.Builder getTestJsonStreamWriterBuilder( .setExecutorProvider(InstantiatingExecutorProvider.newBuilder().build()); } + private JsonStreamWriter.Builder getTestJsonStreamWriterBuilder(String testStream) { + return JsonStreamWriter.newBuilder(testStream, client) + .setChannelProvider(channelProvider) + .setCredentialsProvider(NoCredentialsProvider.create()) + .setExecutorProvider(InstantiatingExecutorProvider.newBuilder().build()); + } + @Test public void testTwoParamNewBuilder_nullSchema() { try { @@ -658,8 +665,13 @@ public void run() throws Throwable { @Test public void testSimpleSchemaUpdate() throws Exception { - try (JsonStreamWriter writer = - getTestJsonStreamWriterBuilder(TEST_STREAM, TABLE_SCHEMA).build()) { + testBigQueryWrite.addResponse( + WriteStream.newBuilder() + .setName(TEST_STREAM) + .setTableSchema(TABLE_SCHEMA) + .setLocation("us") + .build()); + try (JsonStreamWriter writer = getTestJsonStreamWriterBuilder(TEST_STREAM).build()) { testBigQueryWrite.addResponse( AppendRowsResponse.newBuilder() .setAppendResult( @@ -722,7 +734,6 @@ public void testSimpleSchemaUpdate() throws Exception { updatedFoo.put("bar", "bbb"); JSONArray updatedJsonArr = new JSONArray(); updatedJsonArr.put(updatedFoo); - ApiFuture appendFuture4 = writer.append(updatedJsonArr); assertEquals(3L, appendFuture4.get().getAppendResult().getOffset().getValue()); @@ -751,6 +762,88 @@ public void testSimpleSchemaUpdate() throws Exception { } } + @Test + public void testSimpleSchemaUpdate_skipRefreshWriterIfSchemaProvided() throws Exception { + testBigQueryWrite.addResponse( + WriteStream.newBuilder() + .setName(TEST_STREAM) + .setTableSchema(TABLE_SCHEMA) + .setLocation("us") + .build()); + try (JsonStreamWriter writer = + getTestJsonStreamWriterBuilder(TEST_STREAM, TABLE_SCHEMA).build()) { + testBigQueryWrite.addResponse( + AppendRowsResponse.newBuilder() + .setAppendResult( + AppendRowsResponse.AppendResult.newBuilder().setOffset(Int64Value.of(0)).build()) + .setUpdatedSchema(UPDATED_TABLE_SCHEMA) + .build()); + testBigQueryWrite.addResponse(createAppendResponse(1)); + testBigQueryWrite.addResponse(createAppendResponse(2)); + testBigQueryWrite.addResponse(createAppendResponse(3)); + // First append + JSONObject foo = new JSONObject(); + foo.put("foo", "aaa"); + JSONArray jsonArr = new JSONArray(); + jsonArr.put(foo); + + ApiFuture appendFuture1 = writer.append(jsonArr); + ApiFuture appendFuture2 = writer.append(jsonArr); + ApiFuture appendFuture3 = writer.append(jsonArr); + + assertEquals(0L, appendFuture1.get().getAppendResult().getOffset().getValue()); + assertEquals(1L, appendFuture2.get().getAppendResult().getOffset().getValue()); + assertEquals( + 1, + testBigQueryWrite + .getAppendRequests() + .get(0) + .getProtoRows() + .getRows() + .getSerializedRowsCount()); + assertEquals( + testBigQueryWrite + .getAppendRequests() + .get(0) + .getProtoRows() + .getRows() + .getSerializedRows(0), + FooType.newBuilder().setFoo("aaa").build().toByteString()); + + assertEquals(2L, appendFuture3.get().getAppendResult().getOffset().getValue()); + assertEquals( + 1, + testBigQueryWrite + .getAppendRequests() + .get(1) + .getProtoRows() + .getRows() + .getSerializedRowsCount()); + assertEquals( + testBigQueryWrite + .getAppendRequests() + .get(1) + .getProtoRows() + .getRows() + .getSerializedRows(0), + FooType.newBuilder().setFoo("aaa").build().toByteString()); + + // Second append with updated schema. + JSONObject updatedFoo = new JSONObject(); + updatedFoo.put("foo", "aaa"); + updatedFoo.put("bar", "bbb"); + JSONArray updatedJsonArr = new JSONArray(); + updatedJsonArr.put(updatedFoo); + + // Schema update will not happen for writer that has schema explicitly provided. + assertThrows( + AppendSerializationError.class, + () -> { + ApiFuture appendFuture4 = writer.append(updatedJsonArr); + }); + } + } + @Test public void testWithoutIgnoreUnknownFieldsUpdateImmeidateSuccess() throws Exception { TableSchema tableSchema = TableSchema.newBuilder().addFields(0, TEST_INT).build(); @@ -764,6 +857,10 @@ public void testWithoutIgnoreUnknownFieldsUpdateImmeidateSuccess() throws Except .setType(TableFieldSchema.Type.STRING) .setMode(Mode.NULLABLE)) .build(); + + // GetWriteStream is called once and got the updated schema + testBigQueryWrite.addResponse( + WriteStream.newBuilder().setName(TEST_STREAM).setTableSchema(tableSchema).build()); // GetWriteStream is called once and the writer is fixed to accept unknown fields. testBigQueryWrite.addResponse( WriteStream.newBuilder().setName(TEST_STREAM).setTableSchema(updatedSchema).build()); @@ -772,8 +869,7 @@ public void testWithoutIgnoreUnknownFieldsUpdateImmeidateSuccess() throws Except .setAppendResult( AppendRowsResponse.AppendResult.newBuilder().setOffset(Int64Value.of(0)).build()) .build()); - try (JsonStreamWriter writer = - getTestJsonStreamWriterBuilder(TEST_STREAM, tableSchema).build()) { + try (JsonStreamWriter writer = getTestJsonStreamWriterBuilder(TEST_STREAM).build()) { JSONObject foo = new JSONObject(); foo.put("test_int", 10); JSONObject bar = new JSONObject(); @@ -800,6 +896,8 @@ public void testWithoutIgnoreUnknownFieldsUpdateSecondSuccess() throws Exception .setMode(Mode.NULLABLE)) .build(); // GetWriteStream is called once and got the updated schema + testBigQueryWrite.addResponse( + WriteStream.newBuilder().setName(TEST_STREAM).setTableSchema(TABLE_SCHEMA).build()); testBigQueryWrite.addResponse( WriteStream.newBuilder().setName(TEST_STREAM).setTableSchema(updatedSchema).build()); testBigQueryWrite.addResponse( @@ -807,8 +905,7 @@ public void testWithoutIgnoreUnknownFieldsUpdateSecondSuccess() throws Exception .setAppendResult( AppendRowsResponse.AppendResult.newBuilder().setOffset(Int64Value.of(0)).build()) .build()); - try (JsonStreamWriter writer = - getTestJsonStreamWriterBuilder(TEST_STREAM, tableSchema).build()) { + try (JsonStreamWriter writer = getTestJsonStreamWriterBuilder(TEST_STREAM).build()) { JSONObject foo = new JSONObject(); foo.put("test_int", 10); JSONObject bar = new JSONObject(); @@ -826,15 +923,28 @@ public void testSchemaUpdateInMultiplexing_singleConnection() throws Exception { // Set min connection count to be 1 to force sharing connection. ConnectionWorkerPool.setOptions( Settings.builder().setMinConnectionsPerRegion(1).setMaxConnectionsPerRegion(1).build()); + // GetWriteStream is called twice and got the updated schema + testBigQueryWrite.addResponse( + WriteStream.newBuilder() + .setName(TEST_STREAM) + .setTableSchema(TABLE_SCHEMA) + .setLocation("us") + .build()); + testBigQueryWrite.addResponse( + WriteStream.newBuilder() + .setName(TEST_STREAM) + .setTableSchema(TABLE_SCHEMA_2) + .setLocation("us") + .build()); // The following two writers have different stream name and schema, but will share the same // connection . JsonStreamWriter writer1 = - getTestJsonStreamWriterBuilder(TEST_STREAM, TABLE_SCHEMA) + getTestJsonStreamWriterBuilder(TEST_STREAM) .setEnableConnectionPool(true) .setLocation("us") .build(); JsonStreamWriter writer2 = - getTestJsonStreamWriterBuilder(TEST_STREAM_2, TABLE_SCHEMA_2) + getTestJsonStreamWriterBuilder(TEST_STREAM_2) .setEnableConnectionPool(true) .setLocation("us") .build(); @@ -911,14 +1021,27 @@ public void testSchemaUpdateInMultiplexing_multipleWriterForSameStreamName() thr ConnectionWorkerPool.setOptions( Settings.builder().setMinConnectionsPerRegion(1).setMaxConnectionsPerRegion(1).build()); + // GetWriteStream is called twice and got the updated schema + testBigQueryWrite.addResponse( + WriteStream.newBuilder() + .setName(TEST_STREAM) + .setTableSchema(TABLE_SCHEMA) + .setLocation("us") + .build()); + testBigQueryWrite.addResponse( + WriteStream.newBuilder() + .setName(TEST_STREAM) + .setTableSchema(TABLE_SCHEMA) + .setLocation("us") + .build()); // Create two writers writing to the same stream. JsonStreamWriter writer1 = - getTestJsonStreamWriterBuilder(TEST_STREAM, TABLE_SCHEMA) + getTestJsonStreamWriterBuilder(TEST_STREAM) .setEnableConnectionPool(true) .setLocation("us") .build(); JsonStreamWriter writer2 = - getTestJsonStreamWriterBuilder(TEST_STREAM, TABLE_SCHEMA) + getTestJsonStreamWriterBuilder(TEST_STREAM) .setEnableConnectionPool(true) .setLocation("us") .build(); @@ -987,10 +1110,16 @@ public void testSchemaUpdateInMultiplexing_IgnoreUpdateIfTimeStampNewer() throws // Set min connection count to be 1 to force sharing connection. ConnectionWorkerPool.setOptions( Settings.builder().setMinConnectionsPerRegion(1).setMaxConnectionsPerRegion(1).build()); + testBigQueryWrite.addResponse( + WriteStream.newBuilder() + .setName(TEST_STREAM) + .setTableSchema(TABLE_SCHEMA) + .setLocation("us") + .build()); // The following two writers have different stream name and schema, but will share the same - // connection . + // connection. JsonStreamWriter writer1 = - getTestJsonStreamWriterBuilder(TEST_STREAM, TABLE_SCHEMA) + getTestJsonStreamWriterBuilder(TEST_STREAM) .setEnableConnectionPool(true) .setLocation("us") .build(); diff --git a/google-cloud-bigquerystorage/src/test/java/com/google/cloud/bigquery/storage/v1/it/ITBigQueryWriteManualClientTest.java b/google-cloud-bigquerystorage/src/test/java/com/google/cloud/bigquery/storage/v1/it/ITBigQueryWriteManualClientTest.java index a068f6d635..1e73643eb8 100644 --- a/google-cloud-bigquerystorage/src/test/java/com/google/cloud/bigquery/storage/v1/it/ITBigQueryWriteManualClientTest.java +++ b/google-cloud-bigquerystorage/src/test/java/com/google/cloud/bigquery/storage/v1/it/ITBigQueryWriteManualClientTest.java @@ -785,7 +785,7 @@ public void testJsonStreamWriterSchemaUpdate() WriteStream.newBuilder().setType(WriteStream.Type.COMMITTED).build()) .build()); try (JsonStreamWriter jsonStreamWriter = - JsonStreamWriter.newBuilder(writeStream.getName(), writeStream.getTableSchema()).build()) { + JsonStreamWriter.newBuilder(writeStream.getName(), client).build()) { // write the 1st row JSONObject foo = new JSONObject(); foo.put("col1", "aaa"); @@ -895,7 +895,7 @@ public void testJsonStreamWriterSchemaUpdateConcurrent() // Start writing using the JsonWriter try (JsonStreamWriter jsonStreamWriter = - JsonStreamWriter.newBuilder(writeStream.getName(), writeStream.getTableSchema()).build()) { + JsonStreamWriter.newBuilder(writeStream.getName(), client).build()) { int numberOfThreads = 5; ExecutorService streamTaskExecutor = Executors.newFixedThreadPool(5); CountDownLatch latch = new CountDownLatch(numberOfThreads); From c307e4b2d614b4b1ad67cabaa4987bc1a8f9e67c Mon Sep 17 00:00:00 2001 From: Gaole Meng Date: Fri, 7 Jul 2023 13:59:33 -0700 Subject: [PATCH 8/9] feat: improve json stream writer json to proto conversion speed by caching the schema. This will introduce approximately 2x improvement to append speed --- .../clirr-ignored-differences.xml | 12 + .../cloud/bigquery/storage/v1/Exceptions.java | 23 ++ .../storage/v1/JsonToProtoMessage.java | 289 +++++++++++++----- .../storage/v1/SchemaAwareStreamWriter.java | 51 ++-- .../bigquery/storage/v1/ToProtoConverter.java | 5 +- .../storage/v1/JsonToProtoMessageTest.java | 120 ++++++-- 6 files changed, 367 insertions(+), 133 deletions(-) diff --git a/google-cloud-bigquerystorage/clirr-ignored-differences.xml b/google-cloud-bigquerystorage/clirr-ignored-differences.xml index 96d4b3d595..1ce4f651e5 100644 --- a/google-cloud-bigquerystorage/clirr-ignored-differences.xml +++ b/google-cloud-bigquerystorage/clirr-ignored-differences.xml @@ -157,5 +157,17 @@ com/google/cloud/bigquery/storage/v1/JsonStreamWriter boolean isDone() + + 7006 + com/google/cloud/bigquery/storage/v1/ToProtoConverter + com.google.protobuf.DynamicMessage convertToProtoMessage(com.google.protobuf.Descriptors$Descriptor, com.google.cloud.bigquery.storage.v1.TableSchema, java.lang.Object, boolean) + java.util.List + + + 7005 + com/google/cloud/bigquery/storage/v1/ToProtoConverter + com.google.protobuf.DynamicMessage convertToProtoMessage(com.google.protobuf.Descriptors$Descriptor, com.google.cloud.bigquery.storage.v1.TableSchema, java.lang.Object, boolean) + com.google.protobuf.DynamicMessage convertToProtoMessage(com.google.protobuf.Descriptors$Descriptor, com.google.cloud.bigquery.storage.v1.TableSchema, java.lang.Iterable, boolean) + diff --git a/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/Exceptions.java b/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/Exceptions.java index 2f9083e4e9..e67b5daa9c 100644 --- a/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/Exceptions.java +++ b/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/Exceptions.java @@ -259,6 +259,29 @@ public AppendSerializationError( } } + /** This exception is thrown from proto converter to wrap the row index to error mapping. */ + static class RowIndexToErrorException extends IllegalArgumentException { + Map rowIndexToErrorMessage; + + boolean hasDataUnknownError; + + public RowIndexToErrorException( + Map rowIndexToErrorMessage, boolean hasDataUnknownError) { + this.rowIndexToErrorMessage = rowIndexToErrorMessage; + this.hasDataUnknownError = hasDataUnknownError; + } + + // This message should not be exposed to the user directly. + // Please examine individual row's error through `rowIndexToErrorMessage`. + public String getMessage() { + return "The map of row index to error message is " + rowIndexToErrorMessage.toString(); + } + + public boolean hasDataUnknownError() { + return hasDataUnknownError; + } + } + /** This exception is used internally to handle field level parsing errors. */ public static class FieldParseError extends IllegalArgumentException { private final String fieldName; diff --git a/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/JsonToProtoMessage.java b/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/JsonToProtoMessage.java index 3d1e1e0b5d..5ac0a34ae8 100644 --- a/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/JsonToProtoMessage.java +++ b/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/JsonToProtoMessage.java @@ -16,6 +16,7 @@ package com.google.cloud.bigquery.storage.v1; import com.google.api.pathtemplate.ValidationException; +import com.google.cloud.bigquery.storage.v1.Exceptions.RowIndexToErrorException; import com.google.common.base.Preconditions; import com.google.common.collect.ImmutableMap; import com.google.common.primitives.Doubles; @@ -29,7 +30,10 @@ import java.math.BigDecimal; import java.math.RoundingMode; import java.time.LocalDate; +import java.util.ArrayList; +import java.util.HashMap; import java.util.List; +import java.util.Map; import org.json.JSONArray; import org.json.JSONException; import org.json.JSONObject; @@ -121,7 +125,10 @@ public static DynamicMessage convertJsonToProtoMessage( } /** - * Converts input message to Protobuf + * Converts input message to Protobuf. + * + *

WARNING: it's much more efficient to call the other APIs accepting json array if the jsons + * share the same table schema. * * @param protoSchema the schema of the output Protobuf schems. * @param tableSchema tha underlying table schema for which Protobuf is being built. @@ -130,15 +137,37 @@ public static DynamicMessage convertJsonToProtoMessage( * schema should be accepted. * @return Converted message in Protobuf format. */ - @Override public DynamicMessage convertToProtoMessage( Descriptor protoSchema, TableSchema tableSchema, Object json, boolean ignoreUnknownFields) { return convertToProtoMessage(protoSchema, tableSchema, (JSONObject) json, ignoreUnknownFields); } + /** + * Converts Json array to list of Protobuf + * + * @param protoSchema the schema of the output Protobuf schems. + * @param tableSchema tha underlying table schema for which Protobuf is being built. + * @param jsonArray the input JSON array converted to Protobuf. + * @param ignoreUnknownFields flag indicating that the additional fields not present in the output + * schema should be accepted. + * @return Converted message in Protobuf format. + */ + @Override + public List convertToProtoMessage( + Descriptor protoSchema, + TableSchema tableSchema, + Iterable jsonArray, + boolean ignoreUnknownFields) { + return convertToProtoMessage( + protoSchema, tableSchema, (JSONArray) jsonArray, ignoreUnknownFields); + } + /** * Converts Json data to protocol buffer messages given the protocol buffer descriptor. * + *

WARNING: it's much more efficient to call the other APIs accepting json array if the jsons + * share the same table schema. + * * @param protoSchema * @param json * @throws IllegalArgumentException when JSON data is not compatible with proto descriptor. @@ -155,6 +184,9 @@ public DynamicMessage convertToProtoMessage(Descriptor protoSchema, JSONObject j /** * Converts Json data to protocol buffer messages given the protocol buffer descriptor. * + *

WARNING: it's much more efficient to call the other APIs accepting json array if the jsons + * share the same table schema. + * * @param protoSchema * @param tableSchema bigquery table schema is needed for type conversion of DATETIME, TIME, * NUMERIC, BIGNUMERIC @@ -175,6 +207,9 @@ public DynamicMessage convertToProtoMessage( /** * Converts Json data to protocol buffer messages given the protocol buffer descriptor. * + *

WARNING: it's much more efficient to call the other APIs accepting json array if the jsons + * share the same table schema. + * * @param protoSchema * @param tableSchema bigquery table schema is needed for type conversion of DATETIME, TIME, * NUMERIC, BIGNUMERIC @@ -189,11 +224,48 @@ public DynamicMessage convertToProtoMessage( Preconditions.checkNotNull(protoSchema, "Protobuf descriptor is null."); Preconditions.checkNotNull(tableSchema, "TableSchema is null."); Preconditions.checkState(json.length() != 0, "JSONObject is empty."); - return convertToProtoMessage( protoSchema, tableSchema.getFieldsList(), json, "root", ignoreUnknownFields); } + /** + * Converts Json array to list of protocol buffer messages given the protocol buffer descriptor. + * + * @param protoSchema + * @param tableSchema bigquery table schema is needed for type conversion of DATETIME, TIME, + * NUMERIC, BIGNUMERIC + * @param jsonArray + * @param ignoreUnknownFields allows unknown fields in JSON input to be ignored. + * @throws IllegalArgumentException when JSON data is not compatible with proto descriptor. + */ + public List convertToProtoMessage( + Descriptor protoSchema, + TableSchema tableSchema, + JSONArray jsonArray, + boolean ignoreUnknownFields) + throws IllegalArgumentException { + Preconditions.checkNotNull(jsonArray, "jsonArray is null."); + Preconditions.checkNotNull(protoSchema, "Protobuf descriptor is null."); + Preconditions.checkNotNull(tableSchema, "tableSchema is null."); + Preconditions.checkState(jsonArray.length() != 0, "jsonArray is empty."); + + return convertToProtoMessage( + protoSchema, tableSchema.getFieldsList(), jsonArray, "root", ignoreUnknownFields); + } + + private DynamicMessage convertToProtoMessage( + Descriptor protoSchema, + List tableSchema, + JSONObject jsonObject, + String jsonScope, + boolean ignoreUnknownFields) { + JSONArray jsonArray = new JSONArray(); + jsonArray.put(jsonObject); + return convertToProtoMessage( + protoSchema, tableSchema, jsonArray, jsonScope, ignoreUnknownFields) + .get(0); + } + /** * Converts Json data to protocol buffer messages given the protocol buffer descriptor. * @@ -202,84 +274,162 @@ public DynamicMessage convertToProtoMessage( * @param jsonScope Debugging purposes * @throws IllegalArgumentException when JSON data is not compatible with proto descriptor. */ - private DynamicMessage convertToProtoMessage( + private List convertToProtoMessage( Descriptor protoSchema, List tableSchema, - JSONObject json, + JSONArray jsonArray, String jsonScope, boolean ignoreUnknownFields) - throws IllegalArgumentException { - - DynamicMessage.Builder protoMsg = DynamicMessage.newBuilder(protoSchema); - String[] jsonNames = JSONObject.getNames(json); - if (jsonNames == null) { - return protoMsg.build(); - } - for (String jsonName : jsonNames) { - // We want lowercase here to support case-insensitive data writes. - // The protobuf descriptor that is used is assumed to have all lowercased fields - String jsonFieldLocator = jsonName.toLowerCase(); + throws RowIndexToErrorException { + List messageList = new ArrayList<>(); + Map jsonNameToMetadata = new HashMap<>(); + Map rowIndexToErrorMessage = new HashMap<>(); - // If jsonName is not compatible with proto naming convention, we should look by its - // placeholder name. - if (!BigQuerySchemaUtil.isProtoCompatible(jsonFieldLocator)) { - jsonFieldLocator = BigQuerySchemaUtil.generatePlaceholderFieldName(jsonFieldLocator); - } - String currentScope = jsonScope + "." + jsonName; - FieldDescriptor field = protoSchema.findFieldByName(jsonFieldLocator); - if (field == null && !ignoreUnknownFields) { - throw new Exceptions.DataHasUnknownFieldException(currentScope); - } else if (field == null) { - continue; - } - TableFieldSchema fieldSchema = null; - if (tableSchema != null) { - // protoSchema is generated from tableSchema so their field ordering should match. - fieldSchema = tableSchema.get(field.getIndex()); - if (!fieldSchema.getName().toLowerCase().equals(BigQuerySchemaUtil.getFieldName(field))) { - throw new ValidationException( - "Field at index " - + field.getIndex() - + " has mismatch names (" - + fieldSchema.getName() - + ") (" - + field.getName() - + ")"); - } - } + boolean hasDataUnknownError = false; + for (int i = 0; i < jsonArray.length(); i++) { try { - if (!field.isRepeated()) { - fillField( - protoMsg, field, fieldSchema, json, jsonName, currentScope, ignoreUnknownFields); + DynamicMessage.Builder protoMsg = DynamicMessage.newBuilder(protoSchema); + JSONObject jsonObject = jsonArray.getJSONObject(i); + String[] jsonNames = JSONObject.getNames(jsonObject); + if (jsonNames == null) { + messageList.add(protoMsg.build()); + continue; + } + for (String jsonName : jsonNames) { + String currentScope = jsonScope + "." + jsonName; + FieldDescriptorAndFieldTableSchema fieldDescriptorAndFieldTableSchema = + jsonNameToMetadata.computeIfAbsent( + currentScope, + k -> { + return computeDescriptorAndSchema( + currentScope, ignoreUnknownFields, jsonName, protoSchema, tableSchema); + }); + if (fieldDescriptorAndFieldTableSchema == null) { + continue; + } + FieldDescriptor field = fieldDescriptorAndFieldTableSchema.fieldDescriptor; + TableFieldSchema tableFieldSchema = fieldDescriptorAndFieldTableSchema.tableFieldSchema; + try { + if (!field.isRepeated()) { + fillField( + protoMsg, + field, + tableFieldSchema, + jsonObject, + jsonName, + currentScope, + ignoreUnknownFields); + } else { + fillRepeatedField( + protoMsg, + field, + tableFieldSchema, + jsonObject, + jsonName, + currentScope, + ignoreUnknownFields); + } + } catch (Exceptions.FieldParseError ex) { + throw ex; + } catch (Exception ex) { + // This function is recursively called, so this throw will be caught and throw directly + // out by the catch above. + throw new Exceptions.FieldParseError( + currentScope, + tableFieldSchema != null + ? tableFieldSchema.getType().name() + : field.getType().name(), + ex); + } + } + DynamicMessage msg; + try { + msg = protoMsg.build(); + } catch (UninitializedMessageException e) { + String errorMsg = e.getMessage(); + int idxOfColon = errorMsg.indexOf(":"); + String missingFieldName = errorMsg.substring(idxOfColon + 2); + throw new IllegalArgumentException( + String.format( + "JSONObject does not have the required field %s.%s.", + jsonScope, missingFieldName)); + } + messageList.add(msg); + } catch (IllegalArgumentException exception) { + if (exception instanceof Exceptions.DataHasUnknownFieldException) { + hasDataUnknownError = true; + } + if (exception instanceof Exceptions.FieldParseError) { + Exceptions.FieldParseError ex = (Exceptions.FieldParseError) exception; + rowIndexToErrorMessage.put( + i, + "Field " + + ex.getFieldName() + + " failed to convert to " + + ex.getBqType() + + ". Error: " + + ex.getCause().getMessage()); } else { - fillRepeatedField( - protoMsg, field, fieldSchema, json, jsonName, currentScope, ignoreUnknownFields); + rowIndexToErrorMessage.put(i, exception.getMessage()); } - } catch (Exceptions.FieldParseError ex) { - throw ex; - } catch (Exception ex) { - // This function is recursively called, so this throw will be caught and throw directly out - // by the catch - // above. - throw new Exceptions.FieldParseError( - currentScope, - fieldSchema != null ? fieldSchema.getType().name() : field.getType().name(), - ex); } } + if (!rowIndexToErrorMessage.isEmpty()) { + throw new RowIndexToErrorException(rowIndexToErrorMessage, hasDataUnknownError); + } + return messageList; + } - DynamicMessage msg; - try { - msg = protoMsg.build(); - } catch (UninitializedMessageException e) { - String errorMsg = e.getMessage(); - int idxOfColon = errorMsg.indexOf(":"); - String missingFieldName = errorMsg.substring(idxOfColon + 2); - throw new IllegalArgumentException( - String.format( - "JSONObject does not have the required field %s.%s.", jsonScope, missingFieldName)); + private static final class FieldDescriptorAndFieldTableSchema { + TableFieldSchema tableFieldSchema; + + // Field descriptor + FieldDescriptor fieldDescriptor; + } + + private FieldDescriptorAndFieldTableSchema computeDescriptorAndSchema( + String currentScope, + boolean ignoreUnknownFields, + String jsonName, + Descriptor protoSchema, + List tableFieldSchemaList) { + + // We want lowercase here to support case-insensitive data writes. + // The protobuf descriptor that is used is assumed to have all lowercased fields + String jsonFieldLocator = jsonName.toLowerCase(); + + // If jsonName is not compatible with proto naming convention, we should look by its + // placeholder name. + if (!BigQuerySchemaUtil.isProtoCompatible(jsonFieldLocator)) { + jsonFieldLocator = BigQuerySchemaUtil.generatePlaceholderFieldName(jsonFieldLocator); } - return msg; + + FieldDescriptor field = protoSchema.findFieldByName(jsonFieldLocator); + if (field == null && !ignoreUnknownFields) { + throw new Exceptions.DataHasUnknownFieldException(currentScope); + } else if (field == null) { + return null; + } + TableFieldSchema fieldSchema = null; + if (tableFieldSchemaList != null) { + // protoSchema is generated from tableSchema so their field ordering should match. + fieldSchema = tableFieldSchemaList.get(field.getIndex()); + if (!fieldSchema.getName().toLowerCase().equals(BigQuerySchemaUtil.getFieldName(field))) { + throw new ValidationException( + "Field at index " + + field.getIndex() + + " has mismatch names (" + + fieldSchema.getName() + + ") (" + + field.getName() + + ")"); + } + } + FieldDescriptorAndFieldTableSchema fieldDescriptorAndFieldTableSchema = + new FieldDescriptorAndFieldTableSchema(); + fieldDescriptorAndFieldTableSchema.fieldDescriptor = field; + fieldDescriptorAndFieldTableSchema.tableFieldSchema = fieldSchema; + return fieldDescriptorAndFieldTableSchema; } /** @@ -302,7 +452,6 @@ private void fillField( String currentScope, boolean ignoreUnknownFields) throws IllegalArgumentException { - java.lang.Object val = json.get(exactJsonKeyName); if (val == JSONObject.NULL) { return; diff --git a/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/SchemaAwareStreamWriter.java b/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/SchemaAwareStreamWriter.java index 10fceeee68..2c5a79af64 100644 --- a/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/SchemaAwareStreamWriter.java +++ b/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/SchemaAwareStreamWriter.java @@ -21,13 +21,15 @@ import com.google.api.gax.core.ExecutorProvider; import com.google.api.gax.rpc.TransportChannelProvider; import com.google.cloud.bigquery.storage.v1.Exceptions.AppendSerializationError; +import com.google.cloud.bigquery.storage.v1.Exceptions.RowIndexToErrorException; import com.google.common.base.Preconditions; import com.google.protobuf.Descriptors.Descriptor; import com.google.protobuf.Descriptors.DescriptorValidationException; -import com.google.protobuf.Message; +import com.google.protobuf.DynamicMessage; import com.google.rpc.Code; import java.io.IOException; import java.util.HashMap; +import java.util.List; import java.util.Map; import java.util.logging.Logger; import java.util.regex.Matcher; @@ -124,19 +126,23 @@ private void refreshWriter(TableSchema updatedSchema) this.streamWriter = streamWriterBuilder.setWriterSchema(this.protoSchema).build(); } - private Message buildMessage(T item) + private List buildMessage(Iterable items) throws InterruptedException, DescriptorValidationException, IOException { try { return this.toProtoConverter.convertToProtoMessage( - this.descriptor, this.tableSchema, item, ignoreUnknownFields); - } catch (Exceptions.DataHasUnknownFieldException ex) { + this.descriptor, this.tableSchema, items, ignoreUnknownFields); + } catch (RowIndexToErrorException ex) { + // We only retry for data unknown error. + if (!ex.hasDataUnknownError) { + throw ex; + } // Directly return error when stream writer refresh is disabled. if (this.skipRefreshStreamWriter) { throw ex; } LOG.warning( - "Saw unknown field " - + ex.getFieldName() + "Saw unknown field error during proto message conversin within error messages" + + ex.rowIndexToErrorMessage + ", try to refresh the writer with updated schema, stream: " + streamName); GetWriteStreamRequest writeStreamRequest = @@ -147,7 +153,7 @@ private Message buildMessage(T item) WriteStream writeStream = client.getWriteStream(writeStreamRequest); refreshWriter(writeStream.getTableSchema()); return this.toProtoConverter.convertToProtoMessage( - this.descriptor, this.tableSchema, item, ignoreUnknownFields); + this.descriptor, this.tableSchema, items, ignoreUnknownFields); } } /** @@ -169,7 +175,6 @@ public ApiFuture append(Iterable items, long offset) if (!this.skipRefreshStreamWriter && this.streamWriter.getUpdatedSchema() != null) { refreshWriter(this.streamWriter.getUpdatedSchema()); } - ProtoRows.Builder rowsBuilder = ProtoRows.newBuilder(); // Any error in convertToProtoMessage will throw an // IllegalArgumentException/IllegalStateException/NullPointerException. @@ -177,29 +182,15 @@ public ApiFuture append(Iterable items, long offset) // After the conversion is finished an AppendSerializtionError exception that contains all the // conversion errors will be thrown. Map rowIndexToErrorMessage = new HashMap<>(); - int i = -1; - for (T item : items) { - i += 1; - try { - Message protoMessage = buildMessage(item); - rowsBuilder.addSerializedRows(protoMessage.toByteString()); - } catch (IllegalArgumentException exception) { - if (exception instanceof Exceptions.FieldParseError) { - Exceptions.FieldParseError ex = (Exceptions.FieldParseError) exception; - rowIndexToErrorMessage.put( - i, - "Field " - + ex.getFieldName() - + " failed to convert to " - + ex.getBqType() - + ". Error: " - + ex.getCause().getMessage()); - } else { - rowIndexToErrorMessage.put(i, exception.getMessage()); - } - } catch (InterruptedException ex) { - throw new RuntimeException(ex); + try { + List protoMessages = buildMessage(items); + for (DynamicMessage dynamicMessage : protoMessages) { + rowsBuilder.addSerializedRows(dynamicMessage.toByteString()); } + } catch (RowIndexToErrorException exception) { + rowIndexToErrorMessage = exception.rowIndexToErrorMessage; + } catch (InterruptedException ex) { + throw new RuntimeException(ex); } if (!rowIndexToErrorMessage.isEmpty()) { diff --git a/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/ToProtoConverter.java b/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/ToProtoConverter.java index ca17ed11e7..76ef223e24 100644 --- a/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/ToProtoConverter.java +++ b/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/ToProtoConverter.java @@ -17,11 +17,12 @@ import com.google.protobuf.Descriptors; import com.google.protobuf.DynamicMessage; +import java.util.List; public interface ToProtoConverter { - DynamicMessage convertToProtoMessage( + List convertToProtoMessage( Descriptors.Descriptor protoSchema, TableSchema tableSchema, - T inputObject, + Iterable inputObject, boolean ignoreUnknownFields); } diff --git a/google-cloud-bigquerystorage/src/test/java/com/google/cloud/bigquery/storage/v1/JsonToProtoMessageTest.java b/google-cloud-bigquerystorage/src/test/java/com/google/cloud/bigquery/storage/v1/JsonToProtoMessageTest.java index 5c44d014d4..c2fab22c6c 100644 --- a/google-cloud-bigquerystorage/src/test/java/com/google/cloud/bigquery/storage/v1/JsonToProtoMessageTest.java +++ b/google-cloud-bigquerystorage/src/test/java/com/google/cloud/bigquery/storage/v1/JsonToProtoMessageTest.java @@ -20,6 +20,7 @@ import com.google.cloud.bigquery.storage.test.JsonTest.*; import com.google.cloud.bigquery.storage.test.SchemaTest.*; +import com.google.cloud.bigquery.storage.v1.Exceptions.RowIndexToErrorException; import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableMap; import com.google.protobuf.ByteString; @@ -29,6 +30,7 @@ import java.math.BigDecimal; import java.util.ArrayList; import java.util.Collections; +import java.util.List; import java.util.Map; import java.util.logging.Logger; import org.json.JSONArray; @@ -598,7 +600,7 @@ public void testInt32NotMatchInt64() throws Exception { JsonToProtoMessage.INSTANCE.convertToProtoMessage(TestInt32.getDescriptor(), json); Assert.fail("should fail"); } catch (IllegalArgumentException e) { - assertEquals("JSONObject does not have a int32 field at root.int.", e.getMessage()); + assertTrue(e.getMessage().contains("JSONObject does not have a int32 field at root.int.")); } } @@ -619,7 +621,8 @@ public void testDateTimeMismatch() throws Exception { TestDatetime.getDescriptor(), tableSchema, json); Assert.fail("should fail"); } catch (IllegalArgumentException e) { - assertEquals("JSONObject does not have a int64 field at root.datetime.", e.getMessage()); + assertTrue( + e.getMessage().contains("JSONObject does not have a int64 field at root.datetime.")); } } @@ -640,7 +643,8 @@ public void testTimeMismatch() throws Exception { TestTime.getDescriptor(), tableSchema, json); Assert.fail("should fail"); } catch (IllegalArgumentException e) { - assertEquals("JSONObject does not have a int64 field at root.time[0].", e.getMessage()); + assertTrue( + e.getMessage().contains("JSONObject does not have a int64 field at root.time[0].")); } } @@ -811,9 +815,12 @@ public void testAllTypes() throws Exception { assertEquals(protoMsg, AllTypesToCorrectProto.get(entry.getKey())[success]); success += 1; } catch (IllegalArgumentException e) { - assertEquals( - "JSONObject does not have a " + entry.getValue() + " field at root.test_field_type.", - e.getMessage()); + assertTrue( + e.getMessage() + .contains( + "JSONObject does not have a " + + entry.getValue() + + " field at root.test_field_type.")); } } if (entry.getKey() == DoubleType.getDescriptor()) { @@ -846,12 +853,12 @@ public void testAllRepeatedTypesWithLimits() throws Exception { LOG.info(e.getMessage()); assertTrue( e.getMessage() - .equals( + .contains( "JSONObject does not have a " + entry.getValue() + " field at root.test_repeated[0].") || e.getMessage() - .equals("Error: root.test_repeated[0] could not be converted to byte[].")); + .contains("Error: root.test_repeated[0] could not be converted to byte[].")); } } if (entry.getKey() == RepeatedDouble.getDescriptor()) { @@ -897,8 +904,9 @@ public void testRequired() throws Exception { JsonToProtoMessage.INSTANCE.convertToProtoMessage(TestRequired.getDescriptor(), json); Assert.fail("should fail"); } catch (IllegalArgumentException e) { - assertEquals( - "JSONObject does not have the required field root.required_double.", e.getMessage()); + assertTrue( + e.getMessage() + .contains("JSONObject does not have the required field root.required_double.")); } } @@ -929,9 +937,10 @@ public void testStructSimpleFail() throws Exception { JsonToProtoMessage.INSTANCE.convertToProtoMessage(MessageType.getDescriptor(), json); Assert.fail("should fail"); } catch (IllegalArgumentException e) { - assertEquals( - "JSONObject does not have a string field at root.test_field_type.test_field_type.", - e.getMessage()); + assertTrue( + e.getMessage() + .contains( + "JSONObject does not have a string field at root.test_field_type.test_field_type.")); } } @@ -1089,8 +1098,9 @@ public void testStructComplexFail() throws Exception { JsonToProtoMessage.INSTANCE.convertToProtoMessage(ComplexRoot.getDescriptor(), json); Assert.fail("should fail"); } catch (IllegalArgumentException e) { - assertEquals( - "JSONObject does not have a int64 field at root.complex_lvl1.test_int.", e.getMessage()); + assertTrue( + e.getMessage() + .contains("JSONObject does not have a int64 field at root.complex_lvl1.test_int.")); } } @@ -1103,8 +1113,9 @@ public void testRepeatedWithMixedTypes() throws Exception { JsonToProtoMessage.INSTANCE.convertToProtoMessage(RepeatedDouble.getDescriptor(), json); Assert.fail("should fail"); } catch (IllegalArgumentException e) { - assertEquals( - "JSONObject does not have a double field at root.test_repeated[2].", e.getMessage()); + assertTrue( + e.getMessage() + .contains("JSONObject does not have a double field at root.test_repeated[2].")); } } @@ -1165,9 +1176,10 @@ public void testNestedRepeatedComplexFail() throws Exception { JsonToProtoMessage.INSTANCE.convertToProtoMessage(NestedRepeated.getDescriptor(), json); Assert.fail("should fail"); } catch (IllegalArgumentException e) { - assertEquals( - "JSONObject does not have a string field at root.repeated_string.test_repeated[0].", - e.getMessage()); + assertTrue( + e.getMessage() + .contains( + "JSONObject does not have a string field at root.repeated_string.test_repeated[0].")); } } @@ -1198,10 +1210,10 @@ public void testAllowUnknownFieldsError() throws Exception { DynamicMessage protoMsg = JsonToProtoMessage.INSTANCE.convertToProtoMessage(RepeatedInt64.getDescriptor(), json); Assert.fail("Should fail"); - } catch (Exceptions.DataHasUnknownFieldException e) { - assertEquals( - "The source object has fields unknown to BigQuery: root.string.", e.getMessage()); - assertEquals("root.string", e.getFieldName()); + } catch (IllegalArgumentException e) { + assertTrue( + e.getMessage() + .contains("The source object has fields unknown to BigQuery: " + "root.string.")); } } @@ -1262,9 +1274,10 @@ public void testAllowUnknownFieldsSecondLevel() throws Exception { JsonToProtoMessage.INSTANCE.convertToProtoMessage(ComplexLvl1.getDescriptor(), json); Assert.fail("Should fail"); } catch (IllegalArgumentException e) { - assertEquals( - "The source object has fields unknown to BigQuery: root.complex_lvl2.no_match.", - e.getMessage()); + assertTrue( + e.getMessage() + .contains( + "The source object has fields unknown to BigQuery: root.complex_lvl2.no_match.")); } } @@ -1327,9 +1340,9 @@ public void testBadJsonFieldRepeated() throws Exception { JsonToProtoMessage.INSTANCE.convertToProtoMessage( RepeatedBytes.getDescriptor(), ts, json); Assert.fail("Should fail"); - } catch (Exceptions.FieldParseError ex) { - assertEquals(ex.getBqType(), "NUMERIC"); - assertEquals(ex.getFieldName(), "root.test_repeated"); + } catch (RowIndexToErrorException ex) { + assertTrue(ex.rowIndexToErrorMessage.size() == 1); + assertTrue(ex.getMessage().contains("root.test_repeated failed to convert to NUMERIC.")); } } @@ -1354,7 +1367,7 @@ public void testBadJsonFieldIntRepeated() throws Exception { RepeatedInt32.getDescriptor(), ts, json); Assert.fail("Should fail"); } catch (IllegalArgumentException ex) { - assertEquals(ex.getMessage(), "Text 'blah' could not be parsed at index 0"); + assertTrue(ex.getMessage().contains("Text 'blah' could not be parsed at index 0")); } } @@ -1421,6 +1434,51 @@ public void testDoubleAndFloatToNumericConversion() { assertEquals(expectedProto, protoMsg); } + @Test + public void testDoubleAndFloatToNumericConversionWithJsonArray() { + TableSchema ts = + TableSchema.newBuilder() + .addFields( + 0, + TableFieldSchema.newBuilder() + .setName("numeric") + .setType(TableFieldSchema.Type.NUMERIC) + .build()) + .build(); + List protoList = new ArrayList<>(); + int protoNum = 10; + for (int i = 0; i < protoNum; i++) { + protoList.add( + TestNumeric.newBuilder() + .setNumeric( + BigDecimalByteStringEncoder.encodeToNumericByteString( + new BigDecimal("24.678" + i))) + .build()); + } + + JSONArray doubleJsonArray = new JSONArray(); + JSONArray floatJsonArray = new JSONArray(); + for (int i = 0; i < protoNum; i++) { + JSONObject doubleJson = new JSONObject(); + doubleJson.put("numeric", new Double(24.678 + (i * 0.0001))); + doubleJsonArray.put(doubleJson); + + JSONObject floatJson = new JSONObject(); + floatJson.put("numeric", new Float(24.678 + (i * 0.0001))); + floatJsonArray.put(floatJson); + } + + List protoMsgList = + JsonToProtoMessage.INSTANCE.convertToProtoMessage( + TestNumeric.getDescriptor(), ts, doubleJsonArray, false); + assertEquals(protoList, protoMsgList); + + protoMsgList = + JsonToProtoMessage.INSTANCE.convertToProtoMessage( + TestNumeric.getDescriptor(), ts, floatJsonArray, false); + assertEquals(protoList, protoMsgList); + } + @Test public void testBigDecimalToBigNumericConversion() { TableSchema ts = From 5d6b32462249cefbe63ba03f05486352e5bb842a Mon Sep 17 00:00:00 2001 From: Owl Bot Date: Wed, 12 Jul 2023 23:39:16 +0000 Subject: [PATCH 9/9] =?UTF-8?q?=F0=9F=A6=89=20Updates=20from=20OwlBot=20po?= =?UTF-8?q?st-processor?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md --- README.md | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index 0b621554f8..657b93c17d 100644 --- a/README.md +++ b/README.md @@ -50,20 +50,20 @@ If you are using Maven without the BOM, add this to your dependencies: If you are using Gradle 5.x or later, add this to your dependencies: ```Groovy -implementation platform('com.google.cloud:libraries-bom:26.16.0') +implementation platform('com.google.cloud:libraries-bom:26.18.0') implementation 'com.google.cloud:google-cloud-bigquerystorage' ``` If you are using Gradle without BOM, add this to your dependencies: ```Groovy -implementation 'com.google.cloud:google-cloud-bigquerystorage:2.37.2' +implementation 'com.google.cloud:google-cloud-bigquerystorage:2.39.1' ``` If you are using SBT, add this to your dependencies: ```Scala -libraryDependencies += "com.google.cloud" % "google-cloud-bigquerystorage" % "2.37.2" +libraryDependencies += "com.google.cloud" % "google-cloud-bigquerystorage" % "2.39.1" ``` @@ -220,7 +220,7 @@ Java is a registered trademark of Oracle and/or its affiliates. [kokoro-badge-link-5]: http://storage.googleapis.com/cloud-devrel-public/java/badges/java-bigquerystorage/java11.html [stability-image]: https://img.shields.io/badge/stability-stable-green [maven-version-image]: https://img.shields.io/maven-central/v/com.google.cloud/google-cloud-bigquerystorage.svg -[maven-version-link]: https://central.sonatype.com/artifact/com.google.cloud/google-cloud-bigquerystorage/2.37.2 +[maven-version-link]: https://central.sonatype.com/artifact/com.google.cloud/google-cloud-bigquerystorage/2.39.1 [authentication]: https://github.com/googleapis/google-cloud-java#authentication [auth-scopes]: https://developers.google.com/identity/protocols/oauth2/scopes [predefined-iam-roles]: https://cloud.google.com/iam/docs/understanding-roles#predefined_roles