# faucet-source-bigquery
[](https://crates.io/crates/faucet-source-bigquery)
[](https://docs.rs/faucet-source-bigquery)
[](https://github.com/PawanSikawat/faucet-stream/blob/main/rust-toolchain.toml)
[](https://github.com/PawanSikawat/faucet-stream#license)
Google **BigQuery** query source for the [faucet-stream](https://github.com/PawanSikawat/faucet-stream) ecosystem. Runs a SQL statement against BigQuery's [REST API](https://cloud.google.com/bigquery/docs/reference/rest) (`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`](https://crates.io/crates/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
```bash
# As a library:
cargo add faucet-source-bigquery
# In the CLI (opt-in connector feature):
cargo install faucet-cli --features source-bigquery
```
## Quick start
```yaml
# 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
```
```bash
faucet run pipeline.yaml
```
## Configuration reference
| `project_id` | string | — *(required)* | GCP project ID the query is billed to and run against. |
| `auth` | `BigQueryCredentials` | — *(required)* | Authentication — see [Authentication](#authentication). |
| `query` | string | — *(required)* | SQL to execute. May contain positional `?` markers (bound from `params` + matrix context) and `${field.path}` placeholders resolved against the parent-record context at runtime. |
| `use_legacy_sql` | bool | `false` | Use BigQuery's legacy SQL dialect. Leave `false` (Standard SQL) unless the query uses legacy `[project:dataset.table]` references. |
| `location` | string | *(unset)* | Location override for non-`US` jobs (`"EU"`, `"asia-east1"`, …). When unset, BigQuery uses the queried tables' default location. |
| `max_results_per_page` | int | `1000` | Rows per `jobs.getQueryResults` page. Smaller = more HTTP round-trips, less memory; larger = fewer requests, more memory. |
| `params` | array | `[]` | Positional bind parameters, sent in declaration order **before** any context-derived values. Typed from each JSON value. |
| `statement_timeout` | int (seconds) | `60` | Per-statement server-side timeout (`timeoutMs` on `jobs.query`). If the query isn't done within this window, BigQuery returns `jobComplete=false` and the source begins polling. |
| `poll_timeout` | int (seconds) | `300` | Max wall-clock time spent polling `jobs.getQueryResults` for a job still reporting `jobComplete=false`, before failing with `FaucetError::Source`. **`0` disables the cap** (poll forever). Only the completion wait is bounded; `pageToken` paging afterward is unaffected. |
| `batch_size` | int | `1000` | Records per emitted `StreamPage`. **`0` = no batching**: the entire result set is emitted in a single page (good for small lookup tables or sinks that prefer one large request). |
### Authentication
`auth` uses the shared `BigQueryCredentials` enum (the project-wide `{ type, config }` shape):
| `application_default` | *(none)* | Running on GCP (workload identity / metadata server) or after `gcloud auth application-default login`. |
| `service_account_key_path` | `{ path: <file> }` | You have a service-account key file on disk. |
| `service_account_key` | `{ json: <string> }` | You want to inject the key JSON inline, typically via `${env:VAR}` / `${secret:…}` indirection. |
```yaml
# Application Default Credentials (workload identity, gcloud)
auth:
type: application_default
```
```yaml
# Service-account key file
auth:
type: service_account_key_path
config:
path: /etc/secrets/bigquery-sa.json
```
```yaml
# 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
```yaml
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
```yaml
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
```yaml
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:
```bash
faucet schema source bigquery
```
## Library usage
```rust
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
| `Auth` error / 401 / 403 | Credentials invalid or missing scope. Confirm the service account has **BigQuery Data Viewer** + **BigQuery Job User** on the project, or that ADC is initialized (`gcloud auth application-default login`). |
| `403 ... billing` | The `project_id` has no billing enabled, or the SA lacks `bigquery.jobs.create`. Use a project with billing and the Job User role. |
| Job fails with a location error | The queried dataset isn't in the default (`US`) location. Set `location` to the dataset's region (e.g. `EU`). |
| `FaucetError::Source: poll timeout` | The job ran longer than `poll_timeout`. Raise `poll_timeout` (or set `0` to wait indefinitely), and/or raise `statement_timeout`. |
| A `NUMERIC` column arrives as a string | Intentional — `NUMERIC`/`BIGNUMERIC` are decoded as strings to preserve full precision. Cast in a downstream transform if you need a JSON number. |
| Bind parameter compared as text unexpectedly | Each `params` entry is typed from its JSON value. Pass `0.75` (number) not `"0.75"` (string) to bind a `FLOAT64`. |
| Legacy table reference fails to parse | Set `use_legacy_sql: true` for `[project:dataset.table]` syntax. |
## License
Licensed under either of [Apache License, Version 2.0](https://www.apache.org/licenses/LICENSE-2.0) or [MIT license](https://opensource.org/licenses/MIT) at your option.