Skip to main content

pg_stream

Streams changes from a PostgreSQL database using logical replication.

Deprecated

This component is deprecated and may be removed in a future release. Prefer a supported alternative where available.

# Common config fields, showing default values
input:
  label: ""
  pg_stream:
    dsn: "" # No default (required)
    include_transaction_markers: false
    stream_snapshot: false
    snapshot_batch_size: 1000
    schema: "" # No default (required)
    tables: [] # No default (required)
    checkpoint_limit: 1024
    temporary_slot: false
    slot_name: "" # No default (required)
    pg_standby_timeout: "10s"
    pg_wal_monitor_interval: "3s"
    max_parallel_snapshot_tables: 1
    auto_replay_nacks: true
    batching:
      count: 0
      byte_size: 0
      period: ""
      check: ""

Using this field overrides the SSL/TLS settings in the environment and DSN.

Fields​

dsn​

The Data Source Name for the PostgreSQL database in the form of postgres://[user[:password]@][netloc][:port][/dbname][?param1=value1&...]. Please note that Postgres enforces SSL by default, you can override this with the parameter sslmode=disable if required.

Type: string

include_transaction_markers​

When set to true, empty messages with operation types BEGIN and COMMIT are generated for the beginning and end of each transaction. Messages with operation metadata set to "begin" or "commit" will have null message payloads.

Type: bool
Default: false

stream_snapshot​

When set to true, the plugin will first stream a snapshot of all existing data in the database before streaming changes. In order to use this the tables that are being snapshot MUST have a primary key set so that reading from the table can be parallelized. Note that this has no effect if tables is left empty, since the snapshot is only planned for tables listed there.

Type: bool
Default: false

snapshot_batch_size​

The number of rows to fetch in each batch when querying the snapshot.

Type: int
Default: 1000

schema​

The PostgreSQL schema from which to replicate data.

Type: string

tables​

A list of table names to include in the logical replication. Each table should be specified as a separate item.

If left empty, the underlying PostgreSQL publication is created FOR ALL TABLES, which replicates every table in every schema of the database, ignoring schema. This also disables stream_snapshot, since the initial snapshot is only planned for tables listed here.

Type: array of string

checkpoint_limit​

The maximum number of messages that can be processed at a given time. Increasing this limit enables parallel processing and batching at the output level. Any given LSN will not be acknowledged unless all messages under that offset are delivered in order to preserve at least once delivery guarantees.

Type: int
Default: 1024

temporary_slot​

If set to true, creates a temporary replication slot that is automatically dropped when the connection is closed.

Type: bool
Default: false

slot_name​

The name of the PostgreSQL logical replication slot to use. If not provided, a random name will be generated. You can create this slot manually before starting replication if desired.

Note: To avoid needing to grant the replication user permission to create publications, you can manually create the publications ahead of time. This connector uses the naming pattern pglog_stream_<replication_slot_name>, so be sure to create them using this convention.

Type: string

pg_standby_timeout​

Specify the standby timeout before refreshing an idle connection.

Type: string
Default: "10s"

pg_wal_monitor_interval​

How often to report changes to the replication lag.

Type: string
Default: "3s"

max_parallel_snapshot_tables​

Int specifies a number of tables that will be processed in parallel during the snapshot processing stage

Type: int
Default: 1

unchanged_toast_value​

The value to emit when there are unchanged TOAST values in the stream. This occurs for updates and deletes where REPLICA IDENTITY is not FULL.

Type: unknown
Default: null

heartbeat_interval​

The interval at which to write heartbeat messages. Heartbeat messages are needed in scenarios when the subscribed tables are low frequency, but there are other high frequency tables writing. Due to the checkpointing mechanism for replication slots, not having new messages to acknowledge will prevent postgres from reclaiming the write ahead log, which can exhaust the local disk. Having heartbeats allows Expanso Edge to safely acknowledge data periodically and move forward the committed point in the log so it can be reclaimed. Setting the duration to 0s will disable heartbeats entirely. Heartbeats are created by periodically writing logical messages to the write ahead log using pg_logical_emit_message.

Type: string
Default: "1h"

tls​

Custom TLS settings can be used to override system defaults.

Type: object

tls.skip_cert_verify​

Whether to skip server side certificate verification.

Type: bool
Default: false

tls.enable_renegotiation​

Whether to allow the remote server to repeatedly request renegotiation. Enable this option if you're seeing the error message local error: tls: no renegotiation.

Type: bool
Default: false

tls.root_cas​

An optional root certificate authority to use. This is a string, representing a certificate chain from the parent trusted root certificate, to possible intermediate signing certificates, to the host certificate.

Secret

This field contains sensitive information. Use a secret reference rather than a literal value.

Type: string
Default: ""

tls.root_cas_file​

An optional path of a root certificate authority file to use. This is a file, often with a .pem extension, containing a certificate chain from the parent trusted root certificate, to possible intermediate signing certificates, to the host certificate.

Type: string
Default: ""

tls.client_certs​

A list of client certificates to use. For each certificate either the fields cert and key, or cert_file and key_file should be specified, but not both.

Type: array of object
Default: []

tls.client_certs[].cert​

A plain text certificate to use.

Type: string
Default: ""

tls.client_certs[].key​

A plain text certificate key to use.

Secret

This field contains sensitive information. Use a secret reference rather than a literal value.

Type: string
Default: ""

tls.client_certs[].cert_file​

The path of a certificate to use.

Type: string
Default: ""

tls.client_certs[].key_file​

The path of a certificate key to use.

Type: string
Default: ""

tls.client_certs[].password​

A plain text password for when the private key is password encrypted in PKCS#1 or PKCS#8 format. The obsolete pbeWithMD5AndDES-CBC algorithm is not supported for the PKCS#8 format.

Because the obsolete pbeWithMD5AndDES-CBC algorithm does not authenticate the ciphertext, it is vulnerable to padding oracle attacks that can let an attacker recover the plaintext.

Secret

This field contains sensitive information. Use a secret reference rather than a literal value.

Type: string
Default: ""

aws​

AWS IAM authentication configuration for PostgreSQL instances. When enabled, IAM credentials are used to generate temporary authentication tokens instead of a static password.

Type: object

aws.enabled​

Enable AWS IAM authentication for PostgreSQL. When enabled, an IAM authentication token is generated and used as the password.

Type: bool
Default: false

aws.region​

The AWS region where the PostgreSQL instance is located. If no region is specified then the environment default will be used.

Type: string

aws.endpoint​

The PostgreSQL endpoint hostname (e.g., mydb.abc123.us-east-1.rds.amazonaws.com).

Type: string

aws.id​

The ID of credentials to use.

Type: string

aws.secret​

The secret for the credentials being used.

Secret

This field contains sensitive information. Use a secret reference rather than a literal value.

Type: string

aws.token​

The token for the credentials being used, required when using short term credentials.

Secret

This field contains sensitive information. Use a secret reference rather than a literal value.

Type: string

aws.role​

Optional AWS IAM role ARN to assume for authentication. Alternatively, use roles array for role chaining instead.

Type: string

aws.role_external_id​

Optional external ID for the role assumption. Only used with the role field. Alternatively, use roles array for role chaining instead.

Type: string

aws.roles​

Optional array of AWS IAM roles to assume for authentication. Roles can be assumed in sequence, enabling chaining for purposes such as cross-account access. Each role can optionally specify an external ID.

Type: array of object

aws.roles[].role​

AWS IAM role ARN to assume.

Type: string
Default: ""

aws.roles[].role_external_id​

Optional external ID for the role assumption.

Type: string
Default: ""

signal_table_name​

The name of the table used to send control signals to the connector, excluding the schema. The table must exist in the schema configured via the schema field, and must not also appear in tables — the signal table is implicitly added to the publication and excluded from snapshot scans, so listing it in both places is rejected at startup. It must have at least these columns — startup validation checks column names only, not types, so a wrong column type (e.g. data JSONB instead of TEXT) is only caught at runtime, on the first signal row read:

  • id — any type representable as a string (e.g. SERIAL, BIGSERIAL, UUID, VARCHAR)
  • type — should be VARCHAR or another string type — the signal type (see supported signals below)
  • data — should be TEXT — a JSON object containing signal parameters

Create the table with:

CREATE TABLE <schema>.<signal_table_name> (
    id   SERIAL PRIMARY KEY,
    type VARCHAR(32),
    data TEXT
);

Signal rows are published as regular output messages (operation=insert, table=<signal_table_name>). To exclude them from downstream processing, filter on the table metadata field using a mapping processor:

pipeline:
  processors:
    - mapping: |
        root = if @table == "rpcn_signal_table" { deleted() } else { this }

Supported signals

log — recognized and logged when received. The data column must contain a JSON object with a message key, whose value is written to the connector's log output.

INSERT INTO <schema>.<signal_table_name> (type, data) VALUES ('log', '{"message": "Signal message"}');

Type: string
Default: ""

auto_replay_nacks​

Whether messages that are rejected (nacked) at the output level should be automatically replayed indefinitely, eventually resulting in back pressure if the cause of the rejections is persistent. If set to false these messages will instead be deleted. Disabling auto replays can greatly improve memory efficiency of high throughput streams as the original shape of the data can be discarded immediately upon consumption and mutation.

Type: bool
Default: true

batching​

Allows you to configure a batching policy.

Type: object

batching.count​

A number of messages at which the batch should be flushed. If 0 disables count based batching.

Type: int
Default: 0

batching.byte_size​

An amount of bytes at which the batch should be flushed. If 0 disables size based batching.

Type: int
Default: 0

batching.period​

A period in which an incomplete batch should be flushed regardless of its size.

Type: string
Default: ""

batching.check​

A Bloblang query that should return a boolean value indicating whether a message should end a batch.

Type: string
Default: ""

batching.processors​

A list of processors to apply to a batch as it is flushed. This allows you to aggregate and archive the batch however you see fit. Please note that all resulting messages are flushed as a single batch, therefore splitting the batch into smaller batches using these processors is a no-op.

Type: array of processor

tables and slot_name are required, but may be empty​

The field descriptions above explain what happens when tables is left empty and when slot_name is not provided. Both behaviors are reached by writing the field with an empty value, not by leaving it out: omitting either one fails submission with Missing required field.

input:
  pg_stream:
    dsn: postgres://user:pass@host:5432/db
    schema: public
    tables: []      # publication FOR ALL TABLES, ignoring schema; also disables stream_snapshot
    slot_name: ""   # a slot name is generated for you

Write the tables you want, or an empty list to take every table. Same for slot_name: name the slot you created, or leave it as an empty string to have one generated.