bigquery: alter retry strategy for storage sample, improve output
@@ -165,8 +165,10 @@ func printDatum(d interface{}) {
// Go's map implementation returns keys in a random ordering, so we sort
// the keys before accessing.
keys := make([]string, len(m))
+ i := 0
for k := range m {
- keys = append(keys, k)
+ keys[i] = k
+ i++
}
sort.Strings(keys)
for _, key := range keys {
@@ -197,9 +199,13 @@ func valueFromTypeMap(field interface{}) interface{} {
// successfully transmitted.
func processStream(ctx context.Context, client *bqStorage.BigQueryStorageClient, st *bqStoragepb.Stream, ch chan<- *bqStoragepb.AvroRows) error {
var offset int64
- streamRetry := 3
+
+ // Streams may be long-running. Rather than using a global retry for the
+ // stream, implement a retry that resets once progress is made.
+ retryLimit := 3
for {
+ retries := 0
// Send the initiating request to start streaming row blocks.
rowStream, err := client.ReadRows(ctx, &bqStoragepb.ReadRowsRequest{
ReadPosition: &bqStoragepb.StreamPosition{
@@ -217,8 +223,8 @@ func processStream(ctx context.Context, client *bqStorage.BigQueryStorageClient,
return nil
}
if err != nil {
- streamRetry--
- if streamRetry <= 0 {
+ retries++
+ if retries >= retryLimit {
return fmt.Errorf("processStream retries exhausted: %v", err)
}
}
@@ -227,6 +233,8 @@ func processStream(ctx context.Context, client *bqStorage.BigQueryStorageClient,
if rc > 0 {
// Bookmark our progress in case of retries and send the rowblock on the channel.
offset = offset + rc
+ // We're making progress, reset retries.
+ retries = 0
ch <- r.GetAvroRows()
}
}
