DDL logical replication
PolarDB for PostgreSQL (Compatible with Oracle) supports the logical replication of data definition language (DDL) commands.
Background information
Native PostgreSQL synchronizes only table data. You must manually create tables with identical definitions on the publisher and subscriber to ensure correct data synchronization.
PolarDB for PostgreSQL (Compatible with Oracle) extends logical replication to support DDL logical replication. With the publish-subscribe model, you can replicate the CREATE/ALTER/DROP operations of database objects to the subscriber.
Applicable scope
-
Your cluster version must be Oracle syntax compatibility 2.0.
-
To use DDL logical replication, you must set the
wal_levelparameter tological. For more information about how to modify parameters, see Configure cluster parameters.
Syntax
-
CREATE PUBLICATIONCREATE PUBLICATION name [ FOR TABLE [ ONLY ] table_name [ * ] [,...] | FOR ALL TABLES ] [ WITH ( publication_parameter [= value] [,... ] ) ] publication_parameter: ... pubddl = '(none | table | all)'In
publication_parameter, a newpubddlparameter is added. Valid values:none,table, andall.-
none (default): DDL statements are not replicated.
-
table: Only DDL statements related to
tableare replicated.-
CREATE TABLE -
ALTER TABLE -
DROP TABLE -
CREATE TABLE AS
-
-
all: All DDL statements are replicated. The currently supported DDL statements are as follows:
-
ALTER INDEX -
ALTER SEQUENCE -
ALTER TABLE -
ALTER TYPE -
CREATE INDEX -
CREATE SCHEMA -
CREATE SEQUENCE -
CREATE TABLE -
CREATE TABLE AS -
CREATE TYPE -
CREATE TYPE HEADER -
CREATE TYPE BODY -
DROP INDEX -
DROP SCHEMA -
DROP SEQUENCE -
DROP TABLE -
DROP TYPENoteIf you specify
pubddl = 'all', you must also specifyFOR ALL TABLES. Global commands can be run in any database and are not currently replicated. Global commands includeROLE,DATABASE, andTableSpacestatements, and someGrantStmt/RevokeStmtstatements whose target objects are global objects.
-
-
-
CREATE SUBSCRIPTIONCREATE SUBSCRIPTION subscription_name CONNECTION 'conninfo' PUBLICATION publication_name [,...] [ WITH ( subscription_parameter [= value] [,... ] ) ] subscription_parameter: ... dump_schema = false/trueIn
subscription_parameter, a newdump_schemaparameter is added. This parameter allows the existing object definitions on the publisher to bedumped to the subscriber when you create a subscription. The default value is false. Valid values:-
false (default): The existing object definitions on the publisher are not
dumped to the subscriber when you create a subscription. -
true: The existing object definitions on the publisher are
dumped to the subscriber when you create a subscription.
-
dump_schema uses the pg_dump and pg_restore tools. Ensure that the cluster access restrictions allow connections from host='127.0.0.1'. Otherwise, the restore operation fails. The dump files are stored in the pg_logical/schemadumps local directory of the cluster. The files are deleted after a successful restore or if an error occurs.
Parameters
Parameter | Description |
polar_enable_ddl_replication | Enables or disables DDL logical replication. Valid values:
|
polar_enable_debug_ddl_replication | Enables or disables debug mode for DDL replication to print more logs. Valid values: true: Enables debug mode for DDL replication. false (default): Disables debug mode for DDL replication. |
Examples
-
Create a publication that supports DDL.
CREATE PUBLICATION mypub FOR ALL TABLES with (pubddl = 'all');The following result is returned:
CREATE PUBLICATION -
Create a subscription.
CREATE SUBSCRIPTION mysub CONNECTION '$publisher_connstr' PUBLICATION mypub;The following result is returned:
NOTICE: created replication slot "mysub" on publisher CREATE SUBSCRIPTION -
Run SQL statements on the publisher.
# Create a table CREATE TABLE t1(id int,val char(3)); # Insert data INSERT INTO t1 values (1,'a'); INSERT INTO t1 values (2,'b'); INSERT INTO t1 values (3,'c'); # Alter the table ALTER TABLE t1 ADD COLUMN c int GENERATED BY DEFAULT AS IDENTITY, ALTER COLUMN c SET GENERATED ALWAYS; # View the table content SELECT * FROM t1; id | val | c ----+-----+--- 1 | a | 1 2 | b | 2 3 | c | 3 (3 rows) # View the table definition \d+ t1 Table "public.t1" Column | Type | Collation | Nullable | Default | Storage | Compression | Stats target | Description --------+--------------+-----------+----------+------------------------------+----------+-------------+--------------+------------- id | integer | | | | plain | | | val | character(3) | | | | extended | | | c | integer | | not null | generated always as identity | plain | | | Publications: "mypub" Replica Identity: FULL Access method: heap -
Check the replication status on the subscriber.
# View the table content SELECT * FROM t1; id | val | c ----+-----+--- 1 | a | 1 2 | b | 2 3 | c | 3 (3 rows) # View the table definition \d+ t1 Table "public.t1" Column | Type | Collation | Nullable | Default | Storage | Compression | Stats target | Description --------+--------------+-----------+----------+------------------------------+----------+-------------+--------------+------------- id | integer | | | | plain | | | val | character(3) | | | | extended | | | c | integer | | not null | generated always as identity | plain | | | Replica Identity: FULL Access method: heap -
Drop the table on the publisher.
DROP TABLE t1; -
Check the replication status on the subscriber.
SELECT * FROM t1;The following result is returned:
ERROR: relation "t1" does not exist LINE 1: SELECT * FROM t1;
Decoding plugin
The decoding plugin adds the following two callback interfaces.
/*
* Output plugin callbacks
*/
typedef struct OutputPluginCallbacks
{
...
LogicalDecodeDDLMessageCB ddl_cb;
/* streaming of changes */
...
LogicalDecodeStreamDDLMessageCB stream_ddl_cb;
} OutputPluginCallbacks;
/*
* Called for the logical decoding 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 streaming logical decoding DDL messages from in-progress
* transactions.
*/
typedef void (*LogicalDecodeStreamDDLMessageCB) (struct LogicalDecodingContext *ctx,
ReorderBufferTXN *txn,
XLogRecPtr message_lsn,
const char *prefix,
Oid relid,
DeparsedCommandType cmdtype,
Size message_size,
const char *message);test_decoding plugin: The DDL message method is already implemented. Use the test decoding plugin as follows:
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);
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 %{if_not_exists}s %{identity}D (%{table_elements:, }s) %{inherits}s %{tablespace}s %{on_commit}s %{partition_by}s
%{access_method}s %{with_clause}s", "identity": {"objname": "t3", "schemaname": "public"}, "inherits": {"fmt": "INHERITS (%{parents:, }D)", "parents": null, "present": false}, "on_commit": {"fmt": "ON COMMIT %{on_commit_value}s", "present": fal
se, "on_commit_value": null}, "tablespace": {"fmt": "TABLESPACE %{tablespace}I", "present": false, "tablespace": null}, "persistence": "", "with_clause": {"fmt": "WITH", "present": false}, "partition_by": {"fmt": "PARTITION BY %{definition}s", "
present": false, "definition": null}, "access_method": {"fmt": "USING %{access_method}I", "present": false, "access_method": null}, "if_not_exists": "", "table_elements": [{"fmt": "%{name}I %{coltype}T STORAGE %{colstorage}s %{compression}s %{co
llation}s %{not_null}s %{default}s %{identity_column}s %{generated_column}s", "name": "id", "type": "column", "coltype": {"typmod": "", "typarray": false, "typename": "int4", "schemaname": "pg_catalog"}, "default": {"fmt": "DEFAULT", "present":
false}, "not_null": "", "collation": {"fmt": "COLLATE", "present": false}, "colstorage": "plain", "compression": {"fmt": "COMPRESSION %{compression_method}I", "present": false, "compression_method": null}, "identity_column": {"fmt": "%{identity_
type}s", "identity_type": {"fmt": "", "present": false}}, "generated_column": {"fmt": "GENERATED ALWAYS AS", "present": false}}]}
0/C001EE98 | 783 | COMMIT 783
select polar_catalog.ddl_deparse_expand_command('{"fmt": "CREATE %{persistence}s TABLE %{if_not_exists}s %{identity}D (%{table_elements:, }s) %{inherits}s %{tablespace}s %{on_commit}s %{partition_by}s %{access_method}s %{with_clause}s", "identity": {"objname": "t3", "schemaname": "public"}, "inherits": {"fmt": "INHERITS (%{parents:, }D)", "parents": null, "present": false}, "on_commit": {"fmt": "ON COMMIT %{on_commit_value}s", "present": false, "on_commit_value": null}, "tablespace": {"fmt": "TABLESPACE %{tablespace}I", "present": false, "tablespace": null}, "persistence": "", "with_clause": {"fmt": "WITH", "present": false}, "partition_by": {"fmt": "PARTITION BY %{definition}s", "present": false, "definition": null}, "access_method": {"fmt": "USING %{access_method}I", "present": false, "access_method": null}, "if_not_exists": "", "table_elements": [{"fmt": "%{name}I %{coltype}T STORAGE %{colstorage}s %{compression}s %{collation}s %{not_null}s %{default}s %{identity_column}s %{generated_column}s", "name": "id", "type": "column", "coltype": {"typmod": "", "typarray": false, "typename": "int4", "schemaname": "pg_catalog"}, "default": {"fmt": "DEFAULT", "present": false}, "not_null": "", "collation": {"fmt": "COLLATE", "present": false}, "colstorage": "plain", "compression": {"fmt": "COMPRESSION %{compression_method}I", "present": false, "compression_method": null}, "identity_column": {"fmt": "%{identity_type}s", "identity_type": {"fmt": "", "present": false}}, "generated_column": {"fmt": "GENERATED ALWAYS AS", "present": false}}]}');
ddl_deparse_expand_command
-------------------------------------------------------------------------
CREATE TABLE public.t3 (id pg_catalog.int4 STORAGE plain )
(1 row)
New system tables and system functions
-
polar_catalog.ddl_deparse_to_json
-
Definition:
ddl_deparse_to_json(IN pg_ddl_command) RETURN text -
Description: Parses the internal parsetree into a JSON string.
-
Parameters: The input is a parsetree of the pg_ddl_command type. The output is a JSON string of the text type.
-
-
polar_catalog.ddl_deparse_expand_command
-
Definition:
ddl_deparse_expand_command(IN text) RETURN text -
Description: Parses a JSON string into an SQL string.
-
Parameters: The input is a JSON string of the text type. The output is an SQL string of the text type.
-
-
polar_catalog.polar_publication
The definition is as follows:
TABLE polar_publication ( puboid Oid primary key, -- publication oid pubddl "char", -- publication: whether it supports ddl object pubglobal "char", -- publication: whether it supports global objects (to be supported in the future) pubflags int -- reserved flag bits );