pgmqtt

Overview

requires wal_level = logical for CDC.

Version

Build

You can build the RPM / DEB packages for pgmqtt using pig build:

pig build pkg pgmqtt         # build RPM / DEB packages

Install

You can install pgmqtt directly. First, make sure the PGDG and PIGSTY repositories are added and enabled:

pig repo add pgsql -u          # Add repo and update cache

Install the extension using pig or apt/yum/dnf:

• Install
• pig
• dnf
• apt

pig install pgmqtt;          # Install for current active PG version

pig ext install -y pgmqtt -v 18  # PG 18
pig ext install -y pgmqtt -v 17  # PG 17
pig ext install -y pgmqtt -v 16  # PG 16
pig ext install -y pgmqtt -v 15  # PG 15
pig ext install -y pgmqtt -v 14  # PG 14

dnf install -y pgmqtt_18       # PG 18
dnf install -y pgmqtt_17       # PG 17
dnf install -y pgmqtt_16       # PG 16
dnf install -y pgmqtt_15       # PG 15
dnf install -y pgmqtt_14       # PG 14

apt install -y postgresql-18-pgmqtt   # PG 18
apt install -y postgresql-17-pgmqtt   # PG 17
apt install -y postgresql-16-pgmqtt   # PG 16
apt install -y postgresql-15-pgmqtt   # PG 15
apt install -y postgresql-14-pgmqtt   # PG 14

Create Extension:

CREATE EXTENSION pgmqtt;

Usage

Sources:

• pgmqtt 0.4.1 README
• pgmqtt 0.4.1 interfaces
• pgmqtt 0.4.1 configuration
• pgmqtt 0.4.1 limitations
• pgmqtt 0.4.1 release notes

pgmqtt embeds an MQTT broker in PostgreSQL. It can publish INSERT, UPDATE, and DELETE changes through logical decoding and can map inbound MQTT topics and JSON payloads to table writes. Use it when database and MQTT integration justify running a network broker inside the PostgreSQL server process.

Preload and Create the Extension

Set logical WAL and preload the worker, then restart PostgreSQL:

wal_level = logical
shared_preload_libraries = 'pgmqtt'

Create the extension after restart:

CREATE EXTENSION pgmqtt;

Listener addresses, ports, authentication, and TLS settings are read by the background worker. Settings documented as startup-only require a worker/server restart, not just pg_reload_conf().

Publish Table Changes

Create an outbound mapping:

SELECT pgmqtt_add_outbound_mapping(
  'public',
  'orders',
  'orders/{{ op | lower }}',
  '{{ columns | tojson }}',
  1
);

The mapping publishes row changes to topics such as orders/insert. The interface also accepts a QoS and template type where supported. Version 0.4.1 drains CDC changes in batches of up to 4096 records.

Inspect or remove outbound mappings:

SELECT * FROM pgmqtt_list_outbound_mappings();
SELECT pgmqtt_remove_outbound_mapping('public', 'orders');

Write Rows from MQTT

Map captured topic segments and JSON fields to a target table:

SELECT pgmqtt_add_inbound_mapping(
  'sensor/{site_id}/temperature',
  'sensor_readings',
  '{"site_id":"{site_id}","value":"$.temperature"}'::jsonb
);

Inbound mappings support insert and documented upsert/delete modes with options such as target_schema, conflict_columns, mapping_name, and template_type. Grant the worker role only the required table privileges and validate payload types and constraints.

SELECT * FROM pgmqtt_list_inbound_mappings();
SELECT pgmqtt_remove_inbound_mapping('temp_readings');

Administration and Status

SELECT * FROM pgmqtt_status();
SELECT pgmqtt_disconnect_client('device-42');
SELECT pgmqtt_disconnect_role('mqtt_devices');
SELECT pgmqtt_reload_acls('*');

pgmqtt_status reports listener, client, subscription, retained-message, CDC, inbound-write, and dead-letter state. Administrative calls are queued for asynchronous processing by the worker.

Configuration Index

• pgmqtt.mqtt_enabled and pgmqtt.mqtt_port: TCP MQTT listener.
• pgmqtt.ws_enabled and pgmqtt.ws_port: WebSocket listener.
• pgmqtt.tick_interval_ms and pgmqtt.cdc_every_n_ticks: worker cadence.
• pgmqtt.max_client_buffer_bytes: per-client flow-control boundary.
• pgmqtt.debug_log and pgmqtt.metrics_*: diagnostics and metrics integration.
• pgmqtt TLS, JWT, password-authentication, and ACL settings: transport and client access controls; availability differs between community and enterprise features.

Protocol and CDC Boundaries

• MQTT 5.0 and 3.1.1 are supported. QoS 0 and 1 are implemented; requested QoS 2 is downgraded to QoS 1.
• CDC covers INSERT, UPDATE, and DELETE, not DDL or TRUNCATE. DELETE payloads may require REPLICA IDENTITY FULL.
• The CDC ring has a finite capacity of 8192 and drops the oldest records on overflow. The QoS 0 topic buffer is capped at 4096 and also drops oldest entries; QoS 1 buffering can grow without a fixed bound.
• The community edition documents TLS through a proxy, while native TLS and some JWT features are enterprise boundaries. Verify the edition before setting listener expectations.

Version 0.4.1 and Operations

The 0.4 line consolidates HTTP/worker handling and reduces panic paths; 0.4.1 raises CDC batch processing to 4096. These changes improve throughput and structure but do not make the embedded broker lossless under every overload or crash.

Running a broker inside PostgreSQL expands the database network and resource boundary. Isolate listener interfaces, enforce authentication and topic ACLs, monitor worker lag and dropped buffers, and test failover and restart behavior before production use.