逻辑解码支持DDL操作

自MogDB 5.0.8版本开始，逻辑复制功能新增对于DDL操作的支持，减少用户在逻辑复制过程中对表的手动维护，避免由于表结构发生变动导致逻辑复制同步过程出现异常。内核逻辑解码添加序列支持，wal2json、logical_decoding、mppdb_decoding三个解码插件完成序列解码对接。

功能描述

MogDB在逻辑解码过程中支持如下DDL（Data Definition Language，数据库模式定义语言）操作：

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]

逻辑解码支持DDL操作所需配套插件支持：

wal2json
mppdb_decoding
test_decoding

下面的插件中新增对逻辑解码DDL类型日志xl_logical_ddl_message的解析：

pg_xlogdump
mog_xlogdump

注意事项

只支持行存表的DDL操作。
不支持列存、ustore存储引擎。
不支持临时表。
不支持非日志表。
不支持逻辑订阅。
当一条语句中存在多个对象，且多个对象属于不同schema，默认按对象出现的顺序，输出其所属的schema。
部分DDL操作语句由于内核实现原因，会产生一些无需关注的DML语句解析结果。
GUC参数wal_level需要>=logical，且开启enable_ddl_logical_record。
wal2json 只支持format-version==1，不支持format-version==2。

示例

逻辑解码功能（以wal2json为plugin）

设置enable_ddl_logical_record=true，设置wal_level=logical。

输入sql语句如下：

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解码结果：

                                              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

产生的wal对应的wal2json输出：

                                                                   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)

对应pg_xlogdump输出：

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

产生的wal对应的wal2json输出：

                                                                   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)

对应mog_xlogdump输出：

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