v5.0

逻辑解码支持 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]

5.0.10 版本后额外支持如下 DDL 操作：
ALTER TABLE RENAME [CONSTRAINT|INDEX]
ALTER TABLE DROP FOREIGN KEY
ALTER TABLE ADD INDEX
ALTER TABLE ADD CONSTRAINT [USING INDEX]
ALTER TABLE AUTO_INCREMENT
ALTER TABLE DROP [PRIMARY KEY|CONSTRAINT|INDEX]
ALTER TABLE COMMENT
ALTER TABLE SET SCHEMA
ALTER TABLE [MODIFY|MOVE|MERGE|RENAME|RESET|EXCHANGE|SPLIT] PARTITION

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

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

pg_xlogdump
mog_xlogdump

注意事项

只支持行存表的 DDL 操作。
不支持列存、ustore 存储引擎。
不支持临时表。
不支持非日志表。
不支持逻辑订阅。
当一条语句中存在多个对象，且多个对象属于不同 schema，默认按对象出现的顺序，输出其所属的 schema。
部分 DDL 操作语句由于内核实现原因，会产生一些无需关注的 DML 语句解析结果。（5.0.10 版本后不再产生）
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 解码结果：

{
    "change": [
        {
        "kind": "DDL",
        "role": "hewenjian",
        "search_path": ["\"$user\"", "public"],
        "schema": "public",
        "sz": 54,
        "original_DDL_query": "CREATE TABLE logical_tb2(col1 boolean[],col2 boolean);"
        }
    ]
}
{
    "change": [
        {
        "kind": "DDL",
        "role": "hewenjian",
        "search_path": ["\"$user\"", "public"],
        "schema": "public",
        "sz": 23,
        "original_DDL_query": "DROP TABLE logical_tb2;"
        }
    ]
}
{
    "change": [
        {
        "kind": "DDL",
        "role": "hewenjian",
        "search_path": ["\"$user\"", "public"],
        "schema": "public",
        "sz": 630,
        "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": [
        {
        "kind": "DDL",
        "role": "hewenjian",
        "search_path": ["\"$user\"", "public"],
        "schema": "public",
        "sz": 63,
        "original_DDL_query": "CREATE INDEX range_sales_idx1 ON range_sales(product_id) LOCAL;"
        }
    ]
}
{
    "change": [
        {
        "kind": "DDL",
        "role": "hewenjian",
        "search_path": ["\"$user\"", "public"],
        "schema": "public",
        "sz": 61,
        "original_DDL_query": "CREATE INDEX range_sales_idx2 ON range_sales(time_id) GLOBAL;"
        }
    ]
}
{
    "change": [
        {
        "kind": "DDL",
        "role": "hewenjian",
        "search_path": ["\"$user\"", "public"],
        "schema": "public",
        "sz": 30,
        "original_DDL_query": "DROP INDEX range_sales_idx1;"
        }
    ]
}
{
    "change": [
        {
        "kind": "DDL",
        "role": "hewenjian",
        "search_path": ["\"$user\"", "public"],
        "schema": "public",
        "sz": 30,
        "original_DDL_query": "DROP INDEX range_sales_idx2;"
        }
    ]
}
{
    "change": [
        {
        "kind": "DDL",
        "role": "hewenjian",
        "search_path": ["\"$user\"", "public"],
        "schema": "public",
        "sz": 23,
        "original_DDL_query": "DROP TABLE range_sales;"
        }
    ]
}

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

反馈