PostgreSQL

6 min

PostgreSQL Integration enables Manufacturing Connect Edge (MCe) to connect with a PostgreSQL database. The connector writes data from Message Broker into a target table and also supports ad hoc queries from the Analytics IT Hub processor.

Supported communication

Outbound (Publish data from MCe to PostgreSQL): Yes

Inbound (Subscribe data from PostgreSQL to MCe): No

Analytics IT Hub processor support: Yes

Configurations

Field	Details	Required
Name	Connector name.	Yes
Hostname	DNS name or IP of the host.	Yes
Port	Port of PostgreSQL host.	Yes
SSL Mode	Controls certificate validation: disable skips TLS entirely, require encrypts but skips verification, verify-ca validates the CA certificate, and verify-full also validates the hostname.	No
CA Certificate	Add a CA certificate here, or add it to the system-wide certificate list.	No
Certificate	Client certificate for mutual TLS authentication with PostgreSQL. Paste the PEM-encoded certificate here.	No
Private key	Private key corresponding to the client certificate. Must match the certificate provided above.	No
Username	User name for authentication.	Yes
Password	Password for authentication.	No
Database	Database name.	Yes
Schema	Schema of the target table. Leave empty to use the default search path of the database (usually public).	No
Table	Target table to write data into. Must exist in the selected database unless Create table is enabled.	Yes
Mapping	Key is a column in your table. Value is a payload field name (deviceId), the matching default-schema column name (device_id), or static text prefixed with @ (@plant-a). Leave empty to use the default schema. When Enable JSONPath is on, a value starting with $ is read as a JSONPath expression instead of a field name.	No
Enable JSONPath	Read mapping values that start with $ as JSONPath expressions, so a column can be filled from a nested payload field such as $.factory.line.machine.sensors.temperature. Dotted paths, quoted keys, and array indexes are supported. Values that do not start with $ keep their usual meaning, and an expression matching nothing writes NULL.	No
Create table	Create the table if it doesn't exist. Requires the appropriate database permissions.	Yes
Commit timeout	Transaction commit timeout (ms).	No
Batch timeout	The maximum time (in milliseconds) to wait after the previous commit attempt before writing an incomplete batch to the database.	No
Max transaction size	Maximum number of messages before a transaction is committed regardless of timeout.	No
Bulk insert count	Buffer messages and perform bulk insert once the Bulk insert count value is reached.	No
Write mode	How the connector acknowledges messages. sync: a message is acknowledged only after its batch is committed, so with persistent storage on, a failed write is kept and retried. async: a message is acknowledged once buffered, for higher throughput, and a batch that fails to write is dropped. Default: sync.	No
Parallel transactions	The maximum number of batch inserts that run at the same time, each on its own database connection. Values above 1 increase throughput, but rows from different batches can be written out of publish order. Default: 1.	No
Parallel Publish Count	The number of simultaneously publishing messages.	No
Throttling limit	Set a limit on the number of outbound messages per second. If sending data to a server that has restrictions on the rate of incoming messages, this ensures that no messages are lost.	No

Persistent storage configurations

Field	Details	Required
Persistent storage	If enabled, messages in the outbound queue are stored on disk. If there are interruptions in connection or power loss, messages are retained and sent when the connection is restored.	No
Persistent storage size limit	Set the size limit for storing persisted data in GB.	No
Queue Mode	Select the sequence of data transfer: FIFO or LIFO. FIFO (First In First Out) sends the earliest message first. LIFO (Last In First Out) sends the latest message first.	No

JSONPath expressions

Off by default. With it off, every mapping value keeps its usual meaning and nothing about an existing configuration changes. With it on, a mapping value starting with $ is evaluated against the published payload, so a column can be filled from a nested field. Values that do not start with $ still resolve as before: payload field name, default-schema column name, or @ literal.

Given this payload:

{
  "factory": {
    "line": {
      "product": "Engine Block",
      "machine": {
        "status": "running",
        "sensors": { "temperature": 72.4 }
      }
    }
  }
}

Column	Mapping value	Inserted
product	$.factory.line.product	Engine Block
status	$.factory.line.machine.status	running
sensor_temperature	$.factory.line.machine.sensors.temperature	72.4
site	@plant-a	plant-a

Query with the Analytics IT Hub processor

Because this connector supports the Analytics IT Hub processor, an Analytics flow can use the processor to run ad hoc SQL against it. Trigger the processor with a message that names the connector instance and carries the SQL in body. Only connector and body are required; the other fields are optional and carry the defaults shown here, so the whole message can be copied and edited as needed:

{
  "connector": "<postgresql-connector-id>",
  "body": "SELECT * FROM readings WHERE device_id = 'press-01' LIMIT 100",
  "parameters": [],
  "headers": {},
  "response_topic": "cc.results.sql",
  "force_reset_cache": false,
  "stream_name": "",
  "message_id_prefix": ""
}

The connector publishes the query result to the response_topic, one message per row. The IT Hub processor subscribes to that topic and forwards each message to its output, carrying the rows on to the rest of the Analytics flow. For a SELECT that returns a table, the resulting data can also be shown in the processor's tabular view.

The optional fields map to the processor's advanced parameters: parameters supplies positional query parameters, headers passes free-form headers, force_reset_cache drops any cached connector state before the call, stream_name verifies the response topic maps to a JetStream stream, and message_id_prefix seeds the per-chunk message IDs.

PostgreSQL Integration Guide

PostgreSQL SSL Integration Guide

Add a Connector