diff --git a/PORT_REVIEW.md b/PORT_REVIEW.md index c964a4f..709bf9d 100644 --- a/PORT_REVIEW.md +++ b/PORT_REVIEW.md @@ -1,3 +1,19 @@ + + # Port review: `dbt-labs/arrow-adbc` → `bigquery-adbc` Every commit in `dbt-labs/arrow-adbc` (fork) that isn't in diff --git a/go/csv_ingest.go b/go/csv_ingest.go index eb9acaa..3ea9bdb 100644 --- a/go/csv_ingest.go +++ b/go/csv_ingest.go @@ -50,7 +50,7 @@ func (st *statement) executeCSVIngest(ctx context.Context) (array.RecordReader, if err != nil { return nil, -1, fmt.Errorf("[bq] open %q: %w", st.ingestPath, err) } - defer file.Close() + defer func() { _ = file.Close() }() if st.queryConfig.Dst == nil { return nil, -1, adbc.Error{ diff --git a/go/driver.go b/go/driver.go index d971180..7808b3c 100644 --- a/go/driver.go +++ b/go/driver.go @@ -144,13 +144,6 @@ const ( // return null over the Storage Read API. OptionBoolUseStorageApiDisabledClient = "adbc.bigquery.sql.query.use_storage_api_disabled_client" - // ContextKeyUseStorageApiDisabledClient signals to runQuery via - // context that it should fall back to the row-based iterator - // (bigquery.RowIterator + rowsToArrowRecordBatch) instead of the - // Storage Read API, so pseudo-columns like _PARTITIONDATE return - // values instead of nulls. - ContextKeyUseStorageApiDisabledClient = "USE_STORAGE_API_DISABLED_CLIENT" - // TODO (harry): migrate the options with prefix "adbc.bigquery.sql" to use the newer format // TODO (harry): add the old option values to optionRemapping map for backward-compatibility // Copy table options — copy_table.source and copy_table.destination each accept @@ -258,6 +251,16 @@ const ( OptionValueCompressionZSTD = "zstd" ) +// contextKey namespaces context values set by this driver, so they can't +// collide with keys from other packages sharing the same context. +type contextKey string + +// ContextKeyUseStorageApiDisabledClient signals to runQuery via context that +// it should fall back to the row-based iterator (bigquery.RowIterator + +// rowsToArrowRecordBatch) instead of the Storage Read API, so pseudo-columns +// like _PARTITIONDATE return values instead of nulls. +const ContextKeyUseStorageApiDisabledClient contextKey = "USE_STORAGE_API_DISABLED_CLIENT" + var ( infoVendorVersion string diff --git a/go/python_models.go b/go/python_models.go index d7ac321..6836b55 100644 --- a/go/python_models.go +++ b/go/python_models.go @@ -87,7 +87,7 @@ func (st *statement) executeDataprocCreateBatch(ctx context.Context) (array.Reco if err != nil { return nil, -1, fmt.Errorf("[bq] create Dataproc client: %w", err) } - defer client.Close() + defer func() { _ = client.Close() }() req := &dataprocpb.CreateBatchRequest{ Parent: st.createBatchReqParent, @@ -113,7 +113,7 @@ func (st *statement) executeSubmitJobAsOperation(ctx context.Context) (array.Rec if err != nil { return nil, -1, fmt.Errorf("[bq] create Dataproc JobController client: %w", err) } - defer client.Close() + defer func() { _ = client.Close() }() req := &dataprocpb.SubmitJobRequest{ ProjectId: st.dataprocProject, @@ -194,7 +194,7 @@ func (c *connectionImpl) addExecutionIdentityDetails(ctx context.Context, job *a if err != nil { return nil, fmt.Errorf("build http transport: %w", err) } - var tokenSource oauth2.TokenSource = oauth2.StaticTokenSource(&oauth2.Token{}) + tokenSource := oauth2.StaticTokenSource(&oauth2.Token{}) if t, ok := ts.Transport.(*oauth2.Transport); ok { tokenSource = t.Source } @@ -211,7 +211,7 @@ func (c *connectionImpl) addExecutionIdentityDetails(ctx context.Context, job *a if err != nil { return nil, fmt.Errorf("call userinfo: %w", err) } - defer resp.Body.Close() + defer func() { _ = resp.Body.Close() }() body, err := io.ReadAll(resp.Body) if err != nil { return nil, err @@ -246,7 +246,7 @@ func (st *statement) getNotebookTemplateName(ctx context.Context) (string, error if err != nil { return "", fmt.Errorf("[bq] create notebook client: %w", err) } - defer client.Close() + defer func() { _ = client.Close() }() req := &aiplatformpb.ListNotebookRuntimeTemplatesRequest{ Parent: st.createNotebookExecuteJobParent, @@ -289,7 +289,7 @@ func (st *statement) executeCreateNotebookExecutionJob(ctx context.Context) (arr if err != nil { return nil, -1, fmt.Errorf("[bq] create notebook client: %w", err) } - defer client.Close() + defer func() { _ = client.Close() }() templateName := "" if st.createNotebookExecuteJobTemplateId != "" { @@ -334,11 +334,11 @@ func (st *statement) executeCreateNotebookExecutionJob(ctx context.Context) (arr waitCtx, cancel := context.WithTimeout(ctx, time.Duration(st.dataprocPoolingTimeout)*time.Second) defer cancel() - retrievedJob, err := op.Wait(waitCtx) - if err != nil { + if _, err = op.Wait(waitCtx); err != nil { return nil, -1, fmt.Errorf("[bq] notebook execution op: %w", err) } + var retrievedJob *aiplatformpb.NotebookExecutionJob elapsed := time.Duration(0) for { retrievedJob, err = client.GetNotebookExecutionJob(ctx, &aiplatformpb.GetNotebookExecutionJobRequest{Name: jobName}) @@ -366,7 +366,7 @@ func (st *statement) executeCreateNotebookExecutionJob(ctx context.Context) (arr gcsClient, err := st.cnxn.newGCSClient(ctx) if err == nil { - defer gcsClient.Close() + defer func() { _ = gcsClient.Close() }() if data, err := readJSONFromGCS(ctx, gcsLogURI, gcsClient); err == nil && data != nil { processGCSNotebookLog(st.cnxn.Logger, data) } @@ -392,7 +392,7 @@ func readJSONFromGCS(ctx context.Context, gcsURI string, storageClient *storage. if err != nil { return nil, err } - defer reader.Close() + defer func() { _ = reader.Close() }() content, err := io.ReadAll(reader) if err != nil { return nil, err @@ -443,7 +443,7 @@ func (st *statement) writeToGCS(ctx context.Context) (array.RecordReader, int64, if err != nil { return nil, -1, fmt.Errorf("[bq] create GCS client: %w", err) } - defer client.Close() + defer func() { _ = client.Close() }() wc := client.Bucket(st.writeGCSBucket).Object(st.writeGCSObjectName).NewWriter(ctx) if _, err := wc.Write([]byte(st.writeGCSContent)); err != nil { diff --git a/go/row_based_iterator.go b/go/row_based_iterator.go index 912123e..52c8e59 100644 --- a/go/row_based_iterator.go +++ b/go/row_based_iterator.go @@ -66,7 +66,7 @@ func (l *RowBasedArrowIterator) Next() (*bigquery.ArrowRecordBatch, error) { const batchSize = 1000 rows := make([][]bigquery.Value, 0, batchSize) - for i := 0; i < batchSize; i++ { + for range batchSize { var row []bigquery.Value err := l.iter.Next(&row) if err == iterator.Done {