feat(bigquery): switch all timestamp representations to int64 usec (#…

…9368)

With the recent resolutions to the discovery representation, we now correctly propagate the options for controlling timestamp representation.  This PR forces all usages to lossless representations.

bigquery/iterator.go

@@ -280,6 +280,7 @@ func fetchTableResultPage(ctx context.Context, src *rowSource, schema Schema, st
 		}()
 	}
 	call := src.t.c.bqs.Tabledata.List(src.t.ProjectID, src.t.DatasetID, src.t.TableID)
+	call = call.FormatOptionsUseInt64Timestamp(true)
 	setClientHeader(call.Header())
 	if pageToken != "" {
 		call.PageToken(pageToken)
@@ -317,6 +318,7 @@ func fetchJobResultPage(ctx context.Context, src *rowSource, schema Schema, star
 	// reduce data transfered by leveraging api projections
 	projectedFields := []googleapi.Field{"rows", "pageToken", "totalRows"}
 	call := src.j.c.bqs.Jobs.GetQueryResults(src.j.projectID, src.j.jobID).Location(src.j.location).Context(ctx)
+	call = call.FormatOptionsUseInt64Timestamp(true)
 	if schema == nil {
 		// only project schema if we weren't supplied one.
 		projectedFields = append(projectedFields, "schema")

bigquery/job.go

@@ -353,6 +353,7 @@ func (j *Job) read(ctx context.Context, waitForQuery func(context.Context, strin
 func (j *Job) waitForQuery(ctx context.Context, projectID string) (Schema, uint64, error) {
 	// Use GetQueryResults only to wait for completion, not to read results.
 	call := j.c.bqs.Jobs.GetQueryResults(projectID, j.jobID).Location(j.location).Context(ctx).MaxResults(0)
+	call = call.FormatOptionsUseInt64Timestamp(true)
 	setClientHeader(call.Header())
 	backoff := gax.Backoff{
 		Initial:    1 * time.Second,

bigquery/query.go

@@ -470,6 +470,9 @@ func (q *Query) probeFastPath() (*bq.QueryRequest, error) {
 		MaximumBytesBilled: q.QueryConfig.MaxBytesBilled,
 		RequestId:          uid.NewSpace("request", nil).New(),
 		Labels:             q.Labels,
+		FormatOptions: &bq.DataFormatOptions{
+			UseInt64Timestamp: true,
+		},
 	}
 	if q.QueryConfig.DisableQueryCache {
 		qRequest.UseQueryCache = &pfalse

bigquery/query_test.go

@@ -439,6 +439,9 @@ func TestProbeFastPath(t *testing.T) {
 			wantReq: &bq.QueryRequest{
 				Query:        "foo",
 				UseLegacySql: &pfalse,
+				FormatOptions: &bq.DataFormatOptions{
+					UseInt64Timestamp: true,
+				},
 			},
 		},
 		{
@@ -473,6 +476,9 @@ func TestProbeFastPath(t *testing.T) {
 					},
 				},
 				UseQueryCache: &pfalse,
+				FormatOptions: &bq.DataFormatOptions{
+					UseInt64Timestamp: true,
+				},
 			},
 		},
 		{

bigquery/value.go

@@ -18,7 +18,6 @@ import (
 	"encoding/base64"
 	"errors"
 	"fmt"
-	"math"
 	"math/big"
 	"reflect"
 	"strconv"
@@ -955,15 +954,11 @@ func convertBasicType(val string, typ FieldType) (Value, error) {
 	case BooleanFieldType:
 		return strconv.ParseBool(val)
 	case TimestampFieldType:
-		f, err := strconv.ParseFloat(val, 64)
+		i, err := strconv.ParseInt(val, 10, 64)
 		if err != nil {
 			return nil, err
 		}
-		secs := math.Trunc(f)
-		// Timestamps in BigQuery have microsecond precision, so we must
-		// return a round number of microseconds.
-		micros := math.Trunc((f-secs)*1e6 + 0.5)
-		return Value(time.Unix(int64(secs), int64(micros)*1000).UTC()), nil
+		return time.UnixMicro(i), nil
 	case DateFieldType:
 		return civil.ParseDate(val)
 	case TimeFieldType:

bigquery/value_test.go

@@ -78,7 +78,7 @@ func TestConvertTime(t *testing.T) {
 	ts := testTimestamp.Round(time.Millisecond)
 	row := &bq.TableRow{
 		F: []*bq.TableCell{
-			{V: fmt.Sprintf("%.10f", float64(ts.UnixNano())/1e9)},
+			{V: fmt.Sprint(ts.UnixMicro())},
 			{V: testDate.String()},
 			{V: testTime.String()},
 			{V: testDateTime.String()},
@@ -95,15 +95,12 @@ func TestConvertTime(t *testing.T) {
 			t.Errorf("#%d: got:\n%v\nwant:\n%v", i, g, w)
 		}
 	}
-	if got[0].(time.Time).Location() != time.UTC {
-		t.Errorf("expected time zone UTC: got:\n%v", got)
-	}
 }
 
 func TestConvertSmallTimes(t *testing.T) {
 	for _, year := range []int{1600, 1066, 1} {
 		want := time.Date(year, time.January, 1, 0, 0, 0, 0, time.UTC)
-		s := fmt.Sprintf("%.10f", float64(want.Unix()))
+		s := fmt.Sprint(time.Date(year, time.January, 1, 0, 0, 0, 0, time.UTC).UnixMicro())
 		got, err := convertBasicType(s, TimestampFieldType)
 		if err != nil {
 			t.Fatal(err)
@@ -114,44 +111,6 @@ func TestConvertSmallTimes(t *testing.T) {
 	}
 }
 
-func TestConvertTimePrecision(t *testing.T) {
-	tcs := []struct {
-		// Internally, BigQuery stores timestamps as microsecond-precision
-		// floats.
-		bq   float64
-		want time.Time
-	}{
-		{
-			bq:   1555593697.154358,
-			want: time.Unix(1555593697, 154358*1000),
-		},
-		{
-			bq:   1555593697.154359,
-			want: time.Unix(1555593697, 154359*1000),
-		},
-		{
-			bq:   1555593697.154360,
-			want: time.Unix(1555593697, 154360*1000),
-		},
-	}
-	for _, tc := range tcs {
-		bqS := fmt.Sprintf("%.6f", tc.bq)
-		t.Run(bqS, func(t *testing.T) {
-			got, err := convertBasicType(bqS, TimestampFieldType)
-			if err != nil {
-				t.Fatalf("convertBasicType failed: %v", err)
-			}
-			gotT, ok := got.(time.Time)
-			if !ok {
-				t.Fatalf("got a %T from convertBasicType, want a time.Time; got = %v", got, got)
-			}
-			if !gotT.Equal(tc.want) {
-				t.Errorf("got %v from convertBasicType, want %v", gotT, tc.want)
-			}
-		})
-	}
-}
-
 func TestConvertNullValues(t *testing.T) {
 	schema := Schema{{Type: StringFieldType}}
 	row := &bq.TableRow{