bug.go

func (b *bigqueryGithubDataSink) executeQuery(queryStr string, cb func([]bigquery.Value, error) error) (context.CancelFunc, error) {
	if !b.isInitialized() {
		return nil, ErrDataSinkNotInitialized
	}

	transactionContext, cancelFunc := context.WithTimeout(b.clientContext, b.opts.ReadTransactionTimeout)
	query := b.client.Query(queryStr)
	rowIterator, err := query.Read(transactionContext)
	if err != nil {
		return cancelFunc, err
	}
	b.addCancelFunc(cancelFunc)
	defer cancelFunc()

	go func(it *bigquery.RowIterator) {
		defer cancelFunc()

		for {
			var values []bigquery.Value
			err := it.Next(&values)

			if cbErr := cb(values, err); cbErr != nil {
				return
			}

			if err == iterator.Done {
				return
			}
		}
	}(rowIterator)
	return cancelFunc, nil
}