apache · goutamadwant · Sep 14, 2026
diff --git a/CHANGES.md b/CHANGES.md
@@ -92,6 +92,7 @@
 
 ## Bugfixes
 
+* (Java) Fixed BigQuery Storage Write API failed-row timestamp conversion, preserving microsecond precision and UTC formatting ([#40110](https://github.com/apache/beam/issues/40110)).
 * (Java) Fixed the Spark runner firing processing-time timers in reverse timestamp order ([#39824](https://github.com/apache/beam/issues/39824)).
 * (Python) Fixed incorrect profiler options handling on portable runners ([#39613](https://github.com/apache/beam/issues/39613)).
 * (Java) KafkaIO dynamic reads no longer require the obsolete `beam_fn_api` experiment ([#29998](https://github.com/apache/beam/issues/29998)).

diff --git a/...oogle-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryUtils.java b/...oogle-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryUtils.java
@@ -230,6 +230,16 @@ public abstract static class Builder {
           .appendZoneRegionId()
           .toFormatter();
 
+  // TIMESTAMP_FORMATTER accepts alternate separators and must not be used for printing.
+  static final java.time.format.DateTimeFormatter BIGQUERY_TIMESTAMP_MICROS_FORMATTER =
+      new java.time.format.DateTimeFormatterBuilder()
+          .appendPattern("uuuu-MM-dd HH:mm:ss")
+          .appendFraction(java.time.temporal.ChronoField.NANO_OF_SECOND, 0, 6, true)
+          .appendLiteral(" UTC")
+          .toFormatter()
+          .withResolverStyle(java.time.format.ResolverStyle.STRICT)
+          .withZone(ZoneOffset.UTC);
+
   private static final DateTimeFormatter BIGQUERY_TIMESTAMP_PRINTER;
 
   /**
@@ -936,7 +946,11 @@ public static Row toBeamRow(Schema rowSchema, TableSchema bqSchema, TableRow jso
           long nanos = (micros % 1_000_000) * 1_000;
           return java.time.Instant.ofEpochSecond(seconds, nanos);
         } catch (NumberFormatException e) {
-          return java.time.Instant.parse(jsonBQString);
+          try {
+            return java.time.Instant.parse(jsonBQString);
+          } catch (DateTimeParseException e2) {
+            return BIGQUERY_TIMESTAMP_MICROS_FORMATTER.parse(jsonBQString, java.time.Instant::from);
+          }
         }
       } else if (fieldType.isLogicalType(Timestamp.IDENTIFIER)) {
         if (!jsonBQString.contains("UTC")) {

diff --git a/...platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/TableRowToStorageApiProto.java b/...platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/TableRowToStorageApiProto.java
@@ -19,6 +19,7 @@
 
 import static java.util.stream.Collectors.toList;
 import static java.util.stream.Collectors.toSet;
+import static org.apache.beam.sdk.io.gcp.bigquery.BigQueryUtils.BIGQUERY_TIMESTAMP_MICROS_FORMATTER;
 import static org.apache.beam.sdk.io.gcp.bigquery.BigQueryUtils.DATETIME_SPACE_FORMATTER;
 import static org.apache.beam.sdk.io.gcp.bigquery.BigQueryUtils.TIMESTAMP_FORMATTER;
 
@@ -56,6 +57,7 @@
 import java.time.LocalTime;
 import java.time.ZoneOffset;
 import java.time.format.DateTimeParseException;
+import java.time.temporal.ChronoUnit;
 import java.util.AbstractMap;
 import java.util.ArrayList;
 import java.util.Collections;
@@ -1990,7 +1992,7 @@ public static Object jsonValueFromMessageValue(
           long epochSeconds = epochMicros / 1_000_000L;
           long nanoAdjustment = (epochMicros % 1_000_000L) * 1_000L;
           Instant instant = Instant.ofEpochSecond(epochSeconds, nanoAdjustment);
-          return LocalDateTime.ofInstant(instant, ZoneOffset.UTC).format(TIMESTAMP_FORMATTER);
+          return BIGQUERY_TIMESTAMP_MICROS_FORMATTER.format(instant);
         } else if (fieldDescriptor.getType().equals(FieldDescriptor.Type.MESSAGE)) {
           Message message = (Message) fieldValue;
           String messageName = fieldDescriptor.getMessageType().getName();
@@ -2000,7 +2002,8 @@ public static Object jsonValueFromMessageValue(
             long seconds = (long) message.getField(descriptor.findFieldByName("seconds"));
             int nanos = (int) message.getField(descriptor.findFieldByName("nanos"));
             Instant instant = Instant.ofEpochSecond(seconds, nanos);
-            return LocalDateTime.ofInstant(instant, ZoneOffset.UTC).format(TIMESTAMP_FORMATTER);
+            return BIGQUERY_TIMESTAMP_MICROS_FORMATTER.format(
+                instant.truncatedTo(ChronoUnit.MICROS));
           } else if (messageName.equals("TimestampPicos")) {
             Descriptor descriptor = message.getDescriptorForType();
             long seconds = (long) message.getField(descriptor.findFieldByName("seconds"));

diff --git a/...cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOWriteTest.java b/...cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOWriteTest.java
@@ -1335,20 +1335,12 @@ public void runTestWriteAvro(boolean schemaFromView) throws Exception {
                 .set("strval", "test")
                 .set("longval", "1")
                 .set("doubleval", "1.0")
-                .set(
-                    "instantval",
-                    useStorageApi || useStorageApiApproximate
-                        ? "2019-01-01 T00:00:00"
-                        : "2019-01-01 00:00:00 UTC"),
+                .set("instantval", "2019-01-01 00:00:00 UTC"),
             new TableRow()
                 .set("strval", "test2")
                 .set("longval", "2")
                 .set("doubleval", "2.0")
-                .set(
-                    "instantval",
-                    useStorageApi || useStorageApiApproximate
-                        ? "2019-02-01 T00:00:00"
-                        : "2019-02-01 00:00:00 UTC")));
+                .set("instantval", "2019-02-01 00:00:00 UTC")));
   }
 
   @Test
@@ -3876,7 +3868,7 @@ public void testStorageApiErrorsWriteTableRows() throws Exception {
     TableRow goodNested =
         new TableRow()
             .set("number", "42")
-            .set("timestamp", "1970-01-01 T00:00:00.000043")
+            .set("timestamp", "1970-01-01 00:00:00.000043 UTC")
             .set("time", "00:52:07.123456")
             .set("datetime", "2019-08-16T00:52:07.123456")
             .set("date", "2019-08-16")
@@ -4427,7 +4419,7 @@ public void testWriteProtosEncodedValues(boolean directWrite) throws Exception {
                         .setMode("REPEATED")
                         .setFields(tableSchema.getFields())));
 
-    final String timestamp = "1970-01-01 T00:00:00.000043";
+    final String timestamp = "1970-01-01 00:00:00.000043 UTC";
     final String date = "2019-08-16";
     final String numeric = "23";
     final String bignumeric = "123456789012345678";
@@ -4585,7 +4577,7 @@ public void testWriteProtosUnEncodedValues(boolean directWrite) throws Exception
                         .setMode("REPEATED")
                         .setFields(tableSchema.getFields())));
 
-    final String timestamp = "1970-01-01 T00:00:00.000043";
+    final String timestamp = "1970-01-01 00:00:00.000043 UTC";
     final String date = "2019-08-16";
     final String numeric = "23";
     final String bignumeric = "123456789012345678";
@@ -4715,7 +4707,7 @@ public void testWriteProtosWrappedValues(boolean directWrite) throws Exception {
                         .setMode("REPEATED")
                         .setFields(tableSchema.getFields())));
 
-    final String timestamp = "1970-01-01 T00:00:00.000043";
+    final String timestamp = "1970-01-01 00:00:00.000043 UTC";
     long timestampMicros =
         (long)
             TYPE_MAP_PROTO_CONVERTERS

diff --git a/...e-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryUtilsTest.java b/...e-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryUtilsTest.java
@@ -1182,6 +1182,67 @@ public void testToBeamRow_timestamp_micros() {
     assertEquals(expectedRow, beamRowMicros);
   }
 
+  @Test
+  public void testTimestampInputCompatibility() {
+    Schema schema = Schema.builder().addLogicalTypeField("ts", SqlTypes.TIMESTAMP).build();
+    java.time.Instant expected = java.time.Instant.parse("2026-09-03T18:51:43.417123Z");
+    for (Object input :
+        Arrays.asList(
+            "2026-09-03 18:51:43.417123 UTC",
+            "2026-09-03T18:51:43.417123Z",
+            "2026-09-03t18:51:43.417123z",
+            "2026-09-03T20:51:43.417123+02:00",
+            "1788461503417123",
+            1788461503417123L)) {
+      assertEquals(
+          input.toString(),
+          expected,
+          BigQueryUtils.toBeamRow(schema, new TableRow().set("ts", input)).getValue("ts"));
+    }
+    assertEquals(
+        java.time.Instant.parse("1969-12-31T23:59:59.999999Z"),
+        BigQueryUtils.toBeamRow(schema, new TableRow().set("ts", -1L)).getValue("ts"));
+    assertEquals(
+        java.time.Instant.EPOCH,
+        BigQueryUtils.toBeamRow(schema, new TableRow().set("ts", "1970-01-01 00:00:00 UTC"))
+            .getValue("ts"));
+    assertThrows(
+        java.time.format.DateTimeParseException.class,
+        () -> BigQueryUtils.toBeamRow(schema, new TableRow().set("ts", "not a timestamp")));
+    assertThrows(
+        java.time.format.DateTimeParseException.class,
+        () -> BigQueryUtils.toBeamRow(schema, new TableRow().set("ts", "2026-02-30 00:00:00 UTC")));
+  }
+
+  @Test
+  public void testDateTimeTimestampInputCompatibility() {
+    Schema schema = Schema.builder().addDateTimeField("ts").build();
+    for (Object input :
+        Arrays.asList("2026-09-03 18:51:43.417123 UTC", "1788461503.417123", 1788461503.417123)) {
+      assertEquals(
+          input.toString(),
+          1788461503417L,
+          BigQueryUtils.toBeamRow(schema, new TableRow().set("ts", input))
+              .getDateTime("ts")
+              .getMillis());
+    }
+    assertEquals(
+        1000L,
+        BigQueryUtils.toBeamRow(schema, new TableRow().set("ts", 1L))
+            .getDateTime("ts")
+            .getMillis());
+    assertEquals(
+        0L,
+        BigQueryUtils.toBeamRow(schema, new TableRow().set("ts", "-0.000001"))
+            .getDateTime("ts")
+            .getMillis());
+    assertThrows(
+        IllegalArgumentException.class,
+        () ->
+            BigQueryUtils.toBeamRow(
+                schema, new TableRow().set("ts", "2026-09-03 18:51:43.417123456 UTC")));
+  }
+
   @Test
   public void testToTableSpec() {
     TableReference withProject =