MQTT to TimescaleDB: Ingestion Architecture

Scope, reader and the three roles

This guide is for engineers connecting IoT gateways and building systems that publish over MQTT to a TimescaleDB database: a BMS supervisor publishing AHU and chiller points, LoRaWAN or Modbus gateways publishing meter readings, edge devices publishing environmental sensors. It covers the path from broker to hypertable. What TimescaleDB is and how hypertables, compression and continuous aggregates work is in TimescaleDB Explained; the general storage design is in How to Store IoT Sensor Data.

MQTT is an OASIS standard publish/subscribe messaging protocol designed for small devices and constrained networks. In this pipeline it has a narrow role: devices publish to topics, the broker forwards each message to the subscribers whose topic filters match. The broker can queue messages for disconnected subscribers, but it is not a history store and cannot replay yesterday's data to a new consumer. If you need replay or several independent consumers reading at their own pace, put Kafka between the broker and the database; see Kafka vs MQTT and Designing an IoT Data Pipeline with Kafka.

Topic and payload conventions

Agree the conventions before the first device publishes; they are hard to change later.

• Topic per point, for example bms/hq/ahu-01/supply_air_temp: site, asset and point as levels. Subscribers use the single-level wildcard + and the multi-level wildcard # (which must be last) to select subsets, such as bms/hq/# or bms/+/+/power_kw.
• Measurement time in the payload, in UTC with an offset, for example {"ts":"2026-10-05T08:00:00Z","v":13.2,"q":0}. The PUBLISH properties listed in the MQTT 5.0 specification (payload format, message expiry, topic alias, response topic, correlation data, user properties, subscription identifier, content type) include no timestamp, so the protocol does not carry a measurement time for you; it has to be in the payload (or, by your own convention, a user property). The time a message reaches the broker or the database is not the time it was measured, especially after a gateway reconnects and flushes its buffer.
• A point register in the database that maps each topic to a point_id, unit and asset. Do not create database objects from topic names automatically in production; a typo in a device configuration would create a new table or column.
• One value per message, or a small fixed set per device. Large JSON documents with changing keys move schema decisions into every query.

QoS, duplicates and persistent sessions

MQTT defines three delivery levels. QoS 0 is "at most once": messages can be lost; the specification itself gives ambient sensor data as an example where that may be acceptable. QoS 1 is "at least once": messages arrive, but duplicates can occur, and a re-delivered PUBLISH carries the DUP flag. QoS 2 is "exactly once" between the two parties of each hop.

Exactly-once on the hop does not make the whole pipeline exactly-once. The bridge can write a batch to the database and crash before it acknowledges the messages; on restart the broker delivers them again. Telegraf's MQTT input acknowledges messages only after they have been written to the outputs, which is the right order, but it means the database will see some messages twice. Treat every path as at-least-once and deduplicate in the database. This design uses QoS 1 end to end plus an idempotent insert. In the specification, a QoS 1 PUBLISH is answered by a PUBACK (two packets), while QoS 2 needs a four-packet exchange (PUBLISH, PUBREC, PUBREL, PUBCOMP), one more round trip per message, and the specification itself notes an increased overhead associated with QoS 2. Because the database deduplicates anyway, that overhead buys little here; that is a judgement for this design, not a rule.

Persistent sessions and broker queues

For messages to wait while the bridge is down, the bridge must connect with Clean Start set to 0 and a Session Expiry Interval greater than 0 (in MQTT 3.1.1 terms, clean session false), always with the same client ID, and subscribe with QoS 1. With Clean Start 1 and expiry 0, the broker discards the session on disconnect and everything published meanwhile is lost to that subscriber. Telegraf exposes this as persistent_session = true with a fixed client_id and qos = 1.

The broker's queue is finite. Mosquitto's max_queued_messages defaults to 1000 QoS 1 and 2 messages per client, and its built-in persistence to disk is off by default. When a queue limit is reached, the Mosquitto manual says subsequent messages are silently dropped (the wording is under max_queued_bytes, which works together with max_queued_messages: messages are queued until the first limit is reached). As an illustrative assumption, a campus publishing 5,000 points a minute (about 83 a second) would fill a 1,000-message queue in about 12 seconds of bridge downtime. Either size the broker queue and enable persistence for the outage you expect (from your bridge restart and maintenance times and the message rate), or keep a buffer at the edge (gateways that store and forward), or add Kafka. Other brokers behave differently when a queue is full; check yours.

Scaling out with shared subscriptions

One bridge instance is usually enough for building data. To run several, MQTT 5 shared subscriptions ($share/{ShareName}/{filter}) send each matching message to only one session in the group. The specification notes that with shared subscriptions the order of delivery is not guaranteed between clients, so two readings of the same point can be written out of order. With the measurement time in the payload and a (point_id, ts) key that does not matter for storage, but it does for any logic in the bridge that assumes order.

The bridge: Telegraf or a small consumer

Telegraf (InfluxData's collection agent) has an MQTT Consumer input and a PostgreSQL output. The input subscribes, can extract tags from topic levels, and with the JSON v2 parser takes fields and the timestamp from the payload. The output writes to PostgreSQL or compatible servers and manages the schema itself: it creates tables per measurement and adds columns when new tags or fields appear, through SQL templates you can override, including a documented sample that calls create_hypertable. Two documented behaviours matter for this design: if the timestamp path finds nothing, JSON v2 uses the current time; and when the output meets an error it classifies as permanent, it discards that sub-batch.

For those reasons, this design lets Telegraf write to a landing schema that it manages, and run the validated load into the curated hypertable as SQL (next section). The alternative is a small consumer of your own (any MQTT client library plus a PostgreSQL driver) that writes raw topic and payload into a landing table in batches, then acknowledges. It is a few hundred lines to maintain instead of configuration, but nothing is inferred for you.

[[inputs.mqtt_consumer]]
  servers = ["ssl://broker.hq.internal:8883"]
  topics = ["bms/#"]
  qos = 1
  persistent_session = true
  client_id = "telegraf-ingest-hq"
  data_format = "json_v2"

  ## no topic setting: same form as the documentation's field pivoting example
  ## (topics = "/sensors/#" with only measurement and tags under topic_parsing)
  [[inputs.mqtt_consumer.topic_parsing]]
    tags = "_/site/asset/point"

  [[inputs.mqtt_consumer.json_v2]]
    timestamp_path = "ts"
    timestamp_format = "2006-01-02T15:04:05Z07:00"
    [[inputs.mqtt_consumer.json_v2.field]]
      path = "v"
      type = "float"
    [[inputs.mqtt_consumer.json_v2.field]]
      path = "q"
      type = "int"

[[outputs.postgresql]]
  connection = "postgres://telegraf_ingest@db.hq.internal/telemetry?sslmode=verify-full"
  schema = "landing"
  timestamp_column_type = "timestamp with time zone"

Notes on the Telegraf configuration

• Port 8883 with ssl://. The MQTT 5.0 specification notes that TCP ports 8883 and 1883 are registered with IANA for MQTT over TLS and without TLS respectively. The Telegraf documentation shows the ssl:// scheme in its examples (with port 1883) plus tls_ca, tls_cert and tls_key options; use whatever port your broker's TLS listener is configured on.
• [[inputs.mqtt_consumer.json_v2]]. The JSON v2 documentation describes the parser configuration as a json_v2 sub-table of the input plugin; its examples use [[inputs.file.json_v2]], and the block above applies the same pattern to the MQTT input.
• timestamp_format. The documentation accepts unix variants or a Go reference time layout; 2006-01-02T15:04:05Z07:00 is Go's RFC 3339 layout, chosen here to match ISO 8601 timestamps such as 2026-10-05T08:00:00Z. Check it against a sample payload before go-live.
• topic_parsing. The Telegraf documentation does not describe MQTT + wildcards in the topic_parsing topic setting, so this block does not use them. The documentation describes _ for an ignored level and # for a variable-length element; its topic parsing example puts one literal topic in topic, and its field pivoting example subscribes to /sensors/# and gives topic_parsing only measurement and tags, with no topic setting. The block above follows that second form. The documentation does not say how entries without topic are matched, so check the resulting tags on a sample message before go-live; if you prefer the documented literal form, add one topic_parsing entry per literal topic.

Batching

Whichever bridge you use, write in batches. The PostgreSQL documentation recommends several rows per transaction, or COPY, because committing each row separately does a lot of work per row. Telegraf batches by design: its agent settings control batch size and flush interval, and the MQTT input's max_undelivered_messages (default 1000) limits how many messages are read but not yet written, which the documentation says should be chosen together with the agent's metric_batch_size. For a small consumer, flushing every few hundred to few thousand messages or every second or two, whichever comes first, is an illustrative starting point; choose the values from your message rate, acceptable ingest delay and the broker's in-flight limits. Acknowledge only after the commit.

Landing table and an idempotent, validated load

The example uses the small-consumer landing shape: the raw topic and payload text, plus arrival time. A scheduled job (every few seconds to a minute, an illustrative interval set by how fresh dashboards need to be) takes the rows present at its start as one batch, rejects what cannot be loaded with a reason, and inserts the rest with ON CONFLICT DO NOTHING, all in one transaction. The sample contains a QoS 1 redelivery (inbox rows 1 and 4), a non-numeric value, an unknown topic, a payload that is not JSON and a payload without a timestamp.

The checks use a CASE so that the JSON cast is attempted only after pg_input_is_valid() has confirmed the payload is valid JSON; PostgreSQL documents that CASE does not evaluate subexpressions it does not need. A cast inside a plain WHERE ... OR ... could fail the whole batch on one bad message.

SET TimeZone = 'UTC';

CREATE TABLE point (
    point_id    integer GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    mqtt_topic  text NOT NULL UNIQUE,
    unit        text NOT NULL
);

CREATE TABLE reading (
    point_id  integer NOT NULL REFERENCES point (point_id),
    ts        timestamptz NOT NULL,
    value     double precision NOT NULL,
    quality   smallint NOT NULL DEFAULT 0,
    PRIMARY KEY (point_id, ts)
);

CREATE TABLE mqtt_inbox (
    inbox_id     bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    received_at  timestamptz NOT NULL,
    topic        text NOT NULL,
    payload      text NOT NULL
);

CREATE TABLE mqtt_reject (
    inbox_id     bigint PRIMARY KEY,
    topic        text NOT NULL,
    payload      text NOT NULL,
    reason       text NOT NULL
);

INSERT INTO point (mqtt_topic, unit) VALUES
    ('bms/hq/ahu-01/supply_air_temp', 'degC'),
    ('bms/hq/ahu-01/zone_co2',        'ppm'),
    ('bms/hq/ch-01/power_kw',         'kW');

INSERT INTO mqtt_inbox (received_at, topic, payload) VALUES
    ('2026-10-05 08:00:01+00', 'bms/hq/ahu-01/supply_air_temp', '{"ts":"2026-10-05T08:00:00Z","v":13.2,"q":0}'),
    ('2026-10-05 08:00:01+00', 'bms/hq/ahu-01/zone_co2',        '{"ts":"2026-10-05T08:00:00Z","v":612,"q":0}'),
    ('2026-10-05 08:00:02+00', 'bms/hq/ch-01/power_kw',         '{"ts":"2026-10-05T08:00:00Z","v":184.5,"q":0}'),
    ('2026-10-05 08:00:09+00', 'bms/hq/ahu-01/supply_air_temp', '{"ts":"2026-10-05T08:00:00Z","v":13.2,"q":0}'),
    ('2026-10-05 08:01:01+00', 'bms/hq/ahu-01/supply_air_temp', '{"ts":"2026-10-05T08:01:00Z","v":13.4,"q":0}'),
    ('2026-10-05 08:01:01+00', 'bms/hq/ahu-01/zone_co2',        '{"ts":"2026-10-05T08:01:00Z","v":"n/a","q":0}'),
    ('2026-10-05 08:01:02+00', 'bms/hq/ahu-02/supply_air_temp', '{"ts":"2026-10-05T08:01:00Z","v":14.1,"q":0}'),
    ('2026-10-05 08:01:03+00', 'bms/hq/ch-01/power_kw',         'power_kw=186.0'),
    ('2026-10-05 08:01:03+00', 'bms/hq/ahu-01/zone_co2',        '{"v":618,"q":0}');

BEGIN;

-- 1. take the current inbox rows as one batch (intent: rows that arrive meanwhile wait for the next run; run single-session only)
CREATE TEMP TABLE batch ON COMMIT DROP AS
SELECT inbox_id, topic, payload FROM mqtt_inbox ORDER BY inbox_id FOR UPDATE;
DELETE FROM mqtt_inbox i USING batch b WHERE i.inbox_id = b.inbox_id;

-- 2. reject what cannot be loaded, with a reason (CASE tests run in order)
INSERT INTO mqtt_reject (inbox_id, topic, payload, reason)
SELECT inbox_id, topic, payload, reason
FROM (
    SELECT b.inbox_id, b.topic, b.payload,
           CASE WHEN p.point_id IS NULL
                    THEN 'unknown topic'
                WHEN NOT pg_input_is_valid(b.payload, 'jsonb')
                    THEN 'invalid JSON'
                WHEN NOT coalesce((b.payload::jsonb ->> 'v') ~ '^-?[0-9]+(\.[0-9]+)?$', false)
                    THEN 'value missing or not numeric'
                WHEN NOT coalesce(pg_input_is_valid(b.payload::jsonb ->> 'ts', 'timestamptz'), false)
                    THEN 'timestamp missing or invalid'
           END AS reason
    FROM batch b
    LEFT JOIN point p ON p.mqtt_topic = b.topic
) checked
WHERE reason IS NOT NULL;

-- 3. load the rest; QoS 1 redeliveries hit the primary key and are skipped
INSERT INTO reading (point_id, ts, value, quality)
SELECT p.point_id,
       (b.payload::jsonb ->> 'ts')::timestamptz,
       (b.payload::jsonb ->> 'v')::double precision,
       (b.payload::jsonb ->> 'q')::smallint
FROM batch b
JOIN point p ON p.mqtt_topic = b.topic
WHERE NOT EXISTS (SELECT 1 FROM mqtt_reject x WHERE x.inbox_id = b.inbox_id)
ON CONFLICT (point_id, ts) DO NOTHING;

COMMIT;

SELECT count(*) AS inbox_left FROM mqtt_inbox;
--  inbox_left
-- ------------
--           0

SELECT inbox_id, topic, reason FROM mqtt_reject ORDER BY inbox_id;
--  inbox_id |             topic             |            reason
-- ----------+-------------------------------+------------------------------
--         6 | bms/hq/ahu-01/zone_co2        | value missing or not numeric
--         7 | bms/hq/ahu-02/supply_air_temp | unknown topic
--         8 | bms/hq/ch-01/power_kw         | invalid JSON
--         9 | bms/hq/ahu-01/zone_co2        | timestamp missing or invalid

SELECT p.mqtt_topic, r.ts::text AS ts, r.value
FROM reading r JOIN point p USING (point_id)
ORDER BY r.ts, p.point_id;
--           mqtt_topic           |           ts           | value
-- -------------------------------+------------------------+-------
--  bms/hq/ahu-01/supply_air_temp | 2026-10-05 08:00:00+00 |  13.2
--  bms/hq/ahu-01/zone_co2        | 2026-10-05 08:00:00+00 |   612
--  bms/hq/ch-01/power_kw         | 2026-10-05 08:00:00+00 | 184.5
--  bms/hq/ahu-01/supply_air_temp | 2026-10-05 08:01:00+00 |  13.4

-- Without ON CONFLICT, a redelivered reading fails the statement:
INSERT INTO reading (point_id, ts, value) VALUES (1, '2026-10-05 08:00:00+00', 13.2);
-- ERROR:  duplicate key value violates unique constraint "reading_pkey"
-- DETAIL:  Key (point_id, ts)=(1, 2026-10-05 08:00:00+00) already exists.
-- SQLSTATE 23505

Notes on the load

Nine messages produced four readings and four rejects; the ninth, the duplicate of inbox row 1, was skipped by ON CONFLICT DO NOTHING inside the same statement. That works for DO NOTHING; with DO UPDATE the documentation says a statement may not affect the same row twice, so deduplicate the batch first if you upsert. pg_input_is_valid() exists from PostgreSQL 16. The job was run in a single session only, so it is an educational example, not a demonstration that overlapping runs are safe. Two properties do rest on documented behaviour: the delete is scoped to the inbox_id values in the batch, so a row that was not in the batch is never deleted by this run, and under Read Committed a statement sees only data committed before it began, so rows the bridge commits later simply wait for the next run. What was not run is two jobs at once: the FOR UPDATE row locks are intended to make the second wait on the first, but how the two batches then divide the rows was not exercised. PostgreSQL also documents FOR UPDATE SKIP LOCKED for multiple consumers of a queue-like table, while warning that it gives an inconsistent view of the data; it was not run here and this design does not need it.

The safer variant below enforces one load job at a time with a documented mechanism instead of relying on the scheduler: a transaction-level advisory lock taken as the first statement. PostgreSQL documents that transaction-level advisory locks are released automatically at the end of the transaction, so a job that fails and rolls back cannot leave it held, and a second job waits at that statement until the first commits or rolls back, then takes its batch from what is left. Advisory locks only work if every load job uses the same key; the system does not enforce them on the table. Ingest is not blocked. Use the transaction-level form (pg_advisory_xact_lock), not the session-level one, which survives the transaction. This variant ran in a single session with the same results as above; the waiting between two jobs follows from the documentation and was not run here. Keep the reject table small by reviewing it daily: unknown topics usually mean a device was added without registering its points. For the wider topic of duplicates, see Handling Duplicate Telemetry Events in SQL.

BEGIN;
-- only one load job at a time: a second job waits here until the first commits or rolls back
SELECT pg_advisory_xact_lock(hashtext('mqtt_inbox_load'));

-- steps 1 to 3 exactly as in the job above:
-- CREATE TEMP TABLE batch ... FOR UPDATE; DELETE ... USING batch;
-- INSERT INTO mqtt_reject ...; INSERT INTO reading ... ON CONFLICT (point_id, ts) DO NOTHING;

COMMIT;

-- same results as above: inbox_left 0, the same four rejects and four readings
SELECT count(*) AS advisory_locks_held FROM pg_locks WHERE locktype = 'advisory';
--  advisory_locks_held
-- ---------------------
--                    0

The readings table as a TimescaleDB hypertable

In TimescaleDB the curated reading table becomes a hypertable. From TimescaleDB 2.23, CREATE TABLE ... WITH (tsdb.hypertable) partitions on the first timestamp column automatically. In the current reference the columnstore is on by default (tsdb.columnstore defaults to true) and TimescaleDB creates a columnstore policy that converts chunks after one chunk interval; segmentby and orderby set how the columnstore groups and orders data. The default chunk interval is 7 days. A primary key must include the partition column, which (point_id, ts) does. Choose the chunk interval so the indexes of the chunks currently being written fit in about a quarter of memory, as the TimescaleDB sizing guidance puts it; the one-day interval below is an illustrative assumption for one-minute building data, to be checked against your row rate, index sizes and memory. See Choosing Time-Series Partition Intervals for the reasoning.

CREATE TABLE reading (
    point_id  integer NOT NULL,
    ts        timestamptz NOT NULL,
    value     double precision NOT NULL,
    quality   smallint NOT NULL DEFAULT 0,
    PRIMARY KEY (point_id, ts)
) WITH (
    tsdb.hypertable,
    tsdb.segmentby = 'point_id',
    tsdb.orderby   = 'ts DESC'
);

-- Change the chunk interval (applies to chunks created afterwards)
SELECT set_chunk_time_interval('reading', INTERVAL '1 day');

Late data, duplicates and compression

Gateways that reconnect after an outage send old readings, which land in chunks that may already be in the columnstore. The TimescaleDB documentation states that inserts, updates, deletes and upserts work directly on the columnstore, so the same ON CONFLICT DO NOTHING load is expected to keep working; that was not run here, because no TimescaleDB instance was run for this page. TimescaleDB also offers "direct compress" during ingestion, which the current reference marks as a tech preview and not production-ready, and it says tables with unique or exclusion constraints cannot use it, so it does not fit this deduplicating design. For late data in general see Late-Arriving and Out-of-Order Sensor Events, and for roll-ups Continuous Aggregates vs Materialized Views.

Security, monitoring, limitations and when not to use this

Security. Use TLS from devices to broker and from bridge to database. Give each device its own broker credentials and an ACL that allows publishing only to its own topics; give the bridge read access to the topics it ingests. The database role used by the bridge should insert into the landing table only; the load job runs under a separate role that owns the curated tables.

Monitoring. Track broker queue depth for the bridge's session, messages received by the bridge (Telegraf reports a messages_received count through its internal metrics), landing-table backlog, rejects per hour by reason, and ingest lag (load time minus measurement time) per site. Alert when lag or backlog grows, before the broker queue fills.

Limitations. This design gives at-least-once delivery with deduplication by (point_id, ts). Two genuinely different readings with the same point and timestamp (a device that resets its clock) are indistinguishable from a duplicate and the second is dropped; log conflicts if that matters. The broker queue is the only buffer between gateway and bridge unless you add one.

When not to use it. If a handful of devices post data every few minutes, an HTTPS endpoint writing straight to PostgreSQL is simpler than running a broker. If several systems must consume the same stream independently, or you need to replay history into a new store, add Kafka. If your devices already write InfluxDB line protocol and you do not need SQL joins to asset data, compare the options in TimescaleDB vs InfluxDB. Building data that starts in BACnet is covered in BACnet Readings to PostgreSQL; for the overall picture, the time-series overview.

Frequently asked questions

Sources

• OASIS MQTT Version 5.0 specification
• MQTT.org: MQTT, the standard for IoT messaging
• Eclipse Mosquitto: mosquitto.conf manual
• Telegraf documentation: MQTT Consumer input plugin
• Telegraf documentation: JSON v2 input data format
• Telegraf documentation: PostgreSQL output plugin
• TimescaleDB reference: CREATE TABLE
• TimescaleDB documentation: Hypertable indexes
• TimescaleDB documentation: Size hypertable chunks
• TimescaleDB documentation: Understand hypercore
• PostgreSQL 18 documentation: Populating a Database
• PostgreSQL 18 documentation: INSERT
• PostgreSQL 18 documentation: System Information Functions (pg_input_is_valid)
• PostgreSQL 18 documentation: Conditional Expressions
• PostgreSQL 18 documentation: SELECT (locking clause, SKIP LOCKED)
• PostgreSQL 18 documentation: Transaction Isolation
• PostgreSQL 18 documentation: Explicit Locking (advisory locks)

Checked 27 October 2026.

How we research these articles: our editorial method. Research-based from official documentation; SQL examples are run on a local database where the page says so. CodeWithSQL earns nothing from the vendors mentioned.