fix(bigquery-jdbc): request int64 timestamps so the REST path is lossless

java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/BigQueryConnection.java

@@ -1390,15 +1390,14 @@ private BigQuery getBigQueryConnection() {
       bigQueryOptions.setRetrySettings(retry_settings_builder.build());
     }
 
-    // ISO8601_STRING is the only output format carrying the full 12-digit fraction
-    // ("YYYY-MM-DDTHH:MM:SS.FFFFFFFFFFFFZ"). The default FLOAT64 and the INT64 alternative both
-    // cap out at microseconds.
-    if (this.enableTimestampPicos) {
-      bigQueryOptions.setDataFormatOptions(
-          DataFormatOptions.newBuilder()
-              .timestampFormatOptions(DataFormatOptions.TimestampFormatOptions.ISO8601_STRING)
-              .build());
-    }
+    // Default FLOAT64 timestamps lose microseconds far from the epoch. ISO8601_STRING is the only
+    // format carrying picoseconds; BigQuery rejects it combined with useInt64Timestamp.
+    bigQueryOptions.setDataFormatOptions(
+        this.enableTimestampPicos
+            ? DataFormatOptions.newBuilder()
+                .timestampFormatOptions(DataFormatOptions.TimestampFormatOptions.ISO8601_STRING)
+                .build()
+            : DataFormatOptions.newBuilder().useInt64Timestamp(true).build());
 
     if (this.catalog != null) {
       bigQueryOptions.setProjectId(this.catalog);

java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/BigQueryFieldValueListWrapper.java

@@ -58,7 +58,7 @@ static BigQueryFieldValueListWrapper of(
       FieldList fieldList, FieldValueList fieldValueList, boolean[] isComplexColumn) {
     boolean[] flags =
         isComplexColumn != null ? isComplexColumn : createComplexColumnFlags(fieldList);
-    Object[] rowValues = unpackRow(fieldValueList, flags);
+    Object[] rowValues = unpackRow(fieldList, fieldValueList, flags);
     return new BigQueryFieldValueListWrapper(fieldList, null, null, rowValues, false, null);
   }
 
@@ -70,16 +70,13 @@ static boolean[] createComplexColumnFlags(FieldList fieldList) {
     boolean[] isComplex = new boolean[size];
     for (int i = 0; i < size; i++) {
       Field field = fieldList.get(i);
-      isComplex[i] =
-          isArray(field)
-              || isStruct(field)
-              || (field.getType() != null
-                  && field.getType().getStandardType() == StandardSQLTypeName.RANGE);
+      isComplex[i] = isArray(field) || isStruct(field);
     }
     return isComplex;
   }
 
-  static Object[] unpackRow(FieldValueList fieldValueList, boolean[] isComplexColumn) {
+  static Object[] unpackRow(
+      FieldList fieldList, FieldValueList fieldValueList, boolean[] isComplexColumn) {
     if (fieldValueList == null) {
       return null;
     }
@@ -90,7 +87,8 @@ static Object[] unpackRow(FieldValueList fieldValueList, boolean[] isComplexColu
       if (fv == null || fv.isNull()) {
         row[i] = null;
       } else if ((i < isComplexColumn.length && isComplexColumn[i])
-          || fv.getAttribute() != Attribute.PRIMITIVE) {
+          || fv.getAttribute() != Attribute.PRIMITIVE
+          || keepsFieldValue(fieldList, i)) {
         row[i] = fv;
       } else {
         row[i] = fv.getStringValue();
@@ -99,6 +97,16 @@ static Object[] unpackRow(FieldValueList fieldValueList, boolean[] isComplexColu
     return row;
   }
 
+  /** RANGE and TIMESTAMP are read from their FieldValue, e.g. to detect int64 encoding. */
+  private static boolean keepsFieldValue(FieldList fieldList, int index) {
+    if (fieldList == null || index >= fieldList.size()) {
+      return false;
+    }
+    Field field = fieldList.get(index);
+    StandardSQLTypeName type = field.getType() == null ? null : field.getType().getStandardType();
+    return type == StandardSQLTypeName.RANGE || type == StandardSQLTypeName.TIMESTAMP;
+  }
+
   static BigQueryFieldValueListWrapper getNestedFieldValueListWrapper(
       FieldList fieldList, List<FieldValue> arrayFieldValueList, boolean... isLast) {
     boolean isLastFlag = isLast != null && isLast.length == 1 && isLast[0];

java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/BigQueryTemporalUtility.java

@@ -439,17 +439,27 @@ static String formatRange(Field field, Object value, boolean enableTimestampPico
       throws BigQueryJdbcException {
     Object start;
     Object end;
+    boolean int64Timestamps = false;
     if (value instanceof FieldValue) {
       Range range = ((FieldValue) value).getRangeValue();
       start = range.getStart().isNull() ? null : range.getStart().getStringValue();
       end = range.getEnd().isNull() ? null : range.getEnd().getStringValue();
+      int64Timestamps = Boolean.TRUE.equals(((FieldValue) value).getUseInt64Timestamps());
     } else if (value instanceof Map) {
       start = ((Map<?, ?>) value).get("start");
       end = ((Map<?, ?>) value).get("end");
     } else {
       return value == null ? null : value.toString();
     }
     StandardSQLTypeName elementType = getRangeElementType(field, start != null ? start : end);
+    if (int64Timestamps && elementType == StandardSQLTypeName.TIMESTAMP) {
+      try {
+        start = start == null ? null : Long.valueOf((String) start);
+        end = end == null ? null : Long.valueOf((String) end);
+      } catch (NumberFormatException e) {
+        throw new BigQueryJdbcException("Invalid RANGE bound: " + value, e);
+      }
+    }
     return "["
         + formatRangeBound(start, elementType, enableTimestampPicos)
         + ", "

java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/BigQueryTypeRegistry.java

@@ -829,7 +829,7 @@ public static Object convert(Object input, StandardSQLTypeName bqType, ZoneId zo
       if (fv.isNull()) {
         return null;
       }
-      input = fv.getValue();
+      input = unwrap(fv, bqType);
     }
     int ordinal = bqType.ordinal();
     if (ordinal >= DESCRIPTORS_BY_ORDINAL.length || DESCRIPTORS_BY_ORDINAL[ordinal] == null) {
@@ -860,7 +860,7 @@ public static <T> T convert(
       if (fv.isNull()) {
         return null;
       }
-      input = fv.getValue();
+      input = unwrap(fv, bqType);
     }
     int ordinal = bqType.ordinal();
     if (ordinal >= DESCRIPTORS_BY_ORDINAL.length || DESCRIPTORS_BY_ORDINAL[ordinal] == null) {
@@ -886,6 +886,20 @@ public static <T> T convert(
     }
   }
 
+  /** Returns int64-encoded TIMESTAMP values as epoch micros, as the Arrow path delivers them. */
+  private static Object unwrap(FieldValue fv, StandardSQLTypeName bqType)
+      throws BigQueryJdbcException {
+    if (bqType == StandardSQLTypeName.TIMESTAMP
+        && Boolean.TRUE.equals(fv.getUseInt64Timestamps())) {
+      try {
+        return Long.valueOf(fv.getStringValue());
+      } catch (NumberFormatException e) {
+        throw new BigQueryJdbcException("Invalid int64 TIMESTAMP value: " + fv.getValue(), e);
+      }
+    }
+    return fv.getValue();
+  }
+
   private static TypeDescriptor<?> getDescriptorForClass(Class<?> clazz) {
     TypeDescriptor<?> descriptor = DESCRIPTORS_BY_CLASS.get(clazz);
     if (descriptor != null) {

java-bigquery-jdbc/src/test/java/com/google/cloud/bigquery/jdbc/BigQueryConnectionTest.java

@@ -277,10 +277,9 @@ public void testTimestampPicosControlsDataFormatOptions() throws SQLException {
     }
 
     try (BigQueryConnection connection = new BigQueryConnection(BASE_URL)) {
-      assertEquals(
-          DataFormatOptions.TimestampFormatOptions.TIMESTAMP_OUTPUT_FORMAT_UNSPECIFIED,
-          connection.getBigQuery().getOptions().getDataFormatOptions().timestampFormatOptions(),
-          "Default connections must not alter the timestamp wire format");
+      assertTrue(
+          connection.getBigQuery().getOptions().getDataFormatOptions().useInt64Timestamp(),
+          "Default connections must request lossless int64 timestamps");
     }
   }
 

java-bigquery-jdbc/src/test/java/com/google/cloud/bigquery/jdbc/BigQueryFieldValueListWrapperTest.java

@@ -182,7 +182,8 @@ public void testMemoryAllocationReduction() throws Exception {
     long bytesBefore = threadBean.getThreadAllocatedBytes(threadId);
     Object[][] rowBuffers = new Object[rowCount][];
     for (int i = 0; i < rowCount; i++) {
-      rowBuffers[i] = BigQueryFieldValueListWrapper.unpackRow(sampleFvl, isComplexColumn);
+      rowBuffers[i] =
+          BigQueryFieldValueListWrapper.unpackRow(simpleFieldList, sampleFvl, isComplexColumn);
     }
     long allocatedBytes = threadBean.getThreadAllocatedBytes(threadId) - bytesBefore;
 

java-bigquery-jdbc/src/test/java/com/google/cloud/bigquery/jdbc/BigQueryJsonResultSetTest.java

@@ -668,6 +668,28 @@ public void testRangeTimestampPicosEnabled() throws SQLException {
         .isEqualTo("[2023-03-30 11:14:19.123456789123, 2023-03-30 11:14:20.123456789123)");
   }
 
+  @Test
+  public void testInt64Timestamps() throws SQLException {
+    Field rangeField =
+        Field.newBuilder("rangeTs", StandardSQLTypeName.RANGE)
+            .setRangeElementType(FieldElementType.newBuilder().setType("TIMESTAMP").build())
+            .build();
+    FieldList fields = FieldList.of(Field.of("ts", StandardSQLTypeName.TIMESTAMP), rangeField);
+    FieldValueList row =
+        FieldValueList.of(
+            ImmutableList.of(
+                FieldValue.of(Attribute.PRIMITIVE, "253402300799999999", true),
+                FieldValue.of(Attribute.RANGE, Range.of("[-1, 253402300799999999)"), true)),
+            fields);
+    BigQueryJsonResultSet rs = createJsonResultSet(fields, row, false);
+    assertThat(rs.next()).isTrue();
+    assertThat(rs.getString("ts")).isEqualTo("9999-12-31 23:59:59.999999");
+    assertThat(rs.getTimestamp("ts").toInstant())
+        .isEqualTo(Instant.parse("9999-12-31T23:59:59.999999Z"));
+    assertThat(rs.getString("rangeTs"))
+        .isEqualTo("[1969-12-31 23:59:59.999999, 9999-12-31 23:59:59.999999)");
+  }
+
   @Test
   public void testRangeTimestampPicosDisabled() throws SQLException {
     Field rangeField =

java-bigquery-jdbc/src/test/java/com/google/cloud/bigquery/jdbc/it/ITTimestampTimeZoneTest.java

@@ -81,7 +81,8 @@ public class ITTimestampTimeZoneTest extends ITBase {
           + " STRUCT(TIMESTAMP '2025-01-01 12:34:56.123 UTC' AS ts) AS ts_struct,"
           // REST API sends these as scientific notation ("1.0E-6", "-1.0E-6").
           + " TIMESTAMP '1970-01-01 00:00:00.000001 UTC' AS ts_post_epoch,"
-          + " TIMESTAMP '1969-12-31 23:59:59.999999 UTC' AS ts_pre_epoch";
+          + " TIMESTAMP '1969-12-31 23:59:59.999999 UTC' AS ts_pre_epoch,"
+          + " TIMESTAMP '9999-12-31 23:59:59.999999 UTC' AS ts_max";
 
   // Same coercion as an INSERT into a TIMESTAMP column, for TIMESTAMP and STRING parameters.
   private static final String PROBE_TIMESTAMP = "SELECT UNIX_MILLIS(CAST(? AS TIMESTAMP))";
@@ -364,7 +365,12 @@ public void testGetTimestampAndGetObject_returnMoment(String zone) throws SQLExc
               assertEquals(
                   Instant.parse("1969-12-31T23:59:59.999999Z"),
                   rs.getTimestamp("ts_pre_epoch").toInstant(),
-                  label(zone, "getTimestamp just before the epoch")));
+                  label(zone, "getTimestamp just before the epoch")),
+          () ->
+              assertEquals(
+                  Instant.parse("9999-12-31T23:59:59.999999Z"),
+                  rs.getTimestamp("ts_max").toInstant(),
+                  label(zone, "getTimestamp at the maximum TIMESTAMP")));
     }
   }
 
@@ -614,7 +620,12 @@ public void testGetString_returnsUtcDigits(String zone) throws SQLException {
               assertEquals(
                   "1969-12-31 23:59:59.999999",
                   rs.getString("ts_pre_epoch"),
-                  label(zone, "getString just before the epoch")));
+                  label(zone, "getString just before the epoch")),
+          () ->
+              assertEquals(
+                  "9999-12-31 23:59:59.999999",
+                  rs.getString("ts_max"),
+                  label(zone, "getString at the maximum TIMESTAMP")));
     }
   }