Logical Decoding Support for DDL

Starting from MogDB version 5.0.8, the logical replication feature has added support for DDL operations, reducing the manual maintenance of tables during logical replication and preventing issues in the replication synchronization process due to changes in table structure. The kernel logical decoding has added sequence support, and the three decoding plugins wal2json, logical_decoding, and mppdb_decoding have completed the sequence decoding interface.

Feature Description

MogDB supports the following DDL (Data Definition Language, database schema definition language) operations during logical decoding:

CREATE/DROP TABLE|TABLE PARTITION
CREATE/DROP INDEX
TRUNCATE TABLE
ALTER TABLE ADD COLUMN [CONSTRAINT]
ALTER TABLE DROP COLUMN
ALTER TABLE ALTER COLUMN [TYPE|SET NOT NULL|DROP NOT NULL|SET DEFAULT|DROP DEFAULT]
ALTER TABLE [DROP|ADD|TRUNCATE] PARTITION
ALTER TABLE MODIFY COLUMN data_type [ON UPDATE]
ALTER TABLE MODIFY COLUMN [NOT] NULL
ALTER TABLE ADD COLUMN [AFTER|FIRST]

The necessary plugins for logical decoding support of DDL operations are:

wal2json
mppdb_decoding
test_decoding

The following plugins have added new parsing for the logical decoding DDL type log xl_logical_ddl_message:

pg_xlogdump
mog_xlogdump

Notes

Only DDL operations on row-stored tables are supported.
DDL operations on column-stored and Ustore engines are not supported.
DDL operations on temporary tables are not supported.
DDL operations on non-logged tables are not supported.
Logical subscriptions are not supported.
When a statement contains multiple objects and the objects belong to different schemas, the output is in the order they appear, with their respective schemas.
Some DDL operation statements may produce some DML statement parsing results that do not need attention due to kernel implementation reasons.
The GUC parameter wal_level needs to be set to logical or higher, and enable_ddl_logical_record must be enabled.
wal2json only supports format-version==1 and does not support format-version==2.

Example

Logical decoding feature (using wal2json as the plugin)

Set enable_ddl_logical_record=true and wal_level=logical.

Input SQL statements are as follows:

DROP TABLE IF EXISTS range_sales ;
SELECT 'init' FROM pg_create_logical_replication_slot('regression_slot', 'wal2json');

CREATE TABLE logical_tb2(col1 boolean[],col2 boolean);
drop table logical_tb2;
CREATE TABLE range_sales
(
    product_id     INT4 NOT NULL,
    customer_id    INT4 PRIMARY KEY,
    time_id        DATE,
    channel_id     CHAR(1),
    type_id        INT4,
    quantity_sold  NUMERIC(3),
    amount_sold    NUMERIC(10,2)
)
PARTITION BY RANGE (time_id)
(
    PARTITION time_2008 VALUES LESS THAN ('2009-01-01'),
    PARTITION time_2009 VALUES LESS THAN ('2010-01-01'),
    PARTITION time_2010 VALUES LESS THAN ('2011-01-01'),
    PARTITION time_2011 VALUES LESS THAN ('2012-01-01')
);
CREATE INDEX range_sales_idx1 ON range_sales(product_id) LOCAL;
CREATE INDEX range_sales_idx2 ON range_sales(time_id) GLOBAL;

drop INDEX  range_sales_idx1 ;
drop INDEX  range_sales_idx2 ;

drop TABLE range_sales;
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'format-version', '1');
SELECT 'stop' FROM pg_drop_replication_slot('regression_slot');

wal2json decoding result:

                                              data                                               
-------------------------------------------------------------------------------------------------
 {"change":[                                                                                    +
         DDL message: role: hewenjian; search_path: "$user",public; sz: 54; schemaname: public; +
         original DDL query:CREATE TABLE logical_tb2(col1 boolean[],col2 boolean);              +
         ]}
 {"change":[                                                                                    +
         DDL message: role: hewenjian; search_path: "$user",public; sz: 23; schemaname: public; +
         original DDL query:drop table logical_tb2;                                             +
         ]}
 {"change":[                                                                                    +
         DDL message: role: hewenjian; search_path: "$user",public; sz: 502; schemaname: public;+
         original DDL query:CREATE TABLE range_sales                                            +
 (                                                                                              +
     product_id     INT4 NOT NULL,                                                              +
     customer_id    INT4 PRIMARY KEY,                                                           +
     time_id        DATE,                                                                       +
     channel_id     CHAR(1),                                                                    +
     type_id        INT4,                                                                       +
     quantity_sold  NUMERIC(3),                                                                 +
     amount_sold    NUMERIC(10,2)                                                               +
 )                                                                                              +
 PARTITION BY RANGE (time_id)                                                                   +
 (                                                                                              +
     PARTITION time_2008 VALUES LESS THAN ('2009-01-01'),                                       +
     PARTITION time_2009 VALUES LESS THAN ('2010-01-01'),                                       +
     PARTITION time_2010 VALUES LESS THAN ('2011-01-01'),                                       +
     PARTITION time_2011 VALUES LESS THAN ('2012-01-01')                                        +
 );                                                                                             +
         ]}
 {"change":[                                                                                    +
         DDL message: role: hewenjian; search_path: "$user",public; sz: 63; schemaname: public; +
         original DDL query:CREATE INDEX range_sales_idx1 ON range_sales(product_id) LOCAL;     +
         ]}
 {"change":[                                                                                    +
         DDL message: role: hewenjian; search_path: "$user",public; sz: 61; schemaname: public; +
         original DDL query:CREATE INDEX range_sales_idx2 ON range_sales(time_id) GLOBAL;       +
         ]}
 {"change":[                                                                                    +
         DDL message: role: hewenjian; search_path: "$user",public; sz: 30; schemaname: public; +
         original DDL query:drop INDEX  range_sales_idx1 ;                                      +
         ]}
 {"change":[                                                                                    +
         DDL message: role: hewenjian; search_path: "$user",public; sz: 30; schemaname: public; +
         original DDL query:drop INDEX  range_sales_idx2 ;                                      +
         ]}
 {"change":[                                                                                    +
         DDL message: role: hewenjian; search_path: "$user",public; sz: 23; schemaname: public; +
         original DDL query:drop TABLE range_sales;                                             +
         ]}
(8 rows)

pg_xlogdump

The corresponding wal2json output for the generated WAL is:

                                                                   data                                                                    
--------------------------------------------------------------------------------------------------------------------------------------------
 {"change":[                                                                                                                               +
         DDL message: role: hewenjian; search_path: public, new_schema1, new_schema2; sz: 53; schemaname: public, new_schema1, new_schema2;+
         original DDL query:TRUNCATE TABLE range_sales,range_sales1,range_sales2;                                                          +
         ]}
(1 row)

The corresponding pg_xlogdump output:

REDO @ 0/55599A0; LSN 0/5559A80: prev 0/5559918; xid 15966; term 1; len 189; total 223; crc 4229648830; desc: LogicalDDLMessage - prefix "DDL"; role "hewenjian"; search_path "public, new_schema1, new_schema2"; schemaname "public, new_schema1, new_schema2"; payload (53 bytes): 54 52 55 4E 43 41 54 45 20 54 41 42 4C 45 20 72 61 6E 67 65 5F 73 61 6C 65 73 2C 72 61 6E 67 65 5F 73 61 6C 65 73 31 2C 72 61 6E 67 65 5F 73 61 6C 65 73 32 3B

mog_xlogdump

The corresponding wal2json output for the generated WAL is:

                                                                   data                                                                    
--------------------------------------------------------------------------------------------------------------------------------------------
 {"change":[                                                                                                                               +
         DDL message: role: hewenjian; search_path: public, new_schema1, new_schema2; sz: 53; schemaname: public, new_schema1, new_schema2;+
         original DDL query:TRUNCATE TABLE range_sales,range_sales1,range_sales2;                                                          +
         ]}
(1 row)

The corresponding mog_xlogdump output:

REDO @ 0/55599A0; LSN 0/5559A80: prev 0/5559918; xid 15966; term 1; len 189; total 223; crc 4229648830; desc: LogicalDDLMessage - prefix "DDL"; role "hewenjian"; search_path "public, new_schema1, new_schema2"; schemaname "public, new_schema1, new_schema2"; payload (53 bytes): 54 52 55 4E 43 41 54 45 20 54 41 42 4C 45 20 72 61 6E 67 65 5F 73 61 6C 65 73 2C 72 61 6E 67 65 5F 73 61 6C 65 73 31 2C 72 61 6E 67 65 5F 73 61 6C 65 73 32 3B