tighten BQ conversion logic

Author: ahmedabu98

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

@@ -932,7 +932,14 @@ public static Row toBeamRow(Schema rowSchema, TableSchema bqSchema, TableRow jso
           return java.time.Instant.parse(jsonBQString);
         }
       } else if (fieldType.isLogicalType(Timestamp.IDENTIFIER)) {
-        return VAR_PRECISION_FORMATTER.parse(jsonBQString, java.time.Instant::from);
+        try {
+          BigDecimal bd = new BigDecimal(jsonBQString);
+          long seconds = bd.longValue();
+          long nanos = bd.remainder(BigDecimal.ONE).movePointRight(9).longValue();
+          return java.time.Instant.ofEpochSecond(seconds, nanos);
+        } catch (NumberFormatException e) {
+          return VAR_PRECISION_FORMATTER.parse(jsonBQString, java.time.Instant::from);
+        }
       }
     }
 

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

@@ -41,6 +41,7 @@
 import java.time.LocalDate;
 import java.time.LocalDateTime;
 import java.time.LocalTime;
+import java.time.OffsetDateTime;
 import java.util.Arrays;
 import java.util.Base64;
 import java.util.Collections;
@@ -1449,17 +1450,27 @@ public void testToBeamRow_timestampMicros_utcSuffix() {
 
     // BigQuery format with " UTC" suffix
     String timestamp = "2024-08-10 16:52:07.123456 UTC";
-
-    Row beamRow = BigQueryUtils.toBeamRow(schema, new TableRow().set("ts", timestamp));
-
-    java.time.Instant actual = (java.time.Instant) beamRow.getValue("ts");
-    assertEquals(2024, actual.atZone(java.time.ZoneOffset.UTC).getYear());
-    assertEquals(8, actual.atZone(java.time.ZoneOffset.UTC).getMonthValue());
-    assertEquals(10, actual.atZone(java.time.ZoneOffset.UTC).getDayOfMonth());
-    assertEquals(16, actual.atZone(java.time.ZoneOffset.UTC).getHour());
-    assertEquals(52, actual.atZone(java.time.ZoneOffset.UTC).getMinute());
-    assertEquals(7, actual.atZone(java.time.ZoneOffset.UTC).getSecond());
-    assertEquals(123456000, actual.getNano());
+    String parsableTimestamp = "2024-08-10T16:52:07.123456Z";
+    java.time.Instant instant = OffsetDateTime.parse(parsableTimestamp).toInstant();
+    long seconds = instant.getEpochSecond();
+    long micros = instant.getNano() / 1000; // BQ stores in micros precision
+    String value = seconds + "." + micros;
+
+    List<TableRow> testRows =
+        Arrays.asList(new TableRow().set("ts", timestamp), new TableRow().set("ts", value));
+
+    for (TableRow row : testRows) {
+      Row beamRow = BigQueryUtils.toBeamRow(schema, row);
+
+      java.time.Instant actual = (java.time.Instant) beamRow.getValue("ts");
+      assertEquals(2024, actual.atZone(java.time.ZoneOffset.UTC).getYear());
+      assertEquals(8, actual.atZone(java.time.ZoneOffset.UTC).getMonthValue());
+      assertEquals(10, actual.atZone(java.time.ZoneOffset.UTC).getDayOfMonth());
+      assertEquals(16, actual.atZone(java.time.ZoneOffset.UTC).getHour());
+      assertEquals(52, actual.atZone(java.time.ZoneOffset.UTC).getMinute());
+      assertEquals(7, actual.atZone(java.time.ZoneOffset.UTC).getSecond());
+      assertEquals(123456000, actual.getNano());
+    }
   }
 
   @Test

sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/BigQueryMetastoreCatalogIT.java

@@ -58,6 +58,7 @@ public Catalog createCatalog() {
             .put("gcp_project", OPTIONS.getProject())
             .put("gcp_location", "us-central1")
             .put("warehouse", warehouse)
+            .put("io-impl", "org.apache.iceberg.gcp.gcs.GCSFileIO")
             .build(),
         new Configuration());
   }
@@ -115,6 +116,8 @@ public void testWriteToPartitionedAndValidateWithBQQuery()
             .map(tr -> BigQueryUtils.toBeamRow(BEAM_SCHEMA, tr))
             .collect(Collectors.toList());
 
+    System.out.println("returned: " + beamRows);
+    System.out.println("expected: " + inputRows);
     assertThat(beamRows, containsInAnyOrder(inputRows.toArray()));
 
     String queryByPartition =

sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/IcebergCatalogBaseIT.java

@@ -308,8 +308,8 @@ public Row apply(Long num) {
               .addValue(LongStream.range(0, num % 10).boxed().collect(Collectors.toList()))
               .addValue(num % 2 == 0 ? null : nestedRow)
               .addValue(num)
-              .addValue(DateTimeUtil.timestamptzFromMicros(timestampMillis * 1000).toInstant())
-              .addValue(DateTimeUtil.timestampFromMicros(timestampMillis * 1000))
+              .addValue(DateTimeUtil.timestamptzFromMicros(timestampMillis * 1000 + 123456789).toInstant())
+              .addValue(DateTimeUtil.timestampFromMicros(timestampMillis * 1000 + 123456789))
               .addValue(DateTimeUtil.dateFromDays(Integer.parseInt(strNum)))
               .addValue(DateTimeUtil.timeFromMicros(num))
               .build();