Skip to main content

postgres_cdc

Streams data changes from a PostgreSQL database using logical replication.

Captures row-level changes from a PostgreSQL database in real time using logical replication. You choose the schema and tables to replicate, and can optionally stream a snapshot of all existing data before streaming ongoing changes.

# Config fields, showing default values
input:
  label: ""
  postgres_cdc:
    dsn: "" # No default (required)
    include_transaction_markers: false
    stream_snapshot: false
    snapshot_batch_size: 1000
    snapshot_memory_safety_factor: 0
    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
    heartbeat_interval: 1h
    unchanged_toast_value: null
    batching:
      count: 0
      byte_size: 0
      period: ""
      check: ""
      processors: []
    tls:
      skip_cert_verify: false
      enable_renegotiation: false
      root_cas: ""
      root_cas_file: ""
      client_certs: []
    aws:
      enabled: false
      endpoint: ""
      region: ""
      id: ""
      secret: ""
      token: ""
      role: ""
      role_external_id: ""
      roles:
        - role: ""
          role_external_id: ""

Examples​

Stream a snapshot of all existing rows in the selected tables before streaming ongoing changes, using a temporary replication slot.

input:
  postgres_cdc:
    dsn: postgres://user:password@host:5432/dbname
    slot_name: my_slot
    stream_snapshot: true
    snapshot_batch_size: 100000
    temporary_slot: true
    schema: public
    tables:
      - my_table

Fields​

dsn​

The data source name (DSN) of the PostgreSQL database from which you want to stream changes, in the form postgres://[user[:password]@][host][:port][/dbname][?param1=value1&...].

Type: string

include_transaction_markers​

When set to true, empty messages are emitted for BEGIN and COMMIT operations. These messages have null payloads, with the operation recorded in their metadata.

Type: bool
Default: false

stream_snapshot​

When set to true, this input streams a snapshot of all existing data in the source database before streaming data changes. All tables must have primary keys.

Type: bool
Default: false

snapshot_batch_size​

The number of table rows to fetch in each batch when querying the snapshot. Only used when stream_snapshot is true.

Type: int
Default: 1000

snapshot_memory_safety_factor​

A factor that limits the amount of memory used while streaming the snapshot.

Type: float

schema​

The PostgreSQL schema from which to replicate data.

Type: string

tables​

A list of table names to include in the snapshot and logical replication.

Type: array

checkpoint_limit​

The maximum number of messages that can be processed at a given time. Increasing this value enables parallel processing and batching at the output level. Log sequence numbers (LSNs) are not acknowledged until all messages within a checkpoint have been delivered, which preserves at-least-once delivery guarantees.

Type: int
Default: 1024

temporary_slot​

When set to true, the input creates a temporary replication slot that is automatically dropped when the connection closes. This is useful for testing or for avoiding accumulation of data in the slot.

Type: bool
Default: false

slot_name​

The name of the PostgreSQL logical replication slot to use.

Type: string

pg_standby_timeout​

The standby timeout after which an idle connection is refreshed to keep it alive.

Type: string
Default: "10s"

pg_wal_monitor_interval​

How often to report changes to the replication lag and record them as metrics.

Type: string
Default: "3s"

max_parallel_snapshot_tables​

The maximum number of tables that are processed in parallel when the initial snapshot of the source database is taken.

Type: int
Default: 1

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. Set to false to instead drop rejected messages, which can improve memory usage.

Type: bool
Default: true

heartbeat_interval​

The interval between heartbeat messages that are written to the write-ahead log (WAL) using pg_logical_emit_message. This is useful for keeping replication progress current on tables with low activity. Set to 0s to disable heartbeats.

Type: string
Default: "1h"

unchanged_toast_value​

The value to emit in place of unchanged TOAST values when they appear in the message stream. PostgreSQL does not include unchanged TOAST values in the replication stream, so this placeholder lets you detect them downstream.

Type: unknown
Default: null

batching​

Allows you to configure a batching policy for the messages produced by this input.

Type: object

batching.count​

The number of messages after which the batch is flushed. Set to 0 to disable count-based batching.

Type: int
Default: 0

batching.byte_size​

The number of bytes at which the batch is flushed. Set to 0 to disable size-based batching.

Type: int
Default: 0

batching.period​

The period of time after which an incomplete batch is 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.

Type: array
Default: []

tls​

Custom TLS settings can be used to override system defaults for the connection to PostgreSQL.

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.

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 specify values for either the cert and key fields, or the cert_file and key_file fields.

Type: array
Default: []

tls.client_certs.cert​

A plain text certificate to use.

Type: string
Default: ""

tls.client_certs.cert_file​

The path of a certificate to use.

Type: string
Default: ""

tls.client_certs.key​

A plain text certificate key to use. :::warning Secret This field contains sensitive information that usually shouldn't be added to a config directly, read our secrets page for more info. :::

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. :::warning Secret This field contains sensitive information that usually shouldn't be added to a config directly, read our secrets page for more info. :::

Type: string
Default: ""

aws​

Optional AWS configuration used to authenticate to the PostgreSQL database with IAM 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 database password.

Type: bool
Default: false

aws.endpoint​

The PostgreSQL endpoint hostname, for example mydb.abc123.us-east-1.rds.amazonaws.com.

Type: string

aws.region​

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

Type: string

aws.id​

The ID of credentials to use.

Type: string

aws.secret​

The secret for the credentials being used. :::warning Secret This field contains sensitive information that usually shouldn't be added to a config directly, read our secrets page for more info. :::

Type: string

aws.token​

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

Type: string

aws.role​

Optional AWS IAM role ARN to assume for authentication.

Type: string

aws.role_external_id​

Optional external ID for the role assumption. Only used with the role field.

Type: string

aws.roles​

An optional list of AWS IAM roles to assume in sequence, enabling role chaining for cross-account access.

Type: array

aws.roles.role​

The AWS IAM role ARN to assume.

Type: string
Default: ""

aws.roles.role_external_id​

An optional external ID to provide when assuming the role.

Type: string
Default: ""