From d5e66435d389ae0c2dac1d28cf80f8470d751299 Mon Sep 17 00:00:00 2001 From: Masahiko Sawada Date: Mon, 8 Jun 2026 11:25:47 -0700 Subject: [PATCH v3] Add a hook for handling logical messages on subscribers. Previously, pgoutput could emit logical decoding messages, those written with pg_logical_emit_message(), but the built-in subscriber never requested them from the publisher, and apply_handle_message() was only a placeholder that discarded the MESSAGE message type. Acting on such messages therefore meant writing a separate logical decoding client, and giving up the apply worker's streaming, conflict handling and progress tracking. Logical messages are a general-purpose side channel carried in band with the data, useful for replicating deparsed DDL, for carrying change-data-capture context that the row data does not have, and for cache invalidation delivered in commit order relative to the rows it refers to. This commit adds a new "message" subscription option, which asks the publisher for the pgoutput "messages" option, and dispatches each received message to a new LogicalRepMessageHandle_hook. Behavior is unchanged when the option is off or when no handler is installed. A transactional message is applied as a step of the remote transaction it belongs to, so the handler's work commits atomically with that transaction, and transaction streaming and ALTER SUBSCRIPTION ... SKIP apply to it as they do to any other change. A non-transactional message belongs to no remote transaction and is committed on its own. A message may still reach the handler more than once, for example after an apply worker restart, so handlers must be idempotent. Note also that messages are not scoped by publication: every message emitted in the publisher database is sent, subject only to origin filtering, and pg_logical_emit_message() requires no special privileges. A handler must therefore treat the payload as untrusted input, and as an arbitrary string of bytes rather than as text in the subscriber's encoding. The publisher must be running PostgreSQL 14 or later, where pgoutput gainedthe "messages" option; against an older publisher the option is accepted but no message is ever sent. Bump catalog version. Reviewed-by: Bharath Rupireddy Reviewed-by: Fujii Masao Reviewed-by: Chao Li Discussion: Discussion: https://postgr.es/m/CAD21AoCTNGiddikkUcDKj5QLnsg-51bpr-o6L-GTHWZL4ZFYtQ@mail.gmail.com --- doc/src/sgml/catalogs.sgml | 11 + doc/src/sgml/func/func-admin.sgml | 7 +- doc/src/sgml/ref/alter_subscription.sgml | 5 +- doc/src/sgml/ref/create_subscription.sgml | 43 ++++ src/backend/catalog/pg_subscription.c | 1 + src/backend/catalog/system_views.sql | 5 +- src/backend/commands/subscriptioncmds.c | 25 ++- .../libpqwalreceiver/libpqwalreceiver.c | 4 + src/backend/replication/logical/proto.c | 37 ++++ src/backend/replication/logical/worker.c | 87 +++++++- src/bin/pg_dump/pg_dump.c | 16 +- src/bin/pg_dump/pg_dump.h | 1 + src/bin/pg_dump/t/002_pg_dump.pl | 4 +- src/bin/psql/describe.c | 7 +- src/include/catalog/pg_subscription.h | 5 + src/include/replication/logicalproto.h | 13 ++ src/include/replication/logicalworker.h | 31 +++ src/include/replication/walreceiver.h | 1 + src/test/modules/Makefile | 1 + src/test/modules/meson.build | 1 + .../modules/test_logicalmsg_hooks/.gitignore | 4 + .../modules/test_logicalmsg_hooks/Makefile | 20 ++ .../modules/test_logicalmsg_hooks/meson.build | 28 +++ .../test_logicalmsg_hooks/t/001_basic.pl | 207 ++++++++++++++++++ .../test_logicalmsg_hooks.c | 140 ++++++++++++ src/test/regress/expected/subscription.out | 202 +++++++++-------- src/test/regress/sql/subscription.sql | 17 ++ src/tools/pgindent/typedefs.list | 2 + 28 files changed, 818 insertions(+), 107 deletions(-) create mode 100644 src/test/modules/test_logicalmsg_hooks/.gitignore create mode 100644 src/test/modules/test_logicalmsg_hooks/Makefile create mode 100644 src/test/modules/test_logicalmsg_hooks/meson.build create mode 100644 src/test/modules/test_logicalmsg_hooks/t/001_basic.pl create mode 100644 src/test/modules/test_logicalmsg_hooks/test_logicalmsg_hooks.c diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml index 6066c4784f4..78e6ccdf361 100644 --- a/doc/src/sgml/catalogs.sgml +++ b/doc/src/sgml/catalogs.sgml @@ -8737,6 +8737,17 @@ SCRAM-SHA-256$<iteration count>:&l + + + submessage bool + + + If true, the subscription requests the publisher to send logical + decoding messages, which are passed to a handler registered by an + extension on the subscriber. + + + subconninfo text diff --git a/doc/src/sgml/func/func-admin.sgml b/doc/src/sgml/func/func-admin.sgml index 0eae1c1f616..06005e9aa45 100644 --- a/doc/src/sgml/func/func-admin.sgml +++ b/doc/src/sgml/func/func-admin.sgml @@ -1479,7 +1479,12 @@ postgres=# SELECT '0/0'::pg_lsn + pd.segment_number * ps.setting::int + :offset false) controls if the message is immediately flushed to WAL or not. flush has no effect with transactional, as the message's WAL - record is flushed along with its transaction. + record is flushed along with its transaction. In addition to logical + decoding clients, messages emitted by this function are delivered to + subscribers whose subscriptions have + message + enabled, where they are passed to a handler registered by an extension + on the subscriber. diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml index b6f47545682..0e45e7e98ab 100644 --- a/doc/src/sgml/ref/alter_subscription.sgml +++ b/doc/src/sgml/ref/alter_subscription.sgml @@ -309,8 +309,9 @@ ALTER SUBSCRIPTION name RENAME TO < two_phase, retain_dead_tuples, max_retention_duration, - wal_receiver_timeout, and - conflict_log_destination. + wal_receiver_timeout, + conflict_log_destination, and + message. Only a superuser can set password_required = false. diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml index 25a81e2e62a..b8ed5c7816d 100644 --- a/doc/src/sgml/ref/create_subscription.sgml +++ b/doc/src/sgml/ref/create_subscription.sgml @@ -478,6 +478,49 @@ CREATE SUBSCRIPTION subscription_name + + message (boolean) + + + Specifies whether the subscription requests the publisher to send + logical decoding messages, that is, those emitted with + pg_logical_emit_message. + The default is false. + + + Received messages are only acted upon if an extension installed on + the subscriber has registered a handler for them; otherwise they are + received and discarded. A message may be passed to the handler more + than once, for example after an apply worker restart, so a handler + must be prepared to process the same message repeatedly. + + + A transactional message is applied as part of the remote transaction + that emitted it, and is therefore delivered in commit order with + respect to other transactions, not in the order in which it was written. + Note that + streaming + and SKIP + apply to transactional messages just as they do to other changes. A + non-transactional message is delivered as soon as it is decoded, even + if the transaction that emitted it later aborted. + + + Messages are not restricted to the tables or schemas of the + subscribed publications: every message emitted in the publisher + database is sent, subject only to + origin + filtering. Message contents are supplied by whoever calls + pg_logical_emit_message on the publisher, which + requires no special privileges, so a handler should treat them as + untrusted input. A message is an arbitrary string of bytes, and is + transmitted without character set conversion; it may therefore + contain embedded nulls, and need not be valid text in the encoding + of either database. + + + + origin (string) diff --git a/src/backend/catalog/pg_subscription.c b/src/backend/catalog/pg_subscription.c index 5ff61edb989..3618161ea96 100644 --- a/src/backend/catalog/pg_subscription.c +++ b/src/backend/catalog/pg_subscription.c @@ -139,6 +139,7 @@ GetSubscription(Oid subid, bool missing_ok, bool conninfo_needed, sub->maxretention = subform->submaxretention; sub->retentionactive = subform->subretentionactive; sub->conflictlogrelid = subform->subconflictlogrelid; + sub->message = subform->submessage; if (conninfo_needed) { diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql index 090281a03dd..aab999d4ff8 100644 --- a/src/backend/catalog/system_views.sql +++ b/src/backend/catalog/system_views.sql @@ -1538,8 +1538,9 @@ GRANT SELECT (oid, subdbid, subskiplsn, subname, subowner, subenabled, subbinary, substream, subtwophasestate, subdisableonerr, subpasswordrequired, subrunasowner, subfailover, subretaindeadtuples, submaxretention, subretentionactive, - subserver, subconflictlogrelid, subconflictlogdest, subslotname, - subsynccommit, subwalrcvtimeout, subpublications, suborigin) + subserver, subconflictlogrelid, subconflictlogdest, submessage, + subslotname, subsynccommit, subwalrcvtimeout, subpublications, + suborigin) ON pg_subscription TO public; CREATE VIEW pg_stat_subscription_stats AS diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index d3eef2a3efa..c510679fabf 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -83,6 +83,7 @@ #define SUBOPT_LSN 0x00020000 #define SUBOPT_ORIGIN 0x00040000 #define SUBOPT_CONFLICT_LOG_DEST 0x00080000 +#define SUBOPT_MESSAGE 0x00100000 /* check if the 'val' has 'bits' set */ #define IsSet(val, bits) (((val) & (bits)) == (bits)) @@ -109,6 +110,7 @@ typedef struct SubOpts bool runasowner; bool failover; bool retaindeadtuples; + bool message; int32 maxretention; char *origin; ConflictLogDest conflictlogdest; @@ -207,6 +209,8 @@ parse_subscription_options(ParseState *pstate, List *stmt_options, opts->origin = pstrdup(LOGICALREP_ORIGIN_ANY); if (IsSet(supported_opts, SUBOPT_CONFLICT_LOG_DEST)) opts->conflictlogdest = CONFLICT_LOG_DEST_LOG; + if (IsSet(supported_opts, SUBOPT_MESSAGE)) + opts->message = false; /* Parse options */ foreach(lc, stmt_options) @@ -459,6 +463,15 @@ parse_subscription_options(ParseState *pstate, List *stmt_options, opts->conflictlogdest = GetConflictLogDest(val); opts->specified_opts |= SUBOPT_CONFLICT_LOG_DEST; } + else if (IsSet(supported_opts, SUBOPT_MESSAGE) && + strcmp(defel->defname, "message") == 0) + { + if (IsSet(opts->specified_opts, SUBOPT_MESSAGE)) + errorConflictingDefElem(defel, pstate); + + opts->specified_opts |= SUBOPT_MESSAGE; + opts->message = defGetBoolean(defel); + } else ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), @@ -699,7 +712,7 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt, SUBOPT_RETAIN_DEAD_TUPLES | SUBOPT_MAX_RETENTION_DURATION | SUBOPT_WAL_RECEIVER_TIMEOUT | SUBOPT_ORIGIN | - SUBOPT_CONFLICT_LOG_DEST); + SUBOPT_CONFLICT_LOG_DEST | SUBOPT_MESSAGE); parse_subscription_options(pstate, stmt->options, supported_opts, &opts); /* @@ -854,6 +867,7 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt, values[Anum_pg_subscription_subretentionactive - 1] = BoolGetDatum(opts.retaindeadtuples); values[Anum_pg_subscription_subserver - 1] = ObjectIdGetDatum(serverid); + values[Anum_pg_subscription_submessage - 1] = BoolGetDatum(opts.message); if (!OidIsValid(serverid)) values[Anum_pg_subscription_subconninfo - 1] = CStringGetTextDatum(conninfo); @@ -1666,7 +1680,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, SUBOPT_MAX_RETENTION_DURATION | SUBOPT_WAL_RECEIVER_TIMEOUT | SUBOPT_ORIGIN | - SUBOPT_CONFLICT_LOG_DEST); + SUBOPT_CONFLICT_LOG_DEST | + SUBOPT_MESSAGE); break; case ALTER_SUBSCRIPTION_ENABLED: @@ -2053,6 +2068,12 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, } } + if (IsSet(opts.specified_opts, SUBOPT_MESSAGE)) + { + values[Anum_pg_subscription_submessage - 1] = BoolGetDatum(opts.message); + replaces[Anum_pg_subscription_submessage - 1] = true; + } + update_tuple = true; break; } diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c index 029990d9fce..e345b80690f 100644 --- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c +++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c @@ -631,6 +631,10 @@ libpqrcv_startstreaming(WalReceiverConn *conn, appendQuotedLiteral(&cmd, options->proto.logical.origin); } + if (options->proto.logical.messages && + PQserverVersion(conn->streamConn) >= 140000) + appendStringInfo(&cmd, ", messages 'on'"); + pubnames = options->proto.logical.publication_names; pubnames_str = stringlist_to_identifierstr(pubnames); appendStringInfoString(&cmd, ", publication_names "); diff --git a/src/backend/replication/logical/proto.c b/src/backend/replication/logical/proto.c index 86ad97cd937..6461361d68d 100644 --- a/src/backend/replication/logical/proto.c +++ b/src/backend/replication/logical/proto.c @@ -660,6 +660,43 @@ logicalrep_write_message(StringInfo out, TransactionId xid, XLogRecPtr lsn, pq_sendbytes(out, message, sz); } +/* + * Read MESSAGE from stream. + */ +void +logicalrep_read_message(StringInfo in, LogicalRepMessageData *msg_data) +{ + Size len; + char *msg; + uint8 flags; + + /* read and decode flags */ + flags = pq_getmsgint(in, 1); + msg_data->transactional = (flags & MESSAGE_TRANSACTIONAL) != 0; + + msg_data->lsn = pq_getmsgint64(in); + msg_data->prefix = pstrdup(pq_getmsgstring(in)); + + /* read message length */ + len = pq_getmsgint(in, 4); + msg_data->message_size = len; + + /* and data */ + msg = palloc(len + 1); + pq_copymsgbytes(in, msg, len); + + /* + * The payload is an arbitrary string of bytes, so message_size, not + * strlen(), is what gives its length. NULL-terminate it anyway, as + * logicalrep_read_tuple() does for column values: that way it can be + * handed to code expecting a C string without reading past the end, which + * is convenient for logging and debugging. Such code will of course stop + * at the first embedded null, if any. + */ + msg[len] = '\0'; + msg_data->message = msg; +} + /* * Write relation description to the output stream. */ diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index ce4f1b93b66..3b3d421c2b9 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -549,6 +549,9 @@ typedef struct ApplySubXactData static ApplySubXactData subxact_data = {0, 0, InvalidTransactionId, NULL}; +/* Hook for plugins to get control in handling logical decoding messages */ +LogicalRepMessageHandle_hook_type LogicalRepMessageHandle_hook = NULL; + static inline void subxact_filename(char *path, Oid subid, TransactionId xid); static inline void changes_filename(char *path, Oid subid, TransactionId xid); @@ -1694,6 +1697,71 @@ apply_handle_origin(StringInfo s) errmsg_internal("ORIGIN message sent out of order"))); } +/* + * Handle MESSAGE message. + * + * Invoke a hook function if set. + */ +static void +apply_handle_message(StringInfo s) +{ + LogicalRepMessageData msg; + + /* Tablesync worker should never receive MESSAGE */ + if (am_tablesync_worker()) + ereport(ERROR, + errcode(ERRCODE_PROTOCOL_VIOLATION), + errmsg_internal("tablesync worker received a MESSAGE message")); + + if (!LogicalRepMessageHandle_hook) + return; + + if (is_skipping_changes() || + handle_streamed_transaction(LOGICAL_REP_MSG_MESSAGE, s)) + return; + + logicalrep_read_message(s, &msg); + + begin_replication_step(); + + (*LogicalRepMessageHandle_hook) (&msg); + + end_replication_step(); + + /* + * A transactional message is applied as a step of the remote transaction + * that emitted it, and is committed together with it when applying the + * commit message. A non-transactional message belongs to no remote + * transaction, so commit it here. + */ + if (!msg.transactional) + { + Assert(!in_remote_transaction); + Assert(!in_streamed_transaction); + + /* + * The message doesn't belong to any remote transaction, so there is + * no remote commit LSN nor timestamp to record. Clear the state left + * over by the previously applied transaction so that this commit + * doesn't inherit it. + */ + replorigin_xact_clear(false); + + CommitTransactionCommand(); + + pgstat_report_stat(false); + + /* + * Report the flush position only after the local commit is durable. + * Otherwise send_feedback() would report everything received so far + * as flushed, the publisher would advance confirmed_flush_lsn past + * this message, and a crash before the commit reaches disk would lose + * the handler's work without the message ever being sent again. + */ + store_flush_position(msg.lsn, XactLastCommitEnd); + } +} + /* * Initialize fileset (if not already done). * @@ -3846,12 +3914,7 @@ apply_dispatch(StringInfo s) break; case LOGICAL_REP_MSG_MESSAGE: - - /* - * Logical replication does not use generic logical messages yet. - * Although, it could be used by other applications that use this - * output plugin. - */ + apply_handle_message(s); break; case LOGICAL_REP_MSG_STREAM_START: @@ -5128,7 +5191,8 @@ maybe_reread_subscription(void) newsub->passwordrequired != MySubscription->passwordrequired || strcmp(newsub->origin, MySubscription->origin) != 0 || newsub->owner != MySubscription->owner || - !equal(newsub->publications, MySubscription->publications)) + !equal(newsub->publications, MySubscription->publications) || + newsub->message != MySubscription->message) { if (am_parallel_apply_worker()) ereport(LOG, @@ -5614,6 +5678,15 @@ set_stream_options(WalRcvStreamOptions *options, options->proto.logical.twophase = false; options->proto.logical.origin = pstrdup(MySubscription->origin); + + /* + * Logical messages are relation-agnostic, so they don't map cleanly onto + * the tablesync worker of a particular relation, and there would be no + * well-defined ordering between a message and the initial copy. Leave + * them to the (parallel) apply workers. + */ + options->proto.logical.messages = (MySubscription->message && + !am_tablesync_worker()); } /* diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c index 4948e6d80c7..31481720168 100644 --- a/src/bin/pg_dump/pg_dump.c +++ b/src/bin/pg_dump/pg_dump.c @@ -5138,6 +5138,7 @@ getSubscriptions(Archive *fout) int i_subfailover; int i_subretaindeadtuples; int i_submaxretention; + int i_submessage; int i, ntups; @@ -5236,9 +5237,14 @@ getSubscriptions(Archive *fout) " '-1' AS subwalrcvtimeout,\n"); if (fout->remoteVersion >= 190000) - appendPQExpBufferStr(query, " fs.srvname AS subservername\n"); + appendPQExpBufferStr(query, " fs.srvname AS subservername,\n"); else - appendPQExpBufferStr(query, " NULL AS subservername\n"); + appendPQExpBufferStr(query, " NULL AS subservername,\n"); + + if (fout->remoteVersion >= 200000) + appendPQExpBufferStr(query, " s.submessage\n"); + else + appendPQExpBufferStr(query, " false AS submessage\n"); appendPQExpBufferStr(query, "FROM pg_subscription s\n"); @@ -5287,6 +5293,7 @@ getSubscriptions(Archive *fout) i_subpublications = PQfnumber(res, "subpublications"); i_suborigin = PQfnumber(res, "suborigin"); i_suboriginremotelsn = PQfnumber(res, "suboriginremotelsn"); + i_submessage = PQfnumber(res, "submessage"); subinfo = pg_malloc_array(SubscriptionInfo, ntups); @@ -5344,6 +5351,8 @@ getSubscriptions(Archive *fout) else subinfo[i].suboriginremotelsn = pg_strdup(PQgetvalue(res, i, i_suboriginremotelsn)); + subinfo[i].submessage = + (strcmp(PQgetvalue(res, i, i_submessage), "t") == 0); /* Decide whether we want to dump it */ selectDumpableObject(&(subinfo[i].dobj), fout); @@ -5596,6 +5605,9 @@ dumpSubscription(Archive *fout, const SubscriptionInfo *subinfo) if (subinfo->subretaindeadtuples) appendPQExpBufferStr(query, ", retain_dead_tuples = true"); + if (subinfo->submessage) + appendPQExpBufferStr(query, ", message = true"); + if (subinfo->submaxretention) appendPQExpBuffer(query, ", max_retention_duration = %d", subinfo->submaxretention); diff --git a/src/bin/pg_dump/pg_dump.h b/src/bin/pg_dump/pg_dump.h index 5a6726d8b12..30bd1575a26 100644 --- a/src/bin/pg_dump/pg_dump.h +++ b/src/bin/pg_dump/pg_dump.h @@ -721,6 +721,7 @@ typedef struct _SubscriptionInfo bool subrunasowner; bool subfailover; bool subretaindeadtuples; + bool submessage; int submaxretention; char *subservername; char *subconninfo; diff --git a/src/bin/pg_dump/t/002_pg_dump.pl b/src/bin/pg_dump/t/002_pg_dump.pl index 9258948b583..70fe1276d0a 100644 --- a/src/bin/pg_dump/t/002_pg_dump.pl +++ b/src/bin/pg_dump/t/002_pg_dump.pl @@ -3317,9 +3317,9 @@ my %tests = ( create_order => 50, create_sql => 'CREATE SUBSCRIPTION sub3 CONNECTION \'dbname=doesnotexist\' PUBLICATION pub1 - WITH (connect = false, origin = any, streaming = on);', + WITH (connect = false, origin = any, streaming = on, message = on);', regexp => qr/^ - \QCREATE SUBSCRIPTION sub3 CONNECTION 'dbname=doesnotexist' PUBLICATION pub1 WITH (connect = false, slot_name = 'sub3', streaming = on);\E + \QCREATE SUBSCRIPTION sub3 CONNECTION 'dbname=doesnotexist' PUBLICATION pub1 WITH (connect = false, slot_name = 'sub3', streaming = on, message = true);\E /xm, like => { %full_runs, section_post_data => 1, }, unlike => { diff --git a/src/bin/psql/describe.c b/src/bin/psql/describe.c index ad9c8affb4f..cdb6bbb2184 100644 --- a/src/bin/psql/describe.c +++ b/src/bin/psql/describe.c @@ -6894,7 +6894,7 @@ describeSubscriptions(const char *pattern, bool verbose) printQueryOpt myopt = pset.popt; static const bool translate_columns[] = {false, false, false, false, false, false, false, false, false, false, false, false, false, false, - false, false, false, false, false, false, false}; + false, false, false, false, false, false, false, false}; initPQExpBuffer(&buf); @@ -6972,6 +6972,11 @@ describeSubscriptions(const char *pattern, bool verbose) gettext_noop("Retention active")); } + if (pset.sversion >= 200000) + appendPQExpBuffer(&buf, + ", submessage AS \"%s\"\n", + gettext_noop("Message")); + appendPQExpBuffer(&buf, ", subsynccommit AS \"%s\"\n" ", subconninfo AS \"%s\"\n", diff --git a/src/include/catalog/pg_subscription.h b/src/include/catalog/pg_subscription.h index 65ce8e145fb..7c0df9827bf 100644 --- a/src/include/catalog/pg_subscription.h +++ b/src/include/catalog/pg_subscription.h @@ -96,6 +96,9 @@ CATALOG(pg_subscription,6100,SubscriptionRelationId) BKI_SHARED_RELATION BKI_ROW * server */ Oid subconflictlogrelid; /* Relid of the conflict log table. */ + + bool submessage; /* True if the subscription wants to receive + * logical messages. */ #ifdef CATALOG_VARLEN /* variable-length fields start here */ /* @@ -173,6 +176,8 @@ typedef struct Subscription * exceeded max_retention_duration, when * defined */ Oid conflictlogrelid; /* conflict log table Oid */ + bool message; /* True if the subscription wants to receive + * logical messages */ char *conninfo; /* Connection string to the publisher */ char *slotname; /* Name of the replication slot */ char *synccommit; /* Synchronous commit setting for worker */ diff --git a/src/include/replication/logicalproto.h b/src/include/replication/logicalproto.h index 058a955e20c..83741a2080d 100644 --- a/src/include/replication/logicalproto.h +++ b/src/include/replication/logicalproto.h @@ -191,6 +191,18 @@ typedef struct LogicalRepStreamAbortData TimestampTz abort_time; } LogicalRepStreamAbortData; +/* + * Logical decoding message information + */ +typedef struct LogicalRepMessageData +{ + XLogRecPtr lsn; + bool transactional; + const char *prefix; + Size message_size; /* length of message in bytes */ + const char *message; /* payload; may contain embedded nulls */ +} LogicalRepMessageData; + extern void logicalrep_write_begin(StringInfo out, ReorderBufferTXN *txn); extern void logicalrep_read_begin(StringInfo in, LogicalRepBeginData *begin_data); @@ -248,6 +260,7 @@ 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_read_message(StringInfo in, LogicalRepMessageData *msg_data); extern void logicalrep_write_rel(StringInfo out, TransactionId xid, Relation rel, Bitmapset *columns, PublishGencolsType include_gencols_type); diff --git a/src/include/replication/logicalworker.h b/src/include/replication/logicalworker.h index 7d748a28da8..6a68c47f12b 100644 --- a/src/include/replication/logicalworker.h +++ b/src/include/replication/logicalworker.h @@ -16,6 +16,37 @@ extern PGDLLIMPORT volatile sig_atomic_t ParallelApplyMessagePending; +/* avoid include logicalproto.h */ +struct LogicalRepMessageData; + +/* + * Hook for extensions to handle logical decoding messages (see + * pg_logical_emit_message()) received from the publisher. Called only when + * the subscription has the "message" option enabled. + * + * The handler runs in the apply worker, inside a transaction, with an active + * snapshot pushed. For a transactional message the handler's work is part of + * the remote transaction and commits with it; a non-transactional message is + * committed as an independent unit. The same message may be delivered more + * than once, so the handler must be idempotent. + * + * The payload is an arbitrary string of bytes. The WAL record makes no + * distinction between the text and the bytea form of + * pg_logical_emit_message(), and the bytes are transmitted without character + * set conversion, so the payload may contain embedded nulls and bytes that + * are not valid in the subscriber's encoding. Its length is given by + * msg->message_size, not by strlen(); the buffer is null-terminated only for + * convenience, and that null is not part of the payload. + * + * Note that the apply worker runs with an empty search_path. An ERROR thrown + * by the handler restarts the apply worker, or disables the subscription if + * disable_on_error is set. Because the message is delivered again after the + * restart, a handler cannot reject a message by throwing an error; that only + * loops. A message the handler does not want must simply be ignored. + */ +typedef void (*LogicalRepMessageHandle_hook_type) (const struct LogicalRepMessageData *msg); +extern PGDLLIMPORT LogicalRepMessageHandle_hook_type LogicalRepMessageHandle_hook; + extern void ApplyWorkerMain(Datum main_arg); extern void ParallelApplyWorkerMain(Datum main_arg); extern void TableSyncWorkerMain(Datum main_arg); diff --git a/src/include/replication/walreceiver.h b/src/include/replication/walreceiver.h index 5cfe90f9989..eca847c4a8f 100644 --- a/src/include/replication/walreceiver.h +++ b/src/include/replication/walreceiver.h @@ -188,6 +188,7 @@ typedef struct * prepare time */ char *origin; /* Only publish data originating from the * specified origin */ + bool messages; /* Logical message */ } logical; } proto; } WalRcvStreamOptions; diff --git a/src/test/modules/Makefile b/src/test/modules/Makefile index 098bb8142ae..e8a7f50eb38 100644 --- a/src/test/modules/Makefile +++ b/src/test/modules/Makefile @@ -36,6 +36,7 @@ SUBDIRS = \ test_integerset \ test_json_parser \ test_lfind \ + test_logicalmsg_hooks \ test_lwlock_tranches \ test_misc \ test_oat_hooks \ diff --git a/src/test/modules/meson.build b/src/test/modules/meson.build index 4bca42bb370..5f6c0cae77a 100644 --- a/src/test/modules/meson.build +++ b/src/test/modules/meson.build @@ -37,6 +37,7 @@ subdir('test_int128') subdir('test_integerset') subdir('test_json_parser') subdir('test_lfind') +subdir('test_logicalmsg_hooks') subdir('test_lwlock_tranches') subdir('test_misc') subdir('test_oat_hooks') diff --git a/src/test/modules/test_logicalmsg_hooks/.gitignore b/src/test/modules/test_logicalmsg_hooks/.gitignore new file mode 100644 index 00000000000..5dcb3ff9723 --- /dev/null +++ b/src/test/modules/test_logicalmsg_hooks/.gitignore @@ -0,0 +1,4 @@ +# Generated subdirectories +/log/ +/results/ +/tmp_check/ diff --git a/src/test/modules/test_logicalmsg_hooks/Makefile b/src/test/modules/test_logicalmsg_hooks/Makefile new file mode 100644 index 00000000000..af129009b8a --- /dev/null +++ b/src/test/modules/test_logicalmsg_hooks/Makefile @@ -0,0 +1,20 @@ +# src/test/modules/test_logicalmsg_hooks/Makefile + +MODULE_big = test_logicalmsg_hooks +OBJS = \ + $(WIN32RES) \ + test_logicalmsg_hooks.o +PGFILEDESC = "test_logicalmsg_hooks - test logical replication message hooks" + +TAP_TESTS = 1 + +ifdef USE_PGXS +PG_CONFIG = pg_config +PGXS := $(shell $(PG_CONFIG) --pgxs) +include $(PGXS) +else +subdir = src/test/modules/test_logicalmsg_hooks +top_builddir = ../../../.. +include $(top_builddir)/src/Makefile.global +include $(top_srcdir)/contrib/contrib-global.mk +endif diff --git a/src/test/modules/test_logicalmsg_hooks/meson.build b/src/test/modules/test_logicalmsg_hooks/meson.build new file mode 100644 index 00000000000..8084d82e47b --- /dev/null +++ b/src/test/modules/test_logicalmsg_hooks/meson.build @@ -0,0 +1,28 @@ +# Copyright (c) 2022-2026, PostgreSQL Global Development Group + +test_logicalmsg_hooks_sources = files( + 'test_logicalmsg_hooks.c', +) + +if host_system == 'windows' + test_logicalmsg_hooks_sources += rc_lib_gen.process(win32ver_rc, extra_args: [ + '--NAME', 'test_logicalmsg_hooks', + '--FILEDESC', 'test_logicalmsg_hooks - test logical message hooks',]) +endif + +test_logicalmsg_hooks = shared_module('test_logicalmsg_hooks', + test_logicalmsg_hooks_sources, + kwargs: pg_test_mod_args, +) +test_install_libs += test_logicalmsg_hooks + +tests += { + 'name': 'test_logicalmsg_hooks', + 'sd': meson.current_source_dir(), + 'bd': meson.current_build_dir(), + 'tap': { + 'tests': [ + 't/001_basic.pl', + ], + }, +} diff --git a/src/test/modules/test_logicalmsg_hooks/t/001_basic.pl b/src/test/modules/test_logicalmsg_hooks/t/001_basic.pl new file mode 100644 index 00000000000..65d246c5031 --- /dev/null +++ b/src/test/modules/test_logicalmsg_hooks/t/001_basic.pl @@ -0,0 +1,207 @@ +# Copyright (c) 2026, PostgreSQL Global Development Group + +# Test LogicalRepMessageHandle_hook. The test module records every message it +# receives into public.test_logicalmsg_log on the subscriber, so most checks +# here are plain queries against that table. + +use strict; +use warnings FATAL => 'all'; + +use PostgreSQL::Test::Cluster; +use PostgreSQL::Test::Utils; +use Test::More; + +my $pub = PostgreSQL::Test::Cluster->new('publisher'); +$pub->init(allows_streaming => 'logical'); + +# Keep the memory limit low so that the transactions below are streamed. +$pub->append_conf('postgresql.conf', "logical_decoding_work_mem = 64kB\n"); +$pub->start; + +my $sub = PostgreSQL::Test::Cluster->new('subscriber'); +$sub->init(allows_streaming => 'logical'); +$sub->append_conf( + 'postgresql.conf', q{ +shared_preload_libraries = 'test_logicalmsg_hooks' +max_parallel_apply_workers_per_subscription = 2 +}); +$sub->start; + +$pub->safe_psql( + 'postgres', qq{ +CREATE TABLE test (a int); +INSERT INTO test VALUES (1); +CREATE PUBLICATION pub FOR ALL TABLES; +}); + +my $pub_connstr = $pub->connstr . ' dbname=postgres'; + +$sub->safe_psql( + 'postgres', qq{ +CREATE TABLE test (a int primary key); +CREATE SUBSCRIPTION sub CONNECTION '$pub_connstr' PUBLICATION pub WITH (message = true, disable_on_error = true) +}); + +$sub->wait_for_subscription_sync($pub, 'sub'); + +# Number of messages recorded under the given prefix. +sub logged_count +{ + my ($prefix) = @_; + + return $sub->safe_psql('postgres', + "SELECT count(*) FROM test_logicalmsg_log WHERE prefix = '$prefix'"); +} + +# Verify that a transactional message is passed to the hook, and that the +# message data given to the hook is correct. +$pub->safe_psql('postgres', + q{SELECT pg_logical_emit_message(true, 'basic', 'transactional message')} +); +$pub->wait_for_catchup('sub'); + +is( $sub->safe_psql( + 'postgres', + q{SELECT transactional, message_size, message FROM test_logicalmsg_log + WHERE prefix = 'basic'}), + 't|21|transactional message', + "transactional message is passed to the hook"); + +# Verify that a non-transactional message is applied even if the transaction +# that emitted it rolls back. +$pub->safe_psql( + 'postgres', q{ +BEGIN; +SELECT pg_logical_emit_message(false, 'nontransactional', 'emitted then rolled back'); +ROLLBACK; +}); + +# Commit another message to flush WAL, as ROLLBACK doesn't flush. +$pub->safe_psql('postgres', + q{SELECT pg_logical_emit_message(true, 'flush', 'flush')}); +$pub->wait_for_catchup('sub'); + +is( $sub->safe_psql( + 'postgres', + q{SELECT transactional, message FROM test_logicalmsg_log + WHERE prefix = 'nontransactional'}), + 'f|emitted then rolled back', + "non-transactional message of an aborted transaction is applied"); + +# Verify that a transactional message emitted in a subtransaction that is +# rolled back is not passed to the hook. +$pub->safe_psql( + 'postgres', q{ +BEGIN; +SAVEPOINT sp; +SELECT pg_logical_emit_message(true, 'subxact_aborted', 'not delivered'); +ROLLBACK TO SAVEPOINT sp; +SELECT pg_logical_emit_message(true, 'subxact_committed', 'delivered'); +COMMIT; +}); +$pub->wait_for_catchup('sub'); + +is(logged_count('subxact_aborted'), + 0, "message of an aborted subtransaction is not applied"); +is(logged_count('subxact_committed'), + 1, "message of a committed subtransaction is applied"); + +# Verify that transactional messages are applied in streaming transactions, +# both when the leader apply worker serializes the transaction to a file and +# when it hands the transaction to a parallel apply worker. +foreach my $mode ('on', 'parallel') +{ + $sub->safe_psql('postgres', + "ALTER SUBSCRIPTION sub SET (streaming = $mode)"); + + # Reconnect so that the new streaming mode takes effect immediately. + $sub->safe_psql('postgres', 'ALTER SUBSCRIPTION sub DISABLE'); + $sub->safe_psql('postgres', 'ALTER SUBSCRIPTION sub ENABLE'); + + $pub->safe_psql('postgres', + q{SELECT pg_stat_reset_replication_slot('sub')}); + + $pub->safe_psql( + 'postgres', qq{ +BEGIN; +INSERT INTO test SELECT generate_series(1000, 6000); +SELECT pg_logical_emit_message(true, 'streaming_$mode', 'streamed message'); +COMMIT; +}); + $pub->wait_for_catchup('sub'); + + # Make sure the transaction really was streamed, so that this test cannot + # silently degrade into a non-streaming one. + $pub->poll_query_until('postgres', + q{SELECT stream_txns > 0 FROM pg_stat_replication_slots WHERE slot_name = 'sub'} + ) or die "transaction was not streamed"; + + is(logged_count("streaming_$mode"), + 1, + "message in a streamed transaction is applied (streaming = $mode)"); + + $pub->safe_psql('postgres', 'TRUNCATE test'); + $pub->wait_for_catchup('sub'); +} + +$sub->safe_psql('postgres', 'ALTER SUBSCRIPTION sub SET (streaming = off)'); + +# Verify that ALTER SUBSCRIPTION ... SKIP skips a transactional message along +# with the rest of its transaction. The message is emitted before the +# conflicting change, so the hook runs and records the message on the first +# attempt; that work must be rolled back with the failed transaction, and must +# not be redone once the transaction is skipped. +$pub->safe_psql('postgres', 'INSERT INTO test VALUES (1)'); +$pub->wait_for_catchup('sub'); + +my $log_location = -s $sub->logfile; + +$pub->safe_psql( + 'postgres', q{ +BEGIN; +SELECT pg_logical_emit_message(true, 'skipped', 'rolled back, then skipped'); +INSERT INTO test VALUES (1); +COMMIT; +}); + +# Wait until the conflict disables the subscription. +$sub->poll_query_until('postgres', + q{SELECT subenabled = FALSE FROM pg_subscription WHERE subname = 'sub'}); + +# The hook did run before the transaction failed. +ok( $sub->log_contains( + qr/LOG[^\n]+received message: [^\n]+prefix: skipped/, + $log_location), + "hook is called for a message in a transaction that later fails"); + +# ... but its work was rolled back with the transaction. +is(logged_count('skipped'), 0, + "work done by the hook is rolled back with the remote transaction"); + +# Get the finish LSN of the failed transaction. +my $contents = slurp_file($sub->logfile, $log_location); +$contents =~ + qr/conflict detected on relation "public.test".*\n.*DETAIL:.*Could not apply remote change.*\n.*Key already exists in unique index "test_pkey", modified in transaction \d+: key .*, local row .*\n.*CONTEXT:.* for replication target relation "public.test" in transaction \d+, finished at ([[:xdigit:]]+\/[[:xdigit:]]+)/m + or die "could not get error-LSN"; +my $lsn = $1; + +$log_location = -s $sub->logfile; + +# Set skip LSN and re-enable the subscription. +$sub->safe_psql('postgres', qq{ALTER SUBSCRIPTION sub SKIP (lsn = '$lsn')}); +$sub->safe_psql('postgres', 'ALTER SUBSCRIPTION sub ENABLE'); + +# Wait for the failed transaction to be skipped. +$sub->poll_query_until('postgres', + q{SELECT subskiplsn = '0/0' FROM pg_subscription WHERE subname = 'sub'}); + +ok( !$sub->log_contains( + qr/LOG[^\n]+received message: [^\n]+prefix: skipped/, + $log_location), + "hook is not called for a message in a skipped transaction"); +is(logged_count('skipped'), 0, + "message in a skipped transaction is not applied"); + +$sub->stop; +$pub->stop; +done_testing(); diff --git a/src/test/modules/test_logicalmsg_hooks/test_logicalmsg_hooks.c b/src/test/modules/test_logicalmsg_hooks/test_logicalmsg_hooks.c new file mode 100644 index 00000000000..d8d6fc7c619 --- /dev/null +++ b/src/test/modules/test_logicalmsg_hooks/test_logicalmsg_hooks.c @@ -0,0 +1,140 @@ +/*-------------------------------------------------------------------------- + * + * test_logicalmsg_hooks.c + * Code for testing LogicalRepMessageHandle_hook + * + * The handler records every logical decoding message it receives into the + * table public.test_logicalmsg_log, so that tests can assert on the contents + * of a table rather than on the server log. Recording the messages this way + * also exercises the transactional behavior of the hook: a transactional + * message is recorded as part of the remote transaction that emitted it, and + * therefore disappears together with it if that transaction is rolled back or + * skipped, while a non-transactional message is recorded independently. + * + * The handler creates that table itself on the first message, so a test does + * not have to set it up. For a transactional message the creation is part of + * the remote transaction too, and is rolled back with it, so the table does + * not exist until some message has been applied successfully; a test must not + * query it before then. + * + * The messages are also logged, which lets a test tell "the handler ran but + * its work was rolled back" apart from "the handler was never called". + * + * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group + * Portions Copyright (c) 1994, Regents of the University of California + * + * IDENTIFICATION + * src/test/modules/test_logicalmsg_hooks/test_logicalmsg_hooks.c + * + * ------------------------------------------------------------------------- + */ + +#include "postgres.h" + +#include "catalog/namespace.h" +#include "catalog/pg_type.h" +#include "executor/spi.h" +#include "replication/logicalproto.h" +#include "replication/logicalworker.h" +#include "utils/builtins.h" +#include "utils/lsyscache.h" +#include "utils/pg_lsn.h" + +PG_MODULE_MAGIC; + +#define LOG_SCHEMA "public" +#define LOG_TABLE "test_logicalmsg_log" + +static LogicalRepMessageHandle_hook_type prev_logical_message_handler = NULL; +static void test_logical_message_handler(const LogicalRepMessageData *msg); + +/* + * Module load callback + */ +void +_PG_init(void) +{ + prev_logical_message_handler = LogicalRepMessageHandle_hook; + LogicalRepMessageHandle_hook = &test_logical_message_handler; +} + +/* + * Create the log table if not exists. + */ +static void +ensure_log_table_exists(void) +{ + Oid relid; + int ret; + + relid = get_relname_relid(LOG_TABLE, get_namespace_oid(LOG_SCHEMA, true)); + if (OidIsValid(relid)) + return; + + SPI_connect(); + + ret = SPI_execute("CREATE TABLE " LOG_SCHEMA "." LOG_TABLE + "(lsn pg_lsn, transactional boolean, prefix text, message_size int, message text)", + false, 0); + if (ret != SPI_OK_UTILITY) + elog(ERROR, "could not create " LOG_SCHEMA "." LOG_TABLE); + + SPI_finish(); +} + +static void +test_logical_message_handler(const LogicalRepMessageData *msg) +{ + Oid argtypes[5] = {LSNOID, BOOLOID, TEXTOID, INT4OID, TEXTOID}; + Datum values[5]; + int ret; + + /* + * Give a handler installed before ours its turn, so that several + * extensions can act on the same message. + */ + if (prev_logical_message_handler) + (*prev_logical_message_handler) (msg); + + ereport(LOG, + (errmsg("received message: LSN %X/%08X, prefix: %s, size: %zu, transactional: %d", + LSN_FORMAT_ARGS(msg->lsn), msg->prefix, + msg->message_size, msg->transactional))); + + ensure_log_table_exists(); + + /* + * Pass the message as query parameters rather than interpolating it into + * the query text. Message contents come from the publisher and are not to + * be trusted, and the payload may contain characters that would need + * quoting. Note also that message_size, not strlen(), is authoritative + * for the payload length. + * + * The payload is recorded as text because the accompanying test only ever + * emits text messages. A handler must not assume that in general: a + * payload is an arbitrary string of bytes, which may contain embedded + * nulls and bytes that are not valid in the subscriber's encoding. See + * the comments on LogicalRepMessageHandle_hook_type. + */ + values[0] = LSNGetDatum(msg->lsn); + values[1] = BoolGetDatum(msg->transactional); + values[2] = CStringGetTextDatum(msg->prefix); + values[3] = Int32GetDatum((int32) msg->message_size); + values[4] = PointerGetDatum(cstring_to_text_with_len(msg->message, + msg->message_size)); + + SPI_connect(); + + /* + * The apply worker runs with an empty search_path, so the table name must + * be schema-qualified. + */ + ret = SPI_execute_with_args("INSERT INTO " LOG_SCHEMA "." LOG_TABLE + " (lsn, transactional, prefix, message_size, message)" + " VALUES ($1, $2, $3, $4, $5)", + 5, argtypes, values, NULL, false, 0); + if (ret != SPI_OK_INSERT) + elog(ERROR, "could not record logical message: SPI returned %d", ret); + + SPI_finish(); +} diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out index 259db747334..83642fc03ee 100644 --- a/src/test/regress/expected/subscription.out +++ b/src/test/regress/expected/subscription.out @@ -143,18 +143,18 @@ CREATE SUBSCRIPTION regress_testsub4 CONNECTION 'dbname=regress_doesnotexist' PU WARNING: subscription was created, but is not connected HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications. \dRs+ regress_testsub4 - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description -------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- - regress_testsub4 | regress_subscription_user | f | {testpub} | f | parallel | d | f | none | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub4 | regress_subscription_user | f | {testpub} | f | parallel | d | f | none | t | f | f | | f | 0 | f | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | (1 row) ALTER SUBSCRIPTION regress_testsub4 SET (origin = any); \dRs+ regress_testsub4 - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description -------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- - regress_testsub4 | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub4 | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | (1 row) DROP SUBSCRIPTION regress_testsub3; @@ -258,10 +258,10 @@ ALTER SUBSCRIPTION regress_testsub CONNECTION 'foobar'; ERROR: invalid connection string syntax: missing "=" after "foobar" in connection info string \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------------- - regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | test subscription + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+------------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | test subscription (1 row) ALTER SUBSCRIPTION regress_testsub SET PUBLICATION testpub2, testpub3 WITH (refresh = false); @@ -270,10 +270,10 @@ ALTER SUBSCRIPTION regress_testsub SET (slot_name = 'newname'); ALTER SUBSCRIPTION regress_testsub SET (password_required = false); ALTER SUBSCRIPTION regress_testsub SET (run_as_owner = true); \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+------------------------------+------------------+------------+------------------- - regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | f | t | f | | f | 0 | f | off | dbname=regress_doesnotexist2 | -1 | 0/00000000 | test subscription + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+------------------------------+------------------+------------+------------------- + regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | f | t | f | | f | 0 | f | f | off | dbname=regress_doesnotexist2 | -1 | 0/00000000 | test subscription (1 row) ALTER SUBSCRIPTION regress_testsub SET (password_required = true); @@ -289,10 +289,10 @@ ERROR: unrecognized subscription parameter: "create_slot" -- ok ALTER SUBSCRIPTION regress_testsub SKIP (lsn = '0/12345'); \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+------------------------------+------------------+------------+------------------- - regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist2 | -1 | 0/00012345 | test subscription + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+------------------------------+------------------+------------+------------------- + regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | t | f | f | | f | 0 | f | f | off | dbname=regress_doesnotexist2 | -1 | 0/00012345 | test subscription (1 row) -- ok - with lsn = NONE @@ -301,10 +301,10 @@ ALTER SUBSCRIPTION regress_testsub SKIP (lsn = NONE); ALTER SUBSCRIPTION regress_testsub SKIP (lsn = '0/0'); ERROR: invalid WAL location (LSN): 0/0 \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+------------------------------+------------------+------------+------------------- - regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist2 | -1 | 0/00000000 | test subscription + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+------------------------------+------------------+------------+------------------- + regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | t | f | f | | f | 0 | f | f | off | dbname=regress_doesnotexist2 | -1 | 0/00000000 | test subscription (1 row) BEGIN; @@ -340,10 +340,10 @@ ALTER SUBSCRIPTION regress_testsub_foo SET (wal_receiver_timeout = '80s'); ALTER SUBSCRIPTION regress_testsub_foo SET (wal_receiver_timeout = 'foobar'); ERROR: invalid value for parameter "wal_receiver_timeout": "foobar" \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ----------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+------------------------------+------------------+------------+------------------- - regress_testsub_foo | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | t | f | f | | f | 0 | f | local | dbname=regress_doesnotexist2 | 80s | 0/00000000 | test subscription + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +---------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+------------------------------+------------------+------------+------------------- + regress_testsub_foo | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | t | f | f | | f | 0 | f | f | local | dbname=regress_doesnotexist2 | 80s | 0/00000000 | test subscription (1 row) -- rename back to keep the rest simple @@ -372,19 +372,19 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB WARNING: subscription was created, but is not connected HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications. \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- - regress_testsub | regress_subscription_user | f | {testpub} | t | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub | regress_subscription_user | f | {testpub} | t | parallel | d | f | any | t | f | f | | f | 0 | f | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | (1 row) ALTER SUBSCRIPTION regress_testsub SET (binary = false); ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE); \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- - regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | (1 row) DROP SUBSCRIPTION regress_testsub; @@ -396,27 +396,27 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB WARNING: subscription was created, but is not connected HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications. \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- - regress_testsub | regress_subscription_user | f | {testpub} | f | on | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | on | d | f | any | t | f | f | | f | 0 | f | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | (1 row) ALTER SUBSCRIPTION regress_testsub SET (streaming = parallel); \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- - regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | (1 row) ALTER SUBSCRIPTION regress_testsub SET (streaming = false); ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE); \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- - regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | | f | 0 | f | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | (1 row) -- fail - publication already exists @@ -431,10 +431,10 @@ ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub2 WITH (refr ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub2 WITH (refresh = false); ERROR: publication "testpub1" is already in subscription "regress_testsub" \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+-----------------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- - regress_testsub | regress_subscription_user | f | {testpub,testpub1,testpub2} | f | off | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+-----------------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub | regress_subscription_user | f | {testpub,testpub1,testpub2} | f | off | d | f | any | t | f | f | | f | 0 | f | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | (1 row) -- fail - publication used more than once @@ -449,10 +449,10 @@ ERROR: publication "testpub3" is not in subscription "regress_testsub" -- ok - delete publications ALTER SUBSCRIPTION regress_testsub DROP PUBLICATION testpub1, testpub2 WITH (refresh = false); \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- - regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | | f | 0 | f | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | (1 row) DROP SUBSCRIPTION regress_testsub; @@ -488,19 +488,19 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB WARNING: subscription was created, but is not connected HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications. \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- - regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | p | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | p | f | any | t | f | f | | f | 0 | f | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | (1 row) -- we can alter streaming when two_phase enabled ALTER SUBSCRIPTION regress_testsub SET (streaming = true); \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- - regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | t | f | f | | f | 0 | f | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | (1 row) ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE); @@ -510,10 +510,10 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB WARNING: subscription was created, but is not connected HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications. \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- - regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | t | f | f | | f | 0 | f | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | (1 row) ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE); @@ -526,18 +526,18 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB WARNING: subscription was created, but is not connected HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications. \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- - regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | (1 row) ALTER SUBSCRIPTION regress_testsub SET (disable_on_error = true); \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- - regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | t | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | t | any | t | f | f | | f | 0 | f | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | (1 row) ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE); @@ -550,10 +550,10 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB WARNING: subscription was created, but is not connected HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications. \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- - regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | (1 row) ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE); @@ -570,10 +570,10 @@ NOTICE: max_retention_duration is ineffective when retain_dead_tuples is disabl WARNING: subscription was created, but is not connected HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications. \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- - regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 1000 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 1000 | f | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | (1 row) -- fail - max_retention_duration must be non-negative @@ -582,12 +582,38 @@ ERROR: max_retention_duration cannot be negative -- ok ALTER SUBSCRIPTION regress_testsub SET (max_retention_duration = 0); \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- - regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | +(1 row) + +ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE); +DROP SUBSCRIPTION regress_testsub; +-- fail - message must be boolean +CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, message = foo); +ERROR: message requires a Boolean value +-- ok +CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, message = true); +WARNING: subscription was created, but is not connected +HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications. +\dRs+ + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | t | off | dbname=regress_doesnotexist | -1 | 0/00000000 | +(1 row) + +-- ok +ALTER SUBSCRIPTION regress_testsub SET (message = false); +\dRs+ + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | (1 row) +-- cleanup ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE); DROP SUBSCRIPTION regress_testsub; -- let's do some tests with pg_create_subscription rather than superuser diff --git a/src/test/regress/sql/subscription.sql b/src/test/regress/sql/subscription.sql index 7718c742974..55682f255d1 100644 --- a/src/test/regress/sql/subscription.sql +++ b/src/test/regress/sql/subscription.sql @@ -441,6 +441,23 @@ ALTER SUBSCRIPTION regress_testsub SET (max_retention_duration = 0); ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE); DROP SUBSCRIPTION regress_testsub; +-- fail - message must be boolean +CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, message = foo); + +-- ok +CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, message = true); + +\dRs+ + +-- ok +ALTER SUBSCRIPTION regress_testsub SET (message = false); + +\dRs+ + +-- cleanup +ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE); +DROP SUBSCRIPTION regress_testsub; + -- let's do some tests with pg_create_subscription rather than superuser SET SESSION AUTHORIZATION regress_subscription_user3; diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index 85d989f395d..6b452f8c267 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -1707,6 +1707,8 @@ LogicalRepBeginData LogicalRepCommitData LogicalRepCommitPreparedTxnData LogicalRepCtxStruct +LogicalRepMessageData +LogicalRepMessageHandle_hook_type LogicalRepMsgType LogicalRepPartMapEntry LogicalRepPreparedTxnData -- 2.55.0