faucet-sink-postgres

PostgreSQL sink connector for the faucet-stream ecosystem.

Writes JSON records to a PostgreSQL table using either JSONB column mode (storing each record as a single jsonb value) or AutoMap mode (mapping top-level JSON keys directly to table columns). Uses connection pooling via sqlx and efficient multi-row INSERT statements.

write_batch accepts whatever slice the pipeline hands it. When batch_size > 0 and the slice is larger than batch_size, the sink re-chunks internally and issues one multi-row INSERT per chunk; when batch_size = 0, the entire slice is sent in a single INSERT — see Streaming and batching for the tradeoffs.

Installation

[dependencies]
faucet-sink-postgres = "1.0"
tokio = { version = "1", features = ["full"] }

Or via the umbrella crate:

faucet-stream = { version = "1.0", features = ["sink-postgres"] }

Quick Start

use faucet_sink_postgres::{PostgresSink, PostgresSinkConfig};
use faucet_core::Sink;
use serde_json::json;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let config = PostgresSinkConfig::new(
        "postgres://user:password@localhost:5432/mydb",
        "events",
    );

    let sink = PostgresSink::new(config).await?;

    let records = vec![
        json!({"user_id": "u123", "event": "signup", "metadata": {"source": "web"}}),
        json!({"user_id": "u456", "event": "login", "metadata": {"source": "mobile"}}),
    ];

    let rows_written = sink.write_batch(&records).await?;
    println!("Wrote {rows_written} rows");

    Ok(())
}

Configuration

The Debug implementation masks the connection_url with *** to prevent credential leakage in logs.

Streaming and batching

The PostgreSQL sink re-chunks each incoming StreamPage so individual multi-row INSERT statements stay well under Postgres' per-statement bind-parameter limit.

• batch_size > 0 (default 1000) — the sink slices the incoming slice into batch_size-row chunks and issues one multi-row INSERT per chunk. Recommended value is 1000: Postgres' multi-row INSERT sweet spot. Larger chunks rarely add throughput. AutoMap mode binds one parameter per column per row; the sink now splits each chunk further so rows × columns never exceeds Postgres' 65 535 bind-parameter ceiling, so a wide table no longer causes the server to reject the statement regardless of batch_size. JSONB mode binds a single jsonb[] array regardless of row count.
• batch_size = 0 — the "no batching" sentinel. The entire upstream StreamPage is forwarded in a single logical write. Use this when the source already emits page sizes tuned for Postgres — for example a REST source with batch_size: 1000 feeding a JSONB table. AutoMap still sub-splits internally to respect the 65 535-parameter ceiling.

batch_size is purely a chunk-size knob — connection pooling, identifier quoting, and JSONB vs AutoMap behaviour are unchanged.

Column Mapping (PostgresColumnMapping)

Builder Methods

use faucet_sink_postgres::{PostgresSinkConfig, PostgresColumnMapping};

// JSONB mode with custom column name
let config = PostgresSinkConfig::new("postgres://localhost/mydb", "events")
    .column_mapping(PostgresColumnMapping::Jsonb { column: "payload".into() })
    .with_batch_size(1000)
    .max_connections(10);

// AutoMap mode
let config = PostgresSinkConfig::new("postgres://localhost/mydb", "events")
    .column_mapping(PostgresColumnMapping::AutoMap)
    .with_batch_size(250);

Write modes (upsert / delete)

By default the sink appends every record. Set write_mode to upsert or delete to merge or remove rows by a key instead.

Requirements (validated in PostgresSink::new, before any connection is made):

• column_mapping: auto_map is required. Upsert/delete match on real columns, so the key columns must be table columns — not fields buried inside a JSONB blob.
• The key columns must carry a UNIQUE or PRIMARY KEY constraint, since upsert is implemented with INSERT … ON CONFLICT (key) DO UPDATE SET … (non-key columns are set from EXCLUDED; if every column is a key column the clause degrades to DO NOTHING). Without the constraint Postgres rejects the ON CONFLICT target.

Semantics:

• Last-write-wins within a batch. When the same key appears multiple times in one write_batch call, the records are de-duplicated to a single effective action (the final one), so a single statement never hits the same ON CONFLICT target twice. A delete after an upsert (or vice-versa) for the same key resolves to whichever came last.
• write_mode: delete routes every record to a delete by key.
• A record missing a key column (or with a null key value) fails with a typed Sink error. When a dlq: block is configured the good rows are still written (upserts + deletes applied) and only the missing/null-key rows are routed to the DLQ per-row; without a DLQ the whole batch fails. Keys should be present and non-null.

Example YAML config

pipeline:
  source:
    type: postgres-cdc
    config:
      connection_url: postgres://faucet:faucet@localhost:5432/appdb
      slot_name: faucet_slot
      publication_name: faucet_pub
  sink:
    type: postgres
    config:
      connection_url: postgres://writer:pass@localhost:5432/warehouse
      table_name: users
      column_mapping: auto_map
      write_mode: upsert
      key: [id]
      delete_marker:
        field: __op
        values: [d]

The destination table must define the key as a constraint, e.g. CREATE TABLE users (id INT PRIMARY KEY, name TEXT, email TEXT).

Config Loading

use faucet_core::config::{load_json, load_env_file};
use faucet_sink_postgres::PostgresSinkConfig;

// From a JSON file
let config: PostgresSinkConfig = load_json("config.json")?;

// From an .env file with a prefix
let config: PostgresSinkConfig = load_env_file(".env", "PG_SINK")?;

Example JSON config (JSONB mode)

{
  "connection_url": "postgres://writer:s3cret@db.example.com:5432/analytics",
  "table_name": "raw_events",
  "column_mapping": {
    "jsonb": {
      "column": "data"
    }
  },
  "batch_size": 1000,
  "max_connections": 5
}

Example JSON config (AutoMap mode)

{
  "connection_url": "postgres://writer:s3cret@db.example.com:5432/analytics",
  "table_name": "events",
  "column_mapping": "auto_map",
  "batch_size": 1000,
  "max_connections": 10
}

Example .env file

PG_SINK_CONNECTION_URL=postgres://writer:s3cret@db.example.com:5432/analytics
PG_SINK_TABLE_NAME=raw_events
PG_SINK_COLUMN_MAPPING='{"jsonb":{"column":"data"}}'
PG_SINK_BATCH_SIZE=1000
PG_SINK_MAX_CONNECTIONS=5

Config Schema Introspection

use faucet_core::Sink;

let sink = PostgresSink::new(config).await?;
let schema = sink.config_schema();
println!("{}", serde_json::to_string_pretty(&schema)?);

Pipeline Usage

use faucet_core::Pipeline;
use faucet_source_rest::{RestStream, RestStreamConfig};
use faucet_sink_postgres::{PostgresSink, PostgresSinkConfig, PostgresColumnMapping};

let source_config = RestStreamConfig::new("https://api.example.com", "/v1/users");
let source = RestStream::new(source_config);

let sink_config = PostgresSinkConfig::new(
    "postgres://writer:pass@localhost:5432/app",
    "users",
)
.column_mapping(PostgresColumnMapping::AutoMap);

let sink = PostgresSink::new(sink_config).await?;

let result = Pipeline::new(source, sink).run().await?;
println!("Transferred {} records", result.records_written);

Examples

JSONB mode -- store entire records in a single column

This mode is ideal for schemaless ingestion where you want to store raw JSON and query it later with PostgreSQL's JSONB operators.

-- Table schema
CREATE TABLE raw_events (
    id SERIAL PRIMARY KEY,
    data JSONB NOT NULL,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

let config = PostgresSinkConfig::new(
    "postgres://localhost/analytics",
    "raw_events",
)
.column_mapping(PostgresColumnMapping::Jsonb { column: "data".into() })
.with_batch_size(1000)
.max_connections(5);

let sink = PostgresSink::new(config).await?;
sink.write_batch(&records).await?;

AutoMap mode -- map JSON keys to table columns

This mode automatically discovers column names from the table schema and maps matching JSON keys. Columns use SQL-injection-safe quoted identifiers.

-- Table schema
CREATE TABLE events (
    user_id TEXT,
    event TEXT,
    timestamp TIMESTAMPTZ,
    amount NUMERIC
);

let config = PostgresSinkConfig::new(
    "postgres://localhost/analytics",
    "events",
)
.column_mapping(PostgresColumnMapping::AutoMap)
.with_batch_size(1000)
.max_connections(10);

let sink = PostgresSink::new(config).await?;

let records = vec![
    json!({"user_id": "u1", "event": "purchase", "amount": 29.99}),
    json!({"user_id": "u2", "event": "signup"}),  // missing "amount" will be null
];
sink.write_batch(&records).await?;

Connection pooling for high-throughput workloads

let config = PostgresSinkConfig::new(
    "postgres://writer:pass@db-primary.internal:5432/warehouse",
    "metrics",
)
.max_connections(20)
.with_batch_size(1000);

let sink = PostgresSink::new(config).await?;

How It Works

• A connection pool is created in PostgresSink::new() using sqlx::PgPool with the configured max_connections.
• write_batch() slices records into chunks of batch_size (or forwards the whole slice when batch_size = 0) and inserts each chunk using a single multi-row INSERT statement.
• In JSONB mode, inserts use INSERT INTO table (col) SELECT * FROM unnest($1::jsonb[]) for maximum efficiency.
• In AutoMap mode, each column's name and underlying type (udt_name) are queried from the PostgreSQL catalog, scoped via to_regclass to exactly the relation the INSERT targets (the configured schema, else the search_path-resolved table) — so a same-named table in another schema can't pollute the column set. A multi-row INSERT INTO ... VALUES ($1::int4, $2::timestamptz), ... is built dynamically with a per-column cast, and each value is bound as text so the destination column's input function parses it — numbers, booleans, timestamps, uuids land in their native column types, and json/jsonb columns receive JSON text. (Values are not bound as jsonb regardless of column type — doing so previously made the typed-column example above fail at runtime.) The column set is the union of record keys across the batch (in declared table order), so a field present only in a later record is still written; a row missing a column binds SQL NULL.
• All identifiers (table names, column names) are quoted using quote_ident() to prevent SQL injection.

Exactly-once delivery

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

• write_batch_idempotent(records, scope, token) — writes records and UPSERTs the token into a _faucet_commit_token(scope TEXT, token TEXT) watermark table inside the same transaction, so both either commit together or neither does.
• last_committed_token(scope) — reads the current watermark to let the pipeline skip already-committed pages on resume.

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

pipeline:
  source:
    type: postgres-cdc
    config:
      connection_url: postgres://faucet:faucet@localhost:5432/appdb
      slot_name: faucet_slot
      publication_name: faucet_pub
  sink:
    type: postgres
    config:
      connection_url: postgres://writer:pass@localhost:5432/warehouse
      table_name: change_events
      column_mapping: auto_map
  state:
    type: file
    config:
      path: ./state
delivery: exactly_once

See the Exactly-once delivery cookbook for full rationale and the supported source/sink set.

Lineage dataset URI

postgres://<host>:<port>/<db>?table=<schema.table> (credentials stripped) — e.g. postgres://host:5432/app?table=public.orders.

License

Licensed under MIT or Apache-2.0.