faucet-sink-snowflake

Snowflake sink for the faucet-stream ecosystem. Writes JSON records to a Snowflake table over the Snowflake SQL REST API — no driver, no ODBC, no client library, just HTTPS.

Reach for it when you want to land records from any faucet-stream source — a REST API, a database, a file, a queue — into Snowflake with one declarative config. Each batch is parsed and inserted in a single PARSE_JSON + FLATTEN statement, and the JSON array travels as a bound TEXT parameter (PARSE_JSON(?)) rather than interpolated into SQL, so quote characters in your data are safe and cannot inject SQL.

Feature highlights

• Driverless SQL REST API — pure HTTPS against https://{account}.snowflakecomputing.com/api/v2/statements. The HTTP client is built once in new() and reused for every request.
• Two auth methods — JWT key-pair (RS256, minted locally from your RSA private key with the public-key SHA-256 fingerprint in the iss claim) or OAuth bearer tokens from an external identity provider.
• Shared auth providers — the auth field also accepts { ref: <name> }, pointing at a provider in the CLI's top-level auth: catalog (OAuth/bearer tokens only; key-pair JWT is always inline). N matrix rows hitting one IdP then share a single token.
• Set-based batch inserts — one INSERT … SELECT … FROM TABLE(FLATTEN(input => PARSE_JSON(?))) per chunk parses and inserts the whole batch in a single statement.
• Configurable batching — batch_size controls rows per request (default 1000, the documented sweet spot); batch_size: 0 forwards each upstream page untouched.
• Async-execution aware — when Snowflake answers an INSERT with HTTP 202 (queued, not yet run), the sink polls the statement handle until it succeeds before counting rows as written, bounded by poll_timeout.
• Effectively-once delivery — with delivery: exactly_once, each page's INSERT and a commit-token MERGE land in one multi-statement transaction, so a crash/resume never duplicates a page.
• SQL-injection-safe — data is bound, never interpolated; all table/schema/database identifiers are quoted.
• Preflight probe — faucet doctor runs a read-only SELECT 1 to confirm credentials and warehouse access without writing rows.

Installation

# As a library:
cargo add faucet-sink-snowflake

# Via the umbrella crate:
cargo add faucet-stream --features sink-snowflake

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

Quick start

# pipeline.yaml — faucet run pipeline.yaml
version: 1
pipeline:
  source:
    type: s3
    config:
      bucket: my-data-lake
      prefix: raw/events/
      region: us-east-1
      file_format: json_lines
  sink:
    type: snowflake
    config:
      account: xy12345.us-east-1
      warehouse: LOAD_WH
      database: ANALYTICS
      schema: RAW
      table: EVENTS
      auth:
        type: key_pair
        config:
          user: LOADER
          private_key_pem: ${file:./snowflake_key.pem}
      batch_size: 1000

faucet run pipeline.yaml

Configuration reference

Core

Batching & reliability

Authentication

auth uses the shared SnowflakeAuth enum (re-exported from faucet-common-snowflake, so it matches the Snowflake source byte-for-byte) in the project-wide { type, config } shape. The Debug impl masks private_key_pem and token as ***.

# Key-pair JWT (PEM inlined from disk via the file: directive)
auth:
  type: key_pair
  config:
    user: INGEST_USER
    private_key_pem: ${file:./snowflake_key.pem}

# OAuth bearer token via env indirection
auth:
  type: oauth
  config:
    token: ${env:SNOWFLAKE_TOKEN}

# Shared provider from the top-level auth: catalog (OAuth/bearer only)
auth:
  ref: snowflake_idp

Note: a shared auth: { ref } provider must yield a Bearer or Token credential, which maps onto SnowflakeAuth::OAuth. Key-pair JWT is stateless (minted locally from the RSA key) and therefore must always be supplied inline.

Examples

Postgres → Snowflake with key-pair auth

version: 1
name: postgres_to_snowflake
pipeline:
  source:
    type: postgres
    config:
      connection_url: postgres://user:pass@localhost/app
      query: SELECT id, email, created_at FROM users WHERE tenant_id = $1
      params:
        - acme
  sink:
    type: snowflake
    config:
      account: xy12345.us-east-1
      warehouse: INGEST_WH
      database: ANALYTICS
      schema: RAW
      table: USERS
      auth:
        type: key_pair
        config:
          user: INGEST_USER
          private_key_pem: ${file:./snowflake_key.pem}
      batch_size: 500

Latency-sensitive stream with small batches

sink:
  type: snowflake
  config:
    account: xy12345.us-east-1
    warehouse: COMPUTE_WH
    database: MY_DB
    schema: PUBLIC
    table: realtime_events
    auth:
      type: oauth
      config: { token: ${env:SNOWFLAKE_TOKEN} }
    batch_size: 50      # smaller chunks → lower per-row latency

Large load with no re-chunking

sink:
  type: snowflake
  config:
    account: xy12345.us-east-1
    warehouse: LOAD_WH
    database: ANALYTICS
    schema: RAW
    table: EVENTS
    auth:
      type: key_pair
      config:
        user: LOADER
        private_key_pem: ${file:./snowflake_key.pem}
    batch_size: 0          # forward each upstream page as one INSERT
    poll_timeout: 900      # allow up to 15 min for a queued statement to run

Streaming & batching

SnowflakeSink::write_batch re-chunks the incoming records slice into batch_size slices and issues one SQL REST API INSERT per chunk. The default of 1000 matches Snowflake's documented sweet spot for the SQL API — enough rows to amortize the round-trip and statement-parse cost without bloating the request body. Tune up for narrow rows where round-trip latency dominates, down for wide rows where request-body size dominates.

batch_size = 0 is the "no batching" sentinel: write_batch forwards the entire slice as a single INSERT, so the upstream source's StreamPage framing flows through untouched. Use it when the source already emits pages sized for the target (e.g. one large page per write).

Asynchronous execution

Snowflake's SQL REST API may answer a submitted INSERT with HTTP 202 Accepted — the statement was queued but has not yet executed. The sink does not count those rows as written at that point (that would report success before the data is durable). Instead it polls GET /api/v2/statements/{handle} until the statement reports success (code 090001), then returns. The poll loop is bounded by poll_timeout (default 300 s): if the statement is still running after that budget, the write fails with FaucetError::Sink rather than hanging forever. Set poll_timeout: 0 to poll indefinitely.

Effectively-once delivery

SnowflakeSink implements Sink::supports_idempotent_writes (returns true) and the two companion hooks:

• write_batch_idempotent(records, scope, token) — writes the page's records and MERGEs the token into a _faucet_commit_token("scope" STRING PRIMARY KEY, "token" STRING, "updated_at" TIMESTAMP_NTZ) watermark table in the target database/schema, all inside one multi-statement transaction (BEGIN; INSERT; MERGE; COMMIT; with MULTI_STATEMENT_COUNT set on the request), so both either commit together or neither does. The watermark table is created (CREATE TABLE IF NOT EXISTS) as its own request once per sink instance — Snowflake DDL auto-commits, so it can never ride inside the transaction.
• last_committed_token(scope) — reads the current watermark so the pipeline skips already-committed pages on resume.

The whole page is one atomic unit on this path — batch_size re-chunking does not apply to write_batch_idempotent (core issues exactly one token per page; splitting the page across transactions would break the rows-plus-token atomicity). Size the source's batch_size down if a page's JSON payload grows too large for a single SQL REST API request. An empty page still advances the watermark via a commit-only BEGIN; MERGE; COMMIT; transaction.

To use effectively-once delivery, set delivery: exactly_once and pair this sink with a CDC source (postgres-cdc, mysql-cdc, mongodb-cdc) plus a state: block. A DLQ is not permitted in effectively-once mode. All four requirements are validated at config-load time (faucet validate) before any run starts.

version: 1
pipeline:
  source:
    type: postgres-cdc
    config:
      connection_url: postgres://faucet:faucet@localhost:5432/appdb
      slot_name: faucet_slot
      publication_name: faucet_pub
  sink:
    type: snowflake
    config:
      account: xy12345.us-east-1
      warehouse: COMPUTE_WH
      database: ANALYTICS_DB
      schema: PUBLIC
      table: change_events
      auth:
        type: oauth
        config:
          token: ${env:SNOWFLAKE_TOKEN}
  state:
    type: file
    config:
      path: ./state
delivery: exactly_once

See the effectively-once delivery cookbook.

Config loading & schema

Load from YAML/JSON or environment, and inspect the full JSON Schema:

faucet schema sink snowflake

use faucet_core::config::{load_json, load_env_file};
use faucet_sink_snowflake::SnowflakeSinkConfig;

// From a JSON file
let config: SnowflakeSinkConfig = load_json("config.json")?;
// From an .env file with a prefix
let config: SnowflakeSinkConfig = load_env_file(".env", "SNOWFLAKE")?;

SNOWFLAKE_ACCOUNT=xy12345.us-east-1
SNOWFLAKE_WAREHOUSE=COMPUTE_WH
SNOWFLAKE_DATABASE=ANALYTICS_DB
SNOWFLAKE_SCHEMA=PUBLIC
SNOWFLAKE_TABLE=events
SNOWFLAKE_AUTH='{"type":"oauth","config":{"token":"eyJhbGciOiJSUzI1NiIs..."}}'
SNOWFLAKE_BATCH_SIZE=1000

Library usage

use faucet_core::{Pipeline, Sink};
use faucet_sink_snowflake::{SnowflakeAuth, SnowflakeSink, SnowflakeSinkConfig};
use serde_json::json;

# async fn run() -> Result<(), Box<dyn std::error::Error>> {
let config = SnowflakeSinkConfig::new(
    "xy12345.us-east-1",
    "COMPUTE_WH",
    "ANALYTICS_DB",
    "PUBLIC",
    "events",
    SnowflakeAuth::OAuth { token: std::env::var("SNOWFLAKE_TOKEN")? },
)
.with_batch_size(1000);

let sink = SnowflakeSink::new(config)?;

let records = vec![
    json!({"user_id": "u123", "event": "page_view", "ts": "2026-04-02T10:00:00Z"}),
    json!({"user_id": "u456", "event": "click",     "ts": "2026-04-02T10:01:00Z"}),
];
let rows_written = sink.write_batch(&records).await?;
println!("Wrote {rows_written} rows to Snowflake");
# Ok(())
# }

To drive it end-to-end, pair it with any source via Pipeline::new(source, sink).run().await?.

How it works

1. SnowflakeSink::new() builds an HTTP client (reused across all requests). No network call happens at construction.
2. write_batch() splits records into batch_size chunks (or one chunk when batch_size = 0). For each chunk it builds an INSERT using PARSE_JSON(?) + FLATTEN and sends the chunk's JSON array as a bound TEXT parameter, parsing and inserting every row in one statement without interpolating data into the SQL text.
3. Field-to-column mapping: each record's top-level keys project into matching table columns — INSERT INTO "db"."schema"."tbl" ("col1","col2") SELECT value:"col1"::string, value:"col2"::string FROM TABLE(FLATTEN(input => PARSE_JSON(?))). The ::string cast strips the VARIANT's JSON quotes so Snowflake coerces each scalar into the destination column's type on insert (text → number / boolean / timestamp, etc.). The column set comes from the first record; a key absent from a later record is inserted as NULL. Target columns should be scalar — a key targeting a VARIANT/OBJECT/ARRAY column is stringified, not stored as structured JSON. Both column identifiers and JSON path keys are quote-escaped, so record keys cannot inject SQL.
4. The statement targets the fully-qualified "database"."schema"."table" with quoted identifiers.
5. Auth headers are generated per request: an RS256 JWT (1-hour expiry, public-key fingerprint in iss) for key_pair, or Snowflake Token="…" for oauth.
6. Success is confirmed by the 090001 response code; an HTTP 202 enters the async-execution poll loop.

Lineage dataset URI

snowflake://<account>/<database>/<schema>?table=<table> — e.g. snowflake://xy12345.us-east-1/ANALYTICS_DB/PUBLIC?table=events.

Feature flags

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

Troubleshooting / FAQ

See also

• Snowflake source — query Snowflake via the same SQL REST API.
• faucet-common-snowflake — the shared SnowflakeAuth enum and auth helpers.
• Sinks reference — capability matrix across all connectors.
• Authentication cookbook — the shared auth: provider catalog.
• Secrets cookbook — injecting the PEM / token from a secrets manager.

License

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

Arrow columnar bulk load (opt-in, arrow feature)

With a binary built --features arrow, add a bulk_load block to switch the sink onto the Arrow columnar fast path: incoming Arrow RecordBatches are written to Parquet, uploaded to an external stage's backing cloud storage, and loaded with COPY INTO <table> FROM @stage FILE_FORMAT=(TYPE=PARQUET). This runs end-to-end without per-row JSON when the source is also columnar (e.g. parquet → snowflake). It is append-only and independent of the row-path INSERT/exactly-once code, which is unchanged.

sink:
  type: snowflake
  config:
    account: ${env:SNOWFLAKE_ACCOUNT}
    warehouse: LOAD_WH
    database: ANALYTICS
    schema: PUBLIC
    table: EVENTS
    auth: { type: oauth, config: { token: ${env:SNOWFLAKE_TOKEN} } }
    bulk_load:
      stage: ANALYTICS.PUBLIC.EVENTS_STAGE   # existing external stage
      url: s3://my-bucket/snowflake-stage/   # the stage's backing location
      storage_options:                       # object_store keys for the upload
        aws_access_key_id: ${env:AWS_ACCESS_KEY_ID}
        aws_secret_access_key: ${env:AWS_SECRET_ACCESS_KEY}
        aws_region: us-east-1
      match_by_column_name: CASE_INSENSITIVE  # default
      purge: false                            # default; PURGE=TRUE removes staged files

Only external stages are supported — internal-stage PUT is a driver command the SQL REST API cannot run. The Snowflake source has no Arrow path (the v2 SQL API is jsonv2-only). bulk_load on an arrow-off binary is rejected at construction.