Connectors / Databases

PostgreSQL

source

PostgreSQL query source

Install

cargo install faucet-cli --features source-postgres
faucet init --source postgres
# Cargo.toml — use it as a library
cargo add faucet-source-postgres

Starter config

source:
  type: postgres
  config:
    connection_url: ""    # REQUIRED — PostgreSQL connection URL (e.g.
    query: ""    # REQUIRED — SQL query to execute.
    # params: []    # Bind parameters for the query.
    # max_connections: 10    # Maximum number of connections in the pool.
    # batch_size: 1000    # Records per emitted `StreamPage`.
    # shard: null    # Optional primary-key range sharding for clustered (Mode B) execution.
    # read_timeout_secs: 3600    # Longest the source waits on the server for the next row (or for a whole result in `fetch_all`), in seconds, before the read fails.
    # json_big_numbers: null    # What to do with a number inside a JSON / JSONB column that a 64-bit float cannot hold exactly.
    # replication: { type: full }    # one of: full, incremental
      # --- Alternative variants — replace the block above with one of these ---
      # type: incremental
      # column: ""    # REQUIRED
      # initial_value: null    # REQUIRED
    # state_key: ""    # Explicit state-store key for the incremental bookmark.

Capabilities

Replay Non-deterministic replay
Exactly-once replay No
Incremental Yes
Dataset discovery Yes
Reports lag No
Compression No

Config fields

FieldTypeDefaultDescription
connection_url
required
string — PostgreSQL connection URL (e.g. postgres://user:pass@host/db).
query
required
string — SQL query to execute.
params array [] Bind parameters for the query. Defaults to empty.
max_connections integer 10 Maximum number of connections in the pool. Defaults to 10.
batch_size integer 1000 Records per emitted StreamPage. Rows are drained from the sqlx cursor and yielded whenever the buffer reaches this size. Defaults to DEFAULT_BATCH_SIZE.
shard object — Optional primary-key range sharding for clustered (Mode B) execution.
read_timeout_secs integer 3600 Longest the source waits on the server for the next row (or for a whole result in fetch_all), in seconds, before the read fails. This includes the wait for the first row, so a query that computes longer than this before returning anything fails. Defaults to 3600; 0 waits forever.
json_big_numbers object fail What to do with a number inside a JSON column that a 64-bit float cannot represent exactly (more than about 17 significant digits, or beyond the u64 / i64 range): fail (default) fails the read with an error naming the column, string emits the number as a JSON string holding its exact digits (with one warning per column).
replication object type: full Replication mode. Defaults to PostgresReplication::Full.
state_key string — Explicit state-store key for the incremental bookmark. When unset, a key is derived from the connection host and a fingerprint of the database URL and query.
Get started

Your first pipeline runs in five minutes.

Install the CLI, scaffold a config, and move real data — nothing external to stand up.

curl -LsSf https://github.com/faucet-hq/faucet-stream/releases/latest/download/faucet-cli-installer.sh | sh
brew install faucet-hq/faucet-stream/faucet-cli