Postgres Sink

The PostgreSQL sink connector consumes messages from Iggy topics and stores them in a PostgreSQL table, with optional automatic table creation and configurable payload storage.

This page is a curated subset of the documentation. The canonical reference, including SQL query recipes for each payload format, is the upstream postgres_sink README in the apache/iggy repository.

Features

• Automatic Table Creation: Optionally create the target table on startup
• Batch Processing: Insert messages in configurable batches for performance
• Metadata Storage: Store Iggy message metadata (offset, timestamp, topic, etc.)
• Configurable Payload Storage: Store the payload as raw bytes, native JSONB, or text
• Automatic Retries: Transient database errors are retried with a configurable delay
• Connection Pooling: Efficient database connection management

Configuration

Use the broker credentials and runtime setup from the sink guide. From the matching 0.9.0/edge repository root, build the plugin:

cargo build --release -p iggy_connector_postgres_sink

For a local PostgreSQL instance matching the example credentials:

docker run -d --name iggy-postgres -p 127.0.0.1:5432:5432 -e POSTGRES_USER=iggy -e POSTGRES_PASSWORD=iggy -e POSTGRES_DB=iggy postgres:15-alpine
docker exec iggy-postgres pg_isready -U iggy -d iggy

Wait for the readiness command to report accepting connections. Save this connector file in the runtime's connector directory. Replace the URI for your deployment. Startup opens the SQLx pool and runs SELECT 1; it creates the table only when requested.

type = "sink"
key = "postgres-sink"
enabled = true
version = 1
name = "Postgres Sink"
path = "target/release/libiggy_connector_postgres_sink"

[[streams]]
stream = "user_events"
topics = ["events"]
schema = "json"
batch_length = 100
poll_interval = "5ms"
consumer_group = "postgres-sink"

[plugin_config]
connection_string = "postgresql://iggy:iggy@localhost:5432/iggy"
target_table = "iggy_messages"
auto_create_table = true
batch_size = 100
payload_format = "bytea"

Create the Iggy resources and send a message:

./target/release/iggy --username iggy --password iggy stream create user_events
./target/release/iggy --username iggy --password iggy topic create user_events events 1 none 1d
./target/release/iggy --username iggy --password iggy message send --partition-id 0 user_events events '{"user_id":"42","status":"active"}'

Plugin config options

Payload Format

The payload_format option determines the type of the payload column and how the message payload is stored:

Format names are case-insensitive; unrecognized names silently use bytea. JSONB accepts objects, arrays and scalars, but PostgreSQL rejects \u0000 in JSONB and zero bytes in TEXT. JSON decoding and JSONB storage can change the original representation. Message headers are not stored. See PostgreSQL JSON types for JSONB restrictions.

With the default bytea format, valid UTF-8 JSON payloads can still be queried by converting the bytes: convert_from(payload, 'UTF8')::jsonb->>'user_id'. With json, the native operators apply directly: payload->>'user_id'.

Table Schema

With auto_create_table = true and all three metadata flags enabled, the connector creates the following structure if the table is absent. The payload column type follows payload_format:

CREATE TABLE iggy_messages (
    id DECIMAL(39, 0) PRIMARY KEY,
    iggy_offset BIGINT,
    iggy_timestamp TIMESTAMP WITH TIME ZONE,
    iggy_stream TEXT,
    iggy_topic TEXT,
    iggy_partition_id INTEGER,
    iggy_checksum BIGINT,
    iggy_origin_timestamp TIMESTAMP WITH TIME ZONE,
    payload BYTEA,
    created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
);

Disabling include_metadata omits the five offset/timestamp/stream/topic/partition columns. Checksum and origin timestamp flags work independently. id, payload and created_at are always created. With auto_create_table = false, provision a compatible table yourself; startup does not check its existence or schema.

id stores the full unsigned 128-bit message ID as a decimal, without stream, topic, partition or offset in the key. Offset and checksum values are cast to signed 64-bit integers, and partition IDs to signed 32-bit integers; values above those signed ranges appear negative. Message and origin timestamps are interpreted as Unix microseconds. Zero becomes the Unix epoch; a timestamp outside the date library's range falls back to the current time. created_at is generated by PostgreSQL when inserting.

Querying the Data

Run these queries with docker exec -it iggy-postgres psql -U iggy -d iggy, or your PostgreSQL client:

-- Get all messages from a specific stream
SELECT * FROM iggy_messages WHERE iggy_stream = 'user_events';

-- Read a bytea payload as text (for text/JSON payloads)
SELECT id, iggy_offset, convert_from(payload, 'UTF8') as payload_text
FROM iggy_messages
WHERE iggy_stream = 'user_events';

For a JSONB query example, a TEXT search example, and recommended indexes, see the upstream postgres_sink README.

Delivery Semantics

Each poll is split into chunks of at most batch_size messages. One multi-row INSERT writes each chunk; a duplicate primary key, incompatible table, invalid payload or other terminal error rejects that chunk. Later chunks are still attempted. The plugin logs failures, increments its private insertion-error count by the failed chunk size, and returns success after attempting all chunks. Both its processed count and runtime's processed count can therefore include messages that were never stored. Runtime error counts do not expose these insert failures.

Transient insert failures are retried within the chunk. max_retries counts total attempts, and waits grow linearly as retry_delay, twice that delay, and so on. Retried SQLSTATEs are 40001, 40P01, 57P01, 57P02, 57P03, 08000, 08003 and 08006; SQLx I/O and pool-acquisition timeouts are also retried. Other errors stop that chunk's retry loop. Startup has no plugin retry loop.

The sink uses plain INSERT, with no conflict-ignore or upsert clause. Reusing an ID already in the table rejects the entire chunk, including any new IDs alongside it. There is no transaction covering the complete poll. The runtime auto-commits while polling and does not replay failed chunks; a connection failure can leave the write outcome uncertain. These retries do not provide an end-to-end at-least-once guarantee.

Performance Considerations

• Use an appropriate batch_size for your workload (larger batches give better throughput)
• Create indexes on frequently queried columns (iggy_stream, iggy_topic, created_at)
• Set the pool limit with max_connections; it does not report pool usage
• Keep each actual chunk within PostgreSQL's 65,535 query-parameter limit. Each row binds 2 + 5 * include_metadata + include_checksum + include_origin_timestamp parameters, so the default flags allow at most 7,281 rows per statement. Larger chunks are rejected by SQLx; batch_length also limits how many rows reach a poll.