feat: allow java client to handle schema change during same stream name

README.md

@@ -49,20 +49,20 @@ If you are using Maven without 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.4.0')
+implementation platform('com.google.cloud:libraries-bom:26.5.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.28.3'
+implementation 'com.google.cloud:google-cloud-bigquerystorage:2.28.4'
 ```
 
 If you are using SBT, add this to your dependencies:
 
 ```Scala
-libraryDependencies += "com.google.cloud" % "google-cloud-bigquerystorage" % "2.28.3"
+libraryDependencies += "com.google.cloud" % "google-cloud-bigquerystorage" % "2.28.4"
 ```
 
 ## Authentication

google-cloud-bigquerystorage/clirr-ignored-differences.xml

@@ -115,7 +115,7 @@
   <difference>
     <differenceType>7009</differenceType>
     <className>com/google/cloud/bigquery/storage/v1/ConnectionWorkerPool</className>
-    <method>ConnectionWorkerPool(long, long, java.time.Duration, com.google.api.gax.batching.FlowController$LimitExceededBehavior, java.lang.String, com.google.cloud.bigquery.storage.v1.BigQueryWriteClient, boolean)</method>
+    <method>ConnectionWorkerPool(long, long, java.time.Duration, com.google.api.gax.batching.FlowController$LimitExceededBehavior, java.lang.String, com.google.cloud.bigquery.storage.v1.BigQueryWriteSettings)</method>
   </difference>
   <difference>
     <differenceType>7009</differenceType>

google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/ConnectionWorker.java

@@ -222,7 +222,6 @@ public ConnectionWorker(
           Status.fromCode(Code.INVALID_ARGUMENT)
               .withDescription("Writer schema must be provided when building this writer."));
     }
-    this.writerSchema = writerSchema;
     this.maxInflightRequests = maxInflightRequests;
     this.maxInflightBytes = maxInflightBytes;
     this.limitExceededBehavior = limitExceededBehavior;
@@ -432,7 +431,7 @@ private void appendLoop() {
 
     // Indicate whether we are at the first request after switching destination.
     // True means the schema and other metadata are needed.
-    boolean firstRequestForDestinationSwitch = true;
+    boolean firstRequestForTableOrSchemaSwitch = true;
     // Represent whether we have entered multiplexing.
     boolean isMultiplexing = false;
 
@@ -483,25 +482,35 @@ private void appendLoop() {
         resetConnection();
         // Set firstRequestInConnection to indicate the next request to be sent should include
         // metedata. Reset everytime after reconnection.
-        firstRequestForDestinationSwitch = true;
+        firstRequestForTableOrSchemaSwitch = true;
       }
       while (!localQueue.isEmpty()) {
         AppendRowsRequest originalRequest = localQueue.pollFirst().message;
         AppendRowsRequest.Builder originalRequestBuilder = originalRequest.toBuilder();
-
-        // Consider we enter multiplexing if we met a different non empty stream name.
-        if (!originalRequest.getWriteStream().isEmpty()
-            && !streamName.isEmpty()
-            && !originalRequest.getWriteStream().equals(streamName)) {
+        // Always respect the first writer schema seen by the loop.
+        if (writerSchema == null) {
+          writerSchema = originalRequest.getProtoRows().getWriterSchema();
+        }
+        // Consider we enter multiplexing if we met a different non empty stream name or we meet
+        // a new schema for the same stream name.
+        // For the schema comparision we don't use message differencer to speed up the comparing
+        // process. `equals(...)` can bring us false positive, e.g. two repeated field can be
+        // considered the same but is not considered equals(). However as long as it's never provide
+        // false negative we will always correctly pass writer schema to backend.
+        if ((!originalRequest.getWriteStream().isEmpty()
+                && !streamName.isEmpty()
+                && !originalRequest.getWriteStream().equals(streamName))
+            || (originalRequest.getProtoRows().hasWriterSchema()
+                && !originalRequest.getProtoRows().getWriterSchema().equals(writerSchema))) {
           streamName = originalRequest.getWriteStream();
+          writerSchema = originalRequest.getProtoRows().getWriterSchema();
           isMultiplexing = true;
-          firstRequestForDestinationSwitch = true;
+          firstRequestForTableOrSchemaSwitch = true;
         }
 
-        if (firstRequestForDestinationSwitch) {
+        if (firstRequestForTableOrSchemaSwitch) {
           // If we are at the first request for every table switch, including the first request in
           // the connection, we will attach both stream name and table schema to the request.
-          // We don't support change of schema change during multiplexing for the saeme stream name.
           destinationSet.add(streamName);
           if (this.traceId != null) {
             originalRequestBuilder.setTraceId(this.traceId);
@@ -511,17 +520,11 @@ private void appendLoop() {
           originalRequestBuilder.clearWriteStream();
         }
 
-        // We don't use message differencer to speed up the comparing process.
-        // `equals(...)` can bring us false positive, e.g. two repeated field can be considered the
-        // same but is not considered equals(). However as long as it's never provide false negative
-        // we will always correctly pass writer schema to backend.
-        if (firstRequestForDestinationSwitch
-            || !originalRequest.getProtoRows().getWriterSchema().equals(writerSchema)) {
-          writerSchema = originalRequest.getProtoRows().getWriterSchema();
-        } else {
+        // During non table/schema switch requests, clear writer schema.
+        if (!firstRequestForTableOrSchemaSwitch) {
           originalRequestBuilder.getProtoRowsBuilder().clearWriterSchema();
         }
-        firstRequestForDestinationSwitch = false;
+        firstRequestForTableOrSchemaSwitch = false;
 
         // Send should only throw an exception if there is a problem with the request. The catch
         // block will handle this case, and return the exception with the result.

google-cloud-bigquerystorage/src/test/java/com/google/cloud/bigquery/storage/v1/ConnectionWorkerTest.java

@@ -247,10 +247,10 @@ public void testAppendInSameStream_switchSchema() throws Exception {
         // We will get the request as the pattern of:
         // (writer_stream: t1, schema: schema1)
         // (writer_stream: _, schema: _)
-        // (writer_stream: _, schema: schema3)
-        // (writer_stream: _, schema: _)
-        // (writer_stream: _, schema: schema1)
-        // (writer_stream: _, schema: _)
+        // (writer_stream: t1, schema: schema3)
+        // (writer_stream: t1, schema: _)
+        // (writer_stream: t1, schema: schema1)
+        // (writer_stream: t1, schema: _)
         switch (i % 4) {
           case 0:
             if (i == 0) {
@@ -261,19 +261,23 @@ public void testAppendInSameStream_switchSchema() throws Exception {
                 .isEqualTo("foo");
             break;
           case 1:
-            assertThat(serverRequest.getWriteStream()).isEmpty();
+            if (i == 1) {
+              assertThat(serverRequest.getWriteStream()).isEmpty();
+            } else {
+              assertThat(serverRequest.getWriteStream()).isEqualTo(TEST_STREAM_1);
+            }
             // Schema is empty if not at the first request after table switch.
             assertThat(serverRequest.getProtoRows().hasWriterSchema()).isFalse();
             break;
           case 2:
-            assertThat(serverRequest.getWriteStream()).isEmpty();
+            assertThat(serverRequest.getWriteStream()).isEqualTo(TEST_STREAM_1);
             // Schema is populated after table switch.
             assertThat(
                     serverRequest.getProtoRows().getWriterSchema().getProtoDescriptor().getName())
                 .isEqualTo("bar");
             break;
           case 3:
-            assertThat(serverRequest.getWriteStream()).isEmpty();
+            assertThat(serverRequest.getWriteStream()).isEqualTo(TEST_STREAM_1);
             // Schema is empty if not at the first request after table switch.
             assertThat(serverRequest.getProtoRows().hasWriterSchema()).isFalse();
             break;