# faucet-sink-postgres
[](https://crates.io/crates/faucet-sink-postgres)
[](https://docs.rs/faucet-sink-postgres)
PostgreSQL sink connector for the [faucet-stream](https://github.com/PawanSikawat/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](#streaming-and-batching) for the tradeoffs.
## Installation
```toml
[dependencies]
faucet-sink-postgres = "1.0"
tokio = { version = "1", features = ["full"] }
```
Or via the umbrella crate:
```toml
faucet-stream = { version = "1.0", features = ["sink-postgres"] }
```
## Quick Start
```rust
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
| Field | Type | Default | Description |
|-------|------|---------|-------------|
| `connection_url` | `String` | *(required)* | PostgreSQL connection URL (e.g. `postgres://user:pass@host:5432/db`) |
| `table_name` | `String` | *(required)* | Target table name |
| `schema` | `Option<String>` | `null` | Schema (namespace) qualifying `table_name`. When set, both AutoMap column discovery and the `INSERT` target `schema.table_name` explicitly. When unset, the table resolves against the connection's `search_path` |
| `column_mapping` | `PostgresColumnMapping` | `Jsonb { column: "data" }` | How to map JSON records to table columns (see below) |
| `batch_size` | `usize` | `1000` | Maximum rows per multi-row `INSERT`. See [Streaming and batching](#streaming-and-batching) below |
| `max_connections` | `u32` | `5` | Maximum number of connections in the connection pool |
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`)
| Variant | Description |
|---------|-------------|
| `Jsonb { column }` | Insert each record as a single `jsonb` column. The column name defaults to `"data"` but can be overridden. Uses PostgreSQL's `unnest($1::jsonb[])` for efficient batch inserts. |
| `AutoMap` | Map top-level JSON keys directly to table columns. Column names are discovered from the PostgreSQL catalog, scoped (via `to_regclass`) to exactly the relation the `INSERT` targets — the configured `schema` if set, otherwise the `search_path`-resolved table. Only keys that match existing columns are inserted; extra keys are silently ignored. Records with no matching keys are skipped with a warning. |
### Builder Methods
```rust
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.
| Field | Type | Default | Description |
|-------|------|---------|-------------|
| `write_mode` | `"append" \| "upsert" \| "delete"` | `"append"` | How records are applied. |
| `key` | `[String]` | `[]` | Key columns identifying a row. **Required and non-empty** for `upsert`/`delete`. Composite keys are supported. |
| `delete_marker` | `{ field, values: [String] }` | *(none)* | **Upsert only.** Records whose `field` equals one of `values` are deleted by key; all others are upserts. The marker field is stripped from upserted rows before writing. |
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
```yaml
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
```rust
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)
```json
{
"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)
```json
{
"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
```env
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
```rust
use faucet_core::Sink;
let sink = PostgresSink::new(config).await?;
let schema = sink.config_schema();
println!("{}", serde_json::to_string_pretty(&schema)?);
```
## Pipeline Usage
```rust
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.
```sql
-- Table schema
CREATE TABLE raw_events (
id SERIAL PRIMARY KEY,
data JSONB NOT NULL,
created_at TIMESTAMPTZ DEFAULT NOW()
);
```
```rust
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.
```sql
-- Table schema
CREATE TABLE events (
user_id TEXT,
event TEXT,
timestamp TIMESTAMPTZ,
amount NUMERIC
);
```
```rust
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
```rust
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.
```yaml
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](https://pawansikawat.github.io/faucet-stream/cookbook/state.html#exactly-once-delivery) 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.