faucetstream
Get started

Connectors / Databases

PostgreSQL

sink

PostgreSQL sink (JSONB or auto-mapped columns)

Install

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

Starter config

sink:
  type: postgres
  config:
    connection_url: ""    # REQUIRED — PostgreSQL connection URL (e.g.
    table_name: ""    # REQUIRED — Target table name.
    # schema: null    # Optional schema (namespace) qualifying `table_name`.
    # column_mapping: null    # How to map JSON records to table columns.
    # batch_size: 1000    # Maximum rows per multi-row `INSERT` statement.
    # max_connections: 5    # Maximum number of connections in the pool.
    # create_table: true    # Create the target table (and its schema, when `schema:` is set) if it does not exist, inferring the columns from the first written page (#580).
    # write_mode: null    # Write semantics for a sink.
    # key: []    # Key columns.
    # delete_marker: null    # Optional.
    # rollback: null    # Runtime rollback settings (#706): the run id to journal under, the run-id column, and whether to journal before-images / keep a replaced table.
    # write_method: null    # How rows are shipped to PostgreSQL in append mode.
    # scope: null    # Scoped/windowed overwrite (#518): with `write_mode: overwrite`, replace only the rows matching this scope (e.g.
    # write_timeout_secs: 3600    # Longest a page write may wait on the server, in seconds, before it fails.
    # binary_encoding: null    # How a JSON string bound to a binary column becomes bytes (#862 M-064).

Capabilities

Delivery Exactly-once (atomic watermark)
Write modes append, upsert, delete, overwrite
Exactly-once writes Yes
Schema evolution Yes
Scoped cleanup Yes
Staged bulk load No
Run rollback Yes
Compression No

Config fields

FieldTypeDefaultDescription
connection_url
required		 string — PostgreSQL connection URL (e.g. postgres://user:pass@host/db).
table_name
required		 string — Target table name.
schema string — Optional schema (namespace) qualifying table_name.
column_mapping object jsonb: {column: data} How to map JSON records to columns. Defaults to a single jsonb column named data.
batch_size integer 1000 Maximum rows per multi-row INSERT statement. Defaults to DEFAULT_BATCH_SIZE.
max_connections integer 5 Maximum number of connections in the pool. Defaults to 5.
create_table boolean true Create the target table (and its schema, when schema: is set) if it does not exist, inferring the columns from the first written page (#580). Enabled by default: a first-ever sync cannot assume the destination already exists.
write_mode object append Append (default), upsert, or delete.
key array [] Key columns. Required and non-empty for upsert/delete; ignored for append.
delete_marker object — Optional. Upsert only: rows whose field matches one of values are deletes; all others are upserts. The marker field is stripped from upsert rows before writing.
rollback object — Runtime rollback settings (#706): the run id to journal under, the run-id column, and whether to journal before-images / keep a replaced table. Injected per invocation by the CLI when a config has a top-level rollback: block — not meant to be written by hand.
write_method object insert How append-mode rows are shipped: multi-row insert (default) or the copy bulk-load fast-path (COPY … FROM STDIN, typically 5–10× faster for bulk append). copy is append-only — it is rejected with write_mode: upsert|delete — and the exactly-once path always stays on insert so data + watermark commit in one transaction.
scope object — Scoped/windowed overwrite (#518): with write_mode: overwrite, replace only the rows matching this scope (e.g. a date window) instead of truncating the whole table. The prior out-of-scope rows are preserved.
write_timeout_secs integer 3600 Longest a page write may wait on the server, in seconds, before it fails. Defaults to 3600; 0 waits forever. The same bound applies to the commit-token read, schema reads and evolution, and the overwrite staging setup and abort; the bulk statements of a scoped cleanup, a rollback and the overwrite commit are not bounded, because they can legitimately run longer.
binary_encoding object base64 How a string written to a binary column (BLOB/VARBINARY/bytea) becomes bytes: base64 (default) decodes every value — the form every faucet source emits — and fails the row when one is not base64; text stores every string as its text; auto decodes canonical base64 and stores any other string as text, deciding per value, so a text value that happens to be valid base64 ("user", a hex digest) is decoded. Set text when the column is fed plain text.
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
Read the guide → Star on GitHub