Repository: apache/beam
Issue: [Bug]: Storage Write API fails on Batch Pipelines on 2.49

### What happened?

When writing to a BigQuery sink in a Batch pipeline using Storage Write API the pipeline fails due to the following error,

```
java.lang.RuntimeException: org.apache.beam.sdk.util.UserCodeException: java.lang.RuntimeException: Schema field not found: eventid

at org.apache.beam.runners.dataflow.worker.GroupAlsoByWindowsParDoFn$1.output ( org/apache.beam.runners.dataflow.worker/GroupAlsoByWindowsParDoFn.java:187 )
at org.apache.beam.runners.dataflow.worker.GroupAlsoByWindowFnRunner$1.outputWindowedValue ( org/apache.beam.runners.dataflow.worker/GroupAlsoByWindowFnRunner.java:108 )
at org.apache.beam.runners.dataflow.worker.util.BatchGroupAlsoByWindowReshuffleFn.processElement ( org/apache.beam.runners.dataflow.worker.util/BatchGroupAlsoByWindowReshuffleFn.java:56 )
at org.apache.beam.runners.dataflow.worker.util.BatchGroupAlsoByWindowReshuffleFn.processElement ( org/apache.beam.runners.dataflow.worker.util/BatchGroupAlsoByWindowReshuffleFn.java:39 )
at org.apache.beam.runners.dataflow.worker.GroupAlsoByWindowFnRunner.invokeProcessElement ( org/apache.beam.runners.dataflow.worker/GroupAlsoByWindowFnRunner.java:121 )
at org.apache.beam.runners.dataflow.worker.GroupAlsoByWindowFnRunner.processElement ( org/apache.beam.runners.dataflow.worker/GroupAlsoByWindowFnRunner.java:73 )
at org.apache.beam.runners.dataflow.worker.GroupAlsoByWindowsParDoFn.processElement ( org/apache.beam.runners.dataflow.worker/GroupAlsoByWindowsParDoFn.java:117 )
at org.apache.beam.runners.dataflow.worker.util.common.worker.ParDoOperation.process ( org/apache.beam.runners.dataflow.worker.util.common.worker/ParDoOperation.java:44 )
at org.apache.beam.runners.dataflow.worker.util.common.worker.OutputReceiver.process ( org/apache.beam.runners.dataflow.worker.util.common.worker/OutputReceiver.java:54 )
at org.apache.beam.runners.dataflow.worker.util.common.worker.ReadOperation.runReadLoop ( org/apache.beam.runners.dataflow.worker.util.common.worker/ReadOperation.java:218 )
at org.apache.beam.runners.dataflow.worker.util.common.worker.ReadOperation.start ( org/apache.beam.runners.dataflow.worker.util.common.worker/ReadOperation.java:169 )
at org.apache.beam.runners.dataflow.worker.util.common.worker.MapTaskExecutor.execute ( org/apache.beam.runners.dataflow.worker.util.common.worker/MapTaskExecutor.java:83 )
at org.apache.beam.runners.dataflow.worker.BatchDataflowWorker.executeWork ( org/apache.beam.runners.dataflow.worker/BatchDataflowWorker.java:319 )
at org.apache.beam.runners.dataflow.worker.BatchDataflowWorker.doWork ( org/apache.beam.runners.dataflow.worker/BatchDataflowWorker.java:291 )
at org.apache.beam.runners.dataflow.worker.BatchDataflowWorker.getAndPerformWork ( org/apache.beam.runners.dataflow.worker/BatchDataflowWorker.java:221 )
at org.apache.beam.runners.dataflow.worker.DataflowBatchWorkerHarness$WorkerThread.doWork ( org/apache.beam.runners.dataflow.worker/DataflowBatchWorkerHarness.java:147 )
at org.apache.beam.runners.dataflow.worker.DataflowBatchWorkerHarness$WorkerThread.call ( org/apache.beam.runners.dataflow.worker/DataflowBatchWorkerHarness.java:127 )
at org.apache.beam.runners.dataflow.worker.DataflowBatchWorkerHarness$WorkerThread.call ( org/apache.beam.runners.dataflow.worker/DataflowBatchWorkerHarness.java:114 )
at java.util.concurrent.FutureTask.run ( java/util.concurrent/FutureTask.java:264 )
at org.apache.beam.sdk.util.UnboundedScheduledExecutorService$ScheduledFutureTask.run ( org/apache.beam.sdk.util/UnboundedScheduledExecutorService.java:163 )
at java.util.concurrent.ThreadPoolExecutor.runWorker ( java/util.concurrent/ThreadPoolExecutor.java:1128 )
at java.util.concurrent.ThreadPoolExecutor$Worker.run ( java/util.concurrent/ThreadPoolExecutor.java:628 )
at java.lang.Thread.run ( java/lang/Thread.java:834 )
Caused by: org.apache.beam.sdk.util.UserCodeException

at org.apache.beam.sdk.util.UserCodeException.wrap ( UserCodeException.java:39 )
at org.apache.beam.sdk.io.gcp.bigquery.StorageApiConvertMessages$ConvertMessagesDoFn$DoFnInvoker.invokeProcessElement
at org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner.invokeProcessElement ( SimpleDoFnRunner.java:211 )
at org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner.processElement ( SimpleDoFnRunner.java:188 )
at org.apache.beam.runners.dataflow.worker.SimpleParDoFn.processElement ( SimpleParDoFn.java:340 )
at org.apache.beam.runners.dataflow.worker.util.common.worker.ParDoOperation.process ( ParDoOperation.java:44 )
at org.apache.beam.runners.dataflow.worker.util.common.worker.OutputReceiver.process ( OutputReceiver.java:54 )
at org.apache.beam.runners.dataflow.worker.AssignWindowsParDoFnFactory$AssignWindowsParDoFn.processElement ( AssignWindowsParDoFnFactory.java:115 )
at org.apache.beam.runners.dataflow.worker.util.common.worker.ParDoOperation.process ( ParDoOperation.java:44 )
at org.apache.beam.runners.dataflow.worker.util.common.worker.OutputReceiver.process ( OutputReceiver.java:54 )
at org.apache.beam.runners.dataflow.worker.SimpleParDoFn$1.output ( SimpleParDoFn.java:285 )
at org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner.outputWindowedValue ( SimpleDoFnRunner.java:275 )
at org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner.access$900 ( SimpleDoFnRunner.java:85 )
at org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner$DoFnProcessContext.output ( SimpleDoFnRunner.java:423 )
at org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner$DoFnProcessContext.output ( SimpleDoFnRunner.java:411 )
at org.apache.beam.sdk.io.gcp.bigquery.PrepareWrite$1.processElement ( PrepareWrite.java:84 )
at org.apache.beam.sdk.io.gcp.bigquery.PrepareWrite$1$DoFnInvoker.invokeProcessElement
at org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner.invokeProcessElement ( SimpleDoFnRunner.java:211 )
at org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner.processElement ( SimpleDoFnRunner.java:185 )
at org.apache.beam.runners.dataflow.worker.SimpleParDoFn.processElement ( SimpleParDoFn.java:340 )
at org.apache.beam.runners.dataflow.worker.util.common.worker.ParDoOperation.process ( ParDoOperation.java:44 )
at org.apache.beam.runners.dataflow.worker.util.common.worker.OutputReceiver.process ( OutputReceiver.java:54 )
at org.apache.beam.runners.dataflow.worker.SimpleParDoFn$1.output ( SimpleParDoFn.java:285 )
at org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner.outputWindowedValue ( SimpleDoFnRunner.java:275 )
at org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner.access$900 ( SimpleDoFnRunner.java:85 )
at org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner$DoFnProcessContext.output ( SimpleDoFnRunner.java:423 )
at org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner$DoFnProcessContext.output ( SimpleDoFnRunner.java:411 )
at org.apache.beam.sdk.io.jdbc.JdbcIO$ReadFn.processElement ( JdbcIO.java:1528 )
at org.apache.beam.sdk.io.jdbc.JdbcIO$ReadFn$DoFnInvoker.invokeProcessElement
at org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner.invokeProcessElement ( SimpleDoFnRunner.java:211 )
at org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner.processElement ( SimpleDoFnRunner.java:188 )
at org.apache.beam.runners.dataflow.worker.SimpleParDoFn.processElement ( SimpleParDoFn.java:340 )
at org.apache.beam.runners.dataflow.worker.util.common.worker.ParDoOperation.process ( ParDoOperation.java:44 )
at org.apache.beam.runners.dataflow.worker.util.common.worker.OutputReceiver.process ( OutputReceiver.java:54 )
at org.apache.beam.runners.dataflow.worker.SimpleParDoFn$1.output ( SimpleParDoFn.java:285 )
at org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner.outputWindowedValue ( SimpleDoFnRunner.java:275 )
at org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner.access$900 ( SimpleDoFnRunner.java:85 )
at org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner$DoFnProcessContext.output ( SimpleDoFnRunner.java:423 )
at org.apache.beam.sdk.transforms.DoFnOutputReceivers$WindowedContextOutputReceiver.output ( DoFnOutputReceivers.java:76 )
at org.apache.beam.sdk.transforms.MapElements$2.processElement ( MapElements.java:151 )
at org.apache.beam.sdk.transforms.MapElements$2$DoFnInvoker.invokeProcessElement
at org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner.invokeProcessElement ( SimpleDoFnRunner.java:211 )
at org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner.processElement ( SimpleDoFnRunner.java:188 )
at org.apache.beam.runners.dataflow.worker.SimpleParDoFn.processElement ( SimpleParDoFn.java:340 )
at org.apache.beam.runners.dataflow.worker.util.common.worker.ParDoOperation.process ( ParDoOperation.java:44 )
at org.apache.beam.runners.dataflow.worker.util.common.worker.OutputReceiver.process ( OutputReceiver.java:54 )
at org.apache.beam.runners.dataflow.worker.SimpleParDoFn$1.output ( SimpleParDoFn.java:285 )
at org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner.outputWindowedValue ( SimpleDoFnRunner.java:275 )
at org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner.access$900 ( SimpleDoFnRunner.java:85 )
at org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner$DoFnProcessContext.output ( SimpleDoFnRunner.java:423 )
at org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner$DoFnProcessContext.output ( SimpleDoFnRunner.java:411 )
at org.apache.beam.runners.dataflow.ReshuffleOverrideFactory$ReshuffleWithOnlyTrigger$1.processElement ( ReshuffleOverrideFactory.java:86 )
at org.apache.beam.runners.dataflow.ReshuffleOverrideFactory$ReshuffleWithOnlyTrigger$1$DoFnInvoker.invokeProcessElement
at org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner.invokeProcessElement ( SimpleDoFnRunner.java:211 )
at org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner.processElement ( SimpleDoFnRunner.java:188 )
at org.apache.beam.runners.dataflow.worker.SimpleParDoFn.processElement ( SimpleParDoFn.java:340 )
at org.apache.beam.runners.dataflow.worker.util.common.worker.ParDoOperation.process ( ParDoOperation.java:44 )
at org.apache.beam.runners.dataflow.worker.util.common.worker.OutputReceiver.process ( OutputReceiver.java:54 )
at org.apache.beam.runners.dataflow.worker.GroupAlsoByWindowsParDoFn$1.output ( GroupAlsoByWindowsParDoFn.java:185 )
Caused by: java.lang.RuntimeException

at org.apache.beam.sdk.io.gcp.bigquery.TableRowToStorageApiProto$SchemaInformation.getSchemaForField ( TableRowToStorageApiProto.java:371 )
at org.apache.beam.sdk.io.gcp.bigquery.TableRowToStorageApiProto.messageFromMap ( TableRowToStorageApiProto.java:472 )
at org.apache.beam.sdk.io.gcp.bigquery.TableRowToStorageApiProto.messageFromTableRow ( TableRowToStorageApiProto.java:625 )
at org.apache.beam.sdk.io.gcp.bigquery.StorageApiDynamicDestinationsTableRow$TableRowConverter.toMessage ( StorageApiDynamicDestinationsTableRow.java:170 )
at org.apache.beam.sdk.io.gcp.bigquery.StorageApiConvertMessages$ConvertMessagesDoFn.processElement ( StorageApiConvertMessages.java:159 )
```

The field `eventId` exists but for some reason the case of the field changes. 

This does not occur in when using File Loads method. 

### Issue Priority

Priority: 1 (data loss / total loss of function)

### Issue Components

- [ ] Component: Python SDK
- [X] Component: Java SDK
- [ ] Component: Go SDK
- [ ] Component: Typescript SDK
- [x] Component: IO connector
- [ ] Component: Beam examples
- [ ] Component: Beam playground
- [ ] Component: Beam katas
- [ ] Component: Website
- [ ] Component: Spark Runner
- [ ] Component: Flink Runner
- [ ] Component: Samza Runner
- [ ] Component: Twister2 Runner
- [ ] Component: Hazelcast Jet Runner
- [ ] Component: Google Cloud Dataflow Runner

Changed files:
sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOWriteTest.java
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/TableRowToStorageApiProto.java

diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/TableRowToStorageApiProto.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/TableRowToStorageApiProto.java
index d231d84aea..d98f9115cb 100644
--- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/TableRowToStorageApiProto.java
+++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/TableRowToStorageApiProto.java
@@ -344,7 +344,7 @@ public class TableRowToStorageApiProto {
             new SchemaInformation(
                 field, Iterables.concat(this.parentSchemas, ImmutableList.of(this)));
         subFields.add(schemaInformation);
-        subFieldsByName.put(field.getName(), schemaInformation);
+        subFieldsByName.put(field.getName().toLowerCase(), schemaInformation);
       }
     }
 
@@ -365,9 +365,9 @@ public class TableRowToStorageApiProto {
     }
 
     public SchemaInformation getSchemaForField(String name) {
-      SchemaInformation schemaInformation = subFieldsByName.get(name);
+      SchemaInformation schemaInformation = subFieldsByName.get(name.toLowerCase());
       if (schemaInformation == null) {
-        throw new RuntimeException("Schema field not found: " + name);
+        throw new RuntimeException("Schema field not found: " + name.toLowerCase());
       }
       return schemaInformation;
     }
diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOWriteTest.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOWriteTest.java
index 534c1b0c36..6405c0599e 100644
--- a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOWriteTest.java
+++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOWriteTest.java
@@ -1824,7 +1824,7 @@ public class BigQueryIOWriteTest implements Serializable {
                     new SimpleFunction<Long, TableRow>() {
                       @Override
                       public TableRow apply(Long input) {
-                        return new TableRow().set("name", "name " + input).set("number", input);
+                        return new TableRow().set("NaMe", "name " + input).set("numBEr", input);
                       }
                     }))
             .setCoder(TableRowJsonCoder.of());