# faucet-source-snowflake
[](https://crates.io/crates/faucet-source-snowflake)
[](https://docs.rs/faucet-source-snowflake)
[](https://github.com/PawanSikawat/faucet-stream/blob/main/rust-toolchain.toml)
[](https://github.com/PawanSikawat/faucet-stream#license)
**Snowflake** query source for the [faucet-stream](https://github.com/PawanSikawat/faucet-stream) ecosystem. Executes a SQL statement against Snowflake's [SQL REST API](https://docs.snowflake.com/en/developer-guide/sql-api/intro) (`POST /api/v2/statements`), decodes each row into a typed `serde_json::Value`, and streams the result set back partition-by-partition so peak memory stays bounded no matter how large the result is.
Reach for it when you want to pull tables, aggregates, or ad-hoc query results out of Snowflake and land them in any faucet-stream sink — a file, a database, a warehouse, a queue — with one declarative config and no glue code. It talks to Snowflake over HTTPS only, so there is no driver, no ODBC, and no native dependency to build.
## Feature highlights
- **JWT key-pair and OAuth bearer authentication** — sign requests with an RSA key pair (RS256 JWT minted per request) or hand Snowflake an OAuth bearer token from an external IdP. The shared `SnowflakeAuth` enum is re-exported from [`faucet-common-snowflake`](https://crates.io/crates/faucet-common-snowflake) so it matches the Snowflake **sink** byte-for-byte, and credentials are masked in `Debug` output.
- **Server-side partition pagination** — large result sets are paged via `GET /api/v2/statements/{handle}?partition=N`; one HTTP round-trip per partition, with no client-side buffering of partitions you haven't reached yet.
- **Native streaming** — overrides `Source::stream_pages`: partitions are re-framed into `batch_size`-sized pages on the fly, so peak memory is `O(batch_size)` regardless of total row count.
- **Async-statement aware** — when the initial `POST` returns `202 Accepted` (the statement is still running), the source polls the statement handle until the result is ready, bounded by `poll_timeout` so a stuck statement fails cleanly instead of hanging forever.
- **Typed positional bind parameters** — `params` from config plus any `${parent.path}` matrix-context values are sent as Snowflake bind variables, typed from the JSON value (`FIXED` for integers, `REAL` for floats, `BOOLEAN` for bools, `TEXT` otherwise) so a numeric or boolean bind compares correctly against a typed column.
- **Type-aware row decoding** — `FIXED`, `REAL`, `BOOLEAN`, and `VARIANT`/`OBJECT`/`ARRAY` columns are parsed into native JSON shapes; everything else (timestamps, dates, binary) passes through as strings. **Full precision is preserved for `NUMBER`/`DECIMAL`/`NUMERIC`:** a fractional column (any `NUMBER(p,s)` with scale `s > 0`, including all monetary/decimal columns) is decoded as a **string** carrying the exact decimal text, since a JSON number is an `f64` and would silently lose precision — matching the BigQuery source's `NUMERIC`/`BIGNUMERIC` handling. Scale-0 (`NUMBER(p,0)`) values are JSON integers, except a value beyond `u64` which is likewise kept as a string.
- **Shared auth catalog** — OAuth tokens can come from the CLI's top-level `auth:` catalog via `auth: { ref: <name> }`, so many matrix rows hitting one IdP share a single token with single-flight refresh.
## Installation
```bash
# As a library:
cargo add faucet-source-snowflake
# In the CLI (opt-in connector feature):
cargo install faucet-cli --features source-snowflake
```
The connector is opt-in — it is not in the CLI's default feature set. Enable `source-snowflake` (or the `source` / `full` aggregates) to compile it in.
## Quick start
```yaml
# pipeline.yaml — faucet run pipeline.yaml
version: 1
pipeline:
source:
type: snowflake
config:
account: xy12345.us-east-1
warehouse: COMPUTE_WH
database: ANALYTICS
schema: PUBLIC
auth:
type: oauth
config:
token: ${env:SNOWFLAKE_OAUTH_TOKEN}
query: |
SELECT id, name, created_at
FROM events
WHERE created_at >= ?
params:
- "2026-01-01"
sink:
type: jsonl
config:
path: ./events.jsonl
```
```bash
faucet run pipeline.yaml
```
## Configuration reference
### Core
| Field | Type | Default | Description |
|-------|------|---------|-------------|
| `account` | string | — *(required)* | Snowflake account identifier (e.g. `"xy12345.us-east-1"`). Used to build the API hostname and to sign the key-pair JWT. |
| `warehouse` | string | — *(required)* | Virtual warehouse to run the statement on. |
| `database` | string | — *(required)* | Database for the session. |
| `schema` | string | — *(required)* | Schema for the session. |
| `role` | string | *(unset)* | Optional role to assume for the session. When unset, the user's default role is used. |
| `auth` | `AuthSpec<SnowflakeAuth>` | — *(required)* | Authentication — see [Authentication](#authentication). Accepts an inline `{ type, config }` block or `{ ref: <name> }` pointing at a shared provider. |
| `query` | string | — *(required)* | SQL to execute. May contain positional `?` bind markers (bound from `params` + matrix context) and `${field.path}` placeholders resolved against the parent-record context at runtime. |
| `params` | array | `[]` | Positional bind parameters, applied in order **before** any context-derived values. Each entry is typed from its JSON value. |
### Reliability & timeouts
| Field | Type | Default | Description |
|-------|------|---------|-------------|
| `statement_timeout` | int (seconds) | `60` | Per-statement server-side timeout, passed as the `timeout` field on the `POST /api/v2/statements` body. Independent of the HTTP-level request timeout. |
| `poll_timeout` | int (seconds) | `300` | Max wall-clock time the source spends polling an asynchronous statement (one whose initial `POST` returned HTTP 202) before failing with `FaucetError::Source`. **`0` disables the cap** (poll forever). |
### Batching
| Field | Type | Default | Description |
|-------|------|---------|-------------|
| `batch_size` | int | `1000` | Records per emitted `StreamPage`. Snowflake returns the result in *partitions*; the source re-frames them into `batch_size`-sized pages. **`0` = no batching**: the entire result set is buffered and emitted in a single page (good for small lookup tables or sinks that prefer one large request). Capped at `MAX_BATCH_SIZE` (1,000,000); validated at config-load time. |
## Authentication
`auth` uses the shared `SnowflakeAuth` enum (the project-wide `{ type, config }` shape):
| `type` | `config` | Use when |
|--------|----------|----------|
| `oauth` | `{ token: <string> }` | You have an OAuth bearer token from Snowflake or an external IdP. Sent as `Authorization: Snowflake Token="..."` with `X-Snowflake-Authorization-Token-Type: OAUTH`. |
| `key_pair` | `{ user: <string>, private_key_pem: <pem> }` | You authenticate with an RSA key pair registered on the Snowflake user. A fresh RS256 JWT (1-hour expiry) is minted per request and sent as `Authorization: Bearer <jwt>` with `X-Snowflake-Authorization-Token-Type: KEYPAIR_JWT`. |
### OAuth bearer token
```yaml
auth:
type: oauth
config:
token: ${env:SNOWFLAKE_OAUTH_TOKEN}
```
### Key-pair JWT
```yaml
auth:
type: key_pair
config:
user: SVC_LOAD
private_key_pem: ${file:/etc/secrets/snowflake.pem}
```
### Shared provider (CLI `auth:` catalog)
A shared provider must yield a `Bearer` or `Token` credential, which maps onto `SnowflakeAuth::OAuth`. Key-pair JWT is always inline (it cannot be shared via the catalog).
```yaml
auth:
my_snowflake_idp:
type: oauth2
config:
token_url: https://idp.example.com/oauth/token
client_id: ${env:SF_CLIENT_ID}
client_secret: ${secret:SF_CLIENT_SECRET}
pipeline:
source:
type: snowflake
config:
account: xy12345.us-east-1
warehouse: COMPUTE_WH
database: ANALYTICS
schema: PUBLIC
auth: { ref: my_snowflake_idp }
query: SELECT 1
```
## Examples
### Incremental pull with a typed bind parameter
```yaml
source:
type: snowflake
config:
account: xy12345.us-east-1
warehouse: COMPUTE_WH
database: ANALYTICS
schema: PUBLIC
auth:
type: oauth
config: { token: ${env:SNOWFLAKE_OAUTH_TOKEN} }
query: |
SELECT id, name, score
FROM users
WHERE score > ?
params:
- 0.75 # bound as REAL, compares correctly against a numeric column
```
### Large export with no re-chunking
```yaml
source:
type: snowflake
config:
account: xy12345.us-east-1
warehouse: COMPUTE_WH
database: WAREHOUSE
schema: PUBLIC
auth:
type: key_pair
config:
user: SVC_LOAD
private_key_pem: ${file:/etc/secrets/snowflake.pem}
query: SELECT * FROM daily_snapshot
batch_size: 0 # emit the whole result set as one page
```
### Daily window driven by a `${now.*}` token
```yaml
source:
type: snowflake
config:
account: xy12345.us-east-1
warehouse: COMPUTE_WH
database: ANALYTICS
schema: PUBLIC
role: ANALYST
auth:
type: oauth
config: { token: ${env:SNOWFLAKE_OAUTH_TOKEN} }
query: |
SELECT country, COUNT(*) AS n
FROM events
WHERE event_date = ?
GROUP BY country
params:
- "${now.date}" # resolved per run, e.g. 2026-06-17
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 submits the statement, then walks the result partitions reported by Snowflake (the first partition arrives inline in the `POST` response; subsequent partitions are fetched via `GET /api/v2/statements/{handle}?partition=N`). Decoded rows accumulate in a buffer and are yielded as a `StreamPage` every time the buffer reaches `batch_size`, so peak memory is one page. With `batch_size: 0` the entire result set is buffered and emitted in a single page.
This is a one-shot query source — it has no incremental bookmark / resume support, so each run re-executes the query. For incremental loads, encode the watermark in the query (e.g. `WHERE created_at >= ?`) and drive it from a `params` entry, a matrix context, or a `${now.*}` token.
## Config loading & schema introspection
Configs load from YAML/JSON files, environment variables, or `.env` files via the helpers in `faucet_core::config`. Inspect the full JSON Schema with:
```bash
faucet schema source snowflake
```
## Library usage
```rust
use faucet_core::Source;
use faucet_source_snowflake::{SnowflakeAuth, SnowflakeSource, SnowflakeSourceConfig};
# async fn run() -> Result<(), Box<dyn std::error::Error>> {
let cfg = SnowflakeSourceConfig::new(
"xy12345.us-east-1",
"COMPUTE_WH",
"ANALYTICS",
"PUBLIC",
SnowflakeAuth::OAuth { token: std::env::var("SNOWFLAKE_OAUTH_TOKEN")? },
"SELECT id, name FROM events LIMIT 10",
)
.with_role("ANALYST")
.with_batch_size(0);
let rows = SnowflakeSource::new(cfg).fetch_all().await?;
println!("got {} rows", rows.len());
# Ok(())
# }
```
## How it works
1. `new()` builds the reusable HTTPS client and derives the API base URL from `account`.
2. `POST /api/v2/statements` submits the SQL with `statement_timeout` as the body `timeout`, the session `warehouse` / `database` / `schema` / `role`, and any positional bindings (typed from their JSON values). The `Authorization` header is minted per request — a fresh RS256 JWT for key-pair auth, or the OAuth bearer token wrapped as `Snowflake Token="..."`.
3. If the statement is still running, Snowflake returns `202 Accepted` with a handle; the source polls the handle until the result is ready or `poll_timeout` elapses.
4. Result partitions are walked via `?partition=N`; each cell is decoded by its column metadata into a typed JSON value (full precision preserved for large `NUMBER`s).
5. Rows are re-framed into `batch_size` pages and streamed to the pipeline.
## Lineage dataset URI
`snowflake://<account>/<database>/<schema>?query=<sql>` — e.g. `snowflake://xy12345.us-east-1/DB/PUBLIC?query=SELECT id FROM orders`.
## Feature flags
This crate has no optional features of its own; enable it in the CLI/umbrella via the `source-snowflake` feature (included in the `source` and `full` aggregates).
## Troubleshooting / FAQ
| Symptom | Likely cause & fix |
|---------|--------------------|
| `Auth` error / 401 | OAuth token expired or invalid, or the key-pair public key isn't registered on the Snowflake user. Refresh the token, or confirm `ALTER USER … SET RSA_PUBLIC_KEY` matches your `private_key_pem`. |
| `403` / "insufficient privileges" | The session role can't read the queried objects, or `warehouse` access is denied. Set `role` to one with `USAGE` on the warehouse and `SELECT` on the objects. |
| Statement hangs, then fails with `FaucetError::Source: poll timeout` | The statement ran longer than `poll_timeout` (default 300 s). Raise `poll_timeout` (or set `0` to poll indefinitely), and/or resize the warehouse so the query finishes sooner. |
| `390201` / "no active warehouse" | `warehouse` is misspelled, suspended, or the role lacks `USAGE` on it. Verify the warehouse name and grants. |
| Account / host not found (DNS or 404) | `account` is wrong. Use the full account identifier including region/cloud (e.g. `xy12345.us-east-1`, or `orgname-accountname`), not the login URL. |
| A `NUMBER`/`DECIMAL` column arrives as a string | Intentional — a fractional `NUMBER(p,s)` (scale `s > 0`) and a scale-0 value beyond `u64` are decoded as strings to preserve full precision (a JSON number is an `f64`). Cast in a downstream transform if you need a JSON number and can accept the precision loss. |
| 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 `REAL`. |
| `auth: { ref }` rejected for key-pair | Key-pair JWT can't come from the shared `auth:` catalog — only OAuth (`Bearer`/`Token`) credentials map to a shared provider. Inline the `key_pair` block instead. |
## See also
- [Connector reference & capability matrix](https://pawansikawat.github.io/faucet-stream/reference/connectors.html)
- [Authentication cookbook](https://pawansikawat.github.io/faucet-stream/cookbook/auth.html) — shared `auth:` catalog and the `{ type, config }` shape
- [Secrets interpolation](https://pawansikawat.github.io/faucet-stream/cookbook/secrets.html) — `${env:…}` / `${file:…}` / `${secret:…}`
- [`faucet-common-snowflake`](https://crates.io/crates/faucet-common-snowflake) — the shared `SnowflakeAuth` enum and header helpers
- [`faucet-sink-snowflake`](https://crates.io/crates/faucet-sink-snowflake) — write rows back to Snowflake
## 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.