Port over the commits from the legacy bigquery go driver, rebase the latest upstream main

go/bigquery_database.go

@@ -38,12 +38,21 @@ import (
 type databaseImpl struct {
 	driverbase.DatabaseImplBase
 
-	authType        string
-	credentialsType option.CredentialsType
-	credentials     string
-	clientID        string
-	clientSecret    string
-	refreshToken    string
+	authType              string
+	credentialsType       option.CredentialsType
+	credentials           string
+	clientID              string
+	clientSecret          string
+	refreshToken          string
+	accessToken           string
+	accessTokenEndpoint   string
+	accessTokenServerName string
+
+	// External-account (Workload Identity Federation) options.
+	externalAccountAudience         string
+	externalAccountImpersonationURL string
+	externalAccountRequestURL       string
+	externalAccountRequestData      string
 
 	impersonateTargetPrincipal string
 	impersonateDelegates       []string
@@ -65,27 +74,35 @@ type databaseImpl struct {
 
 func (d *databaseImpl) Open(ctx context.Context) (adbc.ConnectionWithContext, error) {
 	conn := &connectionImpl{
-		ConnectionImplBase:         driverbase.NewConnectionImplBase(&d.DatabaseImplBase),
-		authType:                   d.authType,
-		credentialsType:            d.credentialsType,
-		credentials:                d.credentials,
-		clientID:                   d.clientID,
-		clientSecret:               d.clientSecret,
-		refreshToken:               d.refreshToken,
-		impersonateTargetPrincipal: d.impersonateTargetPrincipal,
-		impersonateDelegates:       d.impersonateDelegates,
-		impersonateScopes:          d.impersonateScopes,
-		impersonateLifetime:        d.impersonateLifetime,
-		catalog:                    d.projectID,
-		dbSchema:                   d.datasetID,
-		location:                   d.location,
-		endpoint:                   d.endpoint,
-		storageEndpoint:            d.storageEndpoint,
-		resultRecordBufferSize:     defaultQueryResultBufferSize,
-		prefetchConcurrency:        defaultQueryPrefetchConcurrency,
-		quotaProject:               d.quotaProject,
-		bulkIngestMethod:           d.bulkIngestMethod,
-		bulkIngestCompression:      d.bulkIngestCompression,
+		ConnectionImplBase:    driverbase.NewConnectionImplBase(&d.DatabaseImplBase),
+		authType:              d.authType,
+		credentialsType:       d.credentialsType,
+		credentials:           d.credentials,
+		clientID:              d.clientID,
+		clientSecret:          d.clientSecret,
+		refreshToken:          d.refreshToken,
+		accessToken:           d.accessToken,
+		accessTokenEndpoint:   d.accessTokenEndpoint,
+		accessTokenServerName: d.accessTokenServerName,
+
+		externalAccountAudience:         d.externalAccountAudience,
+		externalAccountImpersonationURL: d.externalAccountImpersonationURL,
+		externalAccountRequestURL:       d.externalAccountRequestURL,
+		externalAccountRequestData:      d.externalAccountRequestData,
+		impersonateTargetPrincipal:      d.impersonateTargetPrincipal,
+		impersonateDelegates:            d.impersonateDelegates,
+		impersonateScopes:               d.impersonateScopes,
+		impersonateLifetime:             d.impersonateLifetime,
+		catalog:                         d.projectID,
+		dbSchema:                        d.datasetID,
+		location:                        d.location,
+		endpoint:                        d.endpoint,
+		storageEndpoint:                 d.storageEndpoint,
+		resultRecordBufferSize:          defaultQueryResultBufferSize,
+		prefetchConcurrency:             defaultQueryPrefetchConcurrency,
+		quotaProject:                    d.quotaProject,
+		bulkIngestMethod:                d.bulkIngestMethod,
+		bulkIngestCompression:           d.bulkIngestCompression,
 	}
 
 	err := conn.newClient(ctx)
@@ -118,6 +135,20 @@ func (d *databaseImpl) GetOption(ctx context.Context, key string) (string, error
 		return d.clientSecret, nil
 	case OptionAuthRefreshToken:
 		return d.refreshToken, nil
+	case OptionStringAuthAccessToken:
+		return d.accessToken, nil
+	case OptionStringAuthAccessTokenEndpoint:
+		return d.accessTokenEndpoint, nil
+	case OptionStringAuthAccessTokenServerName:
+		return d.accessTokenServerName, nil
+	case OptionStringAuthExternalAccountAudience:
+		return d.externalAccountAudience, nil
+	case OptionStringAuthExternalAccountImpersonationURL:
+		return d.externalAccountImpersonationURL, nil
+	case OptionStringAuthExternalAccountRequestURL:
+		return d.externalAccountRequestURL, nil
+	case OptionStringAuthExternalAccountRequestData:
+		return d.externalAccountRequestData, nil
 	case OptionAuthQuotaProject:
 		return d.quotaProject, nil
 	case OptionLocation:
@@ -126,7 +157,7 @@ func (d *databaseImpl) GetOption(ctx context.Context, key string) (string, error
 		return d.projectID, nil
 	case OptionDatasetID:
 		return d.datasetID, nil
-	case OptionEndpoint:
+	case OptionEndpoint, OptionStringAPIEndpoint:
 		return d.endpoint, nil
 	case OptionStorageEndpoint:
 		return d.storageEndpoint, nil
@@ -176,9 +207,12 @@ func (d *databaseImpl) SetOptions(ctx context.Context, options map[string]string
 }
 
 func (d *databaseImpl) hasImpersonationOptions() bool {
+	// Scopes are deliberately not counted: they are base auth scopes that
+	// callers (e.g. dbt-auth) always send, so counting them would make
+	// GetOption report an impersonation lifetime when none is configured.
+	// FIXME: separate base scopes from impersonate scopes, then fix this divergence from the upstream
 	return d.impersonateTargetPrincipal != "" ||
-		len(d.impersonateDelegates) > 0 ||
-		len(d.impersonateScopes) > 0
+		len(d.impersonateDelegates) > 0
 }
 
 func (d *databaseImpl) SetOption(ctx context.Context, key string, value string) error {
@@ -204,7 +238,9 @@ func (d *databaseImpl) SetOption(ctx context.Context, key string, value string)
 			OptionValueAuthTypeJSONCredentialString,
 			OptionValueAuthTypeUserAuthentication,
 			OptionValueAuthTypeAppDefaultCredentials,
-			OptionValueAuthTypeAnonymous:
+			OptionValueAuthTypeAnonymous,
+			OptionValueAuthTypeExternalAccount,
+			OptionValueAuthTypeTemporaryAccessToken:
 			d.authType = value
 		default:
 			return adbc.Error{
@@ -238,14 +274,35 @@ func (d *databaseImpl) SetOption(ctx context.Context, key string, value string)
 		d.clientSecret = value
 	case OptionAuthRefreshToken:
 		d.refreshToken = value
+	case OptionStringAuthAccessToken:
+		d.accessToken = value
+	case OptionStringAuthAccessTokenEndpoint:
+		d.accessTokenEndpoint = value
+	case OptionStringAuthAccessTokenServerName:
+		d.accessTokenServerName = value
+	case OptionStringAuthExternalAccountAudience:
+		d.externalAccountAudience = value
+	case OptionStringAuthExternalAccountImpersonationURL:
+		d.externalAccountImpersonationURL = value
+	case OptionStringAuthExternalAccountRequestURL:
+		d.externalAccountRequestURL = value
+	case OptionStringAuthExternalAccountRequestData:
+		d.externalAccountRequestData = value
 	case OptionAuthQuotaProject:
 		d.quotaProject = value
 	case OptionImpersonateTargetPrincipal:
 		d.impersonateTargetPrincipal = value
 	case OptionImpersonateDelegates:
-		d.impersonateDelegates = strings.Split(value, ",")
+		// Guard against strings.Split("", ",") yielding [""]. These values are
+		// copied into the connection by Open, so an unguarded split here would
+		// defeat the same guard on the connection side.
+		if value != "" {
+			d.impersonateDelegates = strings.Split(value, ",")
+		}
 	case OptionImpersonateScopes:
-		d.impersonateScopes = strings.Split(value, ",")
+		if value != "" {
+			d.impersonateScopes = strings.Split(value, ",")
+		}
 	case OptionImpersonateLifetime:
 		duration, err := time.ParseDuration(value)
 		if err != nil {
@@ -259,7 +316,7 @@ func (d *databaseImpl) SetOption(ctx context.Context, key string, value string)
 		d.projectID = value
 	case OptionDatasetID:
 		d.datasetID = value
-	case OptionEndpoint:
+	case OptionEndpoint, OptionStringAPIEndpoint:
 		d.endpoint = value
 	case OptionStorageEndpoint:
 		d.storageEndpoint = value

go/bulk_ingest.go

@@ -21,6 +21,7 @@ import (
 	"io"
 	"log/slog"
 	"os"
+	"strings"
 
 	"cloud.google.com/go/bigquery"
 	"github.com/adbc-drivers/driverbase-go/driverbase"
@@ -29,6 +30,53 @@ import (
 	"github.com/google/uuid"
 )
 
+// bigQueryFieldTypeFromMetadata maps Arrow types stashed in IPC metadata
+// (BIGQUERY:type) to BigQuery types. These do not align 1:1 with the
+// Standard SQL type names — see google-cloud-go/bigquery/schema.go.
+func bigQueryFieldTypeFromMetadata(v string) (bigquery.FieldType, bool) {
+	// Trim parameter suffixes: e.g. NUMERIC(38,9) or ARRAY<INT64>.
+	t := strings.ToUpper(strings.TrimSpace(v))
+	if idx := strings.IndexAny(t, "<("); idx >= 0 {
+		t = t[:idx]
+	}
+
+	switch t {
+	case "STRING":
+		return bigquery.StringFieldType, true
+	case "BYTES":
+		return bigquery.BytesFieldType, true
+	case "INTEGER", "INT64":
+		return bigquery.IntegerFieldType, true
+	case "FLOAT", "FLOAT64":
+		return bigquery.FloatFieldType, true
+	case "BOOL", "BOOLEAN":
+		return bigquery.BooleanFieldType, true
+	case "TIMESTAMP":
+		return bigquery.TimestampFieldType, true
+	case "DATE":
+		return bigquery.DateFieldType, true
+	case "TIME":
+		return bigquery.TimeFieldType, true
+	case "DATETIME":
+		return bigquery.DateTimeFieldType, true
+	case "NUMERIC", "DECIMAL":
+		return bigquery.NumericFieldType, true
+	case "BIGNUMERIC", "BIGDECIMAL":
+		return bigquery.BigNumericFieldType, true
+	case "GEOGRAPHY":
+		return bigquery.GeographyFieldType, true
+	case "INTERVAL":
+		return bigquery.IntervalFieldType, true
+	case "JSON":
+		return bigquery.JSONFieldType, true
+	case "ARRAY", "RECORD", "STRUCT":
+		// composite; the caller must derive from the Arrow field type.
+		return "", false
+	default:
+		return bigquery.StringFieldType, true
+	}
+}
+
 type bigqueryBulkIngestSink struct {
 	f    *os.File
 	path string
@@ -186,6 +234,16 @@ func arrowFieldToBigQueryField(field arrow.Field) (*bigquery.FieldSchema, error)
 		Required: !field.Nullable,
 	}
 
+	// Use IPC metadata to disambiguate BigQuery logical types when Arrow's
+	// physical type alone is not enough (e.g. DATETIME vs TIMESTAMP,
+	// NUMERIC vs BIGNUMERIC precision, JSON strings vs plain STRING).
+	if v, ok := field.Metadata.GetValue("BIGQUERY:type"); ok {
+		if fieldType, ok := bigQueryFieldTypeFromMetadata(v); ok {
+			bqField.Type = fieldType
+			return bqField, nil
+		}
+	}
+
 	switch field.Type.ID() {
 	case arrow.BINARY, arrow.LARGE_BINARY, arrow.BINARY_VIEW, arrow.FIXED_SIZE_BINARY:
 		bqField.Type = bigquery.BytesFieldType