pglogrepl

pglogrepl is a Go package designed for PostgreSQL logical replication. It is built on top of github.com/jackc/pgx/v5/pgconn for its underlying PostgreSQL connection management.

To use this package effectively, developers should have a solid understanding of the PostgreSQL replication protocol.

PostgreSQL (version 14+) supports streaming large in-progress transactions via the replication protocol using StreamXXXMessage. To enable this feature when using the pgoutput plugin, you must meet the following requirements:

1. Protocol Version: Use at least protocol version 2.
2. Streaming Parameter: Set the streaming parameter to true for the pgoutput plugin.

Controlling when streaming starts

The transition from in-memory decoding to streaming is controlled by the PostgreSQL configuration parameter logical_decoding_work_mem.

• For testing: You can set this to the minimum value (64kB) to force streaming behavior more frequently.
• For production: Do not use the minimum value; use a value appropriate for your workload.

Limitations

• The wal2json plugin does not currently support streaming.

The pglogrepl_demo is a demonstration tool that connects to a PostgreSQL database and logs all messages sent over logical replication.

Prerequisites

• You must provide a connection string via the PGLOGREPL_DEMO_CONN_STRING environment variable.
• The connection must be a superuser, as the demo creates a publication using ON ALL TABLES.

Execution

To run the demo, set the connection string and use go run on the main.go file within the demo directory.

$ PGLOGREPL_DEMO_CONN_STRING="postgres://pglogrepl:secret@127.0.0.1/pglogrepl?replication=database" go run main.go

To execute the package tests, you must provide a replication connection string via the PGLOGREPL_TEST_CONN_STRING environment variable.

If you wish to skip the base backup step (which involves PostgreSQL creating and streaming a backup tar), set PGLOGREPL_SKIP_BASE_BACKUP=true.

Example Command

PGLOGREPL_TEST_CONN_STRING=postgres://pglogrepl:secret@127.0.0.1/pglogrepl?replication=database go test

To run tests for pglogrepl, you must configure a PostgreSQL instance with replication permissions and specific settings.

1. Database and User Setup

Create a dedicated database and a user with the replication attribute:

create database pglogrepl;
create user pglogrepl with replication password 'secret';

If you are using PostgreSQL 15 or newer, you may need to grant access to the public schema for testing purposes:

grant all on schema public to pglogrepl;

2. Configure pg_hba.conf

Add a replication entry to your pg_hba.conf to allow the replication user to connect (example for local connection):

host replication pglogrepl 127.0.0.1/32 md5

3. Configure postgresql.conf

Ensure the following settings are applied in postgresql.conf:

wal_level = logical
max_wal_senders = 5
max_replication_slots = 5

In the V2 protocol, when a message is part of a stream (inStream == true), it includes an Xid (Transaction ID) prefix. The following message types embed InStreamMessageV2WithXid to provide this context:

• LogicalDecodingMessageV2
• RelationMessageV2
• TypeMessageV2
• InsertMessageV2
• UpdateMessageV2
• DeleteMessageV2
• TruncateMessageV2

When ParseV2 is called with inStream: true, these types will have their Xid field populated from the byte stream before decoding the rest of the message content.

The repository includes two demonstration programs to show how to interact with different replication types:

• Logical Replication: See example/pglogrepl_demo for a program that connects to a database and logs all messages sent over logical replication.
• Physical Replication: See example/pgphysrepl_demo for a program that connects to a database and logs all messages sent over physical replication.

Use ParseV2 to decode raw bytes received from PostgreSQL into structured message types for the logical replication protocol version 2.

Important: The inStream parameter The inStream boolean parameter is critical for correct decoding:

• It must be set to true after a StreamStartMessageV2 has been read.
• It must be set to false after a StreamStopMessageV2 has been read.

When inStream is true, the decoder expects and extracts transaction IDs (Xid) from messages that support streamed transactions (like InsertMessageV2, UpdateMessageV2, etc.).

// Example usage pattern
msg, err := pglogrepl.ParseV2(data, inStream)
if err != nil {
	return err
}

switch m := msg.(type) {
case *pglogrepl.StreamStartMessageV2:
	// Handle stream start
case *pglogrepl.InsertMessageV2:
	// Handle insert
}

To process incoming data from a PostgreSQL logical replication stream, use the Parse function. It takes a byte slice (the raw message data) and returns a Message interface. You can then use a type assertion to access the specific message type (e.g., *InsertMessage, *RelationMessage, etc.).

Note: Parse expects the input data to include the first byte, which is the MessageType identifier. The actual decoding of the message body starts from the second byte.

```go
// Example of parsing a raw message
msg, err := pglogrepl.Parse(rawData)
if err != nil {
    log.Fatal(err)
}

switch m := msg.(type) {
case *pglogrepl.InsertMessage:
    fmt.Printf("Inserted into relation %d\n", m.RelationID)
case *pglogrepl.UpdateMessage:
    fmt.Printf("Updated relation %d\n", m.RelationID)
// ... handle other types
}
```埋

When consuming a replication stream, you will receive raw bytes that need to be parsed into structured types:

• ParsePrimaryKeepaliveMessage: Parses a 17-byte message from the server containing ServerWALEnd (LSN), ServerTime, and ReplyRequested flag.
• ParseXLogData: Parses WAL data messages. It extracts WALStart (LSN), ServerWALEnd (LSN), ServerTime, and the actual WALData payload.

// Example parsing XLogData
// buf is the raw byte slice received from the connection
xld, err := pglogrepl.ParseXLogData(buf)
if err != nil {
    // handle error
}
fmt.Printf("WAL Start: %s, Data Size: %d\n", xld.WALStart, len(xld.WALData))