Improve performance of BigQueryIO connector when withPropagateSuccessfulStorageApiWrites(true) is used

CHANGES.md

@@ -63,6 +63,7 @@
 ## I/Os
 
 * Support for X source added (Java/Python) ([#X](https://github.com/apache/beam/issues/X)).
+* Improvements to the performance of BigqueryIO when using withPropagateSuccessfulStorageApiWrites(true) method (Java) ([#31840](https://github.com/apache/beam/pull/31840)).
 
 ## New Features / Improvements
 

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

@@ -32,6 +32,8 @@
 import com.google.cloud.bigquery.storage.v1.WriteStream.Type;
 import com.google.protobuf.ByteString;
 import com.google.protobuf.DescriptorProtos;
+import com.google.protobuf.Descriptors.Descriptor;
+import com.google.protobuf.Descriptors.DescriptorValidationException;
 import com.google.protobuf.DynamicMessage;
 import io.grpc.Status;
 import java.io.IOException;
@@ -834,21 +836,28 @@ long flush(
                   c, BigQuerySinkMetrics.RpcMethod.APPEND_ROWS, shortTableUrn);
 
               if (successfulRowsReceiver != null) {
-                for (int i = 0; i < c.protoRows.getSerializedRowsCount(); ++i) {
-                  ByteString rowBytes = c.protoRows.getSerializedRowsList().get(i);
-                  try {
-                    TableRow row =
-                        TableRowToStorageApiProto.tableRowFromMessage(
-                            DynamicMessage.parseFrom(
-                                TableRowToStorageApiProto.wrapDescriptorProto(
-                                    Preconditions.checkStateNotNull(appendClientInfo)
-                                        .getDescriptor()),
-                                rowBytes),
-                            true);
-                    org.joda.time.Instant timestamp = c.timestamps.get(i);
-                    successfulRowsReceiver.outputWithTimestamp(row, timestamp);
-                  } catch (Exception e) {
-                    LOG.warn("Failure parsing TableRow", e);
+                Descriptor descriptor = null;
+                try {
+                  descriptor =
+                      TableRowToStorageApiProto.wrapDescriptorProto(
+                          Preconditions.checkStateNotNull(appendClientInfo).getDescriptor());
+                } catch (DescriptorValidationException e) {
+                  LOG.warn(
+                      "Failure getting proto descriptor. Successful output will not be produced.",
+                      e);
+                }
+                if (descriptor != null) {
+                  for (int i = 0; i < c.protoRows.getSerializedRowsCount(); ++i) {
+                    ByteString rowBytes = c.protoRows.getSerializedRowsList().get(i);
+                    try {
+                      TableRow row =
+                          TableRowToStorageApiProto.tableRowFromMessage(
+                              DynamicMessage.parseFrom(descriptor, rowBytes), true);
+                      org.joda.time.Instant timestamp = c.timestamps.get(i);
+                      successfulRowsReceiver.outputWithTimestamp(row, timestamp);
+                    } catch (Exception e) {
+                      LOG.warn("Failure parsing TableRow", e);
+                    }
                   }
                 }
               }

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

@@ -1099,7 +1099,7 @@ public static TableRow tableRowFromMessage(Message message, boolean includeCdcCo
       FieldDescriptor fieldDescriptor = field.getKey();
       Object fieldValue = field.getValue();
       if (includeCdcColumns || !StorageApiCDC.COLUMNS.contains(fieldDescriptor.getName())) {
-        tableRow.putIfAbsent(
+        tableRow.put(
             fieldDescriptor.getName(),
             jsonValueFromMessageValue(fieldDescriptor, fieldValue, true));
       }