Skip to content

Commit c0b874a

Browse files
whuffman36whuffman36
authored
feat: Enable Lossless Timestamps in BQ java client lib (#3589)
* feat: Enable Lossless Timestamps in BQ java client lib * Fix Formatting. * Fix tests for FieldValue and FieldValueList. * Add more robust testing to IT test, minor formatting fixes.
1 parent 3eef3a9 commit c0b874a

File tree

9 files changed

+183
-28
lines changed

9 files changed

+183
-28
lines changed

‎google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/BigQueryImpl.java

Lines changed: 11 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1206,15 +1206,15 @@ public TableDataList call() {
12061206
new PageImpl<>(
12071207
new TableDataPageFetcher(tableId, schema, serviceOptions, cursor, pageOptionMap),
12081208
cursor,
1209-
transformTableData(result.getRows(), schema)),
1209+
transformTableData(result.getRows(), schema, serviceOptions.getUseInt64Timestamps())),
12101210
result.getTotalRows());
12111211
} catch (RetryHelper.RetryHelperException e) {
12121212
throw BigQueryException.translateAndThrow(e);
12131213
}
12141214
}
12151215

12161216
private static Iterable<FieldValueList> transformTableData(
1217-
Iterable<TableRow> tableDataPb, final Schema schema) {
1217+
Iterable<TableRow> tableDataPb, final Schema schema, boolean useInt64Timestamps) {
12181218
return ImmutableList.copyOf(
12191219
Iterables.transform(
12201220
tableDataPb != null ? tableDataPb : ImmutableList.<TableRow>of(),
@@ -1223,7 +1223,7 @@ private static Iterable<FieldValueList> transformTableData(
12231223

12241224
@Override
12251225
public FieldValueList apply(TableRow rowPb) {
1226-
return FieldValueList.fromPb(rowPb.getF(), fields);
1226+
return FieldValueList.fromPb(rowPb.getF(), fields, useInt64Timestamps);
12271227
}
12281228
}));
12291229
}
@@ -1347,7 +1347,8 @@ public TableResult query(QueryJobConfiguration configuration, JobOption... optio
13471347

13481348
// If all parameters passed in configuration are supported by the query() method on the backend,
13491349
// put on fast path
1350-
QueryRequestInfo requestInfo = new QueryRequestInfo(configuration);
1350+
QueryRequestInfo requestInfo =
1351+
new QueryRequestInfo(configuration, getOptions().getUseInt64Timestamps());
13511352
if (requestInfo.isFastQuerySupported(null)) {
13521353
String projectId = getOptions().getProjectId();
13531354
QueryRequest content = requestInfo.toPb();
@@ -1420,7 +1421,8 @@ public com.google.api.services.bigquery.model.QueryResponse call() {
14201421
// fetch next pages of results
14211422
new QueryPageFetcher(jobId, schema, getOptions(), cursor, optionMap(options)),
14221423
cursor,
1423-
transformTableData(results.getRows(), schema)))
1424+
transformTableData(
1425+
results.getRows(), schema, getOptions().getUseInt64Timestamps())))
14241426
.setJobId(jobId)
14251427
.setQueryId(results.getQueryId())
14261428
.build();
@@ -1433,7 +1435,8 @@ public com.google.api.services.bigquery.model.QueryResponse call() {
14331435
new PageImpl<>(
14341436
new TableDataPageFetcher(null, schema, getOptions(), null, optionMap(options)),
14351437
null,
1436-
transformTableData(results.getRows(), schema)))
1438+
transformTableData(
1439+
results.getRows(), schema, getOptions().getUseInt64Timestamps())))
14371440
// Return the JobID of the successful job
14381441
.setJobId(
14391442
results.getJobReference() != null ? JobId.fromPb(results.getJobReference()) : null)
@@ -1448,7 +1451,8 @@ public TableResult query(QueryJobConfiguration configuration, JobId jobId, JobOp
14481451

14491452
// If all parameters passed in configuration are supported by the query() method on the backend,
14501453
// put on fast path
1451-
QueryRequestInfo requestInfo = new QueryRequestInfo(configuration);
1454+
QueryRequestInfo requestInfo =
1455+
new QueryRequestInfo(configuration, getOptions().getUseInt64Timestamps());
14521456
if (requestInfo.isFastQuerySupported(jobId)) {
14531457
// Be careful when setting the projectID in JobId, if a projectID is specified in the JobId,
14541458
// the job created by the query method will use that project. This may cause the query to

‎google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/BigQueryOptions.java

Lines changed: 17 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -34,10 +34,11 @@ public class BigQueryOptions extends ServiceOptions<BigQuery, BigQueryOptions> {
3434
private static final int DEFAULT_READ_API_TIME_OUT = 60000;
3535
private static final String BIGQUERY_SCOPE = "https://www.googleapis.com/auth/bigquery";
3636
private static final Set<String> SCOPES = ImmutableSet.of(BIGQUERY_SCOPE);
37-
private static final long serialVersionUID = -2437598817433266049L;
37+
private static final long serialVersionUID = -2437598817433266048L;
3838
private final String location;
3939
// set the option ThrowNotFound when you want to throw the exception when the value not found
4040
private boolean setThrowNotFound;
41+
private boolean useInt64Timestamps;
4142
private String queryPreviewEnabled = System.getenv("QUERY_PREVIEW_ENABLED");
4243

4344
public static class DefaultBigQueryFactory implements BigQueryFactory {
@@ -63,6 +64,7 @@ public ServiceRpc create(BigQueryOptions options) {
6364
public static class Builder extends ServiceOptions.Builder<BigQuery, BigQueryOptions, Builder> {
6465

6566
private String location;
67+
private boolean useInt64Timestamps;
6668

6769
private Builder() {}
6870

@@ -84,6 +86,11 @@ public Builder setLocation(String location) {
8486
return this;
8587
}
8688

89+
public Builder setUseInt64Timestamps(boolean useInt64Timestamps) {
90+
this.useInt64Timestamps = useInt64Timestamps;
91+
return this;
92+
}
93+
8794
@Override
8895
public BigQueryOptions build() {
8996
return new BigQueryOptions(this);
@@ -93,6 +100,7 @@ public BigQueryOptions build() {
93100
private BigQueryOptions(Builder builder) {
94101
super(BigQueryFactory.class, BigQueryRpcFactory.class, builder, new BigQueryDefaults());
95102
this.location = builder.location;
103+
this.useInt64Timestamps = builder.useInt64Timestamps;
96104
}
97105

98106
private static class BigQueryDefaults implements ServiceDefaults<BigQuery, BigQueryOptions> {
@@ -140,6 +148,10 @@ public void setThrowNotFound(boolean setThrowNotFound) {
140148
this.setThrowNotFound = setThrowNotFound;
141149
}
142150

151+
public void setUseInt64Timestamps(boolean useInt64Timestamps) {
152+
this.useInt64Timestamps = useInt64Timestamps;
153+
}
154+
143155
@VisibleForTesting
144156
public void setQueryPreviewEnabled(String queryPreviewEnabled) {
145157
this.queryPreviewEnabled = queryPreviewEnabled;
@@ -149,6 +161,10 @@ public boolean getThrowNotFound() {
149161
return setThrowNotFound;
150162
}
151163

164+
public boolean getUseInt64Timestamps() {
165+
return useInt64Timestamps;
166+
}
167+
152168
@SuppressWarnings("unchecked")
153169
@Override
154170
public Builder toBuilder() {

‎google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/FieldValue.java

Lines changed: 39 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@
2626
import com.google.common.io.BaseEncoding;
2727
import java.io.Serializable;
2828
import java.math.BigDecimal;
29+
import java.math.BigInteger;
2930
import java.math.RoundingMode;
3031
import java.time.Duration;
3132
import java.time.Instant;
@@ -46,10 +47,11 @@
4647
public class FieldValue implements Serializable {
4748

4849
private static final int MICROSECONDS = 1000000;
49-
private static final long serialVersionUID = 469098630191710061L;
50+
private static final long serialVersionUID = 469098630191710062L;
5051

5152
private final Attribute attribute;
5253
private final Object value;
54+
private final Boolean useInt64Timestamps;
5355

5456
/** The field value's attribute, giving information on the field's content type. */
5557
public enum Attribute {
@@ -74,8 +76,13 @@ public enum Attribute {
7476
}
7577

7678
private FieldValue(Attribute attribute, Object value) {
79+
this(attribute, value, false);
80+
}
81+
82+
private FieldValue(Attribute attribute, Object value, Boolean useInt64Timestamps) {
7783
this.attribute = checkNotNull(attribute);
7884
this.value = value;
85+
this.useInt64Timestamps = useInt64Timestamps;
7986
}
8087

8188
/**
@@ -107,6 +114,10 @@ public Object getValue() {
107114
return value;
108115
}
109116

117+
public Boolean getUseInt64Timestamps() {
118+
return useInt64Timestamps;
119+
}
120+
110121
/**
111122
* Returns this field's value as a {@link String}. This method should only be used if the
112123
* corresponding field has primitive type ({@link LegacySQLTypeName#BYTES}, {@link
@@ -207,6 +218,9 @@ public boolean getBooleanValue() {
207218
*/
208219
@SuppressWarnings("unchecked")
209220
public long getTimestampValue() {
221+
if (useInt64Timestamps) {
222+
return new BigInteger(getStringValue()).longValue();
223+
}
210224
// timestamps are encoded in the format 1408452095.22 where the integer part is seconds since
211225
// epoch (e.g. 1408452095.22 == 2014-08-19 07:41:35.220 -05:00)
212226
BigDecimal secondsWithMicro = new BigDecimal(getStringValue());
@@ -317,12 +331,13 @@ public String toString() {
317331
return MoreObjects.toStringHelper(this)
318332
.add("attribute", attribute)
319333
.add("value", value)
334+
.add("useInt64Timestamps", useInt64Timestamps)
320335
.toString();
321336
}
322337

323338
@Override
324339
public final int hashCode() {
325-
return Objects.hash(attribute, value);
340+
return Objects.hash(attribute, value, useInt64Timestamps);
326341
}
327342

328343
@Override
@@ -334,7 +349,9 @@ public final boolean equals(Object obj) {
334349
return false;
335350
}
336351
FieldValue other = (FieldValue) obj;
337-
return attribute == other.attribute && Objects.equals(value, other.value);
352+
return attribute == other.attribute
353+
&& Objects.equals(value, other.value)
354+
&& Objects.equals(useInt64Timestamps, other.useInt64Timestamps);
338355
}
339356

340357
/**
@@ -353,42 +370,52 @@ public final boolean equals(Object obj) {
353370
*/
354371
@BetaApi
355372
public static FieldValue of(Attribute attribute, Object value) {
356-
return new FieldValue(attribute, value);
373+
return of(attribute, value, false);
374+
}
375+
376+
@BetaApi
377+
public static FieldValue of(Attribute attribute, Object value, Boolean useInt64Timestamps) {
378+
return new FieldValue(attribute, value, useInt64Timestamps);
357379
}
358380

359381
static FieldValue fromPb(Object cellPb) {
360-
return fromPb(cellPb, null);
382+
return fromPb(cellPb, null, false);
361383
}
362384

363385
@SuppressWarnings("unchecked")
364-
static FieldValue fromPb(Object cellPb, Field recordSchema) {
386+
static FieldValue fromPb(Object cellPb, Field recordSchema, Boolean useInt64Timestamps) {
365387
if (Data.isNull(cellPb)) {
366-
return FieldValue.of(Attribute.PRIMITIVE, null);
388+
return FieldValue.of(Attribute.PRIMITIVE, null, useInt64Timestamps);
367389
}
368390
if (cellPb instanceof String) {
369391
if ((recordSchema != null)
370392
&& (recordSchema.getType() == LegacySQLTypeName.RANGE)
371393
&& (recordSchema.getRangeElementType() != null)) {
372394
return FieldValue.of(
373-
Attribute.RANGE, Range.of((String) cellPb, recordSchema.getRangeElementType()));
395+
Attribute.RANGE,
396+
Range.of((String) cellPb, recordSchema.getRangeElementType()),
397+
useInt64Timestamps);
374398
}
375-
return FieldValue.of(Attribute.PRIMITIVE, cellPb);
399+
return FieldValue.of(Attribute.PRIMITIVE, cellPb, useInt64Timestamps);
376400
}
377401
if (cellPb instanceof List) {
378-
return FieldValue.of(Attribute.REPEATED, FieldValueList.fromPb((List<Object>) cellPb, null));
402+
return FieldValue.of(
403+
Attribute.REPEATED,
404+
FieldValueList.fromPb((List<Object>) cellPb, null, useInt64Timestamps));
379405
}
380406
if (cellPb instanceof Map) {
381407
Map<String, Object> cellMapPb = (Map<String, Object>) cellPb;
382408
if (cellMapPb.containsKey("f")) {
383409
FieldList subFieldsSchema = recordSchema != null ? recordSchema.getSubFields() : null;
384410
return FieldValue.of(
385411
Attribute.RECORD,
386-
FieldValueList.fromPb((List<Object>) cellMapPb.get("f"), subFieldsSchema));
412+
FieldValueList.fromPb(
413+
(List<Object>) cellMapPb.get("f"), subFieldsSchema, useInt64Timestamps));
387414
}
388415
// This should never be the case when we are processing a first level table field (i.e. a
389416
// row's field, not a record sub-field)
390417
if (cellMapPb.containsKey("v")) {
391-
return FieldValue.fromPb(cellMapPb.get("v"), recordSchema);
418+
return FieldValue.fromPb(cellMapPb.get("v"), recordSchema, useInt64Timestamps);
392419
}
393420
}
394421
throw new IllegalArgumentException("Unexpected table cell format");

‎google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/FieldValueList.java

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -112,6 +112,10 @@ FieldValueList withSchema(FieldList schema) {
112112
}
113113

114114
static FieldValueList fromPb(List<?> rowPb, FieldList schema) {
115+
return fromPb(rowPb, schema, false);
116+
}
117+
118+
static FieldValueList fromPb(List<?> rowPb, FieldList schema, Boolean useInt64Timestamps) {
115119
List<FieldValue> row = new ArrayList<>(rowPb.size());
116120
if (schema != null) {
117121
if (schema.size() != rowPb.size()) {
@@ -120,11 +124,11 @@ static FieldValueList fromPb(List<?> rowPb, FieldList schema) {
120124
Iterator<Field> schemaIter = schema.iterator();
121125
Iterator<?> rowPbIter = rowPb.iterator();
122126
while (rowPbIter.hasNext() && schemaIter.hasNext()) {
123-
row.add(FieldValue.fromPb(rowPbIter.next(), schemaIter.next()));
127+
row.add(FieldValue.fromPb(rowPbIter.next(), schemaIter.next(), useInt64Timestamps));
124128
}
125129
} else {
126130
for (Object cellPb : rowPb) {
127-
row.add(FieldValue.fromPb(cellPb, null));
131+
row.add(FieldValue.fromPb(cellPb, null, useInt64Timestamps));
128132
}
129133
}
130134

‎google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/QueryRequestInfo.java

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616

1717
package com.google.cloud.bigquery;
1818

19+
import com.google.api.services.bigquery.model.DataFormatOptions;
1920
import com.google.api.services.bigquery.model.QueryParameter;
2021
import com.google.api.services.bigquery.model.QueryRequest;
2122
import com.google.cloud.bigquery.QueryJobConfiguration.JobCreationMode;
@@ -42,8 +43,9 @@ final class QueryRequestInfo {
4243
private final Boolean useQueryCache;
4344
private final Boolean useLegacySql;
4445
private final JobCreationMode jobCreationMode;
46+
private final DataFormatOptions formatOptions;
4547

46-
QueryRequestInfo(QueryJobConfiguration config) {
48+
QueryRequestInfo(QueryJobConfiguration config, Boolean useInt64Timestamps) {
4749
this.config = config;
4850
this.connectionProperties = config.getConnectionProperties();
4951
this.defaultDataset = config.getDefaultDataset();
@@ -58,6 +60,7 @@ final class QueryRequestInfo {
5860
this.useLegacySql = config.useLegacySql();
5961
this.useQueryCache = config.useQueryCache();
6062
this.jobCreationMode = config.getJobCreationMode();
63+
this.formatOptions = new DataFormatOptions().setUseInt64Timestamp(useInt64Timestamps);
6164
}
6265

6366
boolean isFastQuerySupported(JobId jobId) {
@@ -122,6 +125,9 @@ QueryRequest toPb() {
122125
if (jobCreationMode != null) {
123126
request.setJobCreationMode(jobCreationMode.toString());
124127
}
128+
if (formatOptions != null) {
129+
request.setFormatOptions(formatOptions);
130+
}
125131
return request;
126132
}
127133

@@ -141,6 +147,7 @@ public String toString() {
141147
.add("useQueryCache", useQueryCache)
142148
.add("useLegacySql", useLegacySql)
143149
.add("jobCreationMode", jobCreationMode)
150+
.add("formatOptions", formatOptions.getUseInt64Timestamp())
144151
.toString();
145152
}
146153

@@ -159,7 +166,8 @@ public int hashCode() {
159166
createSession,
160167
useQueryCache,
161168
useLegacySql,
162-
jobCreationMode);
169+
jobCreationMode,
170+
formatOptions);
163171
}
164172

165173
@Override

0 commit comments

Comments
 (0)