fix(bigquery-spec)!: Move batch_size from the plugin spec to the top level spec (#6092)

erezrokah · web-flow · commit a1d706e839d3 · 2022-12-29T11:37:47.000+02:00
#### Summary Same as #6090 not from a fork so tests can pass. Fixes #6070 With the SDK level batching the configuration that "takes" is now at the top spec level. Setting `batch_size` at the plugin level has no impact
diff --git a/plugins/destination/bigquery/client/client.go b/plugins/destination/bigquery/client/client.go
@@ -17,7 +17,6 @@ type Client struct {
 	logger     zerolog.Logger
 	spec       specs.Destination
 	metrics    destination.Metrics
-	batchSize  int
 	pluginSpec Spec
 	client     *bigquery.Client
 }
@@ -41,7 +40,6 @@ func New(ctx context.Context, logger zerolog.Logger, destSpec specs.Destination)
 	}
 
 	c.pluginSpec = spec
-	c.batchSize = spec.BatchSize
 
 	// the context here is used for token refresh so this is workaround as suggested
 	// https://github.com/googleapis/google-cloud-go/issues/946
diff --git a/plugins/destination/bigquery/client/spec.go b/plugins/destination/bigquery/client/spec.go
@@ -34,18 +34,12 @@ type Spec struct {
 	DatasetLocation       string                 `json:"dataset_location"`
 	TimePartitioning      TimePartitioningOption `json:"time_partitioning"`
 	ServiceAccountKeyJSON string                 `json:"service_account_key_json"`
-	BatchSize             int                    `json:"batch_size,omitempty"`
 }
 
-const defaultBatchSize = 1000
-
 func (s *Spec) SetDefaults() {
 	if s.TimePartitioning == "" {
 		s.TimePartitioning = TimePartitioningOptionNone
 	}
-	if s.BatchSize <= 0 {
-		s.BatchSize = defaultBatchSize
-	}
 }
 
 func (s *Spec) Validate() error {
diff --git a/plugins/destination/bigquery/client/write.go b/plugins/destination/bigquery/client/write.go
@@ -10,7 +10,6 @@ import (
 )
 
 const (
-	batchSize    = 1000
 	writeTimeout = 5 * time.Minute
 )
 
diff --git a/plugins/destination/bigquery/main.go b/plugins/destination/bigquery/main.go
@@ -12,6 +12,6 @@ const (
 )
 
 func main() {
-	p := destination.NewPlugin("bigquery", plugin.Version, client.New, destination.WithManagedWriter())
+	p := destination.NewPlugin("bigquery", plugin.Version, client.New, destination.WithManagedWriter(), destination.WithDefaultBatchSize(1000))
 	serve.Destination(p, serve.WithDestinationSentryDSN(sentryDSN))
 }
diff --git a/website/pages/docs/plugins/destinations/bigquery/overview.md b/website/pages/docs/plugins/destinations/bigquery/overview.md
@@ -90,10 +90,6 @@ This is the top-level spec used by the BigQuery destination plugin.
 
   GCP service account key content. This allows for using different service accounts for the GCP source and BigQuery destination. If using service account keys, it is best to use [environment or file variable substitution](/docs/advanced-topics/environment-variable-substitution).
 
-- `batch_size` (int, optional. Default: 1000)
-
-  Number of rows to insert in a single batch.
-
 ## Underlying library
 
 We use the official [cloud.google.com/go/bigquery](https://pkg.go.dev/cloud.google.com/go/bigquery) package for database connection.
diff --git a/website/pages/docs/reference/destination-spec.md b/website/pages/docs/reference/destination-spec.md
@@ -61,6 +61,14 @@ Specifies the update method to use when inserting rows. The exact semantics depe
 - `overwrite`: Same as `overwrite-delete-stale`, but doesn't delete stale rows from previous `sync`s.
 - `append`: Rows are never overwritten or deleted, only appended.
 
+{/*<!-- vale off -->*/}
+### batch_size
+{/*<!-- vale on -->*/}
+
+(`int`, optional)
+
+The number of resources to insert in a single batch. Only relevant for plugins that utilize batching. Each plugin has its own default value.
+
 ### spec
 
 (`object`, optional)

Original file line number	Diff line number	Diff line change
`@@ -17,7 +17,6 @@ type Client struct {`
`17`	`17`	`logger zerolog.Logger`
`18`	`18`	`spec specs.Destination`
`19`	`19`	`metrics destination.Metrics`
`20`		`- batchSize int`
`21`	`20`	`pluginSpec Spec`
`22`	`21`	`client *bigquery.Client`
`23`	`22`	`}`
`@@ -41,7 +40,6 @@ func New(ctx context.Context, logger zerolog.Logger, destSpec specs.Destination)`
`41`	`40`	`}`
`42`	`41`
`43`	`42`	`c.pluginSpec = spec`
`44`		`- c.batchSize = spec.BatchSize`
`45`	`43`
`46`	`44`	`// the context here is used for token refresh so this is workaround as suggested`
`47`	`45`	`// https://github.com/googleapis/google-cloud-go/issues/946`
Original file line number	Diff line number	Diff line change
`@@ -34,18 +34,12 @@ type Spec struct {`
`34`	`34`	DatasetLocation string `json:"dataset_location"`
`35`	`35`	TimePartitioning TimePartitioningOption `json:"time_partitioning"`
`36`	`36`	ServiceAccountKeyJSON string `json:"service_account_key_json"`
`37`		- BatchSize int `json:"batch_size,omitempty"`
`38`	`37`	`}`
`39`	`38`
`40`		`-const defaultBatchSize = 1000`
`41`		`-`
`42`	`39`	`func (s *Spec) SetDefaults() {`
`43`	`40`	`if s.TimePartitioning == "" {`
`44`	`41`	`s.TimePartitioning = TimePartitioningOptionNone`
`45`	`42`	`}`
`46`		`- if s.BatchSize <= 0 {`
`47`		`- s.BatchSize = defaultBatchSize`
`48`		`- }`
`49`	`43`	`}`
`50`	`44`
`51`	`45`	`func (s *Spec) Validate() error {`
Original file line number	Diff line number	Diff line change
`@@ -10,7 +10,6 @@ import (`
`10`	`10`	`)`
`11`	`11`
`12`	`12`	`const (`
`13`		`- batchSize = 1000`
`14`	`13`	`writeTimeout = 5 * time.Minute`
`15`	`14`	`)`
`16`	`15`
Original file line number	Diff line number	Diff line change
`@@ -12,6 +12,6 @@ const (`
`12`	`12`	`)`
`13`	`13`
`14`	`14`	`func main() {`
`15`		`- p := destination.NewPlugin("bigquery", plugin.Version, client.New, destination.WithManagedWriter())`
	`15`	`+ p := destination.NewPlugin("bigquery", plugin.Version, client.New, destination.WithManagedWriter(), destination.WithDefaultBatchSize(1000))`
`16`	`16`	`serve.Destination(p, serve.WithDestinationSentryDSN(sentryDSN))`
`17`	`17`	`}`