0002-DDL-replication-for-Table-DDL-commands-2023_04_04.patch
application/octet-stream
Filename: 0002-DDL-replication-for-Table-DDL-commands-2023_04_04.patch
Type: application/octet-stream
Part: 3
Patch
Same data as JSON:
GET /api/v1/attachments/:id/patch
the parsed metadata as JSON — format, series position, per-file stats; never the diff bytes.
API reference →
Format: format-patch
Series: patch 0002
Subject: DDL replication for Table DDL commands
| File | + | − |
|---|---|---|
| contrib/test_decoding/expected/ddl.out | 26 | 0 |
| contrib/test_decoding/sql/ddl.sql | 5 | 0 |
| contrib/test_decoding/test_decoding.c | 48 | 0 |
| doc/src/sgml/logical-replication.sgml | 1045 | 0 |
| src/backend/access/rmgrdesc/logicalddlmsgdesc.c | 52 | 0 |
| src/backend/access/rmgrdesc/Makefile | 1 | 0 |
| src/backend/access/rmgrdesc/meson.build | 1 | 0 |
| src/backend/access/transam/rmgr.c | 1 | 0 |
| src/backend/catalog/pg_publication.c | 1 | 0 |
| src/backend/commands/event_trigger.c | 2 | 0 |
| src/backend/commands/publicationcmds.c | 215 | 38 |
| src/backend/replication/logical/ddlmessage.c | 84 | 0 |
| src/backend/replication/logical/ddltrigger.c | 283 | 0 |
| src/backend/replication/logical/decode.c | 41 | 0 |
| src/backend/replication/logical/logical.c | 93 | 0 |
| src/backend/replication/logical/logicalfuncs.c | 24 | 0 |
| src/backend/replication/logical/Makefile | 2 | 0 |
| src/backend/replication/logical/meson.build | 2 | 0 |
| src/backend/replication/logical/proto.c | 43 | 0 |
| src/backend/replication/logical/reorderbuffer.c | 135 | 0 |
| src/backend/replication/logical/worker.c | 240 | 0 |
| src/backend/replication/pgoutput/pgoutput.c | 244 | 24 |
| src/backend/tcop/cmdtag.c | 24 | 2 |
| src/backend/tcop/utility.c | 4 | 2 |
| src/backend/utils/cache/relcache.c | 1 | 0 |
| src/bin/pg_dump/pg_dump.c | 72 | 6 |
| src/bin/pg_dump/pg_dump.h | 1 | 0 |
| src/bin/pg_dump/t/002_pg_dump.pl | 1 | 1 |
| src/bin/pg_waldump/rmgrdesc.c | 1 | 0 |
| src/bin/psql/describe.c | 25 | 8 |
| src/include/access/rmgrlist.h | 1 | 0 |
| src/include/catalog/pg_proc.dat | 20 | 0 |
| src/include/catalog/pg_publication.h | 14 | 16 |
| src/include/commands/publicationcmds.h | 14 | 0 |
| src/include/replication/ddlmessage.h | 60 | 0 |
| src/include/replication/decode.h | 1 | 0 |
| src/include/replication/logicalproto.h | 4 | 0 |
| src/include/replication/output_plugin.h | 27 | 0 |
| src/include/replication/pgoutput.h | 1 | 0 |
| src/include/replication/reorderbuffer.h | 39 | 0 |
| src/include/tcop/cmdtag.h | 3 | 1 |
| src/include/tcop/cmdtaglist.h | 193 | 193 |
| src/include/utils/rel.h | 2 | 0 |
| src/test/regress/expected/psql.out | 3 | 3 |
| src/test/regress/expected/publication.out | 210 | 210 |
| src/tools/pgindent/typedefs.list | 5 | 0 |
From bca208929c3d6b59c2b106493aae80a1273acdfe Mon Sep 17 00:00:00 2001
From: Hou Zhijie <houzj.fnst@cn.fujitsu.com>
Date: Mon, 3 Apr 2023 20:31:23 +0800
Subject: [PATCH 2/6] DDL replication for Table DDL commands
To support DDL replication, it use event trigger and DDL deparsing
facilities. During CREATE PUBLICATION we register a command end trigger that
deparses the DDL (if the DDL is annotated as ddlreplok for DDL replication in
cmdtaglist.h) as a JSON blob, and WAL logs it. The event trigger is automatically
removed at the time of DROP PUBLICATION. The WALSender decodes the WAL and sends
it downstream similar to other DML commands. The subscriber then converts JSON
back to the DDL command string and executes it. In the subscriber, we also add
the newly added rel to pg_subscription_rel so that the DML changes on the new
table can be replicated without having to manually run
"ALTER SUBSCRIPTION ... REFRESH PUBLICATION".
- For non-rewrite ALTER object command and
- CREATE object command:
we deparse the command at ddl_command_end event trigger and WAL log the
deparsed json string. The WALSender decodes the WAL and sends it to
subscriber if the created/altered table is published. It supports most of
ALTER TABLE command except some commands(DDL related to PARTITIONED TABLE
...) that introduced recently which haven't been supported by the current
ddl_deparser, we will support that later.
- For DROP object:
The 'command start' event handler logs a ddl message with the relids of
the tables that are dropped which the output plugin (pgoutput) stores in
its internal data structure after verifying that it is for a table that is
part of the publication. Later the 'command end' event handler sends the
actual drop message. Pgoutput on receiving the command end, only sends out
the drop command only if it is for one of the relids marked for deleting.
The reason we have to do this is because, once the logical decoder
receives the 'command end' message, the relid of the table is no longer
valid as it has been deleted as part of invalidations received for the
drop table command. It is no longer possible to verify if the table is
part of the publication list or not. To make this possible, I have added
two more elements to the ddl xlog and ddl message, (relid and cmdtype).
We could have also handled all this on the subscriber side as well, but
that would mean sending spurious ddl messages for tables that are not part
of the publication.
- For table_rewrite ALTER TABLE command:
(ALTER COLUMN TYPE, ADD COLUMN DEFAULT, SET LOGGED, SET ACCESS METHOD)
we deparse the command and WAL log the deparsed json string at
table_rewrite event trigger. The WALSender decodes the WAL and sends it to
subscriber if the altered table is published. Then, the WALSender will
convert the upcoming rewrite INSERTs to UPDATEs and send them to
subscriber so that the data between publisher and subscriber can always be
consistent. Note that the tables that publish rewrite ddl must have a
replica identity configured in order to be able to replicate the upcoming
rewrite UPDATEs.
We do this way because of two reason:
(1) The data before the rewrite ddl could already be different among
publisher and subscriber. To make sure the extra data in subscriber which
doesn't exist in publisher also get rewritten, we need to let the
subscriber execute the original rewrite ddl to rewrite all the data at
first.
(2) the data after executing rewrite ddl could be different among
publisher and subscriber(due to different functions/operators used during
rewrite), so we need to replicate the rewrite UPDATEs to keep the data
consistent.
TO IMPROVE:
This approach could be improved by letting the subscriber try to update
the extra data itself instead of doing fully rewrite ddl and use the
upcoming rewrite UPDATEs to rewrite the rest data. To achieve this, we
could modify the deparsed json string to temporarily remove the rewrite
part and add some logic in subscriber to update the extra data.
Besides, we may not need to send rewrite changes for all type of rewrite
ddl, for example, it seems fine to skip sending rewrite changes for ALTER
TABLE SET LOGGED as the data in the table doesn't actually be changed. We
could use the deparser and event trigger to filter these ddls and skip
sending rewrite changes for them.
---
contrib/test_decoding/expected/ddl.out | 26 +
contrib/test_decoding/sql/ddl.sql | 5 +
contrib/test_decoding/test_decoding.c | 48 +
doc/src/sgml/logical-replication.sgml | 1045 +++++++++++++++++
src/backend/access/rmgrdesc/Makefile | 1 +
.../access/rmgrdesc/logicalddlmsgdesc.c | 52 +
src/backend/access/rmgrdesc/meson.build | 1 +
src/backend/access/transam/rmgr.c | 1 +
src/backend/catalog/pg_publication.c | 1 +
src/backend/commands/event_trigger.c | 2 +
src/backend/commands/publicationcmds.c | 253 +++-
src/backend/replication/logical/Makefile | 2 +
src/backend/replication/logical/ddlmessage.c | 84 ++
src/backend/replication/logical/ddltrigger.c | 283 +++++
src/backend/replication/logical/decode.c | 41 +
src/backend/replication/logical/logical.c | 93 ++
.../replication/logical/logicalfuncs.c | 24 +
src/backend/replication/logical/meson.build | 2 +
src/backend/replication/logical/proto.c | 43 +
.../replication/logical/reorderbuffer.c | 135 +++
src/backend/replication/logical/worker.c | 240 ++++
src/backend/replication/pgoutput/pgoutput.c | 268 ++++-
src/backend/tcop/cmdtag.c | 26 +-
src/backend/tcop/utility.c | 6 +-
src/backend/utils/cache/relcache.c | 1 +
src/bin/pg_dump/pg_dump.c | 78 +-
src/bin/pg_dump/pg_dump.h | 1 +
src/bin/pg_dump/t/002_pg_dump.pl | 2 +-
src/bin/pg_waldump/rmgrdesc.c | 1 +
src/bin/psql/describe.c | 33 +-
src/include/access/rmgrlist.h | 1 +
src/include/catalog/pg_proc.dat | 20 +
src/include/catalog/pg_publication.h | 30 +-
src/include/commands/publicationcmds.h | 14 +
src/include/replication/ddlmessage.h | 60 +
src/include/replication/decode.h | 1 +
src/include/replication/logicalproto.h | 4 +
src/include/replication/output_plugin.h | 27 +
src/include/replication/pgoutput.h | 1 +
src/include/replication/reorderbuffer.h | 39 +
src/include/tcop/cmdtag.h | 4 +-
src/include/tcop/cmdtaglist.h | 386 +++---
src/include/utils/rel.h | 2 +
src/test/regress/expected/psql.out | 6 +-
src/test/regress/expected/publication.out | 420 +++----
src/tools/pgindent/typedefs.list | 5 +
46 files changed, 3314 insertions(+), 504 deletions(-)
create mode 100644 src/backend/access/rmgrdesc/logicalddlmsgdesc.c
create mode 100644 src/backend/replication/logical/ddlmessage.c
create mode 100644 src/backend/replication/logical/ddltrigger.c
create mode 100644 src/include/replication/ddlmessage.h
diff --git a/contrib/test_decoding/expected/ddl.out b/contrib/test_decoding/expected/ddl.out
index 9a28b5ddc5..0f51f2b41a 100644
--- a/contrib/test_decoding/expected/ddl.out
+++ b/contrib/test_decoding/expected/ddl.out
@@ -831,6 +831,32 @@ SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'inc
------
(0 rows)
+SELECT 'ddl msg1' FROM pg_logical_emit_ddl_message('ddl msg1', 16394, 1, 'msg1');
+ ?column?
+----------
+ ddl msg1
+(1 row)
+
+SELECT 'ddl msg2' FROM pg_logical_emit_ddl_message('ddl msg2', 16394, 1, '{"fmt": "CREATE SCHEMA %{if_not_exists}s %{name}I %{authorization}s", "name": "foo", "authorization": {"fmt": "AUTHORIZATION %{authorization_role}I", "present": false, "authorization_role": null}, "if_not_exists": ""}');
+ ?column?
+----------
+ ddl msg2
+(1 row)
+
+SELECT 'ddl msg3' FROM pg_logical_emit_ddl_message('ddl msg3', 16394, 1, '{"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": "foo", "schemaname": "element_test"}, "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 %{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}, "generated_column": {"fmt": "GENERATED ALWAYS AS", "present": false}}]}}');
+ ?column?
+----------
+ ddl msg3
+(1 row)
+
+SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
+ data
+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
+ message: prefix: ddl msg1, relid: 16394, cmdtype: Drop start, sz: 4 content:msg1
+ message: prefix: ddl msg2, relid: 16394, cmdtype: Drop start, sz: 217 content:{"fmt": "CREATE SCHEMA %{if_not_exists}s %{name}I %{authorization}s", "name": "foo", "authorization": {"fmt": "AUTHORIZATION %{authorization_role}I", "present": false, "authorization_role": null}, "if_not_exists": ""}
+ message: prefix: ddl msg3, relid: 16394, cmdtype: Drop start, sz: 1396 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": "foo", "schemaname": "element_test"}, "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 %{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}, "generated_column": {"fmt": "GENERATED ALWAYS AS", "present": false}}]}}
+(3 rows)
+
SELECT pg_drop_replication_slot('regression_slot');
pg_drop_replication_slot
--------------------------
diff --git a/contrib/test_decoding/sql/ddl.sql b/contrib/test_decoding/sql/ddl.sql
index 4f76bed72c..1ea5a4b271 100644
--- a/contrib/test_decoding/sql/ddl.sql
+++ b/contrib/test_decoding/sql/ddl.sql
@@ -431,6 +431,11 @@ SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'inc
-- done, free logical replication slot
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
+SELECT 'ddl msg1' FROM pg_logical_emit_ddl_message('ddl msg1', 16394, 1, 'msg1');
+SELECT 'ddl msg2' FROM pg_logical_emit_ddl_message('ddl msg2', 16394, 1, '{"fmt": "CREATE SCHEMA %{if_not_exists}s %{name}I %{authorization}s", "name": "foo", "authorization": {"fmt": "AUTHORIZATION %{authorization_role}I", "present": false, "authorization_role": null}, "if_not_exists": ""}');
+SELECT 'ddl msg3' FROM pg_logical_emit_ddl_message('ddl msg3', 16394, 1, '{"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": "foo", "schemaname": "element_test"}, "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 %{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}, "generated_column": {"fmt": "GENERATED ALWAYS AS", "present": false}}]}}');
+SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
+
SELECT pg_drop_replication_slot('regression_slot');
/* check that the slot is gone */
diff --git a/contrib/test_decoding/test_decoding.c b/contrib/test_decoding/test_decoding.c
index 628c6a2595..763d5c007f 100644
--- a/contrib/test_decoding/test_decoding.c
+++ b/contrib/test_decoding/test_decoding.c
@@ -14,9 +14,11 @@
#include "catalog/pg_type.h"
+#include "replication/ddlmessage.h"
#include "replication/logical.h"
#include "replication/origin.h"
+#include "tcop/ddl_deparse.h"
#include "utils/builtins.h"
#include "utils/lsyscache.h"
#include "utils/memutils.h"
@@ -72,6 +74,12 @@ static void pg_decode_message(LogicalDecodingContext *ctx,
ReorderBufferTXN *txn, XLogRecPtr lsn,
bool transactional, const char *prefix,
Size sz, const char *message);
+static void pg_decode_ddl_message(LogicalDecodingContext *ctx,
+ ReorderBufferTXN *txn,
+ XLogRecPtr message_lsn,
+ const char *prefix, Oid relid,
+ DeparsedCommandType cmdtype,
+ Size sz, const char *message);
static bool pg_decode_filter_prepare(LogicalDecodingContext *ctx,
TransactionId xid,
const char *gid);
@@ -135,6 +143,7 @@ _PG_output_plugin_init(OutputPluginCallbacks *cb)
cb->filter_by_origin_cb = pg_decode_filter;
cb->shutdown_cb = pg_decode_shutdown;
cb->message_cb = pg_decode_message;
+ cb->ddl_cb = pg_decode_ddl_message;
cb->filter_prepare_cb = pg_decode_filter_prepare;
cb->begin_prepare_cb = pg_decode_begin_prepare_txn;
cb->prepare_cb = pg_decode_prepare_txn;
@@ -750,6 +759,45 @@ pg_decode_message(LogicalDecodingContext *ctx,
OutputPluginWrite(ctx, true);
}
+static void
+pg_decode_ddl_message(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
+ XLogRecPtr message_lsn, const char *prefix, Oid relid,
+ DeparsedCommandType cmdtype, Size sz, const char *message)
+{
+ OutputPluginPrepareWrite(ctx, true);
+ appendStringInfo(ctx->out, "message: prefix: %s, relid: %u, ",
+ prefix, relid);
+
+ switch(cmdtype)
+ {
+ case DCT_SimpleCmd:
+ appendStringInfo(ctx->out, "cmdtype: Simple, ");
+ break;
+ case DCT_TableDropStart:
+ appendStringInfo(ctx->out, "cmdtype: Drop start, ");
+ break;
+ case DCT_TableDropEnd:
+ appendStringInfo(ctx->out, "cmdtype: Drop end, ");
+ break;
+ case DCT_TableAlter:
+ appendStringInfo(ctx->out, "cmdtype: Alter table, ");
+ break;
+ case DCT_ObjectCreate:
+ appendStringInfo(ctx->out, "cmdtype: Create, ");
+ break;
+ case DCT_ObjectDrop:
+ appendStringInfo(ctx->out, "cmdtype: Drop, ");
+ break;
+ default:
+ appendStringInfo(ctx->out, "cmdtype: Invalid, ");
+ break;
+ }
+
+ appendStringInfo(ctx->out, "sz: %zu content:", sz);
+ appendBinaryStringInfo(ctx->out, message, sz);
+ OutputPluginWrite(ctx, true);
+}
+
static void
pg_decode_stream_start(LogicalDecodingContext *ctx,
ReorderBufferTXN *txn)
diff --git a/doc/src/sgml/logical-replication.sgml b/doc/src/sgml/logical-replication.sgml
index 10ada41d80..6ed38ee6c1 100644
--- a/doc/src/sgml/logical-replication.sgml
+++ b/doc/src/sgml/logical-replication.sgml
@@ -1457,6 +1457,1051 @@ test_sub=# SELECT * FROM t1 ORDER BY id;
</sect1>
+ <sect1 id="logical-replication-ddl">
+ <title>DDL Replication</title>
+ <para>
+ Data Definition Commands (DDLs) can be replicated using logical replication.
+ While enabled this feature automatically replicates supported DDL commands
+ that are successfully executed on a publisher to a subscriber. This is
+ especially useful if you have lots schema changes over time that need replication.
+ </para>
+
+ <para>
+ For example, when enabled a CREATE TABLE command executed on the publisher gets
+ WAL-logged, and forwarded to the subscriber to replay; a subsequent "ALTER
+ SUBSCRIPTION ... REFRESH PUBLICATION" is run on the subscriber database so any
+ following DML changes on the new table can be replicated without a hitch.
+ </para>
+
+ <para>
+ DDL replication is disabled by default, it can be enabled at different levels
+ using the ddl PUBLICATION option. This option currently has two levels and are
+ only allowed to be set if the PUBLICATION is FOR ALL TABLES.
+ </para>
+
+ <itemizedlist>
+ <listitem>
+ <para>
+ table: this option enables replication of Table DDL commands,
+ which include:
+ <itemizedlist>
+ <listitem>
+ <para>
+ CREATE/ALTER/DROP TABLE
+ </para>
+ </listitem>
+
+ <listitem>
+ <para>
+ CREATE TABLE AS
+ </para>
+ </listitem>
+
+ </itemizedlist>
+ </para>
+ </listitem>
+ </itemizedlist>
+
+ <sect2 id="ddl-replication-option-examples">
+ <title>Examples - Setup DDL Replication on the Publisher</title>
+
+ <para>
+ Enable table ddl replication for an existing Publication:
+<programlisting>
+ALTER PUBLICATION mypub SET (ddl = 'table');
+</programlisting></para>
+
+ </sect2>
+
+ <sect2 id="ddl-replication-supported-commands">
+ <title>Supported DDL commands</title>
+
+ <para>
+ The DDL commands supported for logical replication are listed in the following
+ matrix. Note that global commands can be executed at any database and are currently
+ not supported for replication, global commands include ROLE statements, Database
+ statements, TableSpace statements and some of the GrantStmt/RevokeStmt if the target
+ object is a global object.
+ </para>
+
+ <table id="ddl-replication-by-command-tag">
+ <title>DDL Replication Support by Command Tag</title>
+ <tgroup cols="3">
+ <colspec colname="col1" colwidth="2*"/>
+ <colspec colname="col2" colwidth="1*"/>
+ <colspec colname="col3" colwidth="1*"/>
+ <thead>
+ <row>
+ <entry>Command Tag</entry>
+ <entry>For Replication</entry>
+ <entry>Notes</entry>
+ </row>
+ </thead>
+ <tbody>
+ <row>
+ <entry align="left"><literal>ALTER AGGREGATE</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>ALTER CAST</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>ALTER COLLATION</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>ALTER CONVERSION</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>ALTER DATABASE</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>ALTER DEFAULT PRIVILEGES</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>ALTER DOMAIN</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>ALTER EVENT TRIGGER</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>ALTER EXTENSION</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>ALTER FOREIGN DATA WRAPPER</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>ALTER FOREIGN TABLE</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>ALTER FUNCTION</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>ALTER INDEX</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>ALTER LANGUAGE</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>ALTER LARGE OBJECT</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>ALTER MATERIALIZED VIEW</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>ALTER OPERATOR</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>ALTER OPERATOR CLASS</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>ALTER OPERATOR FAMILY</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>ALTER POLICY</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>ALTER PROCEDURE</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>ALTER PUBLICATION</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>ALTER ROLE</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>ALTER ROUTINE</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>ALTER RULER</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>ALTER SCHEMA</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>ALTER SEQUENCE</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>ALTER SERVER</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>ALTER STATISTICS</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>ALTER SUBSCRIPTION</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>ALTER SYSTEM</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>ALTER TABLE</literal></entry>
+ <entry align="center"><literal>X</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>ALTER TABLESPACE</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>ALTER TEXT SEARCH CONFIGURATION</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>ALTER TEXT SEARCH DICTIONARY</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>ALTER TEXT SEARCH PARSER</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>ALTER TEXT SEARCH TEMPLATE</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>ALTER TRANSFORM</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>ALTER TRIGGER</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>ALTER TYPE</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>ALTER USER MAPPING</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>ALTER VIEW</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>ANALYZE</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>BEGIN</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>CALL</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>CHECKPOINT</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>CLOSE</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>CLOSE CURSOR</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>CLOSE CURSOR ALL</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>CLUSTER</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>COMMENT</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>COMMIT</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>COMMIT PREPARED</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>COPY</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>COPY FROM</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>CREATE ACCESS METHOD</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>CREATE AGGREGATE</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>CREATE CAST</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>CREATE COLLATION</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>CREATE CONSTRAINT</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>CREATE CONVERSION</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>CREATE DATABASE</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>CREATE DOMAIN</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>CREATE EVENT TRIGGER</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>CREATE EXTENSION</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>CREATE FOREIGN DATA WRAPPER</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>CREATE FOREIGN TABLE</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>CREATE FUNCTION</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>CREATE INDEX</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>CREATE LANGUAGE</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>CREATE MATERIALIZED VIEW</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>CREATE OPERATOR</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>CREATE OPERATOR CLASS</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>CREATE OPERATOR FAMILY</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>CREATE POLICY</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>CREATE PROCEDURE</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>CREATE PUBLICATION</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>CREATE ROLE</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>CREATE RULE</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>CREATE SCHEMA</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>CREATE SEQUENCE</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>CREATE SERVER</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>CREATE STATISTICS</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>CREATE SUBSCRIPTION</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>CREATE TABLE</literal></entry>
+ <entry align="center"><literal>X</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>CREATE TABLE AS</literal></entry>
+ <entry align="center"><literal>X</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>CREATE TABLESPACE</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>CREATE TEXT SEARCH CONFIGURATION</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>CREATE TEXT SEARCH DICTIONARY</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>CREATE TEXT SEARCH PARSER</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>CREATE TEXT SEARCH TEMPLATE</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>CREATE TRANSFORM</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>CREATE TRIGGER</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>CREATE TYPE</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>CREATE USER MAPPING</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>CREATE VIEW</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>DEALLOCATE</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>DEALLOCATE ALL</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>DECLARE CURSOR</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>DELETE</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>DISCARD</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>DISCARD ALL</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>DISCARD PLANS</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>DISCARD SEQUENCES</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>DISCARD TEMP</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>DO</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>DROP ACCESS METHOD</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>DROP AGGREGATE</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>DROP CAST</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>DROP COLLATION</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>DROP CONSTRAINT</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>DROP CONVERSION</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>DROP DATABASE</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>DROP DOMAIN</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>DROP EVENT TRIGGER</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>DROP EXTENSION</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>DROP FOREIGN DATA WRAPPER</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>DROP FOREIGN TABLE</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>DROP FUNCTION</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>DROP INDEX</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>DROP LANGUAGE</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>DROP MATERIALIZED VIEW</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>DROP OPERATOR</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>DROP OPERATOR CLASS</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>DROP OPERATOR FAMILY</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>DROP OWNED</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>DROP POLICY</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>DROP PROCEDURE</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>DROP PUBLICATION</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>DROP ROLE</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>DROP ROUTINE</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>DROP RULE</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>DROP SCHEMA</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>DROP SEQUENCE</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>DROP SERVER</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>DROP STATISTICS</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>DROP SUBSCRIPTION</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>DROP TABLE</literal></entry>
+ <entry align="center"><literal>X</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>DROP TABLESPACE</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>DROP TEXT SEARCH CONFIGURATION</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>DROP TEXT SEARCH DICTIONARY</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>DROP TEXT SEARCH PARSER</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>DROP TEXT SEARCH TEMPLATE</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>DROP TRANSFORM</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>DROP TRIGGER</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>DROP TYPE</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>DROP USER MAPPING</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>DROP VIEW</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>EXECUTE</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>EXPLAIN</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>FETCH</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>GRANT</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>GRANT ROLE</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>IMPORT FOREIGN SCHEMA</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>INSERT</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>LISTEN</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>LOAD</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>LOCK TABLE</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>MERGE</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>MOVE</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>NOTIFY</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>PREPARE</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>PREPARE TRANSACTION</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>REASSIGN OWNED</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>REFRESH MATERIALIZED VIEW</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>REINDEX</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>RELEASE</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>RESET</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>REVOKE</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>REVOKE ROLE</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>ROLLBACK</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>ROLLBACK PREPARED</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>SAVEPOINT</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>SECURITY LABEL</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>SET</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>SET CONSTRAINTS</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>SHOW</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>START TRANSACTION</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>UNLISTEN</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>UPDATE</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ <row>
+ <entry align="left"><literal>VACUUM</literal></entry>
+ <entry align="center"><literal>-</literal></entry>
+ <entry align="left"></entry>
+ </row>
+ </tbody>
+ </tgroup>
+ </table>
+ </sect2>
+
+ <sect2 id="ddl-replication-ddl-deparser">
+ <title>DDL Deparser</title>
+ <para>
+ The DDL deparser utility is invoked during the replication of DDLs. The DDL
+ deparser is capable of converting a DDL command into formatted JSON blob, with
+ the necessary information to reconstruct the DDL commands at the destination. The
+ benefit of using the deparser output compared to the original command string
+ includes:
+ <itemizedlist>
+ <listitem>
+ <para>
+ Every database object in the deparsed output is schema-qualified, so that
+ there are no ambiguities even in the face of search_path changes.
+ </para>
+ </listitem>
+
+ <listitem>
+ <para>
+ The structured JSON and the formatted output makes it possible for
+ machine editing. This can be useful if the subscriber is on a different
+ PG version and has certain DDL syntax differences which need to be
+ resolved before apply.
+ </para>
+ </listitem>
+
+ </itemizedlist>
+ </para>
+
+ <para>
+ The DDL deparser exposes two SQL functions:
+ <itemizedlist>
+ <listitem>
+ <para>
+ ddl_deparse_to_json: given a CollectedCommand from event trigger,
+ returns a JSON representation of it. The command is expanded fully
+ so that there are no ambiguities even in the face of search_path
+ changes.
+ </para>
+ </listitem>
+ </itemizedlist>
+
+ <itemizedlist>
+ <listitem>
+ <para>
+ ddl_deparse_expand_command: Expand JSON format DDL generated by
+ ddl_deparse_to_json to a plain DDL command.
+ </para>
+ </listitem>
+ </itemizedlist>
+ </para>
+ </sect2>
+
+ </sect1>
+
<sect1 id="logical-replication-conflicts">
<title>Conflicts</title>
diff --git a/src/backend/access/rmgrdesc/Makefile b/src/backend/access/rmgrdesc/Makefile
index f88d72fd86..2ff01e69bf 100644
--- a/src/backend/access/rmgrdesc/Makefile
+++ b/src/backend/access/rmgrdesc/Makefile
@@ -18,6 +18,7 @@ OBJS = \
gistdesc.o \
hashdesc.o \
heapdesc.o \
+ logicalddlmsgdesc.o \
logicalmsgdesc.o \
mxactdesc.o \
nbtdesc.o \
diff --git a/src/backend/access/rmgrdesc/logicalddlmsgdesc.c b/src/backend/access/rmgrdesc/logicalddlmsgdesc.c
new file mode 100644
index 0000000000..05e930c90c
--- /dev/null
+++ b/src/backend/access/rmgrdesc/logicalddlmsgdesc.c
@@ -0,0 +1,52 @@
+/*-------------------------------------------------------------------------
+ *
+ * logicalddlmsgdesc.c
+ * rmgr descriptor routines for replication/logical/ddlmessage.c
+ *
+ * Portions Copyright (c) 2015-2022, PostgreSQL Global Development Group
+ *
+ *
+ * IDENTIFICATION
+ * src/backend/access/rmgrdesc/logicalddlmsgdesc.c
+ *
+ *-------------------------------------------------------------------------
+ */
+#include "postgres.h"
+
+#include "replication/ddlmessage.h"
+
+void
+logicalddlmsg_desc(StringInfo buf, XLogReaderState *record)
+{
+ char *rec = XLogRecGetData(record);
+ uint8 info = XLogRecGetInfo(record) & ~XLR_INFO_MASK;
+
+ if (info == XLOG_LOGICAL_DDL_MESSAGE)
+ {
+ xl_logical_ddl_message *xlrec = (xl_logical_ddl_message *) rec;
+ char *prefix = xlrec->message;
+ char *message = xlrec->message + xlrec->prefix_size;
+ char *sep = "";
+
+ Assert(prefix[xlrec->prefix_size] != '\0');
+
+ appendStringInfo(buf, "prefix \"%s\"; payload (%zu bytes): ",
+ prefix, xlrec->message_size);
+ appendStringInfo(buf, "relid %u cmdtype %u", xlrec->relid, xlrec->cmdtype);
+ /* Write message payload as a series of hex bytes */
+ for (int cnt = 0; cnt < xlrec->message_size; cnt++)
+ {
+ appendStringInfo(buf, "%s%02X", sep, (unsigned char) message[cnt]);
+ sep = " ";
+ }
+ }
+}
+
+const char *
+logicalddlmsg_identify(uint8 info)
+{
+ if ((info & ~XLR_INFO_MASK) == XLOG_LOGICAL_DDL_MESSAGE)
+ return "DDL";
+
+ return NULL;
+}
diff --git a/src/backend/access/rmgrdesc/meson.build b/src/backend/access/rmgrdesc/meson.build
index 166cee67b6..781e2d7713 100644
--- a/src/backend/access/rmgrdesc/meson.build
+++ b/src/backend/access/rmgrdesc/meson.build
@@ -11,6 +11,7 @@ rmgr_desc_sources = files(
'gistdesc.c',
'hashdesc.c',
'heapdesc.c',
+ 'logicalddlmsgdesc.c',
'logicalmsgdesc.c',
'mxactdesc.c',
'nbtdesc.c',
diff --git a/src/backend/access/transam/rmgr.c b/src/backend/access/transam/rmgr.c
index 7d67eda5f7..678e81ae01 100644
--- a/src/backend/access/transam/rmgr.c
+++ b/src/backend/access/transam/rmgr.c
@@ -27,6 +27,7 @@
#include "fmgr.h"
#include "funcapi.h"
#include "miscadmin.h"
+#include "replication/ddlmessage.h"
#include "replication/decode.h"
#include "replication/message.h"
#include "replication/origin.h"
diff --git a/src/backend/catalog/pg_publication.c b/src/backend/catalog/pg_publication.c
index 47637f28ab..d209602e28 100644
--- a/src/backend/catalog/pg_publication.c
+++ b/src/backend/catalog/pg_publication.c
@@ -1028,6 +1028,7 @@ GetPublication(Oid pubid)
pub->pubactions.pubupdate = pubform->pubupdate;
pub->pubactions.pubdelete = pubform->pubdelete;
pub->pubactions.pubtruncate = pubform->pubtruncate;
+ pub->pubactions.pubddl_table = pubform->pubddl_table;
pub->pubviaroot = pubform->pubviaroot;
ReleaseSysCache(tup);
diff --git a/src/backend/commands/event_trigger.c b/src/backend/commands/event_trigger.c
index cfc36b02d0..71d2c43afc 100644
--- a/src/backend/commands/event_trigger.c
+++ b/src/backend/commands/event_trigger.c
@@ -38,6 +38,8 @@
#include "parser/parse_func.h"
#include "parser/parser.h"
#include "pgstat.h"
+#include "replication/ddlmessage.h"
+#include "replication/message.h"
#include "tcop/deparse_utility.h"
#include "tcop/ddl_deparse.h"
#include "tcop/utility.h"
diff --git a/src/backend/commands/publicationcmds.c b/src/backend/commands/publicationcmds.c
index f4ba572697..67f1bcec5c 100644
--- a/src/backend/commands/publicationcmds.c
+++ b/src/backend/commands/publicationcmds.c
@@ -38,10 +38,12 @@
#include "commands/publicationcmds.h"
#include "funcapi.h"
#include "miscadmin.h"
+#include "nodes/makefuncs.h"
#include "nodes/nodeFuncs.h"
#include "parser/parse_clause.h"
#include "parser/parse_collate.h"
#include "parser/parse_relation.h"
+#include "parser/parser.h"
#include "storage/lmgr.h"
#include "utils/acl.h"
#include "utils/array.h"
@@ -82,21 +84,25 @@ static void PublicationDropSchemas(Oid pubid, List *schemas, bool missing_ok);
static void
parse_publication_options(ParseState *pstate,
List *options,
+ bool for_all_tables,
bool *publish_given,
PublicationActions *pubactions,
bool *publish_via_partition_root_given,
- bool *publish_via_partition_root)
+ bool *publish_via_partition_root,
+ bool *ddl_type_given)
{
ListCell *lc;
*publish_given = false;
*publish_via_partition_root_given = false;
+ *ddl_type_given = false;
/* defaults */
pubactions->pubinsert = true;
pubactions->pubupdate = true;
pubactions->pubdelete = true;
pubactions->pubtruncate = true;
+ pubactions->pubddl_table = false;
*publish_via_partition_root = false;
/* Parse options */
@@ -123,7 +129,7 @@ parse_publication_options(ParseState *pstate,
pubactions->pubtruncate = false;
*publish_given = true;
- publish = defGetString(defel);
+ publish = pstrdup(defGetString(defel));
if (!SplitIdentifierString(publish, ',', &publish_list))
ereport(ERROR,
@@ -158,6 +164,42 @@ parse_publication_options(ParseState *pstate,
*publish_via_partition_root_given = true;
*publish_via_partition_root = defGetBoolean(defel);
}
+ else if (strcmp(defel->defname, "ddl") == 0)
+ {
+ char *ddl_level;
+ List *ddl_list;
+ ListCell *lc3;
+
+ if (*ddl_type_given)
+ errorConflictingDefElem(defel, pstate);
+
+ /*
+ * If ddl option was given only the explicitly listed ddl types
+ * should be published.
+ */
+ pubactions->pubddl_table = false;
+
+ *ddl_type_given = true;
+ ddl_level = defGetString(defel);
+
+ if (!SplitIdentifierString(ddl_level, ',', &ddl_list))
+ ereport(ERROR,
+ errcode(ERRCODE_SYNTAX_ERROR),
+ errmsg("invalid list syntax for \"ddl\" option"));
+
+ /* Process the option list. */
+ foreach(lc3, ddl_list)
+ {
+ char *publish_opt = (char *) lfirst(lc3);
+
+ if (strcmp(publish_opt, "table") == 0)
+ pubactions->pubddl_table = true;
+ else
+ ereport(ERROR,
+ errcode(ERRCODE_SYNTAX_ERROR),
+ errmsg("unrecognized \"ddl\" value: \"%s\"", publish_opt));
+ }
+ }
else
ereport(ERROR,
(errcode(ERRCODE_SYNTAX_ERROR),
@@ -599,14 +641,56 @@ check_simple_rowfilter_expr(Node *node, ParseState *pstate)
return check_simple_rowfilter_expr_walker(node, pstate);
}
+/*
+ * Helper function to tranform a where clause.
+ *
+ * Also check the publication row filter expression and throw an error if
+ * anything not permitted or unexpected is encountered.
+ */
+static Node *
+GetTransformWhereClauses(const char *queryString, Relation relation,
+ Node *whereClause, bool check_expr)
+{
+ Node *transformedWhereClause = NULL;
+ ParseNamespaceItem *nsitem;
+ ParseState *pstate;
+
+ /*
+ * A fresh pstate is required so that we only have "this" table in its
+ * rangetable
+ */
+ pstate = make_parsestate(NULL);
+ pstate->p_sourcetext = queryString;
+ nsitem = addRangeTableEntryForRelation(pstate, relation,
+ AccessShareLock, NULL,
+ false, false);
+ addNSItemToQuery(pstate, nsitem, false, true, true);
+
+ transformedWhereClause = transformWhereClause(pstate,
+ copyObject(whereClause),
+ EXPR_KIND_WHERE,
+ "PUBLICATION WHERE");
+
+ /* Fix up collation information */
+ assign_expr_collations(pstate, transformedWhereClause);
+
+ /*
+ * We allow only simple expressions in row filters. See
+ * check_simple_rowfilter_expr_walker.
+ */
+ if (check_expr)
+ check_simple_rowfilter_expr(transformedWhereClause, pstate);
+
+ free_parsestate(pstate);
+
+ return transformedWhereClause;
+}
+
/*
* Transform the publication WHERE expression for all the relations in the list,
* ensuring it is coerced to boolean and necessary collation information is
* added if required, and add a new nsitem/RTE for the associated relation to
* the ParseState's namespace list.
- *
- * Also check the publication row filter expression and throw an error if
- * anything not permitted or unexpected is encountered.
*/
static void
TransformPubWhereClauses(List *tables, const char *queryString,
@@ -616,9 +700,6 @@ TransformPubWhereClauses(List *tables, const char *queryString,
foreach(lc, tables)
{
- ParseNamespaceItem *nsitem;
- Node *whereclause = NULL;
- ParseState *pstate;
PublicationRelInfo *pri = (PublicationRelInfo *) lfirst(lc);
if (pri->whereClause == NULL)
@@ -638,34 +719,8 @@ TransformPubWhereClauses(List *tables, const char *queryString,
errdetail("WHERE clause cannot be used for a partitioned table when %s is false.",
"publish_via_partition_root")));
- /*
- * A fresh pstate is required so that we only have "this" table in its
- * rangetable
- */
- pstate = make_parsestate(NULL);
- pstate->p_sourcetext = queryString;
- nsitem = addRangeTableEntryForRelation(pstate, pri->relation,
- AccessShareLock, NULL,
- false, false);
- addNSItemToQuery(pstate, nsitem, false, true, true);
-
- whereclause = transformWhereClause(pstate,
- copyObject(pri->whereClause),
- EXPR_KIND_WHERE,
- "PUBLICATION WHERE");
-
- /* Fix up collation information */
- assign_expr_collations(pstate, whereclause);
-
- /*
- * We allow only simple expressions in row filters. See
- * check_simple_rowfilter_expr_walker.
- */
- check_simple_rowfilter_expr(whereclause, pstate);
-
- free_parsestate(pstate);
-
- pri->whereClause = whereclause;
+ pri->whereClause = GetTransformWhereClauses(queryString, pri->relation,
+ pri->whereClause, true);
}
}
@@ -728,6 +783,52 @@ CheckPubRelationColumnList(char *pubname, List *tables,
}
}
+/*
+ * Create event trigger which is used for DDL replication.
+ */
+static void
+CreateDDLReplicaEventTrigger(char *eventname, List *commands,
+ ObjectAddress pubaddress, Oid puboid)
+{
+ List *tags = NIL;
+ ListCell *lc;
+ Oid trigger_id;
+ ObjectAddress referenced;
+ CreateEventTrigStmt *ddl_trigger;
+ char trigger_name[NAMEDATALEN];
+ char trigger_func_name[NAMEDATALEN];
+ static const char *trigger_func_prefix = "publication_deparse_%s";
+
+ ddl_trigger = makeNode(CreateEventTrigStmt);
+
+ snprintf(trigger_name, sizeof(trigger_name), PUB_EVENT_TRIG_PREFIX,
+ eventname, puboid);
+ snprintf(trigger_func_name, sizeof(trigger_func_name), trigger_func_prefix,
+ eventname);
+
+ ddl_trigger->trigname = pstrdup(trigger_name);
+ ddl_trigger->eventname = eventname;
+ ddl_trigger->funcname = SystemFuncName(trigger_func_name);
+
+ foreach(lc, commands)
+ {
+ CommandTag cmdtag = lfirst_int(lc);
+ String *tag = makeString(pstrdup(GetCommandTagName(cmdtag)));
+
+ tags = lappend(tags, tag);
+ }
+
+ ddl_trigger->whenclause = list_make1(makeDefElem("tag", (Node *) tags, -1));
+
+ trigger_id = CreateEventTrigger(ddl_trigger);
+
+ /*
+ * Register the event triggers as internally dependent on the publication.
+ */
+ ObjectAddressSet(referenced, EventTriggerRelationId, trigger_id);
+ recordDependencyOn(&referenced, &pubaddress, DEPENDENCY_INTERNAL);
+}
+
/*
* Create new publication.
*/
@@ -741,6 +842,7 @@ CreatePublication(ParseState *pstate, CreatePublicationStmt *stmt)
Datum values[Natts_pg_publication];
HeapTuple tup;
bool publish_given;
+ bool ddl_type_given;
PublicationActions pubactions;
bool publish_via_partition_root_given;
bool publish_via_partition_root;
@@ -781,9 +883,11 @@ CreatePublication(ParseState *pstate, CreatePublicationStmt *stmt)
parse_publication_options(pstate,
stmt->options,
+ stmt->for_all_tables,
&publish_given, &pubactions,
&publish_via_partition_root_given,
- &publish_via_partition_root);
+ &publish_via_partition_root,
+ &ddl_type_given);
puboid = GetNewOidWithIndex(rel, PublicationObjectIndexId,
Anum_pg_publication_oid);
@@ -798,6 +902,8 @@ CreatePublication(ParseState *pstate, CreatePublicationStmt *stmt)
BoolGetDatum(pubactions.pubdelete);
values[Anum_pg_publication_pubtruncate - 1] =
BoolGetDatum(pubactions.pubtruncate);
+ values[Anum_pg_publication_pubddl_table - 1] =
+ BoolGetDatum(pubactions.pubddl_table);
values[Anum_pg_publication_pubviaroot - 1] =
BoolGetDatum(publish_via_partition_root);
@@ -835,6 +941,12 @@ CreatePublication(ParseState *pstate, CreatePublicationStmt *stmt)
{
List *rels;
+ if (pubactions.pubddl_table)
+ ereport(ERROR,
+ errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("cannot add table to publication \"%s\" if DDL replication is enabled",
+ stmt->pubname));
+
rels = OpenTableList(relations);
TransformPubWhereClauses(rels, pstate->p_sourcetext,
publish_via_partition_root);
@@ -858,6 +970,39 @@ CreatePublication(ParseState *pstate, CreatePublicationStmt *stmt)
}
}
+ /*
+ * Create an event trigger to allow logging of DDL statements.
+ *
+ */
+ if (pubactions.pubddl_table)
+ {
+ List *start_commands = list_make1_int(CMDTAG_DROP_TABLE);
+ List *rewrite_commands = list_make1_int(CMDTAG_ALTER_TABLE);
+ List *init_commands = list_make2_int(CMDTAG_CREATE_TABLE_AS,
+ CMDTAG_SELECT_INTO);
+ List *end_commands = NIL;
+
+ end_commands = lappend_int(end_commands, CMDTAG_CREATE_TABLE);
+ end_commands = lappend_int(end_commands, CMDTAG_ALTER_TABLE);
+ end_commands = lappend_int(end_commands, CMDTAG_DROP_TABLE);
+
+ /* Create the ddl_command_end event trigger */
+ CreateDDLReplicaEventTrigger(PUB_TRIG_EVENT1, end_commands,
+ myself, puboid);
+
+ /* Create the ddl_command_start event trigger */
+ CreateDDLReplicaEventTrigger(PUB_TRIG_EVENT2, start_commands,
+ myself, puboid);
+
+ /* Create the table_rewrite event trigger */
+ CreateDDLReplicaEventTrigger(PUB_TRIG_EVENT3, rewrite_commands,
+ myself, puboid);
+
+ /* Create the table_init_write event trigger */
+ CreateDDLReplicaEventTrigger(PUB_TRIG_EVENT4, init_commands,
+ myself, puboid);
+ }
+
table_close(rel, RowExclusiveLock);
InvokeObjectPostCreateHook(PublicationRelationId, puboid, 0);
@@ -882,6 +1027,7 @@ AlterPublicationOptions(ParseState *pstate, AlterPublicationStmt *stmt,
bool replaces[Natts_pg_publication];
Datum values[Natts_pg_publication];
bool publish_given;
+ bool ddl_type_given;
PublicationActions pubactions;
bool publish_via_partition_root_given;
bool publish_via_partition_root;
@@ -890,11 +1036,15 @@ AlterPublicationOptions(ParseState *pstate, AlterPublicationStmt *stmt,
List *root_relids = NIL;
ListCell *lc;
+ pubform = (Form_pg_publication) GETSTRUCT(tup);
+
parse_publication_options(pstate,
stmt->options,
+ pubform->puballtables,
&publish_given, &pubactions,
&publish_via_partition_root_given,
- &publish_via_partition_root);
+ &publish_via_partition_root,
+ &ddl_type_given);
pubform = (Form_pg_publication) GETSTRUCT(tup);
@@ -978,6 +1128,18 @@ AlterPublicationOptions(ParseState *pstate, AlterPublicationStmt *stmt,
}
}
+ if (ddl_type_given && pubactions.pubddl_table)
+ {
+ if (root_relids == NIL)
+ root_relids = GetPublicationRelations(pubform->oid,
+ PUBLICATION_PART_ROOT);
+
+ if (root_relids)
+ ereport(ERROR,
+ errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("DDL replication is only supported in FOR ALL TABLES or FOR TABLES IN SCHEMA publication"));
+ }
+
/* Everything ok, form a new tuple. */
memset(values, 0, sizeof(values));
memset(nulls, false, sizeof(nulls));
@@ -996,6 +1158,9 @@ AlterPublicationOptions(ParseState *pstate, AlterPublicationStmt *stmt,
values[Anum_pg_publication_pubtruncate - 1] = BoolGetDatum(pubactions.pubtruncate);
replaces[Anum_pg_publication_pubtruncate - 1] = true;
+
+ values[Anum_pg_publication_pubddl_table - 1] = BoolGetDatum(pubactions.pubddl_table);
+ replaces[Anum_pg_publication_pubddl_table - 1] = true;
}
if (publish_via_partition_root_given)
@@ -1103,6 +1268,12 @@ AlterPublicationTables(AlterPublicationStmt *stmt, HeapTuple tup,
if (stmt->action == AP_AddObjects)
{
+ if (pubform->pubddl_table)
+ ereport(ERROR,
+ errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("cannot add table to publication \"%s\" if DDL replication is enabled",
+ stmt->pubname));
+
TransformPubWhereClauses(rels, queryString, pubform->pubviaroot);
publish_schema |= is_schema_publication(pubid);
@@ -1121,6 +1292,12 @@ AlterPublicationTables(AlterPublicationStmt *stmt, HeapTuple tup,
List *delrels = NIL;
ListCell *oldlc;
+ if (pubform->pubddl_table)
+ ereport(ERROR,
+ errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("cannot add table to publication \"%s\" if DDL replication is enabled",
+ stmt->pubname));
+
TransformPubWhereClauses(rels, queryString, pubform->pubviaroot);
CheckPubRelationColumnList(stmt->pubname, rels, publish_schema,
diff --git a/src/backend/replication/logical/Makefile b/src/backend/replication/logical/Makefile
index 2dc25e37bb..b79ddd8cdc 100644
--- a/src/backend/replication/logical/Makefile
+++ b/src/backend/replication/logical/Makefile
@@ -16,6 +16,8 @@ override CPPFLAGS := -I$(srcdir) $(CPPFLAGS)
OBJS = \
applyparallelworker.o \
+ ddlmessage.o \
+ ddltrigger.o \
decode.o \
launcher.o \
logical.o \
diff --git a/src/backend/replication/logical/ddlmessage.c b/src/backend/replication/logical/ddlmessage.c
new file mode 100644
index 0000000000..e11e56e5a3
--- /dev/null
+++ b/src/backend/replication/logical/ddlmessage.c
@@ -0,0 +1,84 @@
+/*-------------------------------------------------------------------------
+ *
+ * ddlmessage.c
+ * Logical DDL messages.
+ *
+ * Copyright (c) 2022, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ * src/backend/replication/logical/ddlmessage.c
+ *
+ * NOTES
+ *
+ * Logical DDL messages allow XLOG logging of DDL command strings that
+ * get passed to the logical decoding plugin. In normal XLOG processing they
+ * are same as NOOP.
+ *
+ * Unlike generic logical messages, these DDL messages have only transactional
+ * mode. Note by default DDLs in PostgreSQL are transactional.
+ *
+ * These messages are part of current transaction and will be sent to
+ * decoding plugin using in a same way as DML operations.
+ *
+ * Every message carries prefix to avoid conflicts between different decoding
+ * plugins. The plugin authors must take extra care to use unique prefix,
+ * good options seems to be for example to use the name of the extension.
+ *
+ * ---------------------------------------------------------------------------
+ */
+
+#include "postgres.h"
+
+#include "access/xact.h"
+#include "access/xloginsert.h"
+#include "catalog/namespace.h"
+#include "miscadmin.h"
+#include "nodes/execnodes.h"
+#include "replication/logical.h"
+#include "replication/ddlmessage.h"
+#include "utils/memutils.h"
+
+/*
+ * Write logical decoding DDL message into XLog.
+ */
+XLogRecPtr
+LogLogicalDDLMessage(const char *prefix, Oid relid, DeparsedCommandType cmdtype,
+ const char *message, size_t size)
+{
+ xl_logical_ddl_message xlrec;
+
+ /* Ensure we have a valid transaction id. */
+ Assert(IsTransactionState());
+ GetCurrentTransactionId();
+
+ xlrec.dbId = MyDatabaseId;
+ /* Trailing zero is critical; see logicalddlmsg_desc */
+ xlrec.prefix_size = strlen(prefix) + 1;
+ xlrec.message_size = size;
+ xlrec.relid = relid;
+ xlrec.cmdtype = cmdtype;
+
+ XLogBeginInsert();
+ XLogRegisterData((char *) &xlrec, SizeOfLogicalDDLMessage);
+ XLogRegisterData(unconstify(char *, prefix), xlrec.prefix_size);
+ XLogRegisterData(unconstify(char *, message), size);
+
+ /* Allow origin filtering */
+ XLogSetRecordFlags(XLOG_INCLUDE_ORIGIN);
+
+ return XLogInsert(RM_LOGICALDDLMSG_ID, XLOG_LOGICAL_DDL_MESSAGE);
+}
+
+/*
+ * Redo is basically just noop for logical decoding ddl messages.
+ */
+void
+logicalddlmsg_redo(XLogReaderState *record)
+{
+ uint8 info = XLogRecGetInfo(record) & ~XLR_INFO_MASK;
+
+ if (info != XLOG_LOGICAL_DDL_MESSAGE)
+ elog(PANIC, "logicalddlmsg_redo: unknown op code %u", info);
+
+ /* This is only interesting for logical decoding, see decode.c. */
+}
diff --git a/src/backend/replication/logical/ddltrigger.c b/src/backend/replication/logical/ddltrigger.c
new file mode 100644
index 0000000000..1282340c2b
--- /dev/null
+++ b/src/backend/replication/logical/ddltrigger.c
@@ -0,0 +1,283 @@
+/*-------------------------------------------------------------------------
+ *
+ * ddltrigger.c
+ * Logical DDL messages.
+ *
+ * Copyright (c) 2022, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ * src/backend/replication/logical/ddltrigger.c
+ *
+ * NOTES
+ *
+ * Deparse the ddl command and log it.
+ *
+ * ---------------------------------------------------------------------------
+ */
+
+#include "postgres.h"
+
+#include "access/table.h"
+#include "catalog/pg_class.h"
+#include "commands/event_trigger.h"
+#include "funcapi.h"
+#include "lib/ilist.h"
+#include "replication/ddlmessage.h"
+#include "tcop/ddl_deparse.h"
+#include "utils/fmgrprotos.h"
+#include "utils/lsyscache.h"
+
+extern EventTriggerQueryState *currentEventTriggerState;
+
+/*
+ * Deparse the ddl command and log it prior to
+ * execution. Currently only used for DROP TABLE command
+ * so that catalog can be accessed before being deleted.
+ * This is to check if the table is part of the publication
+ * or not.
+ */
+Datum
+publication_deparse_ddl_command_start(PG_FUNCTION_ARGS)
+{
+ EventTriggerData *trigdata;
+ char *command = psprintf("Drop table command start");
+ DropStmt *stmt;
+ ListCell *cell1;
+
+ if (!CALLED_AS_EVENT_TRIGGER(fcinfo))
+ elog(ERROR, "not fired by event trigger manager");
+
+ trigdata = (EventTriggerData *) fcinfo->context;
+ stmt = (DropStmt *) trigdata->parsetree;
+
+ /* Extract the relid from the parse tree */
+ foreach(cell1, stmt->objects)
+ {
+ char relpersist;
+ Node *object = lfirst(cell1);
+ ObjectAddress address;
+ Relation relation = NULL;
+
+ address = get_object_address(stmt->removeType,
+ object,
+ &relation,
+ AccessExclusiveLock,
+ true);
+
+ /* Object does not exist, nothing to do */
+ if (!relation)
+ continue;
+
+ relpersist = get_rel_persistence(address.objectId);
+
+ /*
+ * Do not generate wal log for commands whose target table is a
+ * temporary or unlogged table.
+ *
+ * XXX We may generate wal logs for unlogged tables in the future so
+ * that unlogged tables can also be created and altered on the
+ * subscriber side. This makes it possible to directly replay the SET
+ * LOGGED command and the incoming rewrite message without creating a
+ * new table.
+ */
+ if (relpersist == RELPERSISTENCE_PERMANENT)
+ LogLogicalDDLMessage("deparse", address.objectId, DCT_TableDropStart,
+ command, strlen(command) + 1);
+
+ if (relation)
+ table_close(relation, NoLock);
+ }
+ return PointerGetDatum(NULL);
+}
+
+/*
+ * publication_deparse_table_rewrite
+ *
+ * Deparse the ddl table rewrite command and log it.
+ */
+Datum
+publication_deparse_table_rewrite(PG_FUNCTION_ARGS)
+{
+ char relpersist;
+ CollectedCommand *cmd;
+ char *json_string;
+
+ if (!CALLED_AS_EVENT_TRIGGER(fcinfo))
+ elog(ERROR, "not fired by event trigger manager");
+
+ cmd = currentEventTriggerState->currentCommand;
+
+ Assert(cmd && cmd->d.alterTable.rewrite);
+
+ relpersist = get_rel_persistence(cmd->d.alterTable.objectId);
+
+ /*
+ * Do not generate wal log for commands whose target table is a temporary
+ * or unlogged table.
+ *
+ * XXX We may generate wal logs for unlogged tables in the future so that
+ * unlogged tables can also be created and altered on the subscriber side.
+ * This makes it possible to directly replay the SET LOGGED command and the
+ * incoming rewrite message without creating a new table.
+ */
+ if (relpersist == RELPERSISTENCE_PERMANENT)
+ {
+ /* Deparse the DDL command and WAL log it to allow decoding of the same. */
+ json_string = deparse_utility_command(cmd, false);
+
+ if (json_string != NULL)
+ LogLogicalDDLMessage("deparse", cmd->d.alterTable.objectId, DCT_TableAlter,
+ json_string, strlen(json_string) + 1);
+ }
+
+ return PointerGetDatum(NULL);
+}
+
+/*
+ * Deparse the ddl command and log it. This function
+ * is called after the execution of the command but before the
+ * transaction commits.
+ */
+Datum
+publication_deparse_ddl_command_end(PG_FUNCTION_ARGS)
+{
+ ListCell *lc;
+ slist_iter iter;
+ DeparsedCommandType type;
+ Oid relid;
+ char relkind;
+
+ if (!CALLED_AS_EVENT_TRIGGER(fcinfo))
+ elog(ERROR, "not fired by event trigger manager");
+
+ foreach(lc, currentEventTriggerState->commandList)
+ {
+ char relpersist = RELPERSISTENCE_PERMANENT;
+ CollectedCommand *cmd = lfirst(lc);
+ char *json_string;
+
+ /* Rewrite DDL has been handled in table_rewrite trigger */
+ if (cmd->d.alterTable.rewrite)
+ {
+ RenameStmt *renameStmt = (RenameStmt *) cmd->parsetree;
+
+ if (renameStmt && renameStmt->relationType != OBJECT_TYPE &&
+ renameStmt->relationType != OBJECT_TABLE)
+ continue;
+ }
+
+ if (cmd->type == SCT_Simple &&
+ !OidIsValid(cmd->d.simple.address.objectId))
+ continue;
+
+ if (cmd->type == SCT_AlterTable)
+ {
+ relid = cmd->d.alterTable.objectId;
+ type = DCT_TableAlter;
+ }
+ else
+ {
+ /* Only SCT_Simple for now */
+ relid = cmd->d.simple.address.objectId;
+ type = DCT_SimpleCmd;
+ }
+
+ relkind = get_rel_relkind(relid);
+ if (relkind)
+ relpersist = get_rel_persistence(relid);
+
+ /*
+ * Do not generate wal log for commands whose target table is a
+ * temporary or unlogged table.
+ *
+ * XXX We may generate wal logs for unlogged tables in the future so
+ * that unlogged tables can also be created and altered on the
+ * subscriber side. This makes it possible to directly replay the SET
+ * LOGGED command and the incoming rewrite message without creating a
+ * new table.
+ */
+ if (relpersist == RELPERSISTENCE_PERMANENT)
+ {
+ /*
+ * Deparse the DDL command and WAL log it to allow decoding of the
+ * same.
+ */
+ json_string = deparse_utility_command(cmd, false);
+
+ if (json_string != NULL)
+ LogLogicalDDLMessage("deparse", relid, type, json_string,
+ strlen(json_string) + 1);
+ }
+ }
+
+ slist_foreach(iter, &(currentEventTriggerState->SQLDropList))
+ {
+ SQLDropObject *obj;
+ DropStmt *stmt;
+ EventTriggerData *trigdata;
+ char *command;
+ DeparsedCommandType cmdtype;
+
+ trigdata = (EventTriggerData *) fcinfo->context;
+ stmt = (DropStmt *) trigdata->parsetree;
+
+ obj = slist_container(SQLDropObject, next, iter.cur);
+
+ if (strcmp(obj->objecttype, "table") == 0)
+ cmdtype = DCT_TableDropEnd;
+ else
+ continue;
+
+ command = deparse_drop_command(obj->objidentity, obj->objecttype,
+ stmt->behavior);
+
+ if (command)
+ LogLogicalDDLMessage("deparse", obj->address.objectId, cmdtype,
+ command, strlen(command) + 1);
+ }
+
+ return PointerGetDatum(NULL);
+}
+
+
+/*
+ * publication_deparse_table_init_write
+ *
+ * Deparse the ddl table create command and log it.
+ */
+Datum
+publication_deparse_table_init_write(PG_FUNCTION_ARGS)
+{
+ char relpersist;
+ CollectedCommand *cmd;
+ char *json_string;
+
+ if (!CALLED_AS_EVENT_TRIGGER(fcinfo))
+ elog(ERROR, "not fired by event trigger manager");
+
+ cmd = currentEventTriggerState->currentCommand;
+ Assert(cmd);
+
+ relpersist = get_rel_persistence(cmd->d.simple.address.objectId);
+
+ /*
+ * Do not generate wal log for commands whose target table is a temporary
+ * table.
+ *
+ * We will generate wal logs for unlogged tables so that unlogged tables
+ * can also be created and altered on the subscriber side. This makes it
+ * possible to directly replay the SET LOGGED command and the incoming
+ * rewrite message without creating a new table.
+ */
+ if (relpersist != RELPERSISTENCE_PERMANENT)
+ return PointerGetDatum(NULL);
+
+ /* Deparse the DDL command and WAL log it to allow decoding of the same. */
+ json_string = deparse_utility_command(cmd, false);
+
+ if (json_string != NULL)
+ LogLogicalDDLMessage("deparse", cmd->d.simple.address.objectId, DCT_SimpleCmd,
+ json_string, strlen(json_string) + 1);
+
+ return PointerGetDatum(NULL);
+}
diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c
index 8fe7bb65f1..c615a23fe5 100644
--- a/src/backend/replication/logical/decode.c
+++ b/src/backend/replication/logical/decode.c
@@ -36,6 +36,7 @@
#include "access/xlogutils.h"
#include "catalog/pg_control.h"
#include "replication/decode.h"
+#include "replication/ddlmessage.h"
#include "replication/logical.h"
#include "replication/message.h"
#include "replication/origin.h"
@@ -613,6 +614,46 @@ logicalmsg_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
message->message + message->prefix_size);
}
+/*
+ * Handle rmgr LOGICALDDLMSG_ID records for DecodeRecordIntoReorderBuffer().
+ */
+void
+logicalddl_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
+{
+ SnapBuild *builder = ctx->snapshot_builder;
+ XLogReaderState *r = buf->record;
+ TransactionId xid = XLogRecGetXid(r);
+ uint8 info = XLogRecGetInfo(r) & ~XLR_INFO_MASK;
+ RepOriginId origin_id = XLogRecGetOrigin(r);
+ xl_logical_ddl_message *message;
+
+ if (info != XLOG_LOGICAL_DDL_MESSAGE)
+ elog(ERROR, "unexpected RM_LOGICALDDLMSG_ID record type: %u", info);
+
+ ReorderBufferProcessXid(ctx->reorder, XLogRecGetXid(r), buf->origptr);
+
+ /*
+ * If we don't have snapshot or we are just fast-forwarding, there is no
+ * point in decoding ddl messages.
+ */
+ if (SnapBuildCurrentState(builder) < SNAPBUILD_FULL_SNAPSHOT ||
+ ctx->fast_forward)
+ return;
+
+ message = (xl_logical_ddl_message *) XLogRecGetData(r);
+
+ if (message->dbId != ctx->slot->data.database ||
+ FilterByOrigin(ctx, origin_id))
+ return;
+
+ if (SnapBuildProcessChange(builder, xid, buf->origptr))
+ ReorderBufferQueueDDLMessage(ctx->reorder, xid, buf->endptr,
+ message->message, /* first part of message is prefix */
+ message->message_size,
+ message->message + message->prefix_size,
+ message->relid, message->cmdtype);
+}
+
/*
* Consolidated commit record handling between the different form of commit
* records.
diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c
index c3ec97a0a6..f56a716567 100644
--- a/src/backend/replication/logical/logical.c
+++ b/src/backend/replication/logical/logical.c
@@ -33,6 +33,7 @@
#include "fmgr.h"
#include "miscadmin.h"
#include "pgstat.h"
+#include "replication/ddlmessage.h"
#include "replication/decode.h"
#include "replication/logical.h"
#include "replication/origin.h"
@@ -73,6 +74,10 @@ static void truncate_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
static void message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
XLogRecPtr message_lsn, bool transactional,
const char *prefix, Size message_size, const char *message);
+static void ddl_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
+ XLogRecPtr message_lsn, const char *prefix,
+ Oid relid, DeparsedCommandType cmdtype,
+ Size message_size, const char *message);
/* streaming callbacks */
static void stream_start_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
@@ -90,6 +95,11 @@ static void stream_change_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn
static void stream_message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
XLogRecPtr message_lsn, bool transactional,
const char *prefix, Size message_size, const char *message);
+static void stream_ddl_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
+ XLogRecPtr message_lsn,
+ const char *prefix,
+ Oid relid, DeparsedCommandType cmdtype,
+ Size message_size, const char *message);
static void stream_truncate_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
int nrelations, Relation relations[], ReorderBufferChange *change);
@@ -223,6 +233,7 @@ StartupDecodingContext(List *output_plugin_options,
ctx->reorder->apply_truncate = truncate_cb_wrapper;
ctx->reorder->commit = commit_cb_wrapper;
ctx->reorder->message = message_cb_wrapper;
+ ctx->reorder->ddl = ddl_cb_wrapper;
/*
* To support streaming, we require start/stop/abort/commit/change
@@ -239,6 +250,7 @@ StartupDecodingContext(List *output_plugin_options,
(ctx->callbacks.stream_commit_cb != NULL) ||
(ctx->callbacks.stream_change_cb != NULL) ||
(ctx->callbacks.stream_message_cb != NULL) ||
+ (ctx->callbacks.stream_ddl_cb != NULL) ||
(ctx->callbacks.stream_truncate_cb != NULL);
/*
@@ -256,6 +268,7 @@ StartupDecodingContext(List *output_plugin_options,
ctx->reorder->stream_commit = stream_commit_cb_wrapper;
ctx->reorder->stream_change = stream_change_cb_wrapper;
ctx->reorder->stream_message = stream_message_cb_wrapper;
+ ctx->reorder->stream_ddl = stream_ddl_cb_wrapper;
ctx->reorder->stream_truncate = stream_truncate_cb_wrapper;
@@ -1233,6 +1246,44 @@ message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
error_context_stack = errcallback.previous;
}
+static void
+ddl_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
+ XLogRecPtr message_lsn,
+ const char *prefix, Oid relid, DeparsedCommandType cmdtype,
+ Size message_size,
+ const char *message)
+{
+ LogicalDecodingContext *ctx = cache->private_data;
+ LogicalErrorCallbackState state;
+ ErrorContextCallback errcallback;
+
+ Assert(!ctx->fast_forward);
+
+ if (ctx->callbacks.ddl_cb == NULL)
+ return;
+
+ /* Push callback + info on the error context stack */
+ state.ctx = ctx;
+ state.callback_name = "ddl";
+ state.report_location = message_lsn;
+ errcallback.callback = output_plugin_error_callback;
+ errcallback.arg = (void *) &state;
+ errcallback.previous = error_context_stack;
+ error_context_stack = &errcallback;
+
+ /* set output state */
+ ctx->accept_writes = true;
+ ctx->write_xid = txn != NULL ? txn->xid : InvalidTransactionId;
+ ctx->write_location = message_lsn;
+
+ /* do the actual work: call callback */
+ ctx->callbacks.ddl_cb(ctx, txn, message_lsn, prefix, relid, cmdtype,
+ message_size, message);
+
+ /* Pop the error context stack */
+ error_context_stack = errcallback.previous;
+}
+
static void
stream_start_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
XLogRecPtr first_lsn)
@@ -1548,6 +1599,48 @@ stream_message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
error_context_stack = errcallback.previous;
}
+static void
+stream_ddl_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
+ XLogRecPtr message_lsn,
+ const char *prefix, Oid relid, DeparsedCommandType cmdtype,
+ Size message_size,
+ const char *message)
+{
+ LogicalDecodingContext *ctx = cache->private_data;
+ LogicalErrorCallbackState state;
+ ErrorContextCallback errcallback;
+
+ Assert(!ctx->fast_forward);
+
+ /* We're only supposed to call this when streaming is supported. */
+ Assert(ctx->streaming);
+
+ /* this callback is optional */
+ if (ctx->callbacks.stream_ddl_cb == NULL)
+ return;
+
+ /* Push callback + info on the error context stack */
+ state.ctx = ctx;
+ state.callback_name = "stream_ddl";
+ state.report_location = message_lsn;
+ errcallback.callback = output_plugin_error_callback;
+ errcallback.arg = (void *) &state;
+ errcallback.previous = error_context_stack;
+ error_context_stack = &errcallback;
+
+ /* set output state */
+ ctx->accept_writes = true;
+ ctx->write_xid = txn != NULL ? txn->xid : InvalidTransactionId;
+ ctx->write_location = message_lsn;
+
+ /* do the actual work: call callback */
+ ctx->callbacks.stream_ddl_cb(ctx, txn, message_lsn, prefix, relid,
+ cmdtype, message_size, message);
+
+ /* Pop the error context stack */
+ error_context_stack = errcallback.previous;
+}
+
static void
stream_truncate_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
int nrelations, Relation relations[],
diff --git a/src/backend/replication/logical/logicalfuncs.c b/src/backend/replication/logical/logicalfuncs.c
index fa1b641a2b..92f5f9357e 100644
--- a/src/backend/replication/logical/logicalfuncs.c
+++ b/src/backend/replication/logical/logicalfuncs.c
@@ -29,6 +29,7 @@
#include "nodes/makefuncs.h"
#include "replication/decode.h"
#include "replication/logical.h"
+#include "replication/ddlmessage.h"
#include "replication/message.h"
#include "storage/fd.h"
#include "utils/array.h"
@@ -388,3 +389,26 @@ pg_logical_emit_message_text(PG_FUNCTION_ARGS)
/* bytea and text are compatible */
return pg_logical_emit_message_bytea(fcinfo);
}
+
+/*
+ * SQL function for writing logical decoding DDL message into WAL.
+ */
+Datum
+pg_logical_emit_ddl_message_bytea(PG_FUNCTION_ARGS)
+{
+ char *prefix = text_to_cstring(PG_GETARG_TEXT_PP(0));
+ Oid relid = PG_GETARG_OID(1);
+ DeparsedCommandType cmdtype = PG_GETARG_INT16(2);
+ char *data = text_to_cstring(PG_GETARG_TEXT_PP(3));
+ XLogRecPtr lsn;
+
+ lsn = LogLogicalDDLMessage(prefix, relid, cmdtype, data, strlen(data));
+ PG_RETURN_LSN(lsn);
+}
+
+Datum
+pg_logical_emit_ddl_message_text(PG_FUNCTION_ARGS)
+{
+ /* bytea and text are compatible */
+ return pg_logical_emit_ddl_message_bytea(fcinfo);
+}
diff --git a/src/backend/replication/logical/meson.build b/src/backend/replication/logical/meson.build
index d48cd4c590..05e775697e 100644
--- a/src/backend/replication/logical/meson.build
+++ b/src/backend/replication/logical/meson.build
@@ -2,6 +2,8 @@
backend_sources += files(
'applyparallelworker.c',
+ 'ddlmessage.c',
+ 'ddltrigger.c',
'decode.c',
'launcher.c',
'logical.c',
diff --git a/src/backend/replication/logical/proto.c b/src/backend/replication/logical/proto.c
index f308713275..1ba3feb187 100644
--- a/src/backend/replication/logical/proto.c
+++ b/src/backend/replication/logical/proto.c
@@ -663,6 +663,47 @@ logicalrep_write_message(StringInfo out, TransactionId xid, XLogRecPtr lsn,
pq_sendbytes(out, message, sz);
}
+/*
+ * Read DDL MESSAGE from stream
+ */
+char *
+logicalrep_read_ddl(StringInfo in, XLogRecPtr *lsn,
+ const char **prefix,
+ Size *sz)
+{
+ uint8 flags;
+ char *msg;
+
+ flags = pq_getmsgint(in, 1);
+ if (flags != 0)
+ elog(ERROR, "unrecognized flags %u in ddl message", flags);
+
+ *lsn = pq_getmsgint64(in);
+ *prefix = pq_getmsgstring(in);
+ *sz = pq_getmsgint(in, 4);
+ msg = (char *) pq_getmsgbytes(in, *sz);
+
+ return msg;
+}
+
+/*
+ * Write DDL MESSAGE to stream
+ */
+void
+logicalrep_write_ddl(StringInfo out, XLogRecPtr lsn,
+ const char *prefix, Size sz, const char *message)
+{
+ uint8 flags = 0;
+
+ pq_sendbyte(out, LOGICAL_REP_MSG_DDL);
+
+ pq_sendint8(out, flags);
+ pq_sendint64(out, lsn);
+ pq_sendstring(out, prefix);
+ pq_sendint32(out, sz);
+ pq_sendbytes(out, message, sz);
+}
+
/*
* Write relation description to the output stream.
*/
@@ -1238,6 +1279,8 @@ logicalrep_message_type(LogicalRepMsgType action)
return "TYPE";
case LOGICAL_REP_MSG_MESSAGE:
return "MESSAGE";
+ case LOGICAL_REP_MSG_DDL:
+ return "DDL";
case LOGICAL_REP_MSG_BEGIN_PREPARE:
return "BEGIN PREPARE";
case LOGICAL_REP_MSG_PREPARE:
diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
index 9f44974473..e88eb2f42b 100644
--- a/src/backend/replication/logical/reorderbuffer.c
+++ b/src/backend/replication/logical/reorderbuffer.c
@@ -94,6 +94,7 @@
#include "lib/binaryheap.h"
#include "miscadmin.h"
#include "pgstat.h"
+#include "replication/ddlmessage.h"
#include "replication/logical.h"
#include "replication/reorderbuffer.h"
#include "replication/slot.h"
@@ -516,6 +517,14 @@ ReorderBufferReturnChange(ReorderBuffer *rb, ReorderBufferChange *change,
pfree(change->data.msg.message);
change->data.msg.message = NULL;
break;
+ case REORDER_BUFFER_CHANGE_DDL:
+ if (change->data.ddl.prefix != NULL)
+ pfree(change->data.ddl.prefix);
+ change->data.ddl.prefix = NULL;
+ if (change->data.ddl.message != NULL)
+ pfree(change->data.ddl.message);
+ change->data.ddl.message = NULL;
+ break;
case REORDER_BUFFER_CHANGE_INVALIDATION:
if (change->data.inval.invalidations)
pfree(change->data.inval.invalidations);
@@ -895,6 +904,36 @@ ReorderBufferQueueMessage(ReorderBuffer *rb, TransactionId xid,
}
}
+/*
+ * A transactional DDL message is queued to be processed upon commit.
+ */
+void
+ReorderBufferQueueDDLMessage(ReorderBuffer *rb, TransactionId xid,
+ XLogRecPtr lsn, const char *prefix,
+ Size message_size, const char *message,
+ Oid relid, DeparsedCommandType cmdtype)
+{
+ MemoryContext oldcontext;
+ ReorderBufferChange *change;
+
+ Assert(TransactionIdIsValid(xid));
+
+ oldcontext = MemoryContextSwitchTo(rb->context);
+
+ change = ReorderBufferGetChange(rb);
+ change->action = REORDER_BUFFER_CHANGE_DDL;
+ change->data.ddl.prefix = pstrdup(prefix);
+ change->data.ddl.relid = relid;
+ change->data.ddl.cmdtype = cmdtype;
+ change->data.ddl.message_size = message_size;
+ change->data.ddl.message = palloc(message_size);
+ memcpy(change->data.ddl.message, message, message_size);
+
+ ReorderBufferQueueChange(rb, xid, lsn, change, false);
+
+ MemoryContextSwitchTo(oldcontext);
+}
+
/*
* AssertTXNLsnOrder
* Verify LSN ordering of transaction lists in the reorderbuffer
@@ -1997,6 +2036,29 @@ ReorderBufferApplyMessage(ReorderBuffer *rb, ReorderBufferTXN *txn,
change->data.msg.message);
}
+/*
+ * Helper function for ReorderBufferProcessTXN for applying the DDL message.
+ */
+static inline void
+ReorderBufferApplyDDLMessage(ReorderBuffer *rb, ReorderBufferTXN *txn,
+ ReorderBufferChange *change, bool streaming)
+{
+ if (streaming)
+ rb->stream_ddl(rb, txn, change->lsn,
+ change->data.ddl.prefix,
+ change->data.ddl.relid,
+ change->data.ddl.cmdtype,
+ change->data.ddl.message_size,
+ change->data.ddl.message);
+ else
+ rb->ddl(rb, txn, change->lsn,
+ change->data.ddl.prefix,
+ change->data.ddl.relid,
+ change->data.ddl.cmdtype,
+ change->data.ddl.message_size,
+ change->data.ddl.message);
+}
+
/*
* Function to store the command id and snapshot at the end of the current
* stream so that we can reuse the same while sending the next stream.
@@ -2379,6 +2441,10 @@ ReorderBufferProcessTXN(ReorderBuffer *rb, ReorderBufferTXN *txn,
ReorderBufferApplyMessage(rb, txn, change, streaming);
break;
+ case REORDER_BUFFER_CHANGE_DDL:
+ ReorderBufferApplyDDLMessage(rb, txn, change, streaming);
+ break;
+
case REORDER_BUFFER_CHANGE_INVALIDATION:
/* Execute the invalidation messages locally */
ReorderBufferExecuteInvalidations(change->data.inval.ninvalidations,
@@ -3835,6 +3901,39 @@ ReorderBufferSerializeChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
change->data.msg.message_size);
data += change->data.msg.message_size;
+ break;
+ }
+ case REORDER_BUFFER_CHANGE_DDL:
+ {
+ char *data;
+ Size prefix_size = strlen(change->data.ddl.prefix) + 1;
+
+ sz += prefix_size + change->data.ddl.message_size +
+ sizeof(Size) + sizeof(Oid) + sizeof(int) + sizeof(Size);
+ ReorderBufferSerializeReserve(rb, sz);
+
+ data = ((char *) rb->outbuf) + sizeof(ReorderBufferDiskChange);
+
+ /* might have been reallocated above */
+ ondisk = (ReorderBufferDiskChange *) rb->outbuf;
+
+ /* write the prefix, relid and cmdtype including the size */
+ memcpy(data, &prefix_size, sizeof(Size));
+ data += sizeof(Size);
+ memcpy(data, &change->data.ddl.relid, sizeof(Oid));
+ data += sizeof(Oid);
+ memcpy(data, &change->data.ddl.cmdtype, sizeof(DeparsedCommandType));
+ data += sizeof(int);
+ memcpy(data, change->data.ddl.prefix, prefix_size);
+ data += prefix_size;
+
+ /* write the message including the size */
+ memcpy(data, &change->data.ddl.message_size, sizeof(Size));
+ data += sizeof(Size);
+ memcpy(data, change->data.ddl.message,
+ change->data.ddl.message_size);
+ data += change->data.ddl.message_size;
+
break;
}
case REORDER_BUFFER_CHANGE_INVALIDATION:
@@ -4149,6 +4248,15 @@ ReorderBufferChangeSize(ReorderBufferChange *change)
sz += prefix_size + change->data.msg.message_size +
sizeof(Size) + sizeof(Size);
+ break;
+ }
+ case REORDER_BUFFER_CHANGE_DDL:
+ {
+ Size prefix_size = strlen(change->data.ddl.prefix) + 1;
+
+ sz += prefix_size + change->data.ddl.message_size +
+ sizeof(Size) + sizeof(Size) + sizeof(Oid) + sizeof(DeparsedCommandType);
+
break;
}
case REORDER_BUFFER_CHANGE_INVALIDATION:
@@ -4426,6 +4534,33 @@ ReorderBufferRestoreChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
change->data.msg.message_size);
data += change->data.msg.message_size;
+ break;
+ }
+ case REORDER_BUFFER_CHANGE_DDL:
+ {
+ Size prefix_size;
+
+ /* read prefix */
+ memcpy(&prefix_size, data, sizeof(Size));
+ data += sizeof(Size);
+ memcpy(&change->data.ddl.relid, data, sizeof(Oid));
+ data += sizeof(Oid);
+ memcpy(&change->data.ddl.cmdtype, data, sizeof(DeparsedCommandType));
+ data += sizeof(int);
+ change->data.ddl.prefix = MemoryContextAlloc(rb->context, prefix_size);
+ memcpy(change->data.ddl.prefix, data, prefix_size);
+ Assert(change->data.ddl.prefix[prefix_size - 1] == '\0');
+ data += prefix_size;
+
+ /* read the message */
+ memcpy(&change->data.msg.message_size, data, sizeof(Size));
+ data += sizeof(Size);
+ change->data.msg.message = MemoryContextAlloc(rb->context,
+ change->data.msg.message_size);
+ memcpy(change->data.msg.message, data,
+ change->data.msg.message_size);
+ data += change->data.msg.message_size;
+
break;
}
case REORDER_BUFFER_CHANGE_INVALIDATION:
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 6fd674b5d6..1aef149204 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -166,6 +166,7 @@
#include "miscadmin.h"
#include "nodes/makefuncs.h"
#include "optimizer/optimizer.h"
+#include "parser/analyze.h"
#include "parser/parse_relation.h"
#include "pgstat.h"
#include "postmaster/bgworker.h"
@@ -191,7 +192,10 @@
#include "storage/lmgr.h"
#include "storage/proc.h"
#include "storage/procarray.h"
+#include "tcop/ddl_deparse.h"
+#include "tcop/pquery.h"
#include "tcop/tcopprot.h"
+#include "tcop/utility.h"
#include "utils/acl.h"
#include "utils/builtins.h"
#include "utils/catcache.h"
@@ -3228,6 +3232,238 @@ apply_handle_truncate(StringInfo s)
end_replication_step();
}
+/*
+ * Special handling for CREATE TABLE AS and SELECT INTO
+ * to not populate data from the source table on the subscriber.
+ * Allow the data to be replicated through INSERTs on the publisher.
+ */
+static void
+preprocess_create_table(RawStmt *command)
+{
+ CommandTag commandTag;
+
+ commandTag = CreateCommandTag((Node *) command);
+
+ switch (commandTag)
+ {
+ case CMDTAG_CREATE_TABLE_AS:
+ case CMDTAG_SELECT_INTO:
+ {
+ CreateTableAsStmt *castmt = (CreateTableAsStmt *) command->stmt;
+
+ if (castmt->objtype == OBJECT_TABLE)
+ {
+ /*
+ * Force skipping data population to avoid data
+ * inconsistency. Data should be replicated from the
+ * publisher instead.
+ */
+ castmt->into->skipData = true;
+ }
+ }
+ break;
+ case CMDTAG_SELECT:
+ {
+ SelectStmt *sstmt = (SelectStmt *) command->stmt;
+
+ if (sstmt->intoClause != NULL)
+ {
+ /*
+ * Force skipping data population to avoid data
+ * inconsistency. Data should be replicated from the
+ * publisher instead.
+ */
+ sstmt->intoClause->skipData = true;
+ }
+ }
+ break;
+ default:
+ break;
+ }
+}
+
+/*
+ * Handle CREATE TABLE command
+ *
+ * Call AddSubscriptionRelState for CREATE LABEL command to set the relstate to
+ * SUBREL_STATE_READY so DML changes on this new table can be replicated without
+ * having to manually run "alter subscription ... refresh publication"
+ */
+static void
+handle_create_table(RawStmt *command)
+{
+ CommandTag commandTag;
+ RangeVar *rv = NULL;
+ Oid relid;
+ Oid relnamespace = InvalidOid;
+ CreateStmt *cstmt;
+ char *schemaname = NULL;
+ char *relname = NULL;
+
+ commandTag = CreateCommandTag((Node *) command);
+ cstmt = (CreateStmt *) command->stmt;
+ rv = cstmt->relation;
+
+ if (commandTag == CMDTAG_CREATE_TABLE)
+ {
+ cstmt = (CreateStmt *) command->stmt;
+ rv = cstmt->relation;
+ }
+ else
+ {
+ return;
+ }
+
+ if (!rv)
+ return;
+
+ schemaname = rv->schemaname;
+ relname = rv->relname;
+
+ if (schemaname != NULL)
+ relnamespace = get_namespace_oid(schemaname, false);
+
+ if (relnamespace != InvalidOid)
+ relid = get_relname_relid(relname, relnamespace);
+ else
+ relid = RelnameGetRelid(relname);
+
+ if (OidIsValid(relid))
+ {
+ AddSubscriptionRelState(MySubscription->oid, relid,
+ SUBREL_STATE_READY,
+ InvalidXLogRecPtr);
+ ereport(DEBUG1,
+ (errmsg_internal("table \"%s\" added to subscription \"%s\"",
+ relname, MySubscription->name)));
+ }
+}
+
+/*
+ * Handle DDL replication messages.
+ */
+static void
+apply_handle_ddl(StringInfo s)
+{
+ XLogRecPtr lsn;
+ const char *prefix = NULL;
+ char *message = NULL;
+ char *ddl_command;
+ Size sz;
+ List *parsetree_list;
+ ListCell *parsetree_item;
+ DestReceiver *receiver;
+ MemoryContext oldcontext;
+ const char *save_debug_query_string = debug_query_string;
+
+ message = logicalrep_read_ddl(s, &lsn, &prefix, &sz);
+
+ /* Make sure we are in a transaction command */
+ begin_replication_step();
+
+ ddl_command = deparse_ddl_json_to_string(message);
+ debug_query_string = ddl_command;
+
+ /* DestNone for logical replication */
+ receiver = CreateDestReceiver(DestNone);
+ parsetree_list = pg_parse_query(ddl_command);
+
+ foreach(parsetree_item, parsetree_list)
+ {
+ List *plantree_list;
+ List *querytree_list;
+ RawStmt *command = (RawStmt *) lfirst(parsetree_item);
+ CommandTag commandTag;
+ MemoryContext per_parsetree_context = NULL;
+ Portal portal;
+ bool snapshot_set = false;
+
+ commandTag = CreateCommandTag((Node *) command);
+
+ /* If we got a cancel signal in parsing or prior command, quit */
+ CHECK_FOR_INTERRUPTS();
+
+ /* Remove data population from the command */
+ preprocess_create_table(command);
+
+ /*
+ * Set up a snapshot if parse analysis/planning will need one.
+ */
+ if (analyze_requires_snapshot(command))
+ {
+ PushActiveSnapshot(GetTransactionSnapshot());
+ snapshot_set = true;
+ }
+
+ /*
+ * We do the work for each parsetree in a short-lived context, to
+ * limit the memory used when there are many commands in the string.
+ */
+ per_parsetree_context =
+ AllocSetContextCreate(CurrentMemoryContext,
+ "execute_sql_string per-statement context",
+ ALLOCSET_DEFAULT_SIZES);
+ oldcontext = MemoryContextSwitchTo(per_parsetree_context);
+
+ querytree_list = pg_analyze_and_rewrite_fixedparams(command,
+ ddl_command,
+ NULL, 0, NULL);
+
+ plantree_list = pg_plan_queries(querytree_list, ddl_command, 0, NULL);
+
+ /* Done with the snapshot used for parsing/planning */
+ if (snapshot_set)
+ PopActiveSnapshot();
+
+ portal = CreatePortal("logical replication", true, true);
+
+ /*
+ * We don't have to copy anything into the portal, because everything
+ * we are passing here is in ApplyMessageContext or the
+ * per_parsetree_context, and so will outlive the portal anyway.
+ */
+ PortalDefineQuery(portal,
+ NULL,
+ ddl_command,
+ commandTag,
+ plantree_list,
+ NULL);
+
+ /*
+ * Start the portal. No parameters here.
+ */
+ PortalStart(portal, NULL, 0, InvalidSnapshot);
+
+ /*
+ * Switch back to transaction context for execution.
+ */
+ MemoryContextSwitchTo(oldcontext);
+
+ (void) PortalRun(portal,
+ FETCH_ALL,
+ true,
+ true,
+ receiver,
+ receiver,
+ NULL);
+
+ PortalDrop(portal, false);
+
+ CommandCounterIncrement();
+
+ /*
+ * Table created by DDL replication (database level) is automatically
+ * added to the subscription here.
+ */
+ handle_create_table(command);
+
+ /* Now we may drop the per-parsetree context, if one was created. */
+ MemoryContextDelete(per_parsetree_context);
+ }
+
+ debug_query_string = save_debug_query_string;
+ end_replication_step();
+}
/*
* Logical replication protocol message dispatcher.
@@ -3293,6 +3529,10 @@ apply_dispatch(StringInfo s)
*/
break;
+ case LOGICAL_REP_MSG_DDL:
+ apply_handle_ddl(s);
+ break;
+
case LOGICAL_REP_MSG_STREAM_START:
apply_handle_stream_start(s);
break;
diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c
index ebaf555d56..7ea9ff51ec 100644
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -18,6 +18,7 @@
#include "catalog/pg_publication_rel.h"
#include "catalog/pg_subscription.h"
#include "commands/defrem.h"
+#include "commands/publicationcmds.h"
#include "commands/subscriptioncmds.h"
#include "executor/executor.h"
#include "fmgr.h"
@@ -55,6 +56,11 @@ static void pgoutput_message(LogicalDecodingContext *ctx,
ReorderBufferTXN *txn, XLogRecPtr message_lsn,
bool transactional, const char *prefix,
Size sz, const char *message);
+static void pgoutput_ddl(LogicalDecodingContext *ctx,
+ ReorderBufferTXN *txn, XLogRecPtr message_lsn,
+ const char *prefix, Oid relid,
+ DeparsedCommandType cmdtype,
+ Size sz, const char *message);
static bool pgoutput_origin_filter(LogicalDecodingContext *ctx,
RepOriginId origin_id);
static void pgoutput_begin_prepare_txn(LogicalDecodingContext *ctx,
@@ -207,6 +213,7 @@ typedef struct RelationSyncEntry
typedef struct PGOutputTxnData
{
bool sent_begin_txn; /* flag indicating whether BEGIN has been sent */
+ List *deleted_relids;
} PGOutputTxnData;
/* Map used to remember which relation schemas we sent. */
@@ -254,6 +261,7 @@ _PG_output_plugin_init(OutputPluginCallbacks *cb)
cb->change_cb = pgoutput_change;
cb->truncate_cb = pgoutput_truncate;
cb->message_cb = pgoutput_message;
+ cb->ddl_cb = pgoutput_ddl;
cb->commit_cb = pgoutput_commit_txn;
cb->begin_prepare_cb = pgoutput_begin_prepare_txn;
@@ -270,6 +278,7 @@ _PG_output_plugin_init(OutputPluginCallbacks *cb)
cb->stream_commit_cb = pgoutput_stream_commit;
cb->stream_change_cb = pgoutput_change;
cb->stream_message_cb = pgoutput_message;
+ cb->stream_ddl_cb = pgoutput_ddl;
cb->stream_truncate_cb = pgoutput_truncate;
/* transaction streaming - two-phase commit */
cb->stream_prepare_cb = pgoutput_stream_prepare_txn;
@@ -425,6 +434,7 @@ pgoutput_startup(LogicalDecodingContext *ctx, OutputPluginOptions *opt,
/* This plugin uses binary protocol. */
opt->output_type = OUTPUT_PLUGIN_BINARY_OUTPUT;
+ opt->receive_rewrites = true;
/*
* This is replication start and not slot initialization.
@@ -505,6 +515,7 @@ pgoutput_startup(LogicalDecodingContext *ctx, OutputPluginOptions *opt,
/* Init publication state. */
data->publications = NIL;
+ data->deleted_relids = NIL;
publications_valid = false;
/*
@@ -533,6 +544,34 @@ pgoutput_startup(LogicalDecodingContext *ctx, OutputPluginOptions *opt,
}
}
+/* Initialize the per-transaction level variable for the given transaction. */
+static void
+init_txn_data(LogicalDecodingContext *ctx, ReorderBufferTXN *txn)
+{
+ PGOutputTxnData *txndata;
+
+ if (txn->output_plugin_private != NULL)
+ return;
+
+ txndata = MemoryContextAllocZero(ctx->context, sizeof(PGOutputTxnData));
+
+ txn->output_plugin_private = txndata;
+}
+
+/* Clean up the per-transaction level variable for the given transaction. */
+static void
+clean_txn_data(ReorderBufferTXN *txn)
+{
+ PGOutputTxnData *txndata = (PGOutputTxnData *) txn->output_plugin_private;
+
+ if (txndata == NULL)
+ return;
+
+ list_free(txndata->deleted_relids);
+ pfree(txndata);
+ txn->output_plugin_private = NULL;
+}
+
/*
* BEGIN callback.
*
@@ -546,10 +585,7 @@ pgoutput_startup(LogicalDecodingContext *ctx, OutputPluginOptions *opt,
static void
pgoutput_begin_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn)
{
- PGOutputTxnData *txndata = MemoryContextAllocZero(ctx->context,
- sizeof(PGOutputTxnData));
-
- txn->output_plugin_private = txndata;
+ init_txn_data(ctx, txn);
}
/*
@@ -594,8 +630,7 @@ pgoutput_commit_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
*/
sent_begin_txn = txndata->sent_begin_txn;
OutputPluginUpdateProgress(ctx, !sent_begin_txn);
- pfree(txndata);
- txn->output_plugin_private = NULL;
+ clean_txn_data(txn);
if (!sent_begin_txn)
{
@@ -637,6 +672,8 @@ pgoutput_prepare_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
OutputPluginPrepareWrite(ctx, true);
logicalrep_write_prepare(ctx->out, txn, prepare_lsn);
OutputPluginWrite(ctx, true);
+
+ clean_txn_data(txn);
}
/*
@@ -1411,6 +1448,19 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
ReorderBufferChangeType action = change->action;
TupleTableSlot *old_slot = NULL;
TupleTableSlot *new_slot = NULL;
+ bool table_rewrite = false;
+
+ /*
+ * For heap rewrites, we might need to replicate them if the rewritten
+ * table publishes rewrite ddl message. So get the actual relation here
+ * and check the pubaction later.
+ */
+ if (relation->rd_rel->relrewrite)
+ {
+ table_rewrite = true;
+ relation = RelationIdGetRelation(relation->rd_rel->relrewrite);
+ targetrel = relation;
+ }
if (!is_publishable_relation(relation))
return;
@@ -1456,6 +1506,13 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
Assert(false);
}
+ /*
+ * We don't publish table rewrite change unless we publish the rewrite ddl
+ * message.
+ */
+ if (table_rewrite && !relentry->pubactions.pubddl_table)
+ return;
+
/* Avoid leaking memory by using and resetting our own context */
old = MemoryContextSwitchTo(data->context);
@@ -1503,7 +1560,8 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
* Updates could be transformed to inserts or deletes based on the results
* of the row filter for old and new tuple.
*/
- if (!pgoutput_row_filter(targetrel, old_slot, &new_slot, relentry, &action))
+ if (!table_rewrite &&
+ !pgoutput_row_filter(targetrel, old_slot, &new_slot, relentry, &action))
goto cleanup;
/*
@@ -1528,8 +1586,18 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
switch (action)
{
case REORDER_BUFFER_CHANGE_INSERT:
- logicalrep_write_insert(ctx->out, xid, targetrel, new_slot,
- data->binary, relentry->columns);
+ /*
+ * Convert the rewrite inserts to updates so that the subscriber
+ * can replay it. This is needed to make sure the data between
+ * publisher and subscriber is consistent.
+ */
+ if (table_rewrite)
+ logicalrep_write_update(ctx->out, xid, targetrel,
+ NULL, new_slot, data->binary,
+ relentry->columns);
+ else
+ logicalrep_write_insert(ctx->out, xid, targetrel, new_slot,
+ data->binary, relentry->columns);
break;
case REORDER_BUFFER_CHANGE_UPDATE:
logicalrep_write_update(ctx->out, xid, targetrel, old_slot,
@@ -1552,6 +1620,9 @@ cleanup:
ancestor = NULL;
}
+ if (table_rewrite)
+ RelationClose(relation);
+
MemoryContextSwitchTo(old);
MemoryContextReset(data->context);
}
@@ -1665,6 +1736,140 @@ pgoutput_message(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
OutputPluginWrite(ctx, true);
}
+/* Check if the given object is published. */
+static bool
+is_object_published(LogicalDecodingContext *ctx, Oid objid)
+{
+ Relation relation = NULL;
+ RelationSyncEntry *relentry;
+ PGOutputData *data = (PGOutputData *) ctx->output_plugin_private;
+
+ /* First check the DDL command filter. */
+ switch (get_rel_relkind(objid))
+ {
+ case RELKIND_RELATION:
+ relation = RelationIdGetRelation(objid);
+ relentry = get_rel_sync_entry(data, relation);
+ RelationClose(relation);
+
+ /*
+ * Skip sending this ddl if we don't publish ddl message or the ddl
+ * need to be published via its root relation.
+ */
+ if (!relentry->pubactions.pubddl_table ||
+ relentry->publish_as_relid != objid)
+ return false;
+
+ break;
+ default:
+ /* unsupported objects */
+ return false;
+ }
+
+ return true;
+}
+
+/*
+ * Send the decoded DDL over wire.
+ */
+static void
+pgoutput_ddl(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
+ XLogRecPtr message_lsn,
+ const char *prefix, Oid relid, DeparsedCommandType cmdtype,
+ Size sz, const char *message)
+{
+ PGOutputTxnData *txndata = (PGOutputTxnData *) txn->output_plugin_private;
+
+ /*
+ * Check if the given object is published. Note that for dropped objects,
+ * we cannot get the required information from the catalog, so we skip the
+ * check for them.
+ */
+ if (cmdtype != DCT_TableDropEnd && !is_object_published(ctx, relid))
+ return;
+
+ switch (cmdtype)
+ {
+ case DCT_TableDropStart:
+ {
+ MemoryContext old;
+
+ init_txn_data(ctx, txn);
+
+ txndata = (PGOutputTxnData *) txn->output_plugin_private;
+
+ /*
+ * On DROP start, add the relid to a deleted_relid list if the
+ * relid is part of a publication that supports ddl
+ * publication. We need this because on DROP end, the relid
+ * will no longer be valid. Later on Drop end, verify that the
+ * drop is for a relid that is on the deleted_rid list, and
+ * only then send the ddl message.
+ */
+ old = MemoryContextSwitchTo(ctx->context);
+ txndata->deleted_relids = lappend_oid(txndata->deleted_relids,
+ relid);
+ MemoryContextSwitchTo(old);
+ }
+ return;
+
+ case DCT_TableDropEnd:
+ if (!list_member_oid(txndata->deleted_relids, relid))
+ return;
+
+ txndata->deleted_relids = list_delete_oid(txndata->deleted_relids,
+ relid);
+ break;
+
+ case DCT_TableAlter:
+
+ /*
+ * For table rewrite ddl, we first send the original ddl message
+ * to subscriber, then convert the upcoming rewrite INSERT to
+ * UPDATE and send them to subscriber so that the data between
+ * publisher and subscriber can always be consistent.
+ *
+ * We do this way because of two reason:
+ *
+ * (1) The data before the rewrite ddl could already be different
+ * among publisher and subscriber. To make sure the extra data in
+ * subscriber which doesn't exist in publisher also get rewritten,
+ * we need to let the subscriber execute the original rewrite ddl
+ * to rewrite all the data at first.
+ *
+ * (2) the data after executing rewrite ddl could be different
+ * among publisher and subscriber(due to different
+ * functions/operators used during rewrite), so we need to
+ * replicate the rewrite UPDATEs to keep the data consistent.
+ *
+ * TO IMPROVE: We could improve this by letting the subscriber
+ * only rewrite the extra data instead of doing fully rewrite and
+ * use the upcoming rewrite UPDATEs to rewrite the rest data.
+ * Besides, we may not need to send rewrite changes for all type
+ * of rewrite ddl, for example, it seems fine to skip sending
+ * rewrite changes for ALTER TABLE SET LOGGED as the data in the
+ * table doesn't actually be changed.
+ */
+ break;
+
+ case DCT_SimpleCmd:
+ /* do nothing */
+ break;
+
+ default:
+ elog(ERROR, "unsupported type %d", cmdtype);
+ break;
+ }
+
+ /* Send BEGIN if we haven't yet */
+ if (txndata && !txndata->sent_begin_txn)
+ pgoutput_send_begin(ctx, txn);
+
+ OutputPluginPrepareWrite(ctx, true);
+ logicalrep_write_ddl(ctx->out, message_lsn, prefix, sz, message);
+ OutputPluginWrite(ctx, true);
+}
+
/*
* Return true if the data is associated with an origin and the user has
* requested the changes that don't have an origin, false otherwise.
@@ -1813,6 +2018,7 @@ pgoutput_stream_abort(struct LogicalDecodingContext *ctx,
OutputPluginWrite(ctx, true);
cleanup_rel_sync_cache(toptxn->xid, false);
+ clean_txn_data(txn);
}
/*
@@ -1838,6 +2044,7 @@ pgoutput_stream_commit(struct LogicalDecodingContext *ctx,
OutputPluginWrite(ctx, true);
cleanup_rel_sync_cache(txn->xid, true);
+ clean_txn_data(txn);
}
/*
@@ -1856,6 +2063,8 @@ pgoutput_stream_prepare_txn(LogicalDecodingContext *ctx,
OutputPluginPrepareWrite(ctx, true);
logicalrep_write_stream_prepare(ctx->out, txn, prepare_lsn);
OutputPluginWrite(ctx, true);
+
+ clean_txn_data(txn);
}
/*
@@ -1941,6 +2150,27 @@ set_schema_sent_in_streamed_txn(RelationSyncEntry *entry, TransactionId xid)
MemoryContextSwitchTo(oldctx);
}
+/* Reload publications if needed. */
+static void
+reload_publications(PGOutputData *data)
+{
+ MemoryContext oldctx;
+
+ if (!publications_valid)
+ {
+ oldctx = MemoryContextSwitchTo(CacheMemoryContext);
+ if (data->publications)
+ {
+ list_free_deep(data->publications);
+ data->publications = NIL;
+ }
+ data->publications = LoadPublications(data->publication_names);
+ MemoryContextSwitchTo(oldctx);
+ publications_valid = true;
+ }
+}
+
+
/*
* Find or create entry in the relation schema cache.
*
@@ -1955,7 +2185,6 @@ get_rel_sync_entry(PGOutputData *data, Relation relation)
{
RelationSyncEntry *entry;
bool found;
- MemoryContext oldctx;
Oid relid = RelationGetRelid(relation);
Assert(RelationSyncCache != NULL);
@@ -1973,7 +2202,8 @@ get_rel_sync_entry(PGOutputData *data, Relation relation)
entry->schema_sent = false;
entry->streamed_txns = NIL;
entry->pubactions.pubinsert = entry->pubactions.pubupdate =
- entry->pubactions.pubdelete = entry->pubactions.pubtruncate = false;
+ entry->pubactions.pubdelete = entry->pubactions.pubtruncate =
+ entry->pubactions.pubddl_table = false;
entry->new_slot = NULL;
entry->old_slot = NULL;
memset(entry->exprstate, 0, sizeof(entry->exprstate));
@@ -2002,19 +2232,7 @@ get_rel_sync_entry(PGOutputData *data, Relation relation)
char relkind = get_rel_relkind(relid);
List *rel_publications = NIL;
- /* Reload publications if needed before use. */
- if (!publications_valid)
- {
- oldctx = MemoryContextSwitchTo(CacheMemoryContext);
- if (data->publications)
- {
- list_free_deep(data->publications);
- data->publications = NIL;
- }
- data->publications = LoadPublications(data->publication_names);
- MemoryContextSwitchTo(oldctx);
- publications_valid = true;
- }
+ reload_publications(data);
/*
* Reset schema_sent status as the relation definition may have
@@ -2031,6 +2249,7 @@ get_rel_sync_entry(PGOutputData *data, Relation relation)
entry->pubactions.pubupdate = false;
entry->pubactions.pubdelete = false;
entry->pubactions.pubtruncate = false;
+ entry->pubactions.pubddl_table = false;
/*
* Tuple slots cleanups. (Will be rebuilt later if needed).
@@ -2144,6 +2363,7 @@ get_rel_sync_entry(PGOutputData *data, Relation relation)
entry->pubactions.pubupdate |= pub->pubactions.pubupdate;
entry->pubactions.pubdelete |= pub->pubactions.pubdelete;
entry->pubactions.pubtruncate |= pub->pubactions.pubtruncate;
+ entry->pubactions.pubddl_table |= pub->pubactions.pubddl_table;
/*
* We want to publish the changes as the top-most ancestor
diff --git a/src/backend/tcop/cmdtag.c b/src/backend/tcop/cmdtag.c
index 4bd713a0b4..ce09c5f233 100644
--- a/src/backend/tcop/cmdtag.c
+++ b/src/backend/tcop/cmdtag.c
@@ -26,10 +26,11 @@ typedef struct CommandTagBehavior
const bool table_rewrite_ok;
const bool display_rowcount; /* should the number of rows affected be
* shown in the command completion string */
+ const bool ddl_replication_ok;
} CommandTagBehavior;
-#define PG_CMDTAG(tag, name, evtrgok, rwrok, rowcnt) \
- { name, (uint8) (sizeof(name) - 1), evtrgok, rwrok, rowcnt },
+#define PG_CMDTAG(tag, name, evtrgok, rwrok, rowcnt, ddlreplok) \
+ { name, (uint8) (sizeof(name) - 1), evtrgok, rwrok, rowcnt, ddlreplok },
static const CommandTagBehavior tag_behavior[COMMAND_TAG_NEXTTAG] = {
#include "tcop/cmdtaglist.h"
@@ -57,6 +58,21 @@ GetCommandTagNameAndLen(CommandTag commandTag, Size *len)
return tag_behavior[commandTag].name;
}
+CommandTag *
+GetCommandTagsForDDLRepl(int *ncommands)
+{
+ CommandTag *ddlrepl_commands = palloc0(COMMAND_TAG_NEXTTAG * sizeof(CommandTag));
+ *ncommands = 0;
+
+ for(int i = 0; i < COMMAND_TAG_NEXTTAG; i++)
+ {
+ if (tag_behavior[i].ddl_replication_ok)
+ ddlrepl_commands[(*ncommands)++] = (CommandTag) i;
+ }
+
+ return ddlrepl_commands;
+}
+
bool
command_tag_display_rowcount(CommandTag commandTag)
{
@@ -75,6 +91,12 @@ command_tag_table_rewrite_ok(CommandTag commandTag)
return tag_behavior[commandTag].table_rewrite_ok;
}
+bool
+command_tag_ddl_replication_ok(CommandTag commandTag)
+{
+ return tag_behavior[commandTag].ddl_replication_ok;
+}
+
/*
* Search CommandTag by name
*
diff --git a/src/backend/tcop/utility.c b/src/backend/tcop/utility.c
index 4501cc337c..578a229ffc 100644
--- a/src/backend/tcop/utility.c
+++ b/src/backend/tcop/utility.c
@@ -2264,6 +2264,8 @@ stringify_objtype(ObjectType objtype, bool isgrant)
return "POLICY";
case OBJECT_PROCEDURE:
return "PROCEDURE";
+ case OBJECT_PUBLICATION:
+ return "PUBLICATION";
case OBJECT_ROLE:
return "ROLE";
case OBJECT_ROUTINE:
@@ -2276,6 +2278,8 @@ stringify_objtype(ObjectType objtype, bool isgrant)
return "SEQUENCE";
case OBJECT_STATISTIC_EXT:
return "STATISTICS";
+ case OBJECT_SUBSCRIPTION:
+ return "SUBSCRIPTION";
case OBJECT_TABLE:
return "TABLE";
case OBJECT_TABLESPACE:
@@ -2304,10 +2308,8 @@ stringify_objtype(ObjectType objtype, bool isgrant)
case OBJECT_DEFACL:
case OBJECT_DOMCONSTRAINT:
case OBJECT_PARAMETER_ACL:
- case OBJECT_PUBLICATION:
case OBJECT_PUBLICATION_NAMESPACE:
case OBJECT_PUBLICATION_REL:
- case OBJECT_SUBSCRIPTION:
case OBJECT_TABCONSTRAINT:
case OBJECT_TRANSFORM:
elog(ERROR, "unsupported object type %d", objtype);
diff --git a/src/backend/utils/cache/relcache.c b/src/backend/utils/cache/relcache.c
index 40140de958..9198da2eb1 100644
--- a/src/backend/utils/cache/relcache.c
+++ b/src/backend/utils/cache/relcache.c
@@ -5720,6 +5720,7 @@ RelationBuildPublicationDesc(Relation relation, PublicationDesc *pubdesc)
pubdesc->pubactions.pubupdate |= pubform->pubupdate;
pubdesc->pubactions.pubdelete |= pubform->pubdelete;
pubdesc->pubactions.pubtruncate |= pubform->pubtruncate;
+ pubdesc->pubactions.pubddl_table |= pubform->pubddl_table;
/*
* Check if all columns referenced in the filter expression are part
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index 6abbcff683..526987418c 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -51,6 +51,7 @@
#include "catalog/pg_largeobject_d.h"
#include "catalog/pg_largeobject_metadata_d.h"
#include "catalog/pg_proc_d.h"
+#include "catalog/pg_publication.h"
#include "catalog/pg_subscription.h"
#include "catalog/pg_trigger_d.h"
#include "catalog/pg_type_d.h"
@@ -4063,6 +4064,7 @@ getPublications(Archive *fout, int *numPublications)
int i_pubupdate;
int i_pubdelete;
int i_pubtruncate;
+ int i_pubddl_table;
int i_pubviaroot;
int i,
ntups;
@@ -4078,23 +4080,29 @@ getPublications(Archive *fout, int *numPublications)
resetPQExpBuffer(query);
/* Get the publications. */
- if (fout->remoteVersion >= 130000)
+ if (fout->remoteVersion >= 160000)
+ appendPQExpBufferStr(query,
+ "SELECT p.tableoid, p.oid, p.pubname, "
+ "p.pubowner, "
+ "p.puballtables, p.pubinsert, p.pubupdate, p.pubdelete, p.pubtruncate, p.pubddl_table, p.pubviaroot "
+ "FROM pg_publication p");
+ else if (fout->remoteVersion >= 130000)
appendPQExpBufferStr(query,
"SELECT p.tableoid, p.oid, p.pubname, "
"p.pubowner, "
- "p.puballtables, p.pubinsert, p.pubupdate, p.pubdelete, p.pubtruncate, p.pubviaroot "
+ "p.puballtables, p.pubinsert, p.pubupdate, p.pubdelete, p.pubtruncate, false as p.pubddl_table, p.pubviaroot "
"FROM pg_publication p");
else if (fout->remoteVersion >= 110000)
appendPQExpBufferStr(query,
"SELECT p.tableoid, p.oid, p.pubname, "
"p.pubowner, "
- "p.puballtables, p.pubinsert, p.pubupdate, p.pubdelete, p.pubtruncate, false AS pubviaroot "
+ "p.puballtables, p.pubinsert, p.pubupdate, p.pubdelete, p.pubtruncate, false as p.pubddl_table, false AS pubviaroot "
"FROM pg_publication p");
else
appendPQExpBufferStr(query,
"SELECT p.tableoid, p.oid, p.pubname, "
"p.pubowner, "
- "p.puballtables, p.pubinsert, p.pubupdate, p.pubdelete, false AS pubtruncate, false AS pubviaroot "
+ "p.puballtables, p.pubinsert, p.pubupdate, p.pubdelete, false AS pubtruncate, false as p.pubddl_table, false AS pubviaroot "
"FROM pg_publication p");
res = ExecuteSqlQuery(fout, query->data, PGRES_TUPLES_OK);
@@ -4110,6 +4118,7 @@ getPublications(Archive *fout, int *numPublications)
i_pubupdate = PQfnumber(res, "pubupdate");
i_pubdelete = PQfnumber(res, "pubdelete");
i_pubtruncate = PQfnumber(res, "pubtruncate");
+ i_pubddl_table = PQfnumber(res, "pubddl_table");
i_pubviaroot = PQfnumber(res, "pubviaroot");
pubinfo = pg_malloc(ntups * sizeof(PublicationInfo));
@@ -4133,6 +4142,8 @@ getPublications(Archive *fout, int *numPublications)
(strcmp(PQgetvalue(res, i, i_pubdelete), "t") == 0);
pubinfo[i].pubtruncate =
(strcmp(PQgetvalue(res, i, i_pubtruncate), "t") == 0);
+ pubinfo[i].pubddl_table =
+ (strcmp(PQgetvalue(res, i, i_pubddl_table), "t") == 0);
pubinfo[i].pubviaroot =
(strcmp(PQgetvalue(res, i, i_pubviaroot), "t") == 0);
@@ -4212,7 +4223,10 @@ dumpPublication(Archive *fout, const PublicationInfo *pubinfo)
first = false;
}
- appendPQExpBufferChar(query, '\'');
+ appendPQExpBufferStr(query, "'");
+
+ if (pubinfo->pubddl_table)
+ appendPQExpBufferStr(query, ", ddl = 'table'");
if (pubinfo->pubviaroot)
appendPQExpBufferStr(query, ", publish_via_partition_root = true");
@@ -7976,6 +7990,50 @@ getTriggers(Archive *fout, TableInfo tblinfo[], int numTables)
destroyPQExpBuffer(tbloids);
}
+/*
+ * getPublicationEventTriggers
+ * get the publication event triggers that should be skipped
+ */
+static void
+getPublicationEventTriggers(Archive *fout, SimpleStringList *skipTriggers)
+{
+ PQExpBuffer query;
+ PGresult *res;
+ int i;
+ int ntups;
+
+ query = createPQExpBuffer();
+
+ appendPQExpBufferStr(query,
+ "SELECT oid FROM pg_publication "
+ "WHERE pubddl_table");
+
+ res = ExecuteSqlQuery(fout, query->data, PGRES_TUPLES_OK);
+ ntups = PQntuples(res);
+
+ for (i = 0; i < ntups; i++)
+ {
+ char *trigname;
+ Oid pubid = atooid(PQgetvalue(res, 0, 0));
+
+ trigname = psprintf(PUB_EVENT_TRIG_PREFIX, PUB_TRIG_EVENT1, pubid);
+ simple_string_list_append(skipTriggers, trigname);
+
+ trigname = psprintf(PUB_EVENT_TRIG_PREFIX, PUB_TRIG_EVENT2, pubid);
+ simple_string_list_append(skipTriggers, trigname);
+
+ trigname = psprintf(PUB_EVENT_TRIG_PREFIX, PUB_TRIG_EVENT3, pubid);
+ simple_string_list_append(skipTriggers, trigname);
+
+ trigname = psprintf(PUB_EVENT_TRIG_PREFIX, PUB_TRIG_EVENT4, pubid);
+ simple_string_list_append(skipTriggers, trigname);
+ }
+
+ PQclear(res);
+
+ destroyPQExpBuffer(query);
+}
+
/*
* getEventTriggers
* get information about event triggers
@@ -7996,6 +8054,7 @@ getEventTriggers(Archive *fout, int *numEventTriggers)
i_evtfname,
i_evtenabled;
int ntups;
+ SimpleStringList skipTriggers = {NULL, NULL};
/* Before 9.3, there are no event triggers */
if (fout->remoteVersion < 90300)
@@ -8004,6 +8063,8 @@ getEventTriggers(Archive *fout, int *numEventTriggers)
return NULL;
}
+ getPublicationEventTriggers(fout, &skipTriggers);
+
query = createPQExpBuffer();
appendPQExpBufferStr(query,
@@ -8048,9 +8109,14 @@ getEventTriggers(Archive *fout, int *numEventTriggers)
evtinfo[i].evtenabled = *(PQgetvalue(res, i, i_evtenabled));
/* Decide whether we want to dump it */
- selectDumpableObject(&(evtinfo[i].dobj), fout);
+ if (simple_string_list_member(&skipTriggers, evtinfo[i].evtname))
+ evtinfo[i].dobj.dump= DUMP_COMPONENT_NONE;
+ else
+ selectDumpableObject(&(evtinfo[i].dobj), fout);
}
+ simple_string_list_destroy(&skipTriggers);
+
PQclear(res);
destroyPQExpBuffer(query);
diff --git a/src/bin/pg_dump/pg_dump.h b/src/bin/pg_dump/pg_dump.h
index ed6ce41ad7..bd5f8fb669 100644
--- a/src/bin/pg_dump/pg_dump.h
+++ b/src/bin/pg_dump/pg_dump.h
@@ -620,6 +620,7 @@ typedef struct _PublicationInfo
bool pubupdate;
bool pubdelete;
bool pubtruncate;
+ bool pubddl_table;
bool pubviaroot;
} PublicationInfo;
diff --git a/src/bin/pg_dump/t/002_pg_dump.pl b/src/bin/pg_dump/t/002_pg_dump.pl
index 42215f82f7..317ab933b1 100644
--- a/src/bin/pg_dump/t/002_pg_dump.pl
+++ b/src/bin/pg_dump/t/002_pg_dump.pl
@@ -2804,7 +2804,7 @@ my %tests = (
create_order => 50,
create_sql => 'CREATE PUBLICATION pub2
FOR ALL TABLES
- WITH (publish = \'\');',
+ WITH (publish = \'\', ddl = \'\');',
regexp => qr/^
\QCREATE PUBLICATION pub2 FOR ALL TABLES WITH (publish = '');\E
/xm,
diff --git a/src/bin/pg_waldump/rmgrdesc.c b/src/bin/pg_waldump/rmgrdesc.c
index 6b8c17bb4c..daf1730252 100644
--- a/src/bin/pg_waldump/rmgrdesc.c
+++ b/src/bin/pg_waldump/rmgrdesc.c
@@ -26,6 +26,7 @@
#include "commands/dbcommands_xlog.h"
#include "commands/sequence.h"
#include "commands/tablespace.h"
+#include "replication/ddlmessage.h"
#include "replication/message.h"
#include "replication/origin.h"
#include "rmgrdesc.h"
diff --git a/src/bin/psql/describe.c b/src/bin/psql/describe.c
index 99e28f607e..a41500deed 100644
--- a/src/bin/psql/describe.c
+++ b/src/bin/psql/describe.c
@@ -6183,7 +6183,7 @@ listPublications(const char *pattern)
PQExpBufferData buf;
PGresult *res;
printQueryOpt myopt = pset.popt;
- static const bool translate_columns[] = {false, false, false, false, false, false, false, false};
+ static const bool translate_columns[] = {false, false, false, false, false, false, false, false, false, false};
if (pset.sversion < 100000)
{
@@ -6200,13 +6200,18 @@ listPublications(const char *pattern)
printfPQExpBuffer(&buf,
"SELECT pubname AS \"%s\",\n"
" pg_catalog.pg_get_userbyid(pubowner) AS \"%s\",\n"
- " puballtables AS \"%s\",\n"
- " pubinsert AS \"%s\",\n"
- " pubupdate AS \"%s\",\n"
- " pubdelete AS \"%s\"",
+ " puballtables AS \"%s\"",
gettext_noop("Name"),
gettext_noop("Owner"),
- gettext_noop("All tables"),
+ gettext_noop("All tables"));
+ if (pset.sversion >= 160000)
+ appendPQExpBuffer(&buf,
+ ",\n pubddl_table AS \"%s\"",
+ gettext_noop("Table DDLs"));
+ appendPQExpBuffer(&buf,
+ ",\n pubinsert AS \"%s\",\n"
+ " pubupdate AS \"%s\",\n"
+ " pubdelete AS \"%s\"",
gettext_noop("Inserts"),
gettext_noop("Updates"),
gettext_noop("Deletes"));
@@ -6218,7 +6223,6 @@ listPublications(const char *pattern)
appendPQExpBuffer(&buf,
",\n pubviaroot AS \"%s\"",
gettext_noop("Via root"));
-
appendPQExpBufferStr(&buf,
"\nFROM pg_catalog.pg_publication\n");
@@ -6308,6 +6312,7 @@ describePublications(const char *pattern)
PGresult *res;
bool has_pubtruncate;
bool has_pubviaroot;
+ bool has_pubddl;
PQExpBufferData title;
printTableContent cont;
@@ -6324,13 +6329,19 @@ describePublications(const char *pattern)
has_pubtruncate = (pset.sversion >= 110000);
has_pubviaroot = (pset.sversion >= 130000);
+ has_pubddl = (pset.sversion >= 160000);
initPQExpBuffer(&buf);
printfPQExpBuffer(&buf,
"SELECT oid, pubname,\n"
" pg_catalog.pg_get_userbyid(pubowner) AS owner,\n"
- " puballtables, pubinsert, pubupdate, pubdelete");
+ " puballtables");
+ if (has_pubddl)
+ appendPQExpBufferStr(&buf,
+ ", pubddl_table");
+ appendPQExpBufferStr(&buf,
+ ", pubinsert, pubupdate, pubdelete");
if (has_pubtruncate)
appendPQExpBufferStr(&buf,
", pubtruncate");
@@ -6388,6 +6399,8 @@ describePublications(const char *pattern)
ncols++;
if (has_pubviaroot)
ncols++;
+ if (has_pubddl)
+ ncols++;
initPQExpBuffer(&title);
printfPQExpBuffer(&title, _("Publication %s"), pubname);
@@ -6395,6 +6408,8 @@ describePublications(const char *pattern)
printTableAddHeader(&cont, gettext_noop("Owner"), true, align);
printTableAddHeader(&cont, gettext_noop("All tables"), true, align);
+ if (has_pubddl)
+ printTableAddHeader(&cont, gettext_noop("Table DDLs"), true, align);
printTableAddHeader(&cont, gettext_noop("Inserts"), true, align);
printTableAddHeader(&cont, gettext_noop("Updates"), true, align);
printTableAddHeader(&cont, gettext_noop("Deletes"), true, align);
@@ -6412,6 +6427,8 @@ describePublications(const char *pattern)
printTableAddCell(&cont, PQgetvalue(res, i, 7), false, false);
if (has_pubviaroot)
printTableAddCell(&cont, PQgetvalue(res, i, 8), false, false);
+ if (has_pubddl)
+ printTableAddCell(&cont, PQgetvalue(res, i, 9), false, false);
if (!puballtables)
{
diff --git a/src/include/access/rmgrlist.h b/src/include/access/rmgrlist.h
index 463bcb67c5..abcbe97593 100644
--- a/src/include/access/rmgrlist.h
+++ b/src/include/access/rmgrlist.h
@@ -47,3 +47,4 @@ PG_RMGR(RM_COMMIT_TS_ID, "CommitTs", commit_ts_redo, commit_ts_desc, commit_ts_i
PG_RMGR(RM_REPLORIGIN_ID, "ReplicationOrigin", replorigin_redo, replorigin_desc, replorigin_identify, NULL, NULL, NULL, NULL)
PG_RMGR(RM_GENERIC_ID, "Generic", generic_redo, generic_desc, generic_identify, NULL, NULL, generic_mask, NULL)
PG_RMGR(RM_LOGICALMSG_ID, "LogicalMessage", logicalmsg_redo, logicalmsg_desc, logicalmsg_identify, NULL, NULL, NULL, logicalmsg_decode)
+PG_RMGR(RM_LOGICALDDLMSG_ID, "LogicalDDLMessage", logicalddlmsg_redo, logicalddlmsg_desc, logicalddlmsg_identify, NULL, NULL, NULL, logicalddl_decode)
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index 742df47381..372414af0f 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -11156,6 +11156,14 @@
proname => 'pg_logical_emit_message', provolatile => 'v', proparallel => 'u',
prorettype => 'pg_lsn', proargtypes => 'bool text bytea',
prosrc => 'pg_logical_emit_message_bytea' },
+{ oid => '3813', descr => 'emit a textual logical decoding message',
+ proname => 'pg_logical_emit_ddl_message', provolatile => 'v', proparallel => 'u',
+ prorettype => 'pg_lsn', proargtypes => 'bool text text',
+ prosrc => 'pg_logical_emit_ddl_message_text' },
+{ oid => '3814', descr => 'emit a binary logical decoding message',
+ proname => 'pg_logical_emit_ddl_message', provolatile => 'v', proparallel => 'u',
+ prorettype => 'pg_lsn', proargtypes => 'text regclass int4 text',
+ prosrc => 'pg_logical_emit_ddl_message_bytea' },
# event triggers
{ oid => '3566', descr => 'list objects dropped by the current command',
@@ -12059,5 +12067,17 @@
{ oid => '4643', descr => 'expand JSON format DDL to a plain text DDL command',
proname => 'ddl_deparse_expand_command', prorettype => 'text',
proargtypes => 'text', prosrc => 'ddl_deparse_expand_command' },
+{ oid => '4644', descr => 'trigger for ddl command deparse end',
+ proname => 'publication_deparse_ddl_command_end', prorettype => 'event_trigger',
+ proargtypes => '', prosrc => 'publication_deparse_ddl_command_end' },
+{ oid => '4645', descr => 'trigger for ddl command deparse start',
+ proname => 'publication_deparse_ddl_command_start', prorettype => 'event_trigger',
+ proargtypes => '', prosrc => 'publication_deparse_ddl_command_start' },
+{ oid => '4646', descr => 'trigger for ddl command deparse table rewrite',
+ proname => 'publication_deparse_table_rewrite', prorettype => 'event_trigger',
+ proargtypes => '', prosrc => 'publication_deparse_table_rewrite' },
+{ oid => '4647', descr => 'trigger for ddl command deparse table init',
+ proname => 'publication_deparse_table_init_write', prorettype => 'event_trigger',
+ proargtypes => '', prosrc => 'publication_deparse_table_init_write' },
]
diff --git a/src/include/catalog/pg_publication.h b/src/include/catalog/pg_publication.h
index 6ecaa2a01e..74b28d7350 100644
--- a/src/include/catalog/pg_publication.h
+++ b/src/include/catalog/pg_publication.h
@@ -18,8 +18,17 @@
#define PG_PUBLICATION_H
#include "catalog/genbki.h"
-#include "catalog/objectaddress.h"
#include "catalog/pg_publication_d.h"
+#include "nodes/pg_list.h"
+
+/* Publication trigger events */
+#define PUB_TRIG_EVENT1 "ddl_command_end"
+#define PUB_TRIG_EVENT2 "ddl_command_start"
+#define PUB_TRIG_EVENT3 "table_rewrite"
+#define PUB_TRIG_EVENT4 "table_init_write"
+
+/* Publication event trigger prefix */
+#define PUB_EVENT_TRIG_PREFIX "pg_deparse_trig_%s_%u"
/* ----------------
* pg_publication definition. cpp turns this into
@@ -54,6 +63,9 @@ CATALOG(pg_publication,6104,PublicationRelationId)
/* true if partition changes are published using root schema */
bool pubviaroot;
+
+ /* true if table ddls are published */
+ bool pubddl_table;
} FormData_pg_publication;
/* ----------------
@@ -72,6 +84,7 @@ typedef struct PublicationActions
bool pubupdate;
bool pubdelete;
bool pubtruncate;
+ bool pubddl_table;
} PublicationActions;
typedef struct PublicationDesc
@@ -103,13 +116,6 @@ typedef struct Publication
PublicationActions pubactions;
} Publication;
-typedef struct PublicationRelInfo
-{
- Relation relation;
- Node *whereClause;
- List *columns;
-} PublicationRelInfo;
-
extern Publication *GetPublication(Oid pubid);
extern Publication *GetPublicationByName(const char *pubname, bool missing_ok);
extern List *GetRelationPublications(Oid relid);
@@ -145,14 +151,6 @@ extern List *GetPubPartitionOptionRelations(List *result,
extern Oid GetTopMostAncestorInPublication(Oid puboid, List *ancestors,
int *ancestor_level);
-extern bool is_publishable_relation(Relation rel);
extern bool is_schema_publication(Oid pubid);
-extern ObjectAddress publication_add_relation(Oid pubid, PublicationRelInfo *pri,
- bool if_not_exists);
-extern ObjectAddress publication_add_schema(Oid pubid, Oid schemaid,
- bool if_not_exists);
-
-extern Bitmapset *pub_collist_to_bitmapset(Bitmapset *columns, Datum pubcols,
- MemoryContext mcxt);
#endif /* PG_PUBLICATION_H */
diff --git a/src/include/commands/publicationcmds.h b/src/include/commands/publicationcmds.h
index 70d5e3680a..a3fca9e64c 100644
--- a/src/include/commands/publicationcmds.h
+++ b/src/include/commands/publicationcmds.h
@@ -22,6 +22,13 @@
/* Same as MAXNUMMESSAGES in sinvaladt.c */
#define MAX_RELCACHE_INVAL_MSGS 4096
+typedef struct PublicationRelInfo
+{
+ Relation relation;
+ Node *whereClause;
+ List *columns;
+} PublicationRelInfo;
+
extern ObjectAddress CreatePublication(ParseState *pstate, CreatePublicationStmt *stmt);
extern void AlterPublication(ParseState *pstate, AlterPublicationStmt *stmt);
extern void RemovePublicationById(Oid pubid);
@@ -35,5 +42,12 @@ extern bool pub_rf_contains_invalid_column(Oid pubid, Relation relation,
List *ancestors, bool pubviaroot);
extern bool pub_collist_contains_invalid_column(Oid pubid, Relation relation,
List *ancestors, bool pubviaroot);
+extern Bitmapset *pub_collist_to_bitmapset(Bitmapset *columns, Datum pubcols,
+ MemoryContext mcxt);
+extern ObjectAddress publication_add_relation(Oid pubid, PublicationRelInfo *pri,
+ bool if_not_exists);
+extern ObjectAddress publication_add_schema(Oid pubid, Oid schemaid,
+ bool if_not_exists);
+extern bool is_publishable_relation(Relation rel);
#endif /* PUBLICATIONCMDS_H */
diff --git a/src/include/replication/ddlmessage.h b/src/include/replication/ddlmessage.h
new file mode 100644
index 0000000000..77df6ea11a
--- /dev/null
+++ b/src/include/replication/ddlmessage.h
@@ -0,0 +1,60 @@
+/*-------------------------------------------------------------------------
+ * ddlmessage.h
+ * Exports from replication/logical/ddlmessage.c
+ *
+ * Copyright (c) 2022, PostgreSQL Global Development Group
+ *
+ * src/include/replication/ddlmessage.h
+ *-------------------------------------------------------------------------
+ */
+#ifndef PG_LOGICAL_DDL_MESSAGE_H
+#define PG_LOGICAL_DDL_MESSAGE_H
+
+#include "access/xlog.h"
+#include "access/xlogdefs.h"
+#include "access/xlogreader.h"
+#include "nodes/nodes.h"
+
+
+/*
+ * Support for keeping track of deparsed commands.
+ */
+typedef enum DeparsedCommandType
+{
+ DCT_SimpleCmd,
+ DCT_TableDropStart,
+ DCT_TableDropEnd,
+ DCT_TableAlter,
+ DCT_ObjectCreate,
+ DCT_ObjectDrop
+} DeparsedCommandType;
+
+/*
+ * Generic logical decoding DDL message wal record.
+ */
+typedef struct xl_logical_ddl_message
+{
+ Oid dbId; /* database Oid emitted from */
+ Size prefix_size; /* length of prefix including null terminator */
+ Oid relid; /* id of the table */
+ DeparsedCommandType cmdtype; /* type of sql command */
+ Size message_size; /* size of the message */
+
+ /*
+ * payload, including null-terminated prefix of length prefix_size
+ */
+ char message[FLEXIBLE_ARRAY_MEMBER];
+} xl_logical_ddl_message;
+
+#define SizeOfLogicalDDLMessage (offsetof(xl_logical_ddl_message, message))
+
+extern XLogRecPtr LogLogicalDDLMessage(const char *prefix, Oid relid, DeparsedCommandType cmdtype,
+ const char *ddl_message, size_t size);
+
+/* RMGR API*/
+#define XLOG_LOGICAL_DDL_MESSAGE 0x00
+void logicalddlmsg_redo(XLogReaderState *record);
+void logicalddlmsg_desc(StringInfo buf, XLogReaderState *record);
+const char *logicalddlmsg_identify(uint8 info);
+
+#endif
diff --git a/src/include/replication/decode.h b/src/include/replication/decode.h
index 14fa921ab4..c9ac708d32 100644
--- a/src/include/replication/decode.h
+++ b/src/include/replication/decode.h
@@ -27,6 +27,7 @@ extern void heap2_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
extern void xact_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
extern void standby_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
extern void logicalmsg_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
+extern void logicalddl_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf);
extern void LogicalDecodingProcessRecord(LogicalDecodingContext *ctx,
XLogReaderState *record);
diff --git a/src/include/replication/logicalproto.h b/src/include/replication/logicalproto.h
index 0ea2df5088..5fb3baaeb1 100644
--- a/src/include/replication/logicalproto.h
+++ b/src/include/replication/logicalproto.h
@@ -66,6 +66,7 @@ typedef enum LogicalRepMsgType
LOGICAL_REP_MSG_RELATION = 'R',
LOGICAL_REP_MSG_TYPE = 'Y',
LOGICAL_REP_MSG_MESSAGE = 'M',
+ LOGICAL_REP_MSG_DDL = 'L',
LOGICAL_REP_MSG_BEGIN_PREPARE = 'b',
LOGICAL_REP_MSG_PREPARE = 'P',
LOGICAL_REP_MSG_COMMIT_PREPARED = 'K',
@@ -246,6 +247,9 @@ extern List *logicalrep_read_truncate(StringInfo in,
bool *cascade, bool *restart_seqs);
extern void logicalrep_write_message(StringInfo out, TransactionId xid, XLogRecPtr lsn,
bool transactional, const char *prefix, Size sz, const char *message);
+extern void logicalrep_write_ddl(StringInfo out, XLogRecPtr lsn,
+ const char *prefix, Size sz, const char *message);
+extern char *logicalrep_read_ddl(StringInfo in, XLogRecPtr *lsn, const char **prefix, Size *sz);
extern void logicalrep_write_rel(StringInfo out, TransactionId xid,
Relation rel, Bitmapset *columns);
extern LogicalRepRelation *logicalrep_read_rel(StringInfo in);
diff --git a/src/include/replication/output_plugin.h b/src/include/replication/output_plugin.h
index 2d89d26586..44baf40720 100644
--- a/src/include/replication/output_plugin.h
+++ b/src/include/replication/output_plugin.h
@@ -90,6 +90,18 @@ typedef void (*LogicalDecodeMessageCB) (struct LogicalDecodingContext *ctx,
Size message_size,
const char *message);
+/*
+ * 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);
+
/*
* Filter changes by origin.
*/
@@ -201,6 +213,19 @@ typedef void (*LogicalDecodeStreamMessageCB) (struct LogicalDecodingContext *ctx
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);
+
/*
* Callback for streaming truncates from in-progress transactions.
*/
@@ -221,6 +246,7 @@ typedef struct OutputPluginCallbacks
LogicalDecodeTruncateCB truncate_cb;
LogicalDecodeCommitCB commit_cb;
LogicalDecodeMessageCB message_cb;
+ LogicalDecodeDDLMessageCB ddl_cb;
LogicalDecodeFilterByOriginCB filter_by_origin_cb;
LogicalDecodeShutdownCB shutdown_cb;
@@ -239,6 +265,7 @@ typedef struct OutputPluginCallbacks
LogicalDecodeStreamCommitCB stream_commit_cb;
LogicalDecodeStreamChangeCB stream_change_cb;
LogicalDecodeStreamMessageCB stream_message_cb;
+ LogicalDecodeStreamDDLMessageCB stream_ddl_cb;
LogicalDecodeStreamTruncateCB stream_truncate_cb;
} OutputPluginCallbacks;
diff --git a/src/include/replication/pgoutput.h b/src/include/replication/pgoutput.h
index b4a8015403..a2cf99b4e4 100644
--- a/src/include/replication/pgoutput.h
+++ b/src/include/replication/pgoutput.h
@@ -25,6 +25,7 @@ typedef struct PGOutputData
uint32 protocol_version;
List *publication_names;
List *publications;
+ List *deleted_relids;
bool binary;
char streaming;
bool messages;
diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h
index e37f5120eb..25a061ac70 100644
--- a/src/include/replication/reorderbuffer.h
+++ b/src/include/replication/reorderbuffer.h
@@ -11,6 +11,8 @@
#include "access/htup_details.h"
#include "lib/ilist.h"
+#include "nodes/nodes.h"
+#include "replication/ddlmessage.h"
#include "storage/sinval.h"
#include "utils/hsearch.h"
#include "utils/relcache.h"
@@ -65,6 +67,7 @@ typedef enum ReorderBufferChangeType
REORDER_BUFFER_CHANGE_INSERT,
REORDER_BUFFER_CHANGE_UPDATE,
REORDER_BUFFER_CHANGE_DELETE,
+ REORDER_BUFFER_CHANGE_DDL,
REORDER_BUFFER_CHANGE_MESSAGE,
REORDER_BUFFER_CHANGE_INVALIDATION,
REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT,
@@ -139,6 +142,16 @@ typedef struct ReorderBufferChange
char *message;
} msg;
+ /* DDL */
+ struct
+ {
+ char *prefix;
+ Size message_size;
+ char *message;
+ Oid relid;
+ DeparsedCommandType cmdtype;
+ } ddl;
+
/* New snapshot, set when action == *_INTERNAL_SNAPSHOT */
Snapshot snapshot;
@@ -470,6 +483,16 @@ typedef void (*ReorderBufferMessageCB) (ReorderBuffer *rb,
const char *prefix, Size sz,
const char *message);
+/* DDL message callback signature */
+typedef void (*ReorderBufferDDLMessageCB) (ReorderBuffer *rb,
+ ReorderBufferTXN *txn,
+ XLogRecPtr message_lsn,
+ const char *prefix,
+ Oid relid,
+ DeparsedCommandType cmdtype,
+ Size sz,
+ const char *message);
+
/* begin prepare callback signature */
typedef void (*ReorderBufferBeginPrepareCB) (ReorderBuffer *rb,
ReorderBufferTXN *txn);
@@ -536,6 +559,17 @@ typedef void (*ReorderBufferStreamMessageCB) (
const char *prefix, Size sz,
const char *message);
+/* stream DDL message callback signature */
+typedef void (*ReorderBufferStreamDDLMessageCB) (
+ ReorderBuffer *rb,
+ ReorderBufferTXN *txn,
+ XLogRecPtr message_lsn,
+ const char *prefix,
+ Oid relid,
+ DeparsedCommandType cmdtype,
+ Size sz,
+ const char *message);
+
/* stream truncate callback signature */
typedef void (*ReorderBufferStreamTruncateCB) (
ReorderBuffer *rb,
@@ -592,6 +626,7 @@ struct ReorderBuffer
ReorderBufferApplyTruncateCB apply_truncate;
ReorderBufferCommitCB commit;
ReorderBufferMessageCB message;
+ ReorderBufferDDLMessageCB ddl;
/*
* Callbacks to be called when streaming a transaction at prepare time.
@@ -611,6 +646,7 @@ struct ReorderBuffer
ReorderBufferStreamCommitCB stream_commit;
ReorderBufferStreamChangeCB stream_change;
ReorderBufferStreamMessageCB stream_message;
+ ReorderBufferStreamDDLMessageCB stream_ddl;
ReorderBufferStreamTruncateCB stream_truncate;
/*
@@ -696,6 +732,9 @@ extern void ReorderBufferQueueMessage(ReorderBuffer *rb, TransactionId xid,
Snapshot snap, XLogRecPtr lsn,
bool transactional, const char *prefix,
Size message_size, const char *message);
+extern void ReorderBufferQueueDDLMessage(ReorderBuffer *, TransactionId, XLogRecPtr lsn,
+ const char *prefix, Size message_size,
+ const char *message, Oid relid, DeparsedCommandType cmdtype);
extern void ReorderBufferCommit(ReorderBuffer *rb, TransactionId xid,
XLogRecPtr commit_lsn, XLogRecPtr end_lsn,
TimestampTz commit_time, RepOriginId origin_id, XLogRecPtr origin_lsn);
diff --git a/src/include/tcop/cmdtag.h b/src/include/tcop/cmdtag.h
index 1e7514dcff..076c27e642 100644
--- a/src/include/tcop/cmdtag.h
+++ b/src/include/tcop/cmdtag.h
@@ -16,7 +16,7 @@
/* buffer size required for command completion tags */
#define COMPLETION_TAG_BUFSIZE 64
-#define PG_CMDTAG(tag, name, evtrgok, rwrok, rowcnt) \
+#define PG_CMDTAG(tag, name, evtrgok, rwrok, rowcnt, ddlreplok) \
tag,
typedef enum CommandTag
@@ -53,9 +53,11 @@ CopyQueryCompletion(QueryCompletion *dst, const QueryCompletion *src)
extern void InitializeQueryCompletion(QueryCompletion *qc);
extern const char *GetCommandTagName(CommandTag commandTag);
extern const char *GetCommandTagNameAndLen(CommandTag commandTag, Size *len);
+extern CommandTag *GetCommandTagsForDDLRepl(int *ncommands);
extern bool command_tag_display_rowcount(CommandTag commandTag);
extern bool command_tag_event_trigger_ok(CommandTag commandTag);
extern bool command_tag_table_rewrite_ok(CommandTag commandTag);
+extern bool command_tag_ddl_replication_ok(CommandTag commandTag);
extern CommandTag GetCommandTagEnum(const char *commandname);
extern Size BuildQueryCompletionString(char *buff, const QueryCompletion *qc,
bool nameonly);
diff --git a/src/include/tcop/cmdtaglist.h b/src/include/tcop/cmdtaglist.h
index e738ac1c09..18b69e4bb3 100644
--- a/src/include/tcop/cmdtaglist.h
+++ b/src/include/tcop/cmdtaglist.h
@@ -23,196 +23,196 @@
* textual name, so that we can bsearch on it; see GetCommandTagEnum().
*/
-/* symbol name, textual name, event_trigger_ok, table_rewrite_ok, rowcount */
-PG_CMDTAG(CMDTAG_UNKNOWN, "???", false, false, false)
-PG_CMDTAG(CMDTAG_ALTER_ACCESS_METHOD, "ALTER ACCESS METHOD", true, false, false)
-PG_CMDTAG(CMDTAG_ALTER_AGGREGATE, "ALTER AGGREGATE", true, false, false)
-PG_CMDTAG(CMDTAG_ALTER_CAST, "ALTER CAST", true, false, false)
-PG_CMDTAG(CMDTAG_ALTER_COLLATION, "ALTER COLLATION", true, false, false)
-PG_CMDTAG(CMDTAG_ALTER_CONSTRAINT, "ALTER CONSTRAINT", true, false, false)
-PG_CMDTAG(CMDTAG_ALTER_CONVERSION, "ALTER CONVERSION", true, false, false)
-PG_CMDTAG(CMDTAG_ALTER_DATABASE, "ALTER DATABASE", false, false, false)
-PG_CMDTAG(CMDTAG_ALTER_DEFAULT_PRIVILEGES, "ALTER DEFAULT PRIVILEGES", true, false, false)
-PG_CMDTAG(CMDTAG_ALTER_DOMAIN, "ALTER DOMAIN", true, false, false)
-PG_CMDTAG(CMDTAG_ALTER_EVENT_TRIGGER, "ALTER EVENT TRIGGER", false, false, false)
-PG_CMDTAG(CMDTAG_ALTER_EXTENSION, "ALTER EXTENSION", true, false, false)
-PG_CMDTAG(CMDTAG_ALTER_FOREIGN_DATA_WRAPPER, "ALTER FOREIGN DATA WRAPPER", true, false, false)
-PG_CMDTAG(CMDTAG_ALTER_FOREIGN_TABLE, "ALTER FOREIGN TABLE", true, false, false)
-PG_CMDTAG(CMDTAG_ALTER_FUNCTION, "ALTER FUNCTION", true, false, false)
-PG_CMDTAG(CMDTAG_ALTER_INDEX, "ALTER INDEX", true, false, false)
-PG_CMDTAG(CMDTAG_ALTER_LANGUAGE, "ALTER LANGUAGE", true, false, false)
-PG_CMDTAG(CMDTAG_ALTER_LARGE_OBJECT, "ALTER LARGE OBJECT", true, false, false)
-PG_CMDTAG(CMDTAG_ALTER_MATERIALIZED_VIEW, "ALTER MATERIALIZED VIEW", true, true, false)
-PG_CMDTAG(CMDTAG_ALTER_OPERATOR, "ALTER OPERATOR", true, false, false)
-PG_CMDTAG(CMDTAG_ALTER_OPERATOR_CLASS, "ALTER OPERATOR CLASS", true, false, false)
-PG_CMDTAG(CMDTAG_ALTER_OPERATOR_FAMILY, "ALTER OPERATOR FAMILY", true, false, false)
-PG_CMDTAG(CMDTAG_ALTER_POLICY, "ALTER POLICY", true, false, false)
-PG_CMDTAG(CMDTAG_ALTER_PROCEDURE, "ALTER PROCEDURE", true, false, false)
-PG_CMDTAG(CMDTAG_ALTER_PUBLICATION, "ALTER PUBLICATION", true, false, false)
-PG_CMDTAG(CMDTAG_ALTER_ROLE, "ALTER ROLE", false, false, false)
-PG_CMDTAG(CMDTAG_ALTER_ROUTINE, "ALTER ROUTINE", true, false, false)
-PG_CMDTAG(CMDTAG_ALTER_RULE, "ALTER RULE", true, false, false)
-PG_CMDTAG(CMDTAG_ALTER_SCHEMA, "ALTER SCHEMA", true, false, false)
-PG_CMDTAG(CMDTAG_ALTER_SEQUENCE, "ALTER SEQUENCE", true, false, false)
-PG_CMDTAG(CMDTAG_ALTER_SERVER, "ALTER SERVER", true, false, false)
-PG_CMDTAG(CMDTAG_ALTER_STATISTICS, "ALTER STATISTICS", true, false, false)
-PG_CMDTAG(CMDTAG_ALTER_SUBSCRIPTION, "ALTER SUBSCRIPTION", true, false, false)
-PG_CMDTAG(CMDTAG_ALTER_SYSTEM, "ALTER SYSTEM", false, false, false)
-PG_CMDTAG(CMDTAG_ALTER_TABLE, "ALTER TABLE", true, true, false)
-PG_CMDTAG(CMDTAG_ALTER_TABLESPACE, "ALTER TABLESPACE", false, false, false)
-PG_CMDTAG(CMDTAG_ALTER_TEXT_SEARCH_CONFIGURATION, "ALTER TEXT SEARCH CONFIGURATION", true, false, false)
-PG_CMDTAG(CMDTAG_ALTER_TEXT_SEARCH_DICTIONARY, "ALTER TEXT SEARCH DICTIONARY", true, false, false)
-PG_CMDTAG(CMDTAG_ALTER_TEXT_SEARCH_PARSER, "ALTER TEXT SEARCH PARSER", true, false, false)
-PG_CMDTAG(CMDTAG_ALTER_TEXT_SEARCH_TEMPLATE, "ALTER TEXT SEARCH TEMPLATE", true, false, false)
-PG_CMDTAG(CMDTAG_ALTER_TRANSFORM, "ALTER TRANSFORM", true, false, false)
-PG_CMDTAG(CMDTAG_ALTER_TRIGGER, "ALTER TRIGGER", true, false, false)
-PG_CMDTAG(CMDTAG_ALTER_TYPE, "ALTER TYPE", true, true, false)
-PG_CMDTAG(CMDTAG_ALTER_USER_MAPPING, "ALTER USER MAPPING", true, false, false)
-PG_CMDTAG(CMDTAG_ALTER_VIEW, "ALTER VIEW", true, false, false)
-PG_CMDTAG(CMDTAG_ANALYZE, "ANALYZE", false, false, false)
-PG_CMDTAG(CMDTAG_BEGIN, "BEGIN", false, false, false)
-PG_CMDTAG(CMDTAG_CALL, "CALL", false, false, false)
-PG_CMDTAG(CMDTAG_CHECKPOINT, "CHECKPOINT", false, false, false)
-PG_CMDTAG(CMDTAG_CLOSE, "CLOSE", false, false, false)
-PG_CMDTAG(CMDTAG_CLOSE_CURSOR, "CLOSE CURSOR", false, false, false)
-PG_CMDTAG(CMDTAG_CLOSE_CURSOR_ALL, "CLOSE CURSOR ALL", false, false, false)
-PG_CMDTAG(CMDTAG_CLUSTER, "CLUSTER", false, false, false)
-PG_CMDTAG(CMDTAG_COMMENT, "COMMENT", true, false, false)
-PG_CMDTAG(CMDTAG_COMMIT, "COMMIT", false, false, false)
-PG_CMDTAG(CMDTAG_COMMIT_PREPARED, "COMMIT PREPARED", false, false, false)
-PG_CMDTAG(CMDTAG_COPY, "COPY", false, false, true)
-PG_CMDTAG(CMDTAG_COPY_FROM, "COPY FROM", false, false, false)
-PG_CMDTAG(CMDTAG_CREATE_ACCESS_METHOD, "CREATE ACCESS METHOD", true, false, false)
-PG_CMDTAG(CMDTAG_CREATE_AGGREGATE, "CREATE AGGREGATE", true, false, false)
-PG_CMDTAG(CMDTAG_CREATE_CAST, "CREATE CAST", true, false, false)
-PG_CMDTAG(CMDTAG_CREATE_COLLATION, "CREATE COLLATION", true, false, false)
-PG_CMDTAG(CMDTAG_CREATE_CONSTRAINT, "CREATE CONSTRAINT", true, false, false)
-PG_CMDTAG(CMDTAG_CREATE_CONVERSION, "CREATE CONVERSION", true, false, false)
-PG_CMDTAG(CMDTAG_CREATE_DATABASE, "CREATE DATABASE", false, false, false)
-PG_CMDTAG(CMDTAG_CREATE_DOMAIN, "CREATE DOMAIN", true, false, false)
-PG_CMDTAG(CMDTAG_CREATE_EVENT_TRIGGER, "CREATE EVENT TRIGGER", false, false, false)
-PG_CMDTAG(CMDTAG_CREATE_EXTENSION, "CREATE EXTENSION", true, false, false)
-PG_CMDTAG(CMDTAG_CREATE_FOREIGN_DATA_WRAPPER, "CREATE FOREIGN DATA WRAPPER", true, false, false)
-PG_CMDTAG(CMDTAG_CREATE_FOREIGN_TABLE, "CREATE FOREIGN TABLE", true, false, false)
-PG_CMDTAG(CMDTAG_CREATE_FUNCTION, "CREATE FUNCTION", true, false, false)
-PG_CMDTAG(CMDTAG_CREATE_INDEX, "CREATE INDEX", true, false, false)
-PG_CMDTAG(CMDTAG_CREATE_LANGUAGE, "CREATE LANGUAGE", true, false, false)
-PG_CMDTAG(CMDTAG_CREATE_MATERIALIZED_VIEW, "CREATE MATERIALIZED VIEW", true, false, false)
-PG_CMDTAG(CMDTAG_CREATE_OPERATOR, "CREATE OPERATOR", true, false, false)
-PG_CMDTAG(CMDTAG_CREATE_OPERATOR_CLASS, "CREATE OPERATOR CLASS", true, false, false)
-PG_CMDTAG(CMDTAG_CREATE_OPERATOR_FAMILY, "CREATE OPERATOR FAMILY", true, false, false)
-PG_CMDTAG(CMDTAG_CREATE_POLICY, "CREATE POLICY", true, false, false)
-PG_CMDTAG(CMDTAG_CREATE_PROCEDURE, "CREATE PROCEDURE", true, false, false)
-PG_CMDTAG(CMDTAG_CREATE_PUBLICATION, "CREATE PUBLICATION", true, false, false)
-PG_CMDTAG(CMDTAG_CREATE_ROLE, "CREATE ROLE", false, false, false)
-PG_CMDTAG(CMDTAG_CREATE_ROUTINE, "CREATE ROUTINE", true, false, false)
-PG_CMDTAG(CMDTAG_CREATE_RULE, "CREATE RULE", true, false, false)
-PG_CMDTAG(CMDTAG_CREATE_SCHEMA, "CREATE SCHEMA", true, false, false)
-PG_CMDTAG(CMDTAG_CREATE_SEQUENCE, "CREATE SEQUENCE", true, false, false)
-PG_CMDTAG(CMDTAG_CREATE_SERVER, "CREATE SERVER", true, false, false)
-PG_CMDTAG(CMDTAG_CREATE_STATISTICS, "CREATE STATISTICS", true, false, false)
-PG_CMDTAG(CMDTAG_CREATE_SUBSCRIPTION, "CREATE SUBSCRIPTION", true, false, false)
-PG_CMDTAG(CMDTAG_CREATE_TABLE, "CREATE TABLE", true, false, false)
-PG_CMDTAG(CMDTAG_CREATE_TABLE_AS, "CREATE TABLE AS", true, false, false)
-PG_CMDTAG(CMDTAG_CREATE_TABLESPACE, "CREATE TABLESPACE", false, false, false)
-PG_CMDTAG(CMDTAG_CREATE_TEXT_SEARCH_CONFIGURATION, "CREATE TEXT SEARCH CONFIGURATION", true, false, false)
-PG_CMDTAG(CMDTAG_CREATE_TEXT_SEARCH_DICTIONARY, "CREATE TEXT SEARCH DICTIONARY", true, false, false)
-PG_CMDTAG(CMDTAG_CREATE_TEXT_SEARCH_PARSER, "CREATE TEXT SEARCH PARSER", true, false, false)
-PG_CMDTAG(CMDTAG_CREATE_TEXT_SEARCH_TEMPLATE, "CREATE TEXT SEARCH TEMPLATE", true, false, false)
-PG_CMDTAG(CMDTAG_CREATE_TRANSFORM, "CREATE TRANSFORM", true, false, false)
-PG_CMDTAG(CMDTAG_CREATE_TRIGGER, "CREATE TRIGGER", true, false, false)
-PG_CMDTAG(CMDTAG_CREATE_TYPE, "CREATE TYPE", true, false, false)
-PG_CMDTAG(CMDTAG_CREATE_USER_MAPPING, "CREATE USER MAPPING", true, false, false)
-PG_CMDTAG(CMDTAG_CREATE_VIEW, "CREATE VIEW", true, false, false)
-PG_CMDTAG(CMDTAG_DEALLOCATE, "DEALLOCATE", false, false, false)
-PG_CMDTAG(CMDTAG_DEALLOCATE_ALL, "DEALLOCATE ALL", false, false, false)
-PG_CMDTAG(CMDTAG_DECLARE_CURSOR, "DECLARE CURSOR", false, false, false)
-PG_CMDTAG(CMDTAG_DELETE, "DELETE", false, false, true)
-PG_CMDTAG(CMDTAG_DISCARD, "DISCARD", false, false, false)
-PG_CMDTAG(CMDTAG_DISCARD_ALL, "DISCARD ALL", false, false, false)
-PG_CMDTAG(CMDTAG_DISCARD_PLANS, "DISCARD PLANS", false, false, false)
-PG_CMDTAG(CMDTAG_DISCARD_SEQUENCES, "DISCARD SEQUENCES", false, false, false)
-PG_CMDTAG(CMDTAG_DISCARD_TEMP, "DISCARD TEMP", false, false, false)
-PG_CMDTAG(CMDTAG_DO, "DO", false, false, false)
-PG_CMDTAG(CMDTAG_DROP_ACCESS_METHOD, "DROP ACCESS METHOD", true, false, false)
-PG_CMDTAG(CMDTAG_DROP_AGGREGATE, "DROP AGGREGATE", true, false, false)
-PG_CMDTAG(CMDTAG_DROP_CAST, "DROP CAST", true, false, false)
-PG_CMDTAG(CMDTAG_DROP_COLLATION, "DROP COLLATION", true, false, false)
-PG_CMDTAG(CMDTAG_DROP_CONSTRAINT, "DROP CONSTRAINT", true, false, false)
-PG_CMDTAG(CMDTAG_DROP_CONVERSION, "DROP CONVERSION", true, false, false)
-PG_CMDTAG(CMDTAG_DROP_DATABASE, "DROP DATABASE", false, false, false)
-PG_CMDTAG(CMDTAG_DROP_DOMAIN, "DROP DOMAIN", true, false, false)
-PG_CMDTAG(CMDTAG_DROP_EVENT_TRIGGER, "DROP EVENT TRIGGER", false, false, false)
-PG_CMDTAG(CMDTAG_DROP_EXTENSION, "DROP EXTENSION", true, false, false)
-PG_CMDTAG(CMDTAG_DROP_FOREIGN_DATA_WRAPPER, "DROP FOREIGN DATA WRAPPER", true, false, false)
-PG_CMDTAG(CMDTAG_DROP_FOREIGN_TABLE, "DROP FOREIGN TABLE", true, false, false)
-PG_CMDTAG(CMDTAG_DROP_FUNCTION, "DROP FUNCTION", true, false, false)
-PG_CMDTAG(CMDTAG_DROP_INDEX, "DROP INDEX", true, false, false)
-PG_CMDTAG(CMDTAG_DROP_LANGUAGE, "DROP LANGUAGE", true, false, false)
-PG_CMDTAG(CMDTAG_DROP_MATERIALIZED_VIEW, "DROP MATERIALIZED VIEW", true, false, false)
-PG_CMDTAG(CMDTAG_DROP_OPERATOR, "DROP OPERATOR", true, false, false)
-PG_CMDTAG(CMDTAG_DROP_OPERATOR_CLASS, "DROP OPERATOR CLASS", true, false, false)
-PG_CMDTAG(CMDTAG_DROP_OPERATOR_FAMILY, "DROP OPERATOR FAMILY", true, false, false)
-PG_CMDTAG(CMDTAG_DROP_OWNED, "DROP OWNED", true, false, false)
-PG_CMDTAG(CMDTAG_DROP_POLICY, "DROP POLICY", true, false, false)
-PG_CMDTAG(CMDTAG_DROP_PROCEDURE, "DROP PROCEDURE", true, false, false)
-PG_CMDTAG(CMDTAG_DROP_PUBLICATION, "DROP PUBLICATION", true, false, false)
-PG_CMDTAG(CMDTAG_DROP_ROLE, "DROP ROLE", false, false, false)
-PG_CMDTAG(CMDTAG_DROP_ROUTINE, "DROP ROUTINE", true, false, false)
-PG_CMDTAG(CMDTAG_DROP_RULE, "DROP RULE", true, false, false)
-PG_CMDTAG(CMDTAG_DROP_SCHEMA, "DROP SCHEMA", true, false, false)
-PG_CMDTAG(CMDTAG_DROP_SEQUENCE, "DROP SEQUENCE", true, false, false)
-PG_CMDTAG(CMDTAG_DROP_SERVER, "DROP SERVER", true, false, false)
-PG_CMDTAG(CMDTAG_DROP_STATISTICS, "DROP STATISTICS", true, false, false)
-PG_CMDTAG(CMDTAG_DROP_SUBSCRIPTION, "DROP SUBSCRIPTION", true, false, false)
-PG_CMDTAG(CMDTAG_DROP_TABLE, "DROP TABLE", true, false, false)
-PG_CMDTAG(CMDTAG_DROP_TABLESPACE, "DROP TABLESPACE", false, false, false)
-PG_CMDTAG(CMDTAG_DROP_TEXT_SEARCH_CONFIGURATION, "DROP TEXT SEARCH CONFIGURATION", true, false, false)
-PG_CMDTAG(CMDTAG_DROP_TEXT_SEARCH_DICTIONARY, "DROP TEXT SEARCH DICTIONARY", true, false, false)
-PG_CMDTAG(CMDTAG_DROP_TEXT_SEARCH_PARSER, "DROP TEXT SEARCH PARSER", true, false, false)
-PG_CMDTAG(CMDTAG_DROP_TEXT_SEARCH_TEMPLATE, "DROP TEXT SEARCH TEMPLATE", true, false, false)
-PG_CMDTAG(CMDTAG_DROP_TRANSFORM, "DROP TRANSFORM", true, false, false)
-PG_CMDTAG(CMDTAG_DROP_TRIGGER, "DROP TRIGGER", true, false, false)
-PG_CMDTAG(CMDTAG_DROP_TYPE, "DROP TYPE", true, false, false)
-PG_CMDTAG(CMDTAG_DROP_USER_MAPPING, "DROP USER MAPPING", true, false, false)
-PG_CMDTAG(CMDTAG_DROP_VIEW, "DROP VIEW", true, false, false)
-PG_CMDTAG(CMDTAG_EXECUTE, "EXECUTE", false, false, false)
-PG_CMDTAG(CMDTAG_EXPLAIN, "EXPLAIN", false, false, false)
-PG_CMDTAG(CMDTAG_FETCH, "FETCH", false, false, true)
-PG_CMDTAG(CMDTAG_GRANT, "GRANT", true, false, false)
-PG_CMDTAG(CMDTAG_GRANT_ROLE, "GRANT ROLE", false, false, false)
-PG_CMDTAG(CMDTAG_IMPORT_FOREIGN_SCHEMA, "IMPORT FOREIGN SCHEMA", true, false, false)
-PG_CMDTAG(CMDTAG_INSERT, "INSERT", false, false, true)
-PG_CMDTAG(CMDTAG_LISTEN, "LISTEN", false, false, false)
-PG_CMDTAG(CMDTAG_LOAD, "LOAD", false, false, false)
-PG_CMDTAG(CMDTAG_LOCK_TABLE, "LOCK TABLE", false, false, false)
-PG_CMDTAG(CMDTAG_MERGE, "MERGE", false, false, true)
-PG_CMDTAG(CMDTAG_MOVE, "MOVE", false, false, true)
-PG_CMDTAG(CMDTAG_NOTIFY, "NOTIFY", false, false, false)
-PG_CMDTAG(CMDTAG_PREPARE, "PREPARE", false, false, false)
-PG_CMDTAG(CMDTAG_PREPARE_TRANSACTION, "PREPARE TRANSACTION", false, false, false)
-PG_CMDTAG(CMDTAG_REASSIGN_OWNED, "REASSIGN OWNED", false, false, false)
-PG_CMDTAG(CMDTAG_REFRESH_MATERIALIZED_VIEW, "REFRESH MATERIALIZED VIEW", true, false, false)
-PG_CMDTAG(CMDTAG_REINDEX, "REINDEX", false, false, false)
-PG_CMDTAG(CMDTAG_RELEASE, "RELEASE", false, false, false)
-PG_CMDTAG(CMDTAG_RESET, "RESET", false, false, false)
-PG_CMDTAG(CMDTAG_REVOKE, "REVOKE", true, false, false)
-PG_CMDTAG(CMDTAG_REVOKE_ROLE, "REVOKE ROLE", false, false, false)
-PG_CMDTAG(CMDTAG_ROLLBACK, "ROLLBACK", false, false, false)
-PG_CMDTAG(CMDTAG_ROLLBACK_PREPARED, "ROLLBACK PREPARED", false, false, false)
-PG_CMDTAG(CMDTAG_SAVEPOINT, "SAVEPOINT", false, false, false)
-PG_CMDTAG(CMDTAG_SECURITY_LABEL, "SECURITY LABEL", true, false, false)
-PG_CMDTAG(CMDTAG_SELECT, "SELECT", false, false, true)
-PG_CMDTAG(CMDTAG_SELECT_FOR_KEY_SHARE, "SELECT FOR KEY SHARE", false, false, false)
-PG_CMDTAG(CMDTAG_SELECT_FOR_NO_KEY_UPDATE, "SELECT FOR NO KEY UPDATE", false, false, false)
-PG_CMDTAG(CMDTAG_SELECT_FOR_SHARE, "SELECT FOR SHARE", false, false, false)
-PG_CMDTAG(CMDTAG_SELECT_FOR_UPDATE, "SELECT FOR UPDATE", false, false, false)
-PG_CMDTAG(CMDTAG_SELECT_INTO, "SELECT INTO", true, false, false)
-PG_CMDTAG(CMDTAG_SET, "SET", false, false, false)
-PG_CMDTAG(CMDTAG_SET_CONSTRAINTS, "SET CONSTRAINTS", false, false, false)
-PG_CMDTAG(CMDTAG_SHOW, "SHOW", false, false, false)
-PG_CMDTAG(CMDTAG_START_TRANSACTION, "START TRANSACTION", false, false, false)
-PG_CMDTAG(CMDTAG_TRUNCATE_TABLE, "TRUNCATE TABLE", false, false, false)
-PG_CMDTAG(CMDTAG_UNLISTEN, "UNLISTEN", false, false, false)
-PG_CMDTAG(CMDTAG_UPDATE, "UPDATE", false, false, true)
-PG_CMDTAG(CMDTAG_VACUUM, "VACUUM", false, false, false)
+/* symbol name, textual name, event_trigger_ok, table_rewrite_ok, rowcount, ddlreplok */
+PG_CMDTAG(CMDTAG_UNKNOWN, "???", false, false, false, false)
+PG_CMDTAG(CMDTAG_ALTER_ACCESS_METHOD, "ALTER ACCESS METHOD", true, false, false, false)
+PG_CMDTAG(CMDTAG_ALTER_AGGREGATE, "ALTER AGGREGATE", true, false, false, true)
+PG_CMDTAG(CMDTAG_ALTER_CAST, "ALTER CAST", true, false, false, true)
+PG_CMDTAG(CMDTAG_ALTER_COLLATION, "ALTER COLLATION", true, false, false, true)
+PG_CMDTAG(CMDTAG_ALTER_CONSTRAINT, "ALTER CONSTRAINT", true, false, false, false)
+PG_CMDTAG(CMDTAG_ALTER_CONVERSION, "ALTER CONVERSION", true, false, false, true)
+PG_CMDTAG(CMDTAG_ALTER_DATABASE, "ALTER DATABASE", false, false, false, false)
+PG_CMDTAG(CMDTAG_ALTER_DEFAULT_PRIVILEGES, "ALTER DEFAULT PRIVILEGES", true, false, false, true)
+PG_CMDTAG(CMDTAG_ALTER_DOMAIN, "ALTER DOMAIN", true, false, false, true)
+PG_CMDTAG(CMDTAG_ALTER_EVENT_TRIGGER, "ALTER EVENT TRIGGER", false, false, false, false)
+PG_CMDTAG(CMDTAG_ALTER_EXTENSION, "ALTER EXTENSION", true, false, false, true)
+PG_CMDTAG(CMDTAG_ALTER_FOREIGN_DATA_WRAPPER, "ALTER FOREIGN DATA WRAPPER", true, false, false, true)
+PG_CMDTAG(CMDTAG_ALTER_FOREIGN_TABLE, "ALTER FOREIGN TABLE", true, false, false, true)
+PG_CMDTAG(CMDTAG_ALTER_FUNCTION, "ALTER FUNCTION", true, false, false, true)
+PG_CMDTAG(CMDTAG_ALTER_INDEX, "ALTER INDEX", true, false, false, true)
+PG_CMDTAG(CMDTAG_ALTER_LANGUAGE, "ALTER LANGUAGE", true, false, false, true)
+PG_CMDTAG(CMDTAG_ALTER_LARGE_OBJECT, "ALTER LARGE OBJECT", true, false, false, false)
+PG_CMDTAG(CMDTAG_ALTER_MATERIALIZED_VIEW, "ALTER MATERIALIZED VIEW", true, true, false, true)
+PG_CMDTAG(CMDTAG_ALTER_OPERATOR, "ALTER OPERATOR", true, false, false, true)
+PG_CMDTAG(CMDTAG_ALTER_OPERATOR_CLASS, "ALTER OPERATOR CLASS", true, false, false, true)
+PG_CMDTAG(CMDTAG_ALTER_OPERATOR_FAMILY, "ALTER OPERATOR FAMILY", true, false, false, true)
+PG_CMDTAG(CMDTAG_ALTER_POLICY, "ALTER POLICY", true, false, false, true)
+PG_CMDTAG(CMDTAG_ALTER_PROCEDURE, "ALTER PROCEDURE", true, false, false, true)
+PG_CMDTAG(CMDTAG_ALTER_PUBLICATION, "ALTER PUBLICATION", true, false, false, true)
+PG_CMDTAG(CMDTAG_ALTER_ROLE, "ALTER ROLE", false, false, false, false)
+PG_CMDTAG(CMDTAG_ALTER_ROUTINE, "ALTER ROUTINE", true, false, false, true)
+PG_CMDTAG(CMDTAG_ALTER_RULE, "ALTER RULE", true, false, false, true)
+PG_CMDTAG(CMDTAG_ALTER_SCHEMA, "ALTER SCHEMA", true, false, false, true)
+PG_CMDTAG(CMDTAG_ALTER_SEQUENCE, "ALTER SEQUENCE", true, false, false, true)
+PG_CMDTAG(CMDTAG_ALTER_SERVER, "ALTER SERVER", true, false, false, true)
+PG_CMDTAG(CMDTAG_ALTER_STATISTICS, "ALTER STATISTICS", true, false, false, true)
+PG_CMDTAG(CMDTAG_ALTER_SUBSCRIPTION, "ALTER SUBSCRIPTION", true, false, false, false)
+PG_CMDTAG(CMDTAG_ALTER_SYSTEM, "ALTER SYSTEM", false, false, false, false)
+PG_CMDTAG(CMDTAG_ALTER_TABLE, "ALTER TABLE", true, true, false, true)
+PG_CMDTAG(CMDTAG_ALTER_TABLESPACE, "ALTER TABLESPACE", false, false, false, false)
+PG_CMDTAG(CMDTAG_ALTER_TEXT_SEARCH_CONFIGURATION, "ALTER TEXT SEARCH CONFIGURATION", true, false, false, true)
+PG_CMDTAG(CMDTAG_ALTER_TEXT_SEARCH_DICTIONARY, "ALTER TEXT SEARCH DICTIONARY", true, false, false, true)
+PG_CMDTAG(CMDTAG_ALTER_TEXT_SEARCH_PARSER, "ALTER TEXT SEARCH PARSER", true, false, false, true)
+PG_CMDTAG(CMDTAG_ALTER_TEXT_SEARCH_TEMPLATE, "ALTER TEXT SEARCH TEMPLATE", true, false, false, true)
+PG_CMDTAG(CMDTAG_ALTER_TRANSFORM, "ALTER TRANSFORM", true, false, false, false)
+PG_CMDTAG(CMDTAG_ALTER_TRIGGER, "ALTER TRIGGER", true, false, false, true)
+PG_CMDTAG(CMDTAG_ALTER_TYPE, "ALTER TYPE", true, true, false, true)
+PG_CMDTAG(CMDTAG_ALTER_USER_MAPPING, "ALTER USER MAPPING", true, false, false, true)
+PG_CMDTAG(CMDTAG_ALTER_VIEW, "ALTER VIEW", true, false, false, true)
+PG_CMDTAG(CMDTAG_ANALYZE, "ANALYZE", false, false, false, false)
+PG_CMDTAG(CMDTAG_BEGIN, "BEGIN", false, false, false, false)
+PG_CMDTAG(CMDTAG_CALL, "CALL", false, false, false, false)
+PG_CMDTAG(CMDTAG_CHECKPOINT, "CHECKPOINT", false, false, false, false)
+PG_CMDTAG(CMDTAG_CLOSE, "CLOSE", false, false, false, false)
+PG_CMDTAG(CMDTAG_CLOSE_CURSOR, "CLOSE CURSOR", false, false, false, false)
+PG_CMDTAG(CMDTAG_CLOSE_CURSOR_ALL, "CLOSE CURSOR ALL", false, false, false, false)
+PG_CMDTAG(CMDTAG_CLUSTER, "CLUSTER", false, false, false, false)
+PG_CMDTAG(CMDTAG_COMMENT, "COMMENT", true, false, false, true)
+PG_CMDTAG(CMDTAG_COMMIT, "COMMIT", false, false, false, false)
+PG_CMDTAG(CMDTAG_COMMIT_PREPARED, "COMMIT PREPARED", false, false, false, false)
+PG_CMDTAG(CMDTAG_COPY, "COPY", false, false, true, false)
+PG_CMDTAG(CMDTAG_COPY_FROM, "COPY FROM", false, false, false, false)
+PG_CMDTAG(CMDTAG_CREATE_ACCESS_METHOD, "CREATE ACCESS METHOD", true, false, false, true)
+PG_CMDTAG(CMDTAG_CREATE_AGGREGATE, "CREATE AGGREGATE", true, false, false, true)
+PG_CMDTAG(CMDTAG_CREATE_CAST, "CREATE CAST", true, false, false, true)
+PG_CMDTAG(CMDTAG_CREATE_COLLATION, "CREATE COLLATION", true, false, false, true)
+PG_CMDTAG(CMDTAG_CREATE_CONSTRAINT, "CREATE CONSTRAINT", true, false, false, false)
+PG_CMDTAG(CMDTAG_CREATE_CONVERSION, "CREATE CONVERSION", true, false, false, true)
+PG_CMDTAG(CMDTAG_CREATE_DATABASE, "CREATE DATABASE", false, false, false, false)
+PG_CMDTAG(CMDTAG_CREATE_DOMAIN, "CREATE DOMAIN", true, false, false, true)
+PG_CMDTAG(CMDTAG_CREATE_EVENT_TRIGGER, "CREATE EVENT TRIGGER", false, false, false, false)
+PG_CMDTAG(CMDTAG_CREATE_EXTENSION, "CREATE EXTENSION", true, false, false, true)
+PG_CMDTAG(CMDTAG_CREATE_FOREIGN_DATA_WRAPPER, "CREATE FOREIGN DATA WRAPPER", true, false, false, true)
+PG_CMDTAG(CMDTAG_CREATE_FOREIGN_TABLE, "CREATE FOREIGN TABLE", true, false, false, true)
+PG_CMDTAG(CMDTAG_CREATE_FUNCTION, "CREATE FUNCTION", true, false, false, true)
+PG_CMDTAG(CMDTAG_CREATE_INDEX, "CREATE INDEX", true, false, false, true)
+PG_CMDTAG(CMDTAG_CREATE_LANGUAGE, "CREATE LANGUAGE", true, false, false, true)
+PG_CMDTAG(CMDTAG_CREATE_MATERIALIZED_VIEW, "CREATE MATERIALIZED VIEW", true, false, false, true)
+PG_CMDTAG(CMDTAG_CREATE_OPERATOR, "CREATE OPERATOR", true, false, false, true)
+PG_CMDTAG(CMDTAG_CREATE_OPERATOR_CLASS, "CREATE OPERATOR CLASS", true, false, false, true)
+PG_CMDTAG(CMDTAG_CREATE_OPERATOR_FAMILY, "CREATE OPERATOR FAMILY", true, false, false, true)
+PG_CMDTAG(CMDTAG_CREATE_POLICY, "CREATE POLICY", true, false, false, true)
+PG_CMDTAG(CMDTAG_CREATE_PROCEDURE, "CREATE PROCEDURE", true, false, false, true)
+PG_CMDTAG(CMDTAG_CREATE_PUBLICATION, "CREATE PUBLICATION", true, false, false, true)
+PG_CMDTAG(CMDTAG_CREATE_ROLE, "CREATE ROLE", false, false, false, false)
+PG_CMDTAG(CMDTAG_CREATE_ROUTINE, "CREATE ROUTINE", true, false, false, false)
+PG_CMDTAG(CMDTAG_CREATE_RULE, "CREATE RULE", true, false, false, true)
+PG_CMDTAG(CMDTAG_CREATE_SCHEMA, "CREATE SCHEMA", true, false, false, true)
+PG_CMDTAG(CMDTAG_CREATE_SEQUENCE, "CREATE SEQUENCE", true, false, false, true)
+PG_CMDTAG(CMDTAG_CREATE_SERVER, "CREATE SERVER", true, false, false, true)
+PG_CMDTAG(CMDTAG_CREATE_STATISTICS, "CREATE STATISTICS", true, false, false, true)
+PG_CMDTAG(CMDTAG_CREATE_SUBSCRIPTION, "CREATE SUBSCRIPTION", true, false, false, false)
+PG_CMDTAG(CMDTAG_CREATE_TABLE, "CREATE TABLE", true, false, false, true)
+PG_CMDTAG(CMDTAG_CREATE_TABLE_AS, "CREATE TABLE AS", true, false, false, false)
+PG_CMDTAG(CMDTAG_CREATE_TABLESPACE, "CREATE TABLESPACE", false, false, false, false)
+PG_CMDTAG(CMDTAG_CREATE_TEXT_SEARCH_CONFIGURATION, "CREATE TEXT SEARCH CONFIGURATION", true, false, false, true)
+PG_CMDTAG(CMDTAG_CREATE_TEXT_SEARCH_DICTIONARY, "CREATE TEXT SEARCH DICTIONARY", true, false, false, true)
+PG_CMDTAG(CMDTAG_CREATE_TEXT_SEARCH_PARSER, "CREATE TEXT SEARCH PARSER", true, false, false, true)
+PG_CMDTAG(CMDTAG_CREATE_TEXT_SEARCH_TEMPLATE, "CREATE TEXT SEARCH TEMPLATE", true, false, false, true)
+PG_CMDTAG(CMDTAG_CREATE_TRANSFORM, "CREATE TRANSFORM", true, false, false, true)
+PG_CMDTAG(CMDTAG_CREATE_TRIGGER, "CREATE TRIGGER", true, false, false, true)
+PG_CMDTAG(CMDTAG_CREATE_TYPE, "CREATE TYPE", true, false, false, true)
+PG_CMDTAG(CMDTAG_CREATE_USER_MAPPING, "CREATE USER MAPPING", true, false, false, true)
+PG_CMDTAG(CMDTAG_CREATE_VIEW, "CREATE VIEW", true, false, false, true)
+PG_CMDTAG(CMDTAG_DEALLOCATE, "DEALLOCATE", false, false, false, false)
+PG_CMDTAG(CMDTAG_DEALLOCATE_ALL, "DEALLOCATE ALL", false, false, false, false)
+PG_CMDTAG(CMDTAG_DECLARE_CURSOR, "DECLARE CURSOR", false, false, false, false)
+PG_CMDTAG(CMDTAG_DELETE, "DELETE", false, false, true, false)
+PG_CMDTAG(CMDTAG_DISCARD, "DISCARD", false, false, false, false)
+PG_CMDTAG(CMDTAG_DISCARD_ALL, "DISCARD ALL", false, false, false, false)
+PG_CMDTAG(CMDTAG_DISCARD_PLANS, "DISCARD PLANS", false, false, false, false)
+PG_CMDTAG(CMDTAG_DISCARD_SEQUENCES, "DISCARD SEQUENCES", false, false, false, false)
+PG_CMDTAG(CMDTAG_DISCARD_TEMP, "DISCARD TEMP", false, false, false, false)
+PG_CMDTAG(CMDTAG_DO, "DO", false, false, false, false)
+PG_CMDTAG(CMDTAG_DROP_ACCESS_METHOD, "DROP ACCESS METHOD", true, false, false, true)
+PG_CMDTAG(CMDTAG_DROP_AGGREGATE, "DROP AGGREGATE", true, false, false, true)
+PG_CMDTAG(CMDTAG_DROP_CAST, "DROP CAST", true, false, false, true)
+PG_CMDTAG(CMDTAG_DROP_COLLATION, "DROP COLLATION", true, false, false, true)
+PG_CMDTAG(CMDTAG_DROP_CONSTRAINT, "DROP CONSTRAINT", true, false, false, false)
+PG_CMDTAG(CMDTAG_DROP_CONVERSION, "DROP CONVERSION", true, false, false, true)
+PG_CMDTAG(CMDTAG_DROP_DATABASE, "DROP DATABASE", false, false, false, false)
+PG_CMDTAG(CMDTAG_DROP_DOMAIN, "DROP DOMAIN", true, false, false, true)
+PG_CMDTAG(CMDTAG_DROP_EVENT_TRIGGER, "DROP EVENT TRIGGER", false, false, false, false)
+PG_CMDTAG(CMDTAG_DROP_EXTENSION, "DROP EXTENSION", true, false, false, true)
+PG_CMDTAG(CMDTAG_DROP_FOREIGN_DATA_WRAPPER, "DROP FOREIGN DATA WRAPPER", true, false, false, true)
+PG_CMDTAG(CMDTAG_DROP_FOREIGN_TABLE, "DROP FOREIGN TABLE", true, false, false, true)
+PG_CMDTAG(CMDTAG_DROP_FUNCTION, "DROP FUNCTION", true, false, false, true)
+PG_CMDTAG(CMDTAG_DROP_INDEX, "DROP INDEX", true, false, false, true)
+PG_CMDTAG(CMDTAG_DROP_LANGUAGE, "DROP LANGUAGE", true, false, false, true)
+PG_CMDTAG(CMDTAG_DROP_MATERIALIZED_VIEW, "DROP MATERIALIZED VIEW", true, false, false, true)
+PG_CMDTAG(CMDTAG_DROP_OPERATOR, "DROP OPERATOR", true, false, false, true)
+PG_CMDTAG(CMDTAG_DROP_OPERATOR_CLASS, "DROP OPERATOR CLASS", true, false, false, true)
+PG_CMDTAG(CMDTAG_DROP_OPERATOR_FAMILY, "DROP OPERATOR FAMILY", true, false, false, true)
+PG_CMDTAG(CMDTAG_DROP_OWNED, "DROP OWNED", true, false, false, false)
+PG_CMDTAG(CMDTAG_DROP_POLICY, "DROP POLICY", true, false, false, true)
+PG_CMDTAG(CMDTAG_DROP_PROCEDURE, "DROP PROCEDURE", true, false, false, true)
+PG_CMDTAG(CMDTAG_DROP_PUBLICATION, "DROP PUBLICATION", true, false, false, true)
+PG_CMDTAG(CMDTAG_DROP_ROLE, "DROP ROLE", false, false, false, false)
+PG_CMDTAG(CMDTAG_DROP_ROUTINE, "DROP ROUTINE", true, false, false, true)
+PG_CMDTAG(CMDTAG_DROP_RULE, "DROP RULE", true, false, false, true)
+PG_CMDTAG(CMDTAG_DROP_SCHEMA, "DROP SCHEMA", true, false, false, true)
+PG_CMDTAG(CMDTAG_DROP_SEQUENCE, "DROP SEQUENCE", true, false, false, true)
+PG_CMDTAG(CMDTAG_DROP_SERVER, "DROP SERVER", true, false, false, true)
+PG_CMDTAG(CMDTAG_DROP_STATISTICS, "DROP STATISTICS", true, false, false, true)
+PG_CMDTAG(CMDTAG_DROP_SUBSCRIPTION, "DROP SUBSCRIPTION", true, false, false, false)
+PG_CMDTAG(CMDTAG_DROP_TABLE, "DROP TABLE", true, false, false, true)
+PG_CMDTAG(CMDTAG_DROP_TABLESPACE, "DROP TABLESPACE", false, false, false, false)
+PG_CMDTAG(CMDTAG_DROP_TEXT_SEARCH_CONFIGURATION, "DROP TEXT SEARCH CONFIGURATION", true, false, false, true)
+PG_CMDTAG(CMDTAG_DROP_TEXT_SEARCH_DICTIONARY, "DROP TEXT SEARCH DICTIONARY", true, false, false, true)
+PG_CMDTAG(CMDTAG_DROP_TEXT_SEARCH_PARSER, "DROP TEXT SEARCH PARSER", true, false, false, true)
+PG_CMDTAG(CMDTAG_DROP_TEXT_SEARCH_TEMPLATE, "DROP TEXT SEARCH TEMPLATE", true, false, false, true)
+PG_CMDTAG(CMDTAG_DROP_TRANSFORM, "DROP TRANSFORM", true, false, false, true)
+PG_CMDTAG(CMDTAG_DROP_TRIGGER, "DROP TRIGGER", true, false, false, true)
+PG_CMDTAG(CMDTAG_DROP_TYPE, "DROP TYPE", true, false, false, true)
+PG_CMDTAG(CMDTAG_DROP_USER_MAPPING, "DROP USER MAPPING", true, false, false, true)
+PG_CMDTAG(CMDTAG_DROP_VIEW, "DROP VIEW", true, false, false, true)
+PG_CMDTAG(CMDTAG_EXECUTE, "EXECUTE", false, false, false, false)
+PG_CMDTAG(CMDTAG_EXPLAIN, "EXPLAIN", false, false, false, false)
+PG_CMDTAG(CMDTAG_FETCH, "FETCH", false, false, true, false)
+PG_CMDTAG(CMDTAG_GRANT, "GRANT", true, false, false, true)
+PG_CMDTAG(CMDTAG_GRANT_ROLE, "GRANT ROLE", false, false, false, false)
+PG_CMDTAG(CMDTAG_IMPORT_FOREIGN_SCHEMA, "IMPORT FOREIGN SCHEMA", true, false, false, true)
+PG_CMDTAG(CMDTAG_INSERT, "INSERT", false, false, true, false)
+PG_CMDTAG(CMDTAG_LISTEN, "LISTEN", false, false, false, false)
+PG_CMDTAG(CMDTAG_LOAD, "LOAD", false, false, false, false)
+PG_CMDTAG(CMDTAG_LOCK_TABLE, "LOCK TABLE", false, false, false, false)
+PG_CMDTAG(CMDTAG_MERGE, "MERGE", false, false, true, false)
+PG_CMDTAG(CMDTAG_MOVE, "MOVE", false, false, true, false)
+PG_CMDTAG(CMDTAG_NOTIFY, "NOTIFY", false, false, false, false)
+PG_CMDTAG(CMDTAG_PREPARE, "PREPARE", false, false, false, false)
+PG_CMDTAG(CMDTAG_PREPARE_TRANSACTION, "PREPARE TRANSACTION", false, false, false, false)
+PG_CMDTAG(CMDTAG_REASSIGN_OWNED, "REASSIGN OWNED", false, false, false, false)
+PG_CMDTAG(CMDTAG_REFRESH_MATERIALIZED_VIEW, "REFRESH MATERIALIZED VIEW", true, false, false, true)
+PG_CMDTAG(CMDTAG_REINDEX, "REINDEX", false, false, false, false)
+PG_CMDTAG(CMDTAG_RELEASE, "RELEASE", false, false, false, false)
+PG_CMDTAG(CMDTAG_RESET, "RESET", false, false, false, false)
+PG_CMDTAG(CMDTAG_REVOKE, "REVOKE", true, false, false, true)
+PG_CMDTAG(CMDTAG_REVOKE_ROLE, "REVOKE ROLE", false, false, false, false)
+PG_CMDTAG(CMDTAG_ROLLBACK, "ROLLBACK", false, false, false, false)
+PG_CMDTAG(CMDTAG_ROLLBACK_PREPARED, "ROLLBACK PREPARED", false, false, false, false)
+PG_CMDTAG(CMDTAG_SAVEPOINT, "SAVEPOINT", false, false, false, false)
+PG_CMDTAG(CMDTAG_SECURITY_LABEL, "SECURITY LABEL", true, false, false, true)
+PG_CMDTAG(CMDTAG_SELECT, "SELECT", false, false, true, false)
+PG_CMDTAG(CMDTAG_SELECT_FOR_KEY_SHARE, "SELECT FOR KEY SHARE", false, false, false, false)
+PG_CMDTAG(CMDTAG_SELECT_FOR_NO_KEY_UPDATE, "SELECT FOR NO KEY UPDATE", false, false, false, false)
+PG_CMDTAG(CMDTAG_SELECT_FOR_SHARE, "SELECT FOR SHARE", false, false, false, false)
+PG_CMDTAG(CMDTAG_SELECT_FOR_UPDATE, "SELECT FOR UPDATE", false, false, false, false)
+PG_CMDTAG(CMDTAG_SELECT_INTO, "SELECT INTO", true, false, false, false)
+PG_CMDTAG(CMDTAG_SET, "SET", false, false, false, false)
+PG_CMDTAG(CMDTAG_SET_CONSTRAINTS, "SET CONSTRAINTS", false, false, false, false)
+PG_CMDTAG(CMDTAG_SHOW, "SHOW", false, false, false, false)
+PG_CMDTAG(CMDTAG_START_TRANSACTION, "START TRANSACTION", false, false, false, false)
+PG_CMDTAG(CMDTAG_TRUNCATE_TABLE, "TRUNCATE TABLE", false, false, false, false)
+PG_CMDTAG(CMDTAG_UNLISTEN, "UNLISTEN", false, false, false, false)
+PG_CMDTAG(CMDTAG_UPDATE, "UPDATE", false, false, true, false)
+PG_CMDTAG(CMDTAG_VACUUM, "VACUUM", false, false, false, false)
diff --git a/src/include/utils/rel.h b/src/include/utils/rel.h
index 31f84e90eb..d75334a819 100644
--- a/src/include/utils/rel.h
+++ b/src/include/utils/rel.h
@@ -17,10 +17,12 @@
#include "access/tupdesc.h"
#include "access/xlog.h"
#include "catalog/catalog.h"
+#include "catalog/objectaddress.h"
#include "catalog/pg_class.h"
#include "catalog/pg_index.h"
#include "catalog/pg_publication.h"
#include "nodes/bitmapset.h"
+#include "nodes/lockoptions.h"
#include "partitioning/partdefs.h"
#include "rewrite/prs2lock.h"
#include "storage/block.h"
diff --git a/src/test/regress/expected/psql.out b/src/test/regress/expected/psql.out
index c00e28361c..1c6f33095c 100644
--- a/src/test/regress/expected/psql.out
+++ b/src/test/regress/expected/psql.out
@@ -6223,9 +6223,9 @@ List of schemas
(0 rows)
\dRp "no.such.publication"
- List of publications
- Name | Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
-------+-------+------------+---------+---------+---------+-----------+----------
+ List of publications
+ Name | Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+------+-------+------------+------------+---------+---------+---------+-----------+----------
(0 rows)
\dRs "no.such.subscription"
diff --git a/src/test/regress/expected/publication.out b/src/test/regress/expected/publication.out
index 427f87ea07..7d86977b29 100644
--- a/src/test/regress/expected/publication.out
+++ b/src/test/regress/expected/publication.out
@@ -30,20 +30,20 @@ ERROR: conflicting or redundant options
LINE 1: ...ub_xxx WITH (publish_via_partition_root = 'true', publish_vi...
^
\dRp
- List of publications
- Name | Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------+--------------------------+------------+---------+---------+---------+-----------+----------
- testpib_ins_trunct | regress_publication_user | f | t | f | f | f | f
- testpub_default | regress_publication_user | f | f | t | f | f | f
+ List of publications
+ Name | Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+--------------------+--------------------------+------------+------------+---------+---------+---------+-----------+----------
+ testpib_ins_trunct | regress_publication_user | f | f | t | f | f | f | f
+ testpub_default | regress_publication_user | f | f | f | t | f | f | f
(2 rows)
ALTER PUBLICATION testpub_default SET (publish = 'insert, update, delete');
\dRp
- List of publications
- Name | Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------+--------------------------+------------+---------+---------+---------+-----------+----------
- testpib_ins_trunct | regress_publication_user | f | t | f | f | f | f
- testpub_default | regress_publication_user | f | t | t | t | f | f
+ List of publications
+ Name | Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+--------------------+--------------------------+------------+------------+---------+---------+---------+-----------+----------
+ testpib_ins_trunct | regress_publication_user | f | f | t | f | f | f | f
+ testpub_default | regress_publication_user | f | f | t | t | t | f | f
(2 rows)
--- adding tables
@@ -87,10 +87,10 @@ RESET client_min_messages;
-- should be able to add schema to 'FOR TABLE' publication
ALTER PUBLICATION testpub_fortable ADD TABLES IN SCHEMA pub_test;
\dRp+ testpub_fortable
- Publication testpub_fortable
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub_fortable
+ Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | f
Tables:
"public.testpub_tbl1"
Tables from schemas:
@@ -99,20 +99,20 @@ Tables from schemas:
-- should be able to drop schema from 'FOR TABLE' publication
ALTER PUBLICATION testpub_fortable DROP TABLES IN SCHEMA pub_test;
\dRp+ testpub_fortable
- Publication testpub_fortable
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub_fortable
+ Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | f
Tables:
"public.testpub_tbl1"
-- should be able to set schema to 'FOR TABLE' publication
ALTER PUBLICATION testpub_fortable SET TABLES IN SCHEMA pub_test;
\dRp+ testpub_fortable
- Publication testpub_fortable
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub_fortable
+ Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | f
Tables from schemas:
"pub_test"
@@ -123,10 +123,10 @@ CREATE PUBLICATION testpub_forschema FOR TABLES IN SCHEMA pub_test;
CREATE PUBLICATION testpub_for_tbl_schema FOR TABLES IN SCHEMA pub_test, TABLE pub_test.testpub_nopk;
RESET client_min_messages;
\dRp+ testpub_for_tbl_schema
- Publication testpub_for_tbl_schema
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub_for_tbl_schema
+ Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | f
Tables:
"pub_test.testpub_nopk"
Tables from schemas:
@@ -135,10 +135,10 @@ Tables from schemas:
-- should be able to add a table of the same schema to the schema publication
ALTER PUBLICATION testpub_forschema ADD TABLE pub_test.testpub_nopk;
\dRp+ testpub_forschema
- Publication testpub_forschema
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub_forschema
+ Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | f
Tables:
"pub_test.testpub_nopk"
Tables from schemas:
@@ -147,10 +147,10 @@ Tables from schemas:
-- should be able to drop the table
ALTER PUBLICATION testpub_forschema DROP TABLE pub_test.testpub_nopk;
\dRp+ testpub_forschema
- Publication testpub_forschema
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub_forschema
+ Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | f
Tables from schemas:
"pub_test"
@@ -161,10 +161,10 @@ ERROR: relation "testpub_nopk" is not part of the publication
-- should be able to set table to schema publication
ALTER PUBLICATION testpub_forschema SET TABLE pub_test.testpub_nopk;
\dRp+ testpub_forschema
- Publication testpub_forschema
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub_forschema
+ Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | f
Tables:
"pub_test.testpub_nopk"
@@ -186,10 +186,10 @@ Publications:
"testpub_foralltables"
\dRp+ testpub_foralltables
- Publication testpub_foralltables
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | t | t | t | f | f | f
+ Publication testpub_foralltables
+ Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | t | f | t | t | f | f | f
(1 row)
DROP TABLE testpub_tbl2;
@@ -201,19 +201,19 @@ CREATE PUBLICATION testpub3 FOR TABLE testpub_tbl3;
CREATE PUBLICATION testpub4 FOR TABLE ONLY testpub_tbl3;
RESET client_min_messages;
\dRp+ testpub3
- Publication testpub3
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub3
+ Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | f
Tables:
"public.testpub_tbl3"
"public.testpub_tbl3a"
\dRp+ testpub4
- Publication testpub4
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub4
+ Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | f
Tables:
"public.testpub_tbl3"
@@ -234,10 +234,10 @@ UPDATE testpub_parted1 SET a = 1;
-- only parent is listed as being in publication, not the partition
ALTER PUBLICATION testpub_forparted ADD TABLE testpub_parted;
\dRp+ testpub_forparted
- Publication testpub_forparted
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub_forparted
+ Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | f
Tables:
"public.testpub_parted"
@@ -252,10 +252,10 @@ ALTER TABLE testpub_parted DETACH PARTITION testpub_parted1;
UPDATE testpub_parted1 SET a = 1;
ALTER PUBLICATION testpub_forparted SET (publish_via_partition_root = true);
\dRp+ testpub_forparted
- Publication testpub_forparted
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | t
+ Publication testpub_forparted
+ Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | t
Tables:
"public.testpub_parted"
@@ -284,10 +284,10 @@ SET client_min_messages = 'ERROR';
CREATE PUBLICATION testpub5 FOR TABLE testpub_rf_tbl1, testpub_rf_tbl2 WHERE (c <> 'test' AND d < 5) WITH (publish = 'insert');
RESET client_min_messages;
\dRp+ testpub5
- Publication testpub5
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | f | f | f | f
+ Publication testpub5
+ Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | f | t | f | f | f | f
Tables:
"public.testpub_rf_tbl1"
"public.testpub_rf_tbl2" WHERE ((c <> 'test'::text) AND (d < 5))
@@ -300,10 +300,10 @@ Tables:
ALTER PUBLICATION testpub5 ADD TABLE testpub_rf_tbl3 WHERE (e > 1000 AND e < 2000);
\dRp+ testpub5
- Publication testpub5
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | f | f | f | f
+ Publication testpub5
+ Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | f | t | f | f | f | f
Tables:
"public.testpub_rf_tbl1"
"public.testpub_rf_tbl2" WHERE ((c <> 'test'::text) AND (d < 5))
@@ -319,10 +319,10 @@ Publications:
ALTER PUBLICATION testpub5 DROP TABLE testpub_rf_tbl2;
\dRp+ testpub5
- Publication testpub5
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | f | f | f | f
+ Publication testpub5
+ Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | f | t | f | f | f | f
Tables:
"public.testpub_rf_tbl1"
"public.testpub_rf_tbl3" WHERE ((e > 1000) AND (e < 2000))
@@ -330,10 +330,10 @@ Tables:
-- remove testpub_rf_tbl1 and add testpub_rf_tbl3 again (another WHERE expression)
ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl3 WHERE (e > 300 AND e < 500);
\dRp+ testpub5
- Publication testpub5
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | f | f | f | f
+ Publication testpub5
+ Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | f | t | f | f | f | f
Tables:
"public.testpub_rf_tbl3" WHERE ((e > 300) AND (e < 500))
@@ -366,10 +366,10 @@ SET client_min_messages = 'ERROR';
CREATE PUBLICATION testpub_syntax1 FOR TABLE testpub_rf_tbl1, ONLY testpub_rf_tbl3 WHERE (e < 999) WITH (publish = 'insert');
RESET client_min_messages;
\dRp+ testpub_syntax1
- Publication testpub_syntax1
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | f | f | f | f
+ Publication testpub_syntax1
+ Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | f | t | f | f | f | f
Tables:
"public.testpub_rf_tbl1"
"public.testpub_rf_tbl3" WHERE (e < 999)
@@ -379,10 +379,10 @@ SET client_min_messages = 'ERROR';
CREATE PUBLICATION testpub_syntax2 FOR TABLE testpub_rf_tbl1, testpub_rf_schema1.testpub_rf_tbl5 WHERE (h < 999) WITH (publish = 'insert');
RESET client_min_messages;
\dRp+ testpub_syntax2
- Publication testpub_syntax2
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | f | f | f | f
+ Publication testpub_syntax2
+ Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | f | t | f | f | f | f
Tables:
"public.testpub_rf_tbl1"
"testpub_rf_schema1.testpub_rf_tbl5" WHERE (h < 999)
@@ -497,10 +497,10 @@ CREATE PUBLICATION testpub6 FOR TABLES IN SCHEMA testpub_rf_schema2;
ALTER PUBLICATION testpub6 SET TABLES IN SCHEMA testpub_rf_schema2, TABLE testpub_rf_schema2.testpub_rf_tbl6 WHERE (i < 99);
RESET client_min_messages;
\dRp+ testpub6
- Publication testpub6
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub6
+ Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | f
Tables:
"testpub_rf_schema2.testpub_rf_tbl6" WHERE (i < 99)
Tables from schemas:
@@ -714,10 +714,10 @@ CREATE PUBLICATION testpub_table_ins WITH (publish = 'insert, truncate');
RESET client_min_messages;
ALTER PUBLICATION testpub_table_ins ADD TABLE testpub_tbl5 (a); -- ok
\dRp+ testpub_table_ins
- Publication testpub_table_ins
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | f | f | t | f
+ Publication testpub_table_ins
+ Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | f | t | f | f | t | f
Tables:
"public.testpub_tbl5" (a)
@@ -891,10 +891,10 @@ CREATE TABLE testpub_tbl_both_filters (a int, b int, c int, PRIMARY KEY (a,c));
ALTER TABLE testpub_tbl_both_filters REPLICA IDENTITY USING INDEX testpub_tbl_both_filters_pkey;
ALTER PUBLICATION testpub_both_filters ADD TABLE testpub_tbl_both_filters (a,c) WHERE (c != 1);
\dRp+ testpub_both_filters
- Publication testpub_both_filters
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub_both_filters
+ Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | f
Tables:
"public.testpub_tbl_both_filters" (a, c) WHERE (c <> 1)
@@ -1099,10 +1099,10 @@ ERROR: relation "testpub_tbl1" is already member of publication "testpub_fortbl
CREATE PUBLICATION testpub_fortbl FOR TABLE testpub_tbl1;
ERROR: publication "testpub_fortbl" already exists
\dRp+ testpub_fortbl
- Publication testpub_fortbl
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub_fortbl
+ Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | f
Tables:
"pub_test.testpub_nopk"
"public.testpub_tbl1"
@@ -1140,10 +1140,10 @@ Publications:
"testpub_fortbl"
\dRp+ testpub_default
- Publication testpub_default
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | f | f
+ Publication testpub_default
+ Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | f | t | t | t | f | f
Tables:
"pub_test.testpub_nopk"
"public.testpub_tbl1"
@@ -1221,10 +1221,10 @@ REVOKE CREATE ON DATABASE regression FROM regress_publication_user2;
DROP TABLE testpub_parted;
DROP TABLE testpub_tbl1;
\dRp+ testpub_default
- Publication testpub_default
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | f | f
+ Publication testpub_default
+ Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | f | t | t | t | f | f
(1 row)
-- fail - must be owner of publication
@@ -1234,20 +1234,20 @@ ERROR: must be owner of publication testpub_default
RESET ROLE;
ALTER PUBLICATION testpub_default RENAME TO testpub_foo;
\dRp testpub_foo
- List of publications
- Name | Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
--------------+--------------------------+------------+---------+---------+---------+-----------+----------
- testpub_foo | regress_publication_user | f | t | t | t | f | f
+ List of publications
+ Name | Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+-------------+--------------------------+------------+------------+---------+---------+---------+-----------+----------
+ testpub_foo | regress_publication_user | f | f | t | t | t | f | f
(1 row)
-- rename back to keep the rest simple
ALTER PUBLICATION testpub_foo RENAME TO testpub_default;
ALTER PUBLICATION testpub_default OWNER TO regress_publication_user2;
\dRp testpub_default
- List of publications
- Name | Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
------------------+---------------------------+------------+---------+---------+---------+-----------+----------
- testpub_default | regress_publication_user2 | f | t | t | t | f | f
+ List of publications
+ Name | Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+-----------------+---------------------------+------------+------------+---------+---------+---------+-----------+----------
+ testpub_default | regress_publication_user2 | f | f | t | t | t | f | f
(1 row)
-- adding schemas and tables
@@ -1263,19 +1263,19 @@ CREATE TABLE "CURRENT_SCHEMA"."CURRENT_SCHEMA"(id int);
SET client_min_messages = 'ERROR';
CREATE PUBLICATION testpub1_forschema FOR TABLES IN SCHEMA pub_test1;
\dRp+ testpub1_forschema
- Publication testpub1_forschema
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub1_forschema
+ Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | f
Tables from schemas:
"pub_test1"
CREATE PUBLICATION testpub2_forschema FOR TABLES IN SCHEMA pub_test1, pub_test2, pub_test3;
\dRp+ testpub2_forschema
- Publication testpub2_forschema
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub2_forschema
+ Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | f
Tables from schemas:
"pub_test1"
"pub_test2"
@@ -1289,44 +1289,44 @@ CREATE PUBLICATION testpub6_forschema FOR TABLES IN SCHEMA "CURRENT_SCHEMA", CUR
CREATE PUBLICATION testpub_fortable FOR TABLE "CURRENT_SCHEMA"."CURRENT_SCHEMA";
RESET client_min_messages;
\dRp+ testpub3_forschema
- Publication testpub3_forschema
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub3_forschema
+ Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | f
Tables from schemas:
"public"
\dRp+ testpub4_forschema
- Publication testpub4_forschema
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub4_forschema
+ Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | f
Tables from schemas:
"CURRENT_SCHEMA"
\dRp+ testpub5_forschema
- Publication testpub5_forschema
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub5_forschema
+ Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | f
Tables from schemas:
"CURRENT_SCHEMA"
"public"
\dRp+ testpub6_forschema
- Publication testpub6_forschema
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub6_forschema
+ Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | f
Tables from schemas:
"CURRENT_SCHEMA"
"public"
\dRp+ testpub_fortable
- Publication testpub_fortable
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub_fortable
+ Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | f
Tables:
"CURRENT_SCHEMA.CURRENT_SCHEMA"
@@ -1360,10 +1360,10 @@ ERROR: schema "testpub_view" does not exist
-- dropping the schema should reflect the change in publication
DROP SCHEMA pub_test3;
\dRp+ testpub2_forschema
- Publication testpub2_forschema
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub2_forschema
+ Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | f
Tables from schemas:
"pub_test1"
"pub_test2"
@@ -1371,20 +1371,20 @@ Tables from schemas:
-- renaming the schema should reflect the change in publication
ALTER SCHEMA pub_test1 RENAME to pub_test1_renamed;
\dRp+ testpub2_forschema
- Publication testpub2_forschema
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub2_forschema
+ Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | f
Tables from schemas:
"pub_test1_renamed"
"pub_test2"
ALTER SCHEMA pub_test1_renamed RENAME to pub_test1;
\dRp+ testpub2_forschema
- Publication testpub2_forschema
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub2_forschema
+ Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | f
Tables from schemas:
"pub_test1"
"pub_test2"
@@ -1392,10 +1392,10 @@ Tables from schemas:
-- alter publication add schema
ALTER PUBLICATION testpub1_forschema ADD TABLES IN SCHEMA pub_test2;
\dRp+ testpub1_forschema
- Publication testpub1_forschema
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub1_forschema
+ Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | f
Tables from schemas:
"pub_test1"
"pub_test2"
@@ -1404,10 +1404,10 @@ Tables from schemas:
ALTER PUBLICATION testpub1_forschema ADD TABLES IN SCHEMA non_existent_schema;
ERROR: schema "non_existent_schema" does not exist
\dRp+ testpub1_forschema
- Publication testpub1_forschema
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub1_forschema
+ Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | f
Tables from schemas:
"pub_test1"
"pub_test2"
@@ -1416,10 +1416,10 @@ Tables from schemas:
ALTER PUBLICATION testpub1_forschema ADD TABLES IN SCHEMA pub_test1;
ERROR: schema "pub_test1" is already member of publication "testpub1_forschema"
\dRp+ testpub1_forschema
- Publication testpub1_forschema
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub1_forschema
+ Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | f
Tables from schemas:
"pub_test1"
"pub_test2"
@@ -1427,10 +1427,10 @@ Tables from schemas:
-- alter publication drop schema
ALTER PUBLICATION testpub1_forschema DROP TABLES IN SCHEMA pub_test2;
\dRp+ testpub1_forschema
- Publication testpub1_forschema
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub1_forschema
+ Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | f
Tables from schemas:
"pub_test1"
@@ -1438,10 +1438,10 @@ Tables from schemas:
ALTER PUBLICATION testpub1_forschema DROP TABLES IN SCHEMA pub_test2;
ERROR: tables from schema "pub_test2" are not part of the publication
\dRp+ testpub1_forschema
- Publication testpub1_forschema
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub1_forschema
+ Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | f
Tables from schemas:
"pub_test1"
@@ -1449,29 +1449,29 @@ Tables from schemas:
ALTER PUBLICATION testpub1_forschema DROP TABLES IN SCHEMA non_existent_schema;
ERROR: schema "non_existent_schema" does not exist
\dRp+ testpub1_forschema
- Publication testpub1_forschema
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub1_forschema
+ Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | f
Tables from schemas:
"pub_test1"
-- drop all schemas
ALTER PUBLICATION testpub1_forschema DROP TABLES IN SCHEMA pub_test1;
\dRp+ testpub1_forschema
- Publication testpub1_forschema
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub1_forschema
+ Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | f
(1 row)
-- alter publication set multiple schema
ALTER PUBLICATION testpub1_forschema SET TABLES IN SCHEMA pub_test1, pub_test2;
\dRp+ testpub1_forschema
- Publication testpub1_forschema
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub1_forschema
+ Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | f
Tables from schemas:
"pub_test1"
"pub_test2"
@@ -1480,10 +1480,10 @@ Tables from schemas:
ALTER PUBLICATION testpub1_forschema SET TABLES IN SCHEMA non_existent_schema;
ERROR: schema "non_existent_schema" does not exist
\dRp+ testpub1_forschema
- Publication testpub1_forschema
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub1_forschema
+ Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | f
Tables from schemas:
"pub_test1"
"pub_test2"
@@ -1492,10 +1492,10 @@ Tables from schemas:
-- removing the duplicate schemas
ALTER PUBLICATION testpub1_forschema SET TABLES IN SCHEMA pub_test1, pub_test1;
\dRp+ testpub1_forschema
- Publication testpub1_forschema
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub1_forschema
+ Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | f
Tables from schemas:
"pub_test1"
@@ -1574,18 +1574,18 @@ SET client_min_messages = 'ERROR';
CREATE PUBLICATION testpub3_forschema;
RESET client_min_messages;
\dRp+ testpub3_forschema
- Publication testpub3_forschema
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub3_forschema
+ Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | f
(1 row)
ALTER PUBLICATION testpub3_forschema SET TABLES IN SCHEMA pub_test1;
\dRp+ testpub3_forschema
- Publication testpub3_forschema
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub3_forschema
+ Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | f
Tables from schemas:
"pub_test1"
@@ -1595,20 +1595,20 @@ CREATE PUBLICATION testpub_forschema_fortable FOR TABLES IN SCHEMA pub_test1, TA
CREATE PUBLICATION testpub_fortable_forschema FOR TABLE pub_test2.tbl1, TABLES IN SCHEMA pub_test1;
RESET client_min_messages;
\dRp+ testpub_forschema_fortable
- Publication testpub_forschema_fortable
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub_forschema_fortable
+ Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | f
Tables:
"pub_test2.tbl1"
Tables from schemas:
"pub_test1"
\dRp+ testpub_fortable_forschema
- Publication testpub_fortable_forschema
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub_fortable_forschema
+ Owner | All tables | Table DDLs | Inserts | Updates | Deletes | Truncates | Via root
+--------------------------+------------+------------+---------+---------+---------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | f
Tables:
"pub_test2.tbl1"
Tables from schemas:
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 09466deeea..bc75b10cf1 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -562,6 +562,7 @@ DefElemAction
DefaultACLInfo
DefineStmt
DeleteStmt
+DeparsedCommandType
DependencyGenerator
DependencyGeneratorData
DependencyType
@@ -1453,6 +1454,8 @@ LogicalDecodeBeginPrepareCB
LogicalDecodeChangeCB
LogicalDecodeCommitCB
LogicalDecodeCommitPreparedCB
+LogicalDecodeDDLMessageCB
+LogicalDecodeStreamDDLMessageCB
LogicalDecodeFilterByOriginCB
LogicalDecodeFilterPrepareCB
LogicalDecodeMessageCB
@@ -2310,6 +2313,8 @@ ReorderBufferChange
ReorderBufferChangeType
ReorderBufferCommitCB
ReorderBufferCommitPreparedCB
+ReorderBufferDDLMessageCB
+ReorderBufferStreamDDLMessageCB
ReorderBufferDiskChange
ReorderBufferIterTXNEntry
ReorderBufferIterTXNState
--
2.30.0.windows.2