Add Dynamic Schema support to StorageWriteToBigQuery

.github/trigger_files/beam_PostCommit_Python_Xlang_Gcp_Dataflow.json

@@ -1,4 +1,4 @@
 {
   "comment": "Modify this file in a trivial way to cause this test suite to run",
-  "modification": 16
+  "modification": 21
 }

.github/trigger_files/beam_PostCommit_Python_Xlang_IO_Direct.json

@@ -1,4 +1,4 @@
 {
   "comment": "Modify this file in a trivial way to cause this test suite to run",
-  "modification": 2
+  "modification": 7
 }

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

@@ -168,9 +168,11 @@ public static DynamicMessage messageFromBeamRow(
     for (int i = 0; i < row.getFieldCount(); ++i) {
       Field beamField = beamSchema.getField(i);
       FieldDescriptor fieldDescriptor =
-          Preconditions.checkNotNull(
-              descriptor.findFieldByName(beamField.getName().toLowerCase()),
-              beamField.getName().toLowerCase());
+          descriptor.findFieldByName(beamField.getName().toLowerCase());
+      if (fieldDescriptor == null) {
+        // Field in the union row is not present in the destination table's descriptor; skip it.
+        continue;
+      }
       @Nullable Object value = messageValueFromRowValue(fieldDescriptor, beamField, i, row);
       if (value != null) {
         builder.setField(fieldDescriptor, value);
@@ -330,7 +332,8 @@ private static Object toProtoValue(
       FieldDescriptor fieldDescriptor, FieldType beamFieldType, Object value) {
     switch (beamFieldType.getTypeName()) {
       case ROW:
-        return messageFromBeamRow(fieldDescriptor.getMessageType(), (Row) value, null, -1);
+        return messageFromBeamRow(
+            fieldDescriptor.getMessageType(), (Row) value, null, (String) null);
       case ARRAY:
       case ITERABLE:
         Iterable<Object> iterable = (Iterable<Object>) value;
@@ -419,14 +422,21 @@ static Object mapEntryToProtoValue(
     DynamicMessage.Builder builder = DynamicMessage.newBuilder(descriptor);
     FieldDescriptor keyFieldDescriptor =
         Preconditions.checkNotNull(descriptor.findFieldByName("key"));
-    @Nullable Object key = toProtoValue(keyFieldDescriptor, keyFieldType, entryValue.getKey());
+    @Nullable
+    Object key =
+        entryValue.getKey() != null
+            ? toProtoValue(keyFieldDescriptor, keyFieldType, entryValue.getKey())
+            : null;
     if (key != null) {
       builder.setField(keyFieldDescriptor, key);
     }
     FieldDescriptor valueFieldDescriptor =
         Preconditions.checkNotNull(descriptor.findFieldByName("value"));
     @Nullable
-    Object value = toProtoValue(valueFieldDescriptor, valueFieldType, entryValue.getValue());
+    Object value =
+        entryValue.getValue() != null
+            ? toProtoValue(valueFieldDescriptor, valueFieldType, entryValue.getValue())
+            : null;
     if (value != null) {
       builder.setField(valueFieldDescriptor, value);
     }

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

@@ -4011,11 +4011,13 @@ private <DestinationT> WriteResult expandTyped(
         // TODO: If the user provided a schema, we should use that. There are things that can be
         // specified in a
         // BQ schema that don't have exact matches in a Beam schema (e.g. GEOGRAPHY types).
-        TableSchema tableSchema = BigQueryUtils.toTableSchema(input.getSchema());
-        dynamicDestinations =
-            new ConstantSchemaDestinations<>(
-                dynamicDestinations,
-                StaticValueProvider.of(BigQueryHelpers.toJsonString(tableSchema)));
+        if (!hasSchema) {
+          TableSchema tableSchema = BigQueryUtils.toTableSchema(input.getSchema());
+          dynamicDestinations =
+              new ConstantSchemaDestinations<>(
+                  dynamicDestinations,
+                  StaticValueProvider.of(BigQueryHelpers.toJsonString(tableSchema)));
+        }
       } else if (writeProtoClass != null) {
         if (!hasSchema) {
           try {
@@ -4487,6 +4489,7 @@ static void clearStaticCaches() throws ExecutionException, InterruptedException
     CreateTables.clearCreatedTables();
     TwoLevelMessageConverterCache.clear();
     StorageApiDynamicDestinationsTableRow.clearSchemaCache();
+    StorageApiDynamicDestinationsBeamRow.clearSchemaCache();
     StorageApiWriteUnshardedRecords.clearCache();
     StorageApiWritesShardedRecords.clearCache();
   }

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

@@ -17,13 +17,16 @@
  */
 package org.apache.beam.sdk.io.gcp.bigquery;
 
+import com.google.api.services.bigquery.model.TableReference;
 import com.google.api.services.bigquery.model.TableRow;
 import com.google.cloud.bigquery.storage.v1.TableSchema;
 import com.google.protobuf.DescriptorProtos;
 import com.google.protobuf.Descriptors.Descriptor;
 import com.google.protobuf.Message;
 import java.io.IOException;
+import java.util.concurrent.ExecutionException;
 import org.apache.beam.sdk.io.gcp.bigquery.BigQueryServices.DatasetService;
+import org.apache.beam.sdk.io.gcp.bigquery.BigQueryServices.WriteStreamService;
 import org.apache.beam.sdk.options.PipelineOptions;
 import org.apache.beam.sdk.schemas.Schema;
 import org.apache.beam.sdk.transforms.SerializableBiFunction;
@@ -32,10 +35,22 @@
 import org.apache.beam.sdk.values.Row;
 import org.checkerframework.checker.nullness.qual.NonNull;
 import org.checkerframework.checker.nullness.qual.Nullable;
+import org.joda.time.Duration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /** Storage API DynamicDestinations used when the input is a Beam Row. */
 class StorageApiDynamicDestinationsBeamRow<T, DestinationT extends @NonNull Object>
     extends StorageApiDynamicDestinations<T, DestinationT> {
+  private static final Logger LOG =
+      LoggerFactory.getLogger(StorageApiDynamicDestinationsBeamRow.class);
+  private static final TableSchemaCache SCHEMA_CACHE =
+      new TableSchemaCache(Duration.standardSeconds(1));
+
+  static {
+    SCHEMA_CACHE.start();
+  }
+
   private final TableSchema tableSchema;
   private final SerializableFunction<T, Row> toRow;
   private final @Nullable
@@ -59,21 +74,58 @@ class StorageApiDynamicDestinationsBeamRow<T, DestinationT extends @NonNull Obje
     this.usesCdc = usesCdc;
   }
 
+  static void clearSchemaCache() throws ExecutionException, InterruptedException {
+    SCHEMA_CACHE.clear();
+  }
+
   @Override
   public MessageConverter<T> getMessageConverter(
       DestinationT destination,
       PipelineOptions pipelineOptions,
-      DatasetService datasetService,
-      BigQueryServices.WriteStreamService writeStreamService)
+      @Nullable DatasetService datasetService,
+      @Nullable WriteStreamService writeStreamService)
       throws Exception {
-    return new BeamRowConverter();
+    TableSchema destinationProtoSchema = null;
+    com.google.api.services.bigquery.model.TableSchema destSchema = getSchema(destination);
+    com.google.api.services.bigquery.model.TableSchema schemaToUse = destSchema;
+
+    TableDestination tableDestination = getTable(destination);
+    TableReference tableReference =
+        tableDestination != null ? tableDestination.getTableReference() : null;
+
+    if (tableReference != null && datasetService != null) {
+      try {
+        com.google.api.services.bigquery.model.TableSchema bqSchema =
+            SCHEMA_CACHE.getSchema(tableReference, datasetService);
+        if (bqSchema != null) {
+          if (schemaToUse == null
+              || TableRowToStorageApiProto.hasExtraFields(schemaToUse, bqSchema)) {
+            schemaToUse = bqSchema;
+          }
+        }
+      } catch (Exception e) {
+        LOG.warn("Could not fetch schema from BigQuery for table {}", tableReference, e);
+      }
+    }
+
+    if (schemaToUse != null) {
+      destinationProtoSchema = TableRowToStorageApiProto.schemaToProtoTableSchema(schemaToUse);
+    }
+
+    if (destinationProtoSchema == null) {
+      destinationProtoSchema = this.tableSchema;
+    }
+
+    return new BeamRowConverter(destinationProtoSchema);
   }
 
   class BeamRowConverter implements MessageConverter<T> {
+    final TableSchema tableSchema;
     final Descriptor descriptor;
     final @Nullable Descriptor cdcDescriptor;
 
-    BeamRowConverter() throws Exception {
+    BeamRowConverter(TableSchema tableSchema) throws Exception {
+      this.tableSchema = tableSchema;
       this.descriptor =
           TableRowToStorageApiProto.getDescriptorFromTableSchema(tableSchema, true, false);
       if (usesCdc) {
@@ -131,5 +183,4 @@ public TableRow toFailsafeTableRow(T element) {
       }
     }
   }
-  ;
 }

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

@@ -32,7 +32,6 @@
 import org.apache.beam.sdk.options.PipelineOptions;
 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.MoreObjects;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Supplier;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Suppliers;
 import org.checkerframework.checker.nullness.qual.NonNull;
@@ -94,8 +93,29 @@ public MessageConverter<T> getMessageConverter(
           }
         };
 
+    TableSchema destSchema = getSchema(destination);
+    TableSchema schemaToUse = destSchema;
+
+    TableDestination tableDestination = getTable(destination);
+    TableReference tableReference =
+        tableDestination != null ? tableDestination.getTableReference() : null;
+
+    if (tableReference != null && datasetService != null) {
+      try {
+        TableSchema bqSchema = SCHEMA_CACHE.getSchema(tableReference, datasetService);
+        if (bqSchema != null) {
+          if (schemaToUse == null
+              || TableRowToStorageApiProto.hasExtraFields(schemaToUse, bqSchema)) {
+            schemaToUse = bqSchema;
+          }
+        }
+      } catch (Exception e) {
+        // Schema cache lookup is best-effort fallback for dynamic destinations.
+      }
+    }
+
     return schemaUpdateOptions.isEmpty()
-        ? getConverter.apply(getSchema(destination))
+        ? getConverter.apply(schemaToUse)
         : new SchemaUpgradingTableRowConverter(
             getConverter, options, datasetService, writeStreamService);
   }
@@ -200,9 +220,7 @@ class TableRowConverter implements MessageConverter<T> {
       } else {
         // Make sure we register this schema with the cache, unless there's already a more
         // up-to-date schema.
-        localTableSchema =
-            MoreObjects.firstNonNull(
-                SCHEMA_CACHE.putSchemaIfAbsent(tableReference, localTableSchema), localTableSchema);
+        SCHEMA_CACHE.putSchemaIfAbsent(tableReference, localTableSchema);
       }
       this.tableSchema = localTableSchema;
       this.protoTableSchema = TableRowToStorageApiProto.schemaToProtoTableSchema(tableSchema);

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

@@ -887,7 +887,7 @@ public static Descriptor wrapDescriptorProto(DescriptorProto descriptorProto)
         if (unknownFields != null) {
           unknownFields.set(key, entry.getValue());
         }
-        if (ignoreUnknownValues) {
+        if (ignoreUnknownValues || entry.getValue() == null) {
           continue;
         } else {
           String prefix = schemaInformation.getFullName();
@@ -2087,4 +2087,45 @@ private static boolean isProtoFieldTypeInteger(FieldDescriptor.Type type) {
         return false;
     }
   }
+
+  static boolean hasExtraFields(
+      com.google.api.services.bigquery.model.@Nullable TableSchema clientSchema,
+      com.google.api.services.bigquery.model.@Nullable TableSchema bqSchema) {
+    if (clientSchema == null || clientSchema.getFields() == null) {
+      return false;
+    }
+    if (bqSchema == null || bqSchema.getFields() == null) {
+      return false;
+    }
+    java.util.Map<String, com.google.api.services.bigquery.model.TableFieldSchema> bqFieldMap =
+        bqSchema.getFields().stream()
+            .collect(
+                java.util.stream.Collectors.toMap(
+                    com.google.api.services.bigquery.model.TableFieldSchema::getName,
+                    f -> f,
+                    (a, b) -> a));
+    for (com.google.api.services.bigquery.model.TableFieldSchema clientField :
+        clientSchema.getFields()) {
+      com.google.api.services.bigquery.model.TableFieldSchema bqField =
+          bqFieldMap.get(clientField.getName());
+      if (bqField == null) {
+        return true;
+      }
+      if ("RECORD".equalsIgnoreCase(clientField.getType())
+          || "STRUCT".equalsIgnoreCase(clientField.getType())) {
+        if (clientField.getFields() != null && bqField.getFields() != null) {
+          com.google.api.services.bigquery.model.TableSchema nestedClient =
+              new com.google.api.services.bigquery.model.TableSchema()
+                  .setFields(clientField.getFields());
+          com.google.api.services.bigquery.model.TableSchema nestedBq =
+              new com.google.api.services.bigquery.model.TableSchema()
+                  .setFields(bqField.getFields());
+          if (hasExtraFields(nestedClient, nestedBq)) {
+            return true;
+          }
+        }
+      }
+    }
+    return false;
+  }
 }

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

@@ -53,6 +53,7 @@
 import org.apache.beam.sdk.values.TypeDescriptors;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Strings;
+import org.checkerframework.checker.nullness.qual.Nullable;
 import org.joda.time.Duration;
 
 /**
@@ -241,20 +242,35 @@ public PCollectionRowTuple expand(PCollectionRowTuple input) {
             Schema.of(
                 Field.of("failed_row", FieldType.row(inputSchema)),
                 Field.of("error_message", FieldType.STRING));
+        boolean isDynamicDestinations = configuration.getTable().equals(DYNAMIC_DESTINATIONS);
+        @Nullable Schema recordSchema =
+            isDynamicDestinations ? inputSchema.getField(RECORD).getType().getRowSchema() : null;
         PCollection<Row> failedRowsWithErrors =
             result
                 .getFailedStorageApiInserts()
                 .apply(
                     "Construct failed rows and errors",
                     MapElements.into(TypeDescriptors.rows())
                         .via(
-                            (storageError) ->
-                                Row.withSchema(errorSchema)
-                                    .withFieldValue("error_message", storageError.getErrorMessage())
-                                    .withFieldValue(
-                                        "failed_row",
-                                        BigQueryUtils.toBeamRow(inputSchema, storageError.getRow()))
-                                    .build()))
+                            (storageError) -> {
+                              Row failedRow;
+                              if (isDynamicDestinations && recordSchema != null) {
+                                Row recordRow =
+                                    BigQueryUtils.toBeamRow(recordSchema, storageError.getRow());
+                                failedRow =
+                                    Row.withSchema(inputSchema)
+                                        .withFieldValue(DESTINATION, "")
+                                        .withFieldValue(RECORD, recordRow)
+                                        .build();
+                              } else {
+                                failedRow =
+                                    BigQueryUtils.toBeamRow(inputSchema, storageError.getRow());
+                              }
+                              return Row.withSchema(errorSchema)
+                                  .withFieldValue("error_message", storageError.getErrorMessage())
+                                  .withFieldValue("failed_row", failedRow)
+                                  .build();
+                            }))
                 .setRowSchema(errorSchema);
         return PCollectionRowTuple.of("post_write", postWrite)
             .and(configuration.getErrorHandling().getOutput(), failedRowsWithErrors);