DDL logical replication

Quick start

-- On the publisher: create a publication that replicates all DDL
CREATE PUBLICATION mypub FOR ALL TABLES WITH (pubddl = 'all');

-- On the subscriber: create a subscription
CREATE SUBSCRIPTION mysub
  CONNECTION '<publisher_connection_string>'
  PUBLICATION mypub;

Replace <publisher_connection_string> with the actual connection string of your publisher cluster, for example: host=<host> port=5432 dbname=<dbname> user=<user> password=<password>.

Prerequisites

Before you begin, ensure that you have:

• wal_level set to logical. For instructions, see Configure cluster parameters.

• polar_enable_ddl_replication set to true (the default).

How it works

1. Create a publication with pubddl set to table or all to enable DDL replication.

2. Create a subscription on the subscriber node. Optionally set dump_schema = true to copy existing object definitions from the publisher at subscription creation time.

3. Execute DDL statements on the publisher. PolarDB captures them in the write-ahead log (WAL) and forwards them to subscribers.

4. Verify that the schema changes appear on the subscriber.

Configure DDL replication

CREATE PUBLICATION name
    [ FOR TABLE [ ONLY ] table_name [ * ] [, ...]
      | FOR ALL TABLES ]
    [ WITH ( publication_parameter [= value] [, ... ] ) ]
publication_parameter:
    ...
    pubddl = 'none | table | all'

CREATE SUBSCRIPTION subscription_name
    CONNECTION 'conninfo'
    PUBLICATION publication_name [, ...]
    [ WITH ( subscription_parameter [= value] [, ... ] ) ]
subscription_parameter:
    ...
    dump_schema = false | true

Parameters

Limitations

• When pubddl = 'all', the publication must use FOR ALL TABLES. Table-specific publications are not supported with pubddl = 'all'.

• Global statements (ROLE, DATABASE, TABLESPACE, GRANT, and REVOKE on global objects) are not replicated. They execute locally on all databases.

• dump_schema = true requires the cluster to accept connections on host='127.0.0.1'. If this connection is blocked, the restore fails.

Example: end-to-end DDL replication

This example shows how DDL changes — including ALTER TABLE and DROP TABLE — propagate from publisher to subscriber.

Step 1: Create a publication that replicates all DDL.

CREATE PUBLICATION mypub FOR ALL TABLES WITH (pubddl = 'all');

Expected output:

CREATE PUBLICATION

Step 2: Create a subscription.

CREATE SUBSCRIPTION mysub
  CONNECTION '$publisher_connstr'
  PUBLICATION mypub;

Expected output:

NOTICE:  created replication slot "mysub" on publisher
CREATE SUBSCRIPTION

Step 3: Run DDL and DML on the publisher.

-- Create a table
CREATE TABLE t1 (id int, val char(3));

-- Insert rows
INSERT INTO t1 VALUES (1, 'a');
INSERT INTO t1 VALUES (2, 'b');
INSERT INTO t1 VALUES (3, 'c');

-- Add a generated column
ALTER TABLE t1
  ADD COLUMN c int GENERATED BY DEFAULT AS IDENTITY,
  ALTER COLUMN c SET GENERATED ALWAYS;

-- Verify on the publisher
SELECT * FROM t1;

Expected output:

 id | val | c
----+-----+---
  1 | a   | 1
  2 | b   | 2
  3 | c   | 3
(3 rows)

Step 4: Verify replication on the subscriber.

SELECT * FROM t1;

Expected output — data and schema match the publisher:

 id | val | c
----+-----+---
  1 | a   | 1
  2 | b   | 2
  3 | c   | 3
(3 rows)

Step 5: Drop the table on the publisher.

DROP TABLE t1;

Step 6: Verify the drop propagated to the subscriber.

SELECT * FROM t1;

Expected output:

ERROR:  relation "t1" does not exist
LINE 1: SELECT * FROM t1;

The DROP TABLE was replicated. The table no longer exists on the subscriber.

Decoding extension

PolarDB adds two callback interfaces to OutputPluginCallbacks for logical decoding plugins:

typedef struct OutputPluginCallbacks
{
    ...
    LogicalDecodeDDLMessageCB       ddl_cb;
    ...
    LogicalDecodeStreamDDLMessageCB stream_ddl_cb;
    ...
} OutputPluginCallbacks;

/* Callback for committed DDL messages */
typedef void (*LogicalDecodeDDLMessageCB) (
    struct LogicalDecodingContext *ctx,
    ReorderBufferTXN *txn,
    XLogRecPtr message_lsn,
    const char *prefix,
    Oid relid,
    DeparsedCommandType cmdtype,
    Size message_size,
    const char *message
);

/* Callback for in-progress transaction DDL messages (streaming) */
typedef void (*LogicalDecodeStreamDDLMessageCB) (
    struct LogicalDecodingContext *ctx,
    ReorderBufferTXN *txn,
    XLogRecPtr message_lsn,
    const char *prefix,
    Oid relid,
    DeparsedCommandType cmdtype,
    Size message_size,
    const char *message
);

The DDL message method is implemented in the test_decoding extension. The following example creates a publication, creates a replication slot, creates a table, and reads the decoded DDL output:

CREATE PUBLICATION mypub FOR ALL TABLES WITH (pubddl = 'all');

SELECT * FROM pg_create_logical_replication_slot('regression_slot', 'test_decoding', false, true);

CREATE TABLE t3 (id int);

SELECT * FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL);

The output includes a message field containing the DDL in JSON format:

    lsn     | xid |                    data
------------+-----+--------------------------------------------
 0/C001BF10 | 783 | BEGIN 783
 0/C001EBC0 | 783 | message: prefix: deparse, relid: 16418, cmdtype: Simple, sz: 1505 content:{"fmt": "CREATE %{persistence}s TABLE ..."}
 0/C001EE98 | 783 | COMMIT 783

To expand a DDL JSON message into a SQL string, use polar_catalog.ddl_deparse_expand_command:

SELECT polar_catalog.ddl_deparse_expand_command('<json_string>');

Expected output:

               ddl_deparse_expand_command
---------------------------------------------------------
 CREATE  TABLE  public.t3 (id pg_catalog.int4 STORAGE plain      )
(1 row)

System functions and tables

TABLE polar_publication
(
    puboid    Oid  PRIMARY KEY,  -- Publication OID
    pubddl    "char",            -- DDL replication scope for the publication
    pubglobal "char",            -- Global object replication scope (coming soon)
    pubflags  int                -- Reserved flag bits
);

What's next

• Configure cluster parameters — set wal_level and other replication-related parameters