Struct BigQuerySink 

pub struct BigQuerySink { /* private fields */ }

Available on crate feature sink-bigquery only.

A sink that writes JSON records to a BigQuery table using the streaming insert API (tabledata.insertAll).

Implementations

impl BigQuerySink

pub async fn new( config: BigQuerySinkConfig, ) -> Result<BigQuerySink, FaucetError>

Create a new BigQuery sink from the given configuration.

This initialises the BigQuery client and authenticates with GCP. Returns a FaucetError::Auth if authentication fails.

Trait Implementations

impl Sink for BigQuerySink

fn supports_staged_load(&self) -> bool

BigQuery bulk-loads via a GCS-staged Parquet load job under bulk_load (#528). Advertise the staging capability.

fn check<'life0, 'life1, 'async_trait>( &'life0 self, ctx: &'life1 CheckContext, ) -> Pin<Box<dyn Future<Output = Result<CheckReport, FaucetError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, BigQuerySink: 'async_trait,

Preflight check (faucet doctor).

Runs a single read-only tables.get against the configured project_id.dataset_id.table_id using the already-authenticated client built in BigQuerySink::new. This mints/uses the access token and confirms the credentials can read the target table’s metadata — without inserting any rows. Auth, missing-dataset, missing-table, and permission errors all surface as a Fail probe with a remediation hint. The access token is never included in the reason or hint.

fn write_batch<'life0, 'life1, 'async_trait>( &'life0 self, records: &'life1 [Value], ) -> Pin<Box<dyn Future<Output = Result<usize, FaucetError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, BigQuerySink: 'async_trait,

Write records to BigQuery.

When config.batch_size > 0 and the input slice is larger than batch_size, the slice is split into chunks of batch_size rows and each chunk is sent as a separate tabledata.insertAll call. When config.batch_size == 0, the entire slice is sent in a single insertAll request — useful when upstream StreamPages are already sized for BigQuery’s per-request limits.

fn write_batch_partial<'life0, 'life1, 'async_trait>( &'life0 self, records: &'life1 [Value], ) -> Pin<Box<dyn Future<Output = Result<Vec<Result<(), FaucetError>>, FaucetError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, BigQuerySink: 'async_trait,

Write records to BigQuery, returning a per-row outcome vector.

Unlike write_batch, which collapses all insertErrors into a single FaucetError, this method maps each row to Ok(()) if BigQuery accepted it or Err(FaucetError::Sink(...)) if BigQuery reported a per-row error for it. This allows the pipeline’s DLQ router to quarantine only the rows that BigQuery actually rejected while keeping already-committed siblings out of the dead-letter queue.

Transport-level or HTTP-level failures (e.g. network errors, 4xx/5xx responses) are still returned as an outer Err because no rows can be considered committed in that case.

Chunking follows the same batch_size semantics as write_batch: batch_size == 0 sends the entire slice in one call; batch_size > 0 splits the slice into chunks and concatenates the per-row outcomes in input order.

fn begin_overwrite<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<(), FaucetError>> + Send + 'async_trait>>

where 'life0: 'async_trait, BigQuerySink: 'async_trait,

Create the staging table as an empty structural clone of the target (CREATE OR REPLACE TABLE temp LIKE target, which also copies partitioning/clustering), dropping any leftover staging from a crashed run. Bucket-free: no GCS staging is involved. The target must already exist (LIKE requires it) — overwrite replaces its rows, not its schema.

fn commit_overwrite<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<(), FaucetError>> + Send + 'async_trait>>

where 'life0: 'async_trait, BigQuerySink: 'async_trait,

Atomically replace the destination in one BigQuery multi-statement transaction — TRUNCATE TABLE target; INSERT INTO target SELECT * FROM temp; — so a failure rolls back and the prior rows survive. TRUNCATE+INSERT (rather than CREATE OR REPLACE … AS SELECT) preserves the target’s own partitioning, clustering, and description. The staging table is dropped afterwards. Staging was loaded via the query path, so there is no streaming buffer to miss.

fn abort_overwrite<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<(), FaucetError>> + Send + 'async_trait>>

where 'life0: 'async_trait, BigQuerySink: 'async_trait,

Drop the staging table so a failed/cancelled overwrite leaves nothing behind. Best-effort — the destination was never touched.

fn supports_cleanup(&self) -> bool

Scoped cleanup is always available: a BigQuery table’s scope and key predicates address real columns, and the exactly-once/upsert paths already require the target table to carry a defined schema.

fn last_committed_token<'life0, 'life1, 'async_trait>( &'life0 self, scope: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<String>, FaucetError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, BigQuerySink: 'async_trait,

Read the last durably-committed token for scope from the watermark table, so the pipeline can skip already-committed pages on resume.

fn write_batch_idempotent<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, records: &'life1 [Value], scope: &'life2 str, token: &'life3 str, ) -> Pin<Box<dyn Future<Output = Result<usize, FaucetError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait, BigQuerySink: 'async_trait,

Atomically write records and record token for scope in one BigQuery multi-statement transaction: a typed INSERT … SELECT FROM UNNEST(JSON_QUERY_ARRAY(@payload)) plus a watermark MERGE. Either both the rows and the token commit, or neither does — so a crash/resume skips the already-committed page (zero duplicates) and a failed page replays cleanly.

The entire page is one atomic unit (no batch_size re-chunking — core issues exactly one token per page), so the page must serialize within BigQuery’s ~10 MB jobs.query request limit.

fn current_schema<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Option<Value>, FaucetError>> + Send + 'async_trait>>

where 'life0: 'async_trait, BigQuerySink: 'async_trait,

Read the live destination schema via a schema-only tables.get, mapped to an infer_schema-shaped object so the drift policy can diff a page against the real table.

Returns Ok(None) when the target table does not exist yet (404) or carries no field definitions — both mean “no schema to diff against”, so the drift pass treats every page column as new.

fn evolve_schema<'life0, 'life1, 'async_trait>( &'life0 self, evolution: &'life1 SchemaEvolution, ) -> Pin<Box<dyn Future<Output = Result<(), FaucetError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, BigQuerySink: 'async_trait,

Apply an additive schema evolution to the target table via ALTER TABLE DDL (issue #194):

• additions → ADD COLUMN IF NOT EXISTS <col> <type>
• widenings → ALTER COLUMN <col> SET DATA TYPE <type>
• nullability relaxations → ALTER COLUMN <col> DROP NOT NULL

Each statement runs as its own jobs.query job, verified to completion via the authoritative job-status check. Every statement is idempotent so concurrent runs converge. The cached schema is invalidated afterwards so the next page re-fetches the evolved table.

fn supports_columnar(&self) -> bool

Available on crate feature arrow only.

Columnar load-job is available only when a bulk_load staging config is set and the write mode is append (#380). Load jobs are append/truncate only; upsert/delete stay on the Value MERGE path, so the pipeline never negotiates the columnar loop for them.

fn write_batch_columnar<'life0, 'life1, 'async_trait>( &'life0 self, batch: &'life1 RecordBatch, ) -> Pin<Box<dyn Future<Output = Result<usize, FaucetError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, BigQuerySink: 'async_trait,

Available on crate feature arrow only.

Write one Arrow RecordBatch by encoding it to Parquet, staging it on GCS, and running a BigQuery PARQUET load job to completion. Append-only. The body lives in load.rs (pure cloud I/O — a GCS-SDK staging upload + live load job — that can’t run in CI, so codecov.yml excludes that file exactly as it does the GCS connectors).

fn connector_name(&self) -> &'static str

Stable identifier used as the connector label on metrics and the connector attribute on spans. See Source::connector_name.

fn config_schema(&self) -> Value

Return a JSON Schema describing the configuration this sink accepts.

fn dataset_uri(&self) -> String

Logical dataset identity for lineage emission, following OpenLineage naming conventions (https://openlineage.io/docs/spec/naming).

fn supported_write_modes(&self) -> &'static [WriteMode]

Write modes this sink can apply. Default: append-only. Sinks that implement key-based merge override this to include WriteMode::Upsert / WriteMode::Delete. The CLI rejects a configured mode that is not in this set at config-load time.

fn is_overwrite(&self) -> bool

Whether this sink instance is configured for full-destination replacement (WriteMode::Overwrite).

fn dedups_by_key(&self) -> bool

Whether this sink instance is configured to dedup by key — i.e. write_mode: upsert (or delete) with a non-empty key, so re-applying a record with the same key converges instead of duplicating. Default: false.

fn cleanup_scope<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, scope: &'life1 BTreeMap<String, Value>, seen: &'life2 SeenKeys, ) -> Pin<Box<dyn Future<Output = Result<u64, FaucetError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, BigQuerySink: 'async_trait,

Delete rows matching scope whose key is not in seen.

fn supports_idempotent_writes(&self) -> bool

Whether this sink can durably commit a page’s rows and a commit token in a single atomic transaction. Default: false (at-least-once only).

fn supports_schema_evolution(&self) -> bool

Whether this sink can apply additive/widening DDL via evolve_schema. Default: false. The CLI rejects on_drift: evolve against a sink that returns false at config-load.

fn flush<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<(), FaucetError>> + Send + 'async_trait>>

where 'life0: 'async_trait, Self: 'async_trait,

Flush any buffered data to the destination.

fn sink_guarantee(&self) -> SinkGuarantee

The strongest delivery guarantee this sink can uphold — see SinkGuarantee.