faucet-source-bigquery

Google BigQuery query source for the faucet-stream ecosystem. Runs a SQL statement against BigQuery's REST API (jobs.query + jobs.getQueryResults), decodes each row into a typed serde_json::Value, and streams the result set back page-by-page so memory stays bounded no matter how large the query result is.

Reach for it when you want to pull analytics tables, aggregates, or ad-hoc query results out of BigQuery and land them in any faucet-stream sink — a file, a database, a warehouse, a queue — with one declarative config and no glue code.

Feature highlights

• Three credential modes — Application Default Credentials, a service-account key file, or inline service-account JSON. The shared BigQueryCredentials enum is re-exported from faucet-common-bigquery so it matches the BigQuery sink byte-for-byte.
• Server-side pagination — pages through arbitrarily large result sets via pageToken; rows are re-framed into pages of batch_size for streaming, so peak memory is O(batch_size).
• Async-job aware — when BigQuery returns jobComplete=false (the statement ran past statement_timeout), the source polls jobs.getQueryResults until the job finishes, bounded by poll_timeout so a job that never completes fails cleanly instead of hanging forever.
• Type-aware row decoding — INTEGER/INT64 → JSON number, FLOAT/FLOAT64 → number, BOOLEAN/BOOL → bool, NUMERIC/BIGNUMERIC → string (full precision preserved), RECORD/STRUCT → nested object, REPEATED → array, JSON → parsed value, everything else → string.
• Positional bind parameters — params from config plus any ${parent.path} matrix-context values are sent as BigQuery POSITIONAL parameters, typed from the JSON value (INT64 / FLOAT64 / BOOL / STRING) so a numeric or boolean bind compares correctly against a typed column.
• Standard or legacy SQL — use_legacy_sql toggle; defaults to Standard SQL.
• Client built once — the authenticated BigQuery client is constructed in new() and reused for every request.

Installation

# As a library:
cargo add faucet-source-bigquery

# In the CLI (opt-in connector feature):
cargo install faucet-cli --features source-bigquery

Quick start

# pipeline.yaml — faucet run pipeline.yaml
version: 1
pipeline:
  source:
    type: bigquery
    config:
      project_id: my-project
      auth:
        type: service_account_key_path
        config:
          path: /etc/secrets/bigquery-sa.json
      query: |
        SELECT user_id, event_name, occurred_at
        FROM `my-project.analytics.events`
        WHERE occurred_at >= ?
      params:
        - "2026-01-01"
  sink:
    type: jsonl
    config:
      path: ./events.jsonl

faucet run pipeline.yaml

Configuration reference

Authentication

auth uses the shared BigQueryCredentials enum (the project-wide { type, config } shape):

# Application Default Credentials (workload identity, gcloud)
auth:
  type: application_default

# Service-account key file
auth:
  type: service_account_key_path
  config:
    path: /etc/secrets/bigquery-sa.json

# Inline service-account JSON via env indirection
auth:
  type: service_account_key
  config:
    json: ${env:GCP_SERVICE_ACCOUNT_JSON}

Examples

Incremental pull with a typed bind parameter

source:
  type: bigquery
  config:
    project_id: my-project
    auth: { type: application_default }
    query: |
      SELECT id, name, score
      FROM `my-project.app.users`
      WHERE score > ?
    params:
      - 0.75          # bound as FLOAT64, compares correctly against a numeric column

Large export with no re-chunking

source:
  type: bigquery
  config:
    project_id: my-project
    auth:
      type: service_account_key_path
      config: { path: /etc/secrets/bigquery-sa.json }
    query: SELECT * FROM `my-project.warehouse.daily_snapshot`
    location: EU
    max_results_per_page: 10000
    batch_size: 0      # emit the whole result set as one page

Long-running aggregate that may exceed the statement timeout

source:
  type: bigquery
  config:
    project_id: my-project
    auth: { type: application_default }
    query: |
      SELECT country, COUNT(*) AS n
      FROM `my-project.analytics.events`
      GROUP BY country
    statement_timeout: 30     # return fast; if not done, poll…
    poll_timeout: 600         # …for up to 10 minutes before failing

Streaming & batching

The source overrides Source::stream_pages: it runs the job, then walks jobs.getQueryResults by pageToken, buffering decoded rows and yielding a StreamPage every time the buffer reaches batch_size. Memory is bounded at one page. With batch_size: 0 the entire result set is buffered and emitted in a single page. max_results_per_page controls the HTTP page size independently of the emitted page size.

This is a one-shot query source — it has no incremental bookmark / resume support (each run re-executes the query). For incremental loads, encode the watermark in the query (e.g. WHERE occurred_at >= ?) and drive it from a matrix context or ${now.*} token.

Config loading & schema

Load from YAML/JSON or environment. Inspect the full JSON Schema with:

faucet schema source bigquery

Library usage

use faucet_core::Source;
use faucet_source_bigquery::{BigQueryCredentials, BigQuerySource, BigQuerySourceConfig};

# async fn run() -> Result<(), Box<dyn std::error::Error>> {
let cfg = BigQuerySourceConfig::new(
    "my-project",
    BigQueryCredentials::ApplicationDefault,
    "SELECT COUNT(*) AS n FROM analytics.events",
)
.with_location("EU")
.with_batch_size(0);

let rows = BigQuerySource::new(cfg).await?.fetch_all().await?;
println!("rows: {rows:?}");
# Ok(())
# }

How it works

1. new() resolves BigQueryCredentials and builds an authenticated client once.
2. jobs.query submits the SQL with statement_timeout as timeoutMs and any positional parameters.
3. If jobComplete=false, the source polls jobs.getQueryResults until the job finishes or poll_timeout elapses.
4. Result pages are walked via pageToken; each cell is decoded by its schema type into a typed JSON value.
5. Rows are re-framed into batch_size pages and streamed to the pipeline.

Lineage dataset URI

bigquery://<project_id>?query=<sql> — e.g. bigquery://my-project?query=SELECT id FROM events.

Feature flags

This crate has no optional features of its own; enable it in the CLI/umbrella via the source-bigquery feature.

Troubleshooting / FAQ

License

Licensed under either of Apache License, Version 2.0 or MIT license at your option.