bigquery: add gcp_bigquery_write_api enterprise output

New BatchOutput backed by the BigQuery Storage Write API for higher
throughput and lower latency than the legacy streaming API.

- JSON or protobuf input with auto schema fetch + proto conversion
- Per-table managed stream cache with idle eviction sweeper
- gRPC error classification: permanent failures returned as BatchError
- Service-account impersonation via target_principal + delegates
- Configurable stream idle timeout and sweep interval
- Endpoint overrides for local emulators
- Prometheus-style _total counter metrics + batch latency timer

Author: squiidz

docs/modules/components/pages/outputs/gcp_bigquery_write_api.adoc

@@ -0,0 +1,332 @@
+= gcp_bigquery_write_api
+:type: output
+:status: stable
+:categories: ["GCP","Services"]
+
+
+
+////
+     THIS FILE IS AUTOGENERATED!
+
+     To make changes, edit the corresponding source file under:
+
+     https://github.com/redpanda-data/connect/tree/main/internal/impl/<provider>.
+
+     And:
+
+     https://github.com/redpanda-data/connect/tree/main/cmd/tools/docs_gen/templates/plugin.adoc.tmpl
+////
+
+// © 2026 Redpanda Data Inc.
+
+
+component_type_dropdown::[]
+
+
+Streams data into BigQuery using the Storage Write API.
+
+Introduced in version 4.90.0.
+
+
+[tabs]
+======
+Common::
++
+--
+
+```yml
+# Common config fields, showing default values
+output:
+  label: ""
+  gcp_bigquery_write_api:
+    project: ""
+    dataset: "" # No default (required)
+    table: "" # No default (required)
+    message_format: json
+    max_in_flight: 64
+    batching:
+      count: 0
+      byte_size: 0
+      period: ""
+      check: ""
+    credentials_json: ""
+```
+
+--
+Advanced::
++
+--
+
+```yml
+# All config fields, showing default values
+output:
+  label: ""
+  gcp_bigquery_write_api:
+    project: ""
+    dataset: "" # No default (required)
+    table: "" # No default (required)
+    message_format: json
+    max_in_flight: 64
+    batching:
+      count: 0
+      byte_size: 0
+      period: ""
+      check: ""
+      processors: [] # No default (optional)
+    credentials_json: ""
+    target_principal: ""
+    delegates: []
+    stream_idle_timeout: 5m
+    stream_sweep_interval: 1m
+    endpoint:
+      http: ""
+      grpc: ""
+```
+
+--
+======
+
+Writes messages to a BigQuery table using the Storage Write API.
+This provides higher throughput and lower latency than the legacy streaming API or load jobs.
+
+Messages can be formatted as JSON (default) or raw protobuf bytes.
+When using JSON format the component automatically fetches the table schema and converts each message to the corresponding proto representation.
+
+WARNING: The proto3 JSON mapping encodes int64 and uint64 values as strings.
+JSON messages with integer fields must use string values (e.g. `"age": "30"` not `"age": 30`).
+Otherwise the write will fail with an unmarshalling error.
+
+When batching is enabled the table name is resolved from the first message in each batch.
+All messages in the same batch are written to that table.
+
+
+== Fields
+
+=== `project`
+
+The GCP project ID. If empty, the project is auto-detected from the environment.
+
+
+*Type*: `string`
+
+*Default*: `""`
+
+=== `dataset`
+
+The BigQuery dataset ID.
+
+
+*Type*: `string`
+
+
+=== `table`
+
+The BigQuery table ID. Supports interpolation functions. When batching, resolved from the first message in each batch.
+This field supports xref:configuration:interpolation.adoc#bloblang-queries[interpolation functions].
+
+
+*Type*: `string`
+
+
+=== `message_format`
+
+The format of input messages. Use 'json' to have the component convert JSON to proto automatically. Use 'protobuf' to supply raw proto-encoded bytes.
+
+
+*Type*: `string`
+
+*Default*: `"json"`
+
+Options:
+`json`
+, `protobuf`
+.
+
+=== `max_in_flight`
+
+The maximum number of messages to have in flight at a given time. Increase this to improve throughput.
+
+
+*Type*: `int`
+
+*Default*: `64`
+
+=== `batching`
+
+Allows you to configure a xref:configuration:batching.adoc[batching policy].
+
+
+*Type*: `object`
+
+
+```yml
+# Examples
+
+batching:
+  byte_size: 5000
+  count: 0
+  period: 1s
+
+batching:
+  count: 10
+  period: 1s
+
+batching:
+  check: this.contains("END BATCH")
+  count: 0
+  period: 1m
+```
+
+=== `batching.count`
+
+A number of messages at which the batch should be flushed. If `0` disables count based batching.
+
+
+*Type*: `int`
+
+*Default*: `0`
+
+=== `batching.byte_size`
+
+An amount of bytes at which the batch should be flushed. If `0` disables size based batching.
+
+
+*Type*: `int`
+
+*Default*: `0`
+
+=== `batching.period`
+
+A period in which an incomplete batch should be flushed regardless of its size.
+
+
+*Type*: `string`
+
+*Default*: `""`
+
+```yml
+# Examples
+
+period: 1s
+
+period: 1m
+
+period: 500ms
+```
+
+=== `batching.check`
+
+A xref:guides:bloblang/about.adoc[Bloblang query] that should return a boolean value indicating whether a message should end a batch.
+
+
+*Type*: `string`
+
+*Default*: `""`
+
+```yml
+# Examples
+
+check: this.type == "end_of_transaction"
+```
+
+=== `batching.processors`
+
+A list of xref:components:processors/about.adoc[processors] to apply to a batch as it is flushed. This allows you to aggregate and archive the batch however you see fit. Please note that all resulting messages are flushed as a single batch, therefore splitting the batch into smaller batches using these processors is a no-op.
+
+
+*Type*: `array`
+
+
+```yml
+# Examples
+
+processors:
+  - archive:
+      format: concatenate
+
+processors:
+  - archive:
+      format: lines
+
+processors:
+  - archive:
+      format: json_array
+```
+
+=== `credentials_json`
+
+An optional JSON string containing GCP credentials. If empty, credentials are loaded from the environment.
+[CAUTION]
+====
+This field contains sensitive information that usually shouldn't be added to a config directly, read our xref:configuration:secrets.adoc[secrets page for more info].
+====
+
+
+
+*Type*: `string`
+
+*Default*: `""`
+
+=== `target_principal`
+
+Service account email to impersonate. When set, the output obtains tokens acting as this service account. Requires the caller to have roles/iam.serviceAccountTokenCreator on the target.
+
+
+*Type*: `string`
+
+*Default*: `""`
+
+=== `delegates`
+
+Optional delegation chain for chained service account impersonation. Each service account must be granted roles/iam.serviceAccountTokenCreator on the next in the chain.
+
+
+*Type*: `array`
+
+*Default*: `[]`
+
+=== `stream_idle_timeout`
+
+How long a cached stream can remain unused before being closed. Relevant when the table field uses interpolation to route to many tables.
+
+
+*Type*: `string`
+
+*Default*: `"5m"`
+
+=== `stream_sweep_interval`
+
+How often to check for idle streams to close.
+
+
+*Type*: `string`
+
+*Default*: `"1m"`
+
+=== `endpoint`
+
+Optional endpoint overrides for the BigQuery and Storage Write API clients.
+
+
+*Type*: `object`
+
+
+=== `endpoint.http`
+
+Override the BigQuery HTTP endpoint. Useful for local emulators.
+
+
+*Type*: `string`
+
+*Default*: `""`
+
+=== `endpoint.grpc`
+
+Override the BigQuery Storage gRPC endpoint. Useful for local emulators.
+
+
+*Type*: `string`
+
+*Default*: `""`
+
+

go.mod

@@ -24,7 +24,7 @@ require (
 	buf.build/gen/go/redpandadata/otel/protocolbuffers/go v1.36.11-20260323171043-3635d3966b23.1
 	buf.build/go/hyperpb v0.1.3
 	cloud.google.com/go/aiplatform v1.121.0
-	cloud.google.com/go/bigquery v1.74.0
+	cloud.google.com/go/bigquery v1.75.0
 	cloud.google.com/go/pubsub v1.50.1
 	cloud.google.com/go/spanner v1.88.0
 	cloud.google.com/go/storage v1.62.1

go.sum

@@ -72,8 +72,8 @@ cloud.google.com/go/bigquery v1.4.0/go.mod h1:S8dzgnTigyfTmLBfrtrhyYhwRxG72rYxvf
 cloud.google.com/go/bigquery v1.5.0/go.mod h1:snEHRnqQbz117VIFhE8bmtwIDY80NLUZUMb4Nv6dBIg=
 cloud.google.com/go/bigquery v1.7.0/go.mod h1://okPTzCYNXSlb24MZs83e2Do+h+VXtc4gLoIoXIAPc=
 cloud.google.com/go/bigquery v1.8.0/go.mod h1:J5hqkt3O0uAFnINi6JXValWIb1v0goeZM77hZzJN/fQ=
-cloud.google.com/go/bigquery v1.74.0 h1:Q6bAMv+eyvufOpIrfrYxhM46qq1D3ZQTdgUDQqKS+n8=
-cloud.google.com/go/bigquery v1.74.0/go.mod h1:iViO7Cx3A/cRKcHNRsHB3yqGAMInFBswrE9Pxazsc90=
+cloud.google.com/go/bigquery v1.75.0 h1:gI4AgIhXNZ8hxvPDOp4hLGUnpNBjoBor6POSLcrdWkY=
+cloud.google.com/go/bigquery v1.75.0/go.mod h1:zNCHWok+hfTgKCwNqT+V7GH/YmFFgZqjzljKCZBJTWc=
 cloud.google.com/go/compute v0.1.0/go.mod h1:GAesmwr110a34z04OlxYkATPBEfVhkymfTBXtfbBFow=
 cloud.google.com/go/compute v1.2.0/go.mod h1:xlogom/6gr8RJGBe7nT2eGsQYAFUbbv8dbC29qE3Xmw=
 cloud.google.com/go/compute v1.3.0/go.mod h1:cCZiE1NHEtai4wiufUhW8I8S1JKkAnhnQJWM7YD99wM=