faucet-source-snowflake

Snowflake query source for the faucet-stream ecosystem. Executes a SQL statement against Snowflake's SQL REST API (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 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

# 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

# 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

faucet run pipeline.yaml

Configuration reference

Core

Reliability & timeouts

Batching

Authentication

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

OAuth bearer token

auth:
  type: oauth
  config:
    token: ${env:SNOWFLAKE_OAUTH_TOKEN}

Key-pair JWT

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).

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

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

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

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:

faucet schema source snowflake

Library usage

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 NUMBERs).
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

See also

• Connector reference & capability matrix
• Authentication cookbook — shared auth: catalog and the { type, config } shape
• Secrets interpolation — ${env:…} / ${file:…} / ${secret:…}
• faucet-common-snowflake — the shared SnowflakeAuth enum and header helpers
• faucet-sink-snowflake — write rows back to Snowflake

License

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