faucet-source-postgres

A PostgreSQL query source for the faucet-stream ecosystem. Runs any SQL SELECT against a PostgreSQL database and streams the result set back row-by-row as serde_json::Value records.

Built on sqlx with a reusable connection pool and a true row cursor (Query::fetch), so the source never buffers the whole result set in memory and the downstream sink starts writing as soon as the first batch is parsed off the wire. Reach for it to extract tables, joins, aggregates, or ad-hoc query results out of Postgres and land them in any faucet-stream sink — a file, an object store, a warehouse, a queue — with one declarative config and no glue code.

Feature highlights

• Streaming row cursor — stream_pages drives a sqlx cursor and yields batch_size-sized pages; peak client memory is O(batch_size), independent of total row count.
• Connection pooling — a PgPoolOptions pool sized by max_connections is built once in new() and reused for every fetch.
• Type-aware row decoding — integers, floats, booleans, text, json/jsonb, timestamp(tz), date/time, uuid, numeric/decimal, and bytea are each converted to the right JSON shape (full numeric precision preserved as strings; binary base64-encoded).
• Positional bind parameters — params from config are bound as native scalar types ($1, $2, …), so a numeric or boolean bind compares correctly against a typed column instead of being coerced to jsonb.
• Matrix-context binding — ${parent.path} tokens in the query are rewritten to additional positional bind markers and filled per parent record, so the same query template runs once per row produced by a parent in a matrix pipeline.
• Credential redaction — the connection URL is masked in Debug output and stripped from the emitted lineage dataset URI.

Installation

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

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

source-postgres is an opt-in feature — it is not in the CLI default build. Enable it (or the umbrella source aggregate) when you need this connector.

Quick start

# pipeline.yaml — faucet run pipeline.yaml
version: 1
pipeline:
  source:
    type: postgres
    config:
      connection_url: postgres://user:pass@localhost:5432/app
      query: SELECT id, name, email FROM users WHERE active = true
  sink:
    type: jsonl
    config:
      path: ./users.jsonl

faucet run pipeline.yaml

Configuration reference

All fields live under pipeline.source.config.

Core

Reliability & batching

Examples

Parameterized query

Bind a literal value into the query. Parameters are positional ($1 maps to params[0]):

version: 1
name: shipped_orders
pipeline:
  source:
    type: postgres
    config:
      connection_url: postgres://user:pass@localhost:5432/app
      query: SELECT * FROM orders WHERE status = $1 AND total > $2
      params:
        - shipped
        - 100.0
  sink:
    type: jsonl
    config:
      path: ./shipped_orders.jsonl

Archive a large table to S3, tuned pool and page size

version: 1
name: postgres_to_s3
pipeline:
  source:
    type: postgres
    config:
      connection_url: postgres://user:pass@localhost:5432/app
      query: SELECT * FROM events WHERE created_at < $1
      params:
        - "2026-01-01T00:00:00Z"
      max_connections: 12
      batch_size: 5000
  sink:
    type: s3
    config:
      bucket: my-archive-bucket
      prefix: events/2025/
      region: us-east-1
      file_extension: .jsonl
      max_records_per_file: 50000
      concurrency: 16

Small lookup table, single page (load-job friendly)

Set batch_size: 0 to emit the whole result set as one page — ideal feeding a sink that prefers one large request (SQL COPY, a BigQuery load job, a Snowflake stage upload):

version: 1
pipeline:
  source:
    type: postgres
    config:
      connection_url: postgres://user:pass@localhost:5432/app
      query: SELECT code, label FROM country_codes
      batch_size: 0
  sink:
    type: bigquery
    config:
      project_id: my-project
      dataset_id: ref
      table_id: country_codes

Secrets-manager interpolation

Keep the credentials out of the config file — the CLI resolves ${vault:…} / ${aws-sm:…} / ${gcp-sm:…} / ${azure-kv:…} at load time when built with the matching secrets-* feature:

pipeline:
  source:
    type: postgres
    config:
      connection_url: ${vault:secret/data/warehouse#dsn}
      query: SELECT * FROM customers

Streaming & batching

PostgresSource::stream_pages drives a sqlx row cursor (Query::fetch) without buffering the full result. Rows are accumulated into a batch_size buffer and yielded as a StreamPage once the buffer fills; the trailing partial page (if any) is yielded after the cursor drains. The pipeline writes each page to the sink as it arrives, so memory is bounded by O(batch_size) on both ends.

batch_size = 0 is the "no batching" sentinel — the cursor is drained completely and the entire result set is emitted in a single StreamPage. Use it for small lookup tables, or for downstream sinks (SQL COPY, BigQuery load jobs, Snowflake stage uploads) that prefer one large request to many small ones.

The trait-level batch_size argument to stream_pages is informational; the source always uses its own config field as the authoritative knob, so a pipeline-supplied hint cannot silently override an explicit config value.

This is a query source with no incremental-replication mode — it runs the configured query once and streams the result. Every emitted page therefore carries bookmark: None; there is no resume/state bookmark, no exactly-once delivery, and no upsert/write modes (those are sink concerns). For change data capture, see faucet-source-postgres-cdc, which captures logical-replication changes and is resumable via a state store.

Note — Postgres' wire protocol sends rows from a simple SELECT in a single response (no server-side cursor by default). The streaming implementation bounds client-side memory at O(batch_size) and lets the sink begin writing as soon as the first batch is parsed off the wire.

Supported column types

Columns are converted to JSON values by probing the row's value with each candidate Rust type in turn:

Matrix-context binding

In a parent/child matrix pipeline, the query may reference fields from each parent record with ${parent_id.dotted.path} tokens. At runtime these are rewritten to additional positional bind markers (appended after the static params) and filled per parent record, so a single query template fans out into one parameterized execution per parent row — never string-interpolated, so it is SQL-injection safe.

matrix:
  - id: parent
    source: { ref: tenants }              # produces rows with a `tenant_id`
  - id: child
    parent: parent
    source:
      type: postgres
      config:
        connection_url: postgres://user:pass@localhost:5432/app
        query: SELECT * FROM orders WHERE tenant_id = ${parent.tenant_id}

Config loading

Load a config from JSON, env vars, or a .env file via the helpers in faucet_core::config:

use faucet_core::config::{load_json, load_env_file};
use faucet_source_postgres::PostgresSourceConfig;

# fn example() -> Result<(), Box<dyn std::error::Error>> {
let config: PostgresSourceConfig = load_json("config.json")?;
let config: PostgresSourceConfig = load_env_file(".env", "PG_SOURCE")?;
# Ok(()) }

{
  "connection_url": "postgres://analytics:password@db.example.com:5432/warehouse",
  "query": "SELECT id, name, created_at, metadata FROM events WHERE created_at > $1 ORDER BY created_at",
  "params": ["2025-01-01T00:00:00Z"],
  "max_connections": 5,
  "batch_size": 5000
}

PG_SOURCE_CONNECTION_URL=postgres://user:password@localhost:5432/mydb
PG_SOURCE_QUERY=SELECT * FROM users
PG_SOURCE_MAX_CONNECTIONS=10

Schema introspection

Print the JSON Schema for this source's config (drives faucet validate and faucet init):

faucet schema source postgres

Library usage

use faucet_source_postgres::{PostgresSource, PostgresSourceConfig};
use faucet_core::Source;
use serde_json::json;

# async fn example() -> Result<(), Box<dyn std::error::Error>> {
let config = PostgresSourceConfig::new(
    "postgres://user:pass@localhost:5432/app",
    "SELECT * FROM orders WHERE status = $1 AND total > $2",
)
.params(vec![json!("shipped"), json!(100.0)])
.with_max_connections(20)
.with_batch_size(5000);

let source = PostgresSource::new(config).await?;

// Drain the whole result set:
let orders = source.fetch_all().await?;
for order in &orders {
    println!("{order}");
}
# Ok(()) }

To wire it into a streaming pipeline, hand the source to faucet_core::Pipeline (or run_stream) together with any sink — each StreamPage is written as it arrives.

How it works

• Pool reuse — PostgresSource::new validates batch_size, then builds one PgPool (PgPoolOptions::max_connections(...)) and stores it on the struct. Every fetch borrows a connection from the pool; nothing is reconnected per call.
• True cursor streaming — stream_pages calls Query::fetch (a sqlx cursor) and pulls rows with try_next(), converting each PgRow to a JSON object keyed by column name. Rows are batched into a reusable buffer and the buffer is swapped (mem::replace) on each yield to avoid reallocation churn.
• Typed binding — config and context parameters are bound as native scalars (string / i64 / f64 / bool / null), never as raw jsonb. Binding a serde_json::Value directly would encode it as jsonb and break comparisons against typed columns (e.g. WHERE id = $1 against an integer column).
• Errors — connection and query failures surface as FaucetError::Config with the underlying sqlx message; an out-of-range batch_size is rejected at construction.

Lineage dataset URI

postgres://<host>:<port>/<db>?query=<sql> (credentials stripped) — e.g. postgres://host:5432/app?query=SELECT id FROM orders.

Feature flags

This crate has no optional features of its own. Enable it in the CLI or umbrella crate via the source-postgres feature.

Troubleshooting / FAQ

See also

• Connector reference · Choosing a connector
• faucet-source-postgres-cdc — resumable change data capture via logical replication
• faucet-sink-postgres — the PostgreSQL sink (insert / upsert / delete)
• faucet-source-mysql · faucet-source-sqlite · faucet-source-mssql

Sharded execution (cluster Mode B)

Under faucet serve --cluster, a top-level shard: { count: N } block splits this source into contiguous primary-key ranges that different cluster workers process concurrently. Opt in by naming an integer-typed key column:

shard:
  count: 8
pipeline:
  source:
    type: postgres
    config:
      connection_url: ${env:PG_URL}
      query: "SELECT * FROM events"
      shard: { key: id }   # integer column to range-partition on

The coordinator computes MIN(key) / MAX(key) once and splits that range into half-open slices ("key" in the generated predicate, injection-safe). The boundary shards stay open-ended so rows inserted outside the captured range during the run are still read, and exactly one shard additionally matches key IS NULL so nullable keys are never silently dropped. Each shard keeps its own state key ({run}::{shard}), so a reassigned shard resumes where its previous owner left off.

Outside the cluster coordinator the shard config has no effect — a plain faucet run streams the whole query.

License

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