[Test] fix(bigquery): handle field named "f" in tableRowFromMessage

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

@@ -58,6 +58,8 @@ abstract class AppendClientInfo {
 
   abstract DescriptorProtos.DescriptorProto getDescriptor();
 
+  abstract boolean getUseEnhancedTableRowConversion();
+
   @AutoValue.Builder
   abstract static class Builder {
     abstract Builder setStreamAppendClient(@Nullable BigQueryServices.StreamAppendClient value);
@@ -74,6 +76,8 @@ abstract static class Builder {
 
     abstract Builder setStreamName(@Nullable String name);
 
+    abstract Builder setUseEnhancedTableRowConversion(boolean value);
+
     abstract AppendClientInfo build();
   };
 
@@ -84,13 +88,23 @@ static AppendClientInfo of(
       DescriptorProtos.DescriptorProto descriptor,
       Consumer<BigQueryServices.StreamAppendClient> closeAppendClient)
       throws Exception {
+    return of(tableSchema, descriptor, closeAppendClient, false);
+  }
+
+  static AppendClientInfo of(
+      TableSchema tableSchema,
+      DescriptorProtos.DescriptorProto descriptor,
+      Consumer<BigQueryServices.StreamAppendClient> closeAppendClient,
+      boolean useEnhancedTableRowConversion)
+      throws Exception {
     return new AutoValue_AppendClientInfo.Builder()
         .setTableSchema(tableSchema)
         .setCloseAppendClient(closeAppendClient)
         .setJsonTableSchema(TableRowToStorageApiProto.protoSchemaToTableSchema(tableSchema))
         .setSchemaInformation(
             TableRowToStorageApiProto.SchemaInformation.fromTableSchema(tableSchema))
         .setDescriptor(descriptor)
+        .setUseEnhancedTableRowConversion(useEnhancedTableRowConversion)
         .build();
   }
 
@@ -106,6 +120,20 @@ static AppendClientInfo of(
         closeAppendClient);
   }
 
+  static AppendClientInfo of(
+      TableSchema tableSchema,
+      Consumer<BigQueryServices.StreamAppendClient> closeAppendClient,
+      boolean includeCdcColumns,
+      boolean useEnhancedTableRowConversion)
+      throws Exception {
+    return of(
+        tableSchema,
+        TableRowToStorageApiProto.descriptorSchemaFromTableSchema(
+            tableSchema, true, includeCdcColumns),
+        closeAppendClient,
+        useEnhancedTableRowConversion);
+  }
+
   public AppendClientInfo withNoAppendClient() {
     return toBuilder().setStreamAppendClient(null).build();
   }
@@ -173,7 +201,9 @@ public TableRow toTableRow(ByteString protoBytes, Predicate<String> includeField
           DynamicMessage.parseFrom(
               TableRowToStorageApiProto.wrapDescriptorProto(getDescriptor()), protoBytes),
           true,
-          includeField);
+          includeField,
+          "",
+          getUseEnhancedTableRowConversion());
     } catch (Exception e) {
       throw new RuntimeException(e);
     }

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

@@ -2451,6 +2451,8 @@ public enum Method {
 
     abstract @Nullable String getWriteTempDataset();
 
+    abstract @Nullable Boolean getUseEnhancedTableRowConversion();
+
     abstract @Nullable SerializableFunction<T, RowMutationInformation>
         getRowMutationInformationFn();
 
@@ -2576,6 +2578,8 @@ abstract Builder<T> setBadRecordErrorHandler(
 
       abstract Builder<T> setBadRecordRouter(BadRecordRouter badRecordRouter);
 
+      abstract Builder<T> setUseEnhancedTableRowConversion(Boolean useEnhancedTableRowConversion);
+
       abstract Write<T> build();
     }
 
@@ -3326,6 +3330,20 @@ public Write<T> withWriteTempDataset(String writeTempDataset) {
       return toBuilder().setWriteTempDataset(writeTempDataset).build();
     }
 
+    /**
+     * Enables enhanced table row conversion for better handling of nested fields and complex data
+     * types. When enabled, uses improved conversion logic that provides better compatibility with
+     * BigQuery's data model. When disabled (default), uses the legacy conversion behavior for
+     * backward compatibility.
+     *
+     * @param useEnhancedTableRowConversion true to enable enhanced conversion, false for legacy
+     *     behavior
+     * @return the updated Write transform
+     */
+    public Write<T> withEnhancedTableRowConversion(boolean useEnhancedTableRowConversion) {
+      return toBuilder().setUseEnhancedTableRowConversion(useEnhancedTableRowConversion).build();
+    }
+
     public Write<T> withErrorHandler(ErrorHandler<BadRecord, ?> errorHandler) {
       return toBuilder()
           .setBadRecordErrorHandler(errorHandler)
@@ -3939,7 +3957,10 @@ private <DestinationT> WriteResult continueExpandTyped(
                   getRowMutationInformationFn() != null,
                   getCreateDisposition(),
                   getIgnoreUnknownValues(),
-                  getAutoSchemaUpdate());
+                  getAutoSchemaUpdate(),
+                  getUseEnhancedTableRowConversion() != null
+                      ? getUseEnhancedTableRowConversion()
+                      : false);
         }
 
         int numShards = getStorageApiNumStreams(bqOptions);

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

@@ -29,7 +29,6 @@
 import org.apache.beam.sdk.io.gcp.bigquery.BigQueryServices.DatasetService;
 import org.apache.beam.sdk.transforms.SerializableFunction;
 import org.apache.beam.sdk.util.Preconditions;
-import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Predicates;
 import org.checkerframework.checker.nullness.qual.NonNull;
 
 /** Storage API DynamicDestinations used when the input is a compiled protocol buffer. */
@@ -108,7 +107,10 @@ public TableRow toFailsafeTableRow(T element) {
                   TableRowToStorageApiProto.wrapDescriptorProto(descriptorProto),
                   element.toByteArray()),
               true,
-              Predicates.alwaysTrue());
+              org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Predicates
+                  .alwaysTrue(),
+              "",
+              false);
         } catch (Exception e) {
           throw new RuntimeException(e);
         }

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

@@ -42,6 +42,7 @@ public class StorageApiDynamicDestinationsTableRow<T, DestinationT extends @NonN
   private final CreateDisposition createDisposition;
   private final boolean ignoreUnknownValues;
   private final boolean autoSchemaUpdates;
+  private final boolean useEnhancedTableRowConversion;
   private static final TableSchemaCache SCHEMA_CACHE =
       new TableSchemaCache(Duration.standardSeconds(1));
 
@@ -56,20 +57,26 @@ public class StorageApiDynamicDestinationsTableRow<T, DestinationT extends @NonN
       boolean usesCdc,
       CreateDisposition createDisposition,
       boolean ignoreUnknownValues,
-      boolean autoSchemaUpdates) {
+      boolean autoSchemaUpdates,
+      boolean useEnhancedTableRowConversion) {
     super(inner);
     this.formatFunction = formatFunction;
     this.formatRecordOnFailureFunction = formatRecordOnFailureFunction;
     this.usesCdc = usesCdc;
     this.createDisposition = createDisposition;
     this.ignoreUnknownValues = ignoreUnknownValues;
     this.autoSchemaUpdates = autoSchemaUpdates;
+    this.useEnhancedTableRowConversion = useEnhancedTableRowConversion;
   }
 
   static void clearSchemaCache() throws ExecutionException, InterruptedException {
     SCHEMA_CACHE.clear();
   }
 
+  public boolean getUseEnhancedTableRowConversion() {
+    return useEnhancedTableRowConversion;
+  }
+
   @Override
   public MessageConverter<T> getMessageConverter(
       DestinationT destination, DatasetService datasetService) throws Exception {
@@ -189,7 +196,8 @@ public StorageApiWritePayload toMessage(
               allowMissingFields,
               unknownFields,
               changeType,
-              changeSequenceNum);
+              changeSequenceNum,
+              useEnhancedTableRowConversion);
       return StorageApiWritePayload.of(
           msg.toByteArray(),
           unknownFields,

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

@@ -663,7 +663,9 @@ long flush(
                               getAppendClientInfo(true, null).getDescriptor()),
                           rowBytes),
                       true,
-                      successfulRowsPredicate);
+                      successfulRowsPredicate,
+                      "",
+                      getEnhancedConversionFlag());
             }
             org.joda.time.Instant timestamp = insertTimestamps.get(i);
             failedRowsReceiver.outputWithTimestamp(
@@ -748,7 +750,9 @@ long flush(
                                           .getDescriptor()),
                                   protoBytes),
                               true,
-                              Predicates.alwaysTrue());
+                              Predicates.alwaysTrue(),
+                              "",
+                              getEnhancedConversionFlag());
                     }
                     element =
                         new BigQueryStorageApiInsertError(
@@ -900,7 +904,9 @@ long flush(
                           TableRowToStorageApiProto.tableRowFromMessage(
                               DynamicMessage.parseFrom(descriptor, rowBytes),
                               true,
-                              successfulRowsPredicate);
+                              successfulRowsPredicate,
+                              "",
+                              getEnhancedConversionFlag());
                       org.joda.time.Instant timestamp = c.timestamps.get(i);
                       successfulRowsReceiver.outputWithTimestamp(row, timestamp);
                     } catch (Exception e) {
@@ -967,6 +973,14 @@ void postFlush() {
     private int streamAppendClientCount;
     private final @Nullable Map<String, String> bigLakeConfiguration;
 
+    private boolean getEnhancedConversionFlag() {
+      if (dynamicDestinations instanceof StorageApiDynamicDestinationsTableRow) {
+        return ((StorageApiDynamicDestinationsTableRow<?, ?>) dynamicDestinations)
+            .getUseEnhancedTableRowConversion();
+      }
+      return false;
+    }
+
     WriteRecordsDoFn(
         String operationName,
         StorageApiDynamicDestinations<ElementT, DestinationT> dynamicDestinations,