From 283569252e34b0da2db397532b671e9a11258a1d Mon Sep 17 00:00:00 2001 From: Masahiko Sawada Date: Mon, 8 Jun 2026 11:25:47 -0700 Subject: [PATCH v5] 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 "messages" 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 gained the "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: 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/logical-replication.sgml | 27 +++ 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 | 6 + src/include/replication/logicalproto.h | 13 ++ src/include/replication/logicalworker.h | 42 ++++ 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 + 29 files changed, 857 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 78cce7e370d..6180b60eeb7 100644 --- a/doc/src/sgml/catalogs.sgml +++ b/doc/src/sgml/catalogs.sgml @@ -8231,6 +8231,17 @@ SCRAM-SHA-256$<iteration count>:&l + + + submessages 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 64b0e7bb972..d64a8cca536 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 + messages + enabled, where they are passed to a handler registered by an extension + on the subscriber. diff --git a/doc/src/sgml/logical-replication.sgml b/doc/src/sgml/logical-replication.sgml index 4701a3d9d18..6783ae03e38 100644 --- a/doc/src/sgml/logical-replication.sgml +++ b/doc/src/sgml/logical-replication.sgml @@ -2661,6 +2661,33 @@ CONTEXT: processing remote data for replication origin "pg_16395" during "INSER not need privileges to SET ROLE to the sequence owner. + + Logical decoding messages are an exception to the user switching described + above. If the subscription has messages + enabled, the handler registered for them runs with the privileges of the + subscription owner. Messages are not restricted to the tables or schemas + of the subscribed publications either, and by default any role may call + pg_logcal_emit_message(). + Any role that can connect to the publisher database can therefore choose + what the handler is given, and a handler that acts on what it receives does + so with the subscription owner's privileges. Restrict who can emit messages + on the publisher. Revoke the default privilege and grant it to a role created + for this purpose: + +REVOKE EXECUTE ON FUNCTION pg_logical_emit_message(boolean, text, text, boolean) FROM PUBLIC; +REVOKE EXECUTE ON FUNCTION pg_logical_emit_message(boolean, text, bytea, boolean) FROM PUBLIC; +GRANT EXECUTE ON FUNCTION pg_logical_emit_message(boolean, text, text, boolean) TO message_emitter; +GRANT EXECUTE ON FUNCTION pg_logical_emit_message(boolean, text, bytea, boolean) TO message_emitter; + + Function privileges are per database, so this has to be done in each database + that such a subscription replicates from. No role that is not trusted to drive + the handler should be a member of the new role, since a role that can + SET ROLE to it can emit whatever it likes. Where a message + has to be emitted on behalf of an ordinary user, for example from an event + trigger that captures that user's DDL, use a SECURITY DEFINER + function owned by that role rather than granting the privilege more widely. + + On the publisher, privileges are only checked once at the start of a replication connection and are not re-checked as each change record is read. diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml index 545264e8a0a..72b1b0fcd83 100644 --- a/doc/src/sgml/ref/alter_subscription.sgml +++ b/doc/src/sgml/ref/alter_subscription.sgml @@ -302,8 +302,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 + messages. 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 6248c228a64..1c38daaf434 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 + + messages (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 + by default any role may do, 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 60c15b22194..138cdecd436 100644 --- a/src/backend/catalog/pg_subscription.c +++ b/src/backend/catalog/pg_subscription.c @@ -133,6 +133,7 @@ GetSubscription(Oid subid, bool missing_ok) sub->maxretention = subform->submaxretention; sub->retentionactive = subform->subretentionactive; sub->conflictlogrelid = subform->subconflictlogrelid; + sub->messages = subform->submessages; /* Get slotname */ datum = SysCacheGetAttr(SUBSCRIPTIONOID, diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql index ad340887f54..728eacae112 100644 --- a/src/backend/catalog/system_views.sql +++ b/src/backend/catalog/system_views.sql @@ -1539,8 +1539,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, submessages, + 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 22a61dca65d..fb5b3b71831 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -84,6 +84,7 @@ #define SUBOPT_LSN 0x00020000 #define SUBOPT_ORIGIN 0x00040000 #define SUBOPT_CONFLICT_LOG_DEST 0x00080000 +#define SUBOPT_MESSAGES 0x00100000 /* check if the 'val' has 'bits' set */ #define IsSet(val, bits) (((val) & (bits)) == (bits)) @@ -110,6 +111,7 @@ typedef struct SubOpts bool runasowner; bool failover; bool retaindeadtuples; + bool messages; int32 maxretention; char *origin; ConflictLogDest conflictlogdest; @@ -208,6 +210,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_MESSAGES)) + opts->messages = false; /* Parse options */ foreach(lc, stmt_options) @@ -460,6 +464,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_MESSAGES) && + strcmp(defel->defname, "messages") == 0) + { + if (IsSet(opts->specified_opts, SUBOPT_MESSAGES)) + errorConflictingDefElem(defel, pstate); + + opts->specified_opts |= SUBOPT_MESSAGES; + opts->messages = defGetBoolean(defel); + } else ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), @@ -700,7 +713,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_MESSAGES); parse_subscription_options(pstate, stmt->options, supported_opts, &opts); /* @@ -872,6 +885,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_submessages - 1] = BoolGetDatum(opts.messages); if (stmt->conninfo) { Assert(stmt->conninfo == conninfo && !OidIsValid(serverid)); @@ -1708,7 +1722,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_MESSAGES); break; case ALTER_SUBSCRIPTION_ENABLED: @@ -2115,6 +2130,12 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, } } + if (IsSet(opts.specified_opts, SUBOPT_MESSAGES)) + { + values[Anum_pg_subscription_submessages - 1] = BoolGetDatum(opts.messages); + replaces[Anum_pg_subscription_submessages - 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 7781bb1c168..43a069c6ca0 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -560,6 +560,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); @@ -1710,6 +1713,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; + + begin_replication_step(); + + logicalrep_read_message(s, &msg); + + (*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). * @@ -3872,12 +3940,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: @@ -5163,7 +5226,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->messages != MySubscription->messages) { if (am_parallel_apply_worker()) ereport(LOG, @@ -5654,6 +5718,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->messages && + !am_tablesync_worker()); } /* diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c index 388c3b9c346..f3615d66ec2 100644 --- a/src/bin/pg_dump/pg_dump.c +++ b/src/bin/pg_dump/pg_dump.c @@ -5136,6 +5136,7 @@ getSubscriptions(Archive *fout) int i_subfailover; int i_subretaindeadtuples; int i_submaxretention; + int i_submessages; int i, ntups; @@ -5234,9 +5235,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.submessages\n"); + else + appendPQExpBufferStr(query, " false AS submessages\n"); appendPQExpBufferStr(query, "FROM pg_subscription s\n"); @@ -5285,6 +5291,7 @@ getSubscriptions(Archive *fout) i_subpublications = PQfnumber(res, "subpublications"); i_suborigin = PQfnumber(res, "suborigin"); i_suboriginremotelsn = PQfnumber(res, "suboriginremotelsn"); + i_submessages = PQfnumber(res, "submessages"); subinfo = pg_malloc_array(SubscriptionInfo, ntups); @@ -5342,6 +5349,8 @@ getSubscriptions(Archive *fout) else subinfo[i].suboriginremotelsn = pg_strdup(PQgetvalue(res, i, i_suboriginremotelsn)); + subinfo[i].submessages = + (strcmp(PQgetvalue(res, i, i_submessages), "t") == 0); /* Decide whether we want to dump it */ selectDumpableObject(&(subinfo[i].dobj), fout); @@ -5594,6 +5603,9 @@ dumpSubscription(Archive *fout, const SubscriptionInfo *subinfo) if (subinfo->subretaindeadtuples) appendPQExpBufferStr(query, ", retain_dead_tuples = true"); + if (subinfo->submessages) + appendPQExpBufferStr(query, ", messages = 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 2bbb5d5773b..4f703e40133 100644 --- a/src/bin/pg_dump/pg_dump.h +++ b/src/bin/pg_dump/pg_dump.h @@ -719,6 +719,7 @@ typedef struct _SubscriptionInfo bool subrunasowner; bool subfailover; bool subretaindeadtuples; + bool submessages; 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 1299c837063..780f6415141 100644 --- a/src/bin/pg_dump/t/002_pg_dump.pl +++ b/src/bin/pg_dump/t/002_pg_dump.pl @@ -3305,9 +3305,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, messages = 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, messages = 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 dc9b3841592..171c5c189e3 100644 --- a/src/bin/psql/describe.c +++ b/src/bin/psql/describe.c @@ -6804,7 +6804,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); @@ -6882,6 +6882,11 @@ describeSubscriptions(const char *pattern, bool verbose) gettext_noop("Retention active")); } + if (pset.sversion >= 200000) + appendPQExpBuffer(&buf, + ", submessages AS \"%s\"\n", + gettext_noop("Messages")); + 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 d2781a0b837..6465d79bee5 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 submessages; /* True if the subscription wants to receive + * logical messages. */ #ifdef CATALOG_VARLEN /* variable-length fields start here */ /* @@ -173,6 +176,9 @@ typedef struct Subscription * exceeded max_retention_duration, when * defined */ Oid conflictlogrelid; /* conflict log table Oid */ + bool messages; /* 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 */ char *walrcvtimeout; /* wal_receiver_timeout 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..c1f0c81c0df 100644 --- a/src/include/replication/logicalworker.h +++ b/src/include/replication/logicalworker.h @@ -16,6 +16,48 @@ extern PGDLLIMPORT volatile sig_atomic_t ParallelApplyMessagePending; +/* + * Forward-declared so that this header, which is included by several files + * unrelated to logical replication, does not have to include logicalproto.h. + * A handler that dereferences the struct must include that header itself. + */ +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 "messages" 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. + * + * The handler always runs with the privileges of the subscription owner. + * The payload is chosen by whoever called pg_logical_emit_message() on the + * publisher, which by default any role may do, so a handler that acts on the + * payload gives every role on the publisher the subscription owner's + * privileges. A handler that needs less than that can lower them itself + * with SwitchToUntrustedUser(). + * + * 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..6831bd1f751 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 messages */ } logical; } proto; } WalRcvStreamOptions; diff --git a/src/test/modules/Makefile b/src/test/modules/Makefile index 71a2e65ad70..c58d53fdf79 100644 --- a/src/test/modules/Makefile +++ b/src/test/modules/Makefile @@ -37,6 +37,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 77e1a2810e5..e7a0d351292 100644 --- a/src/test/modules/meson.build +++ b/src/test/modules/meson.build @@ -38,6 +38,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..a9ff6171ad6 --- /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 (messages = 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 681dc66dca6..52487149a81 100644 --- a/src/test/regress/expected/subscription.out +++ b/src/test/regress/expected/subscription.out @@ -139,18 +139,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 | Messages | 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 | Messages | 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; @@ -255,10 +255,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 | Messages | 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); @@ -267,10 +267,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 | Messages | 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); @@ -286,10 +286,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 | Messages | 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 @@ -298,10 +298,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 | Messages | 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; @@ -337,10 +337,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 | Messages | 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 @@ -369,19 +369,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 | Messages | 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 | Messages | 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; @@ -393,27 +393,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 | Messages | 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 | Messages | 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 | Messages | 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 @@ -428,10 +428,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 | Messages | 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 @@ -446,10 +446,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 | Messages | 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; @@ -485,19 +485,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 | Messages | 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 | Messages | 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); @@ -507,10 +507,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 | Messages | 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); @@ -523,18 +523,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 | Messages | 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 | Messages | 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); @@ -547,10 +547,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 | Messages | 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); @@ -567,10 +567,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 | Messages | 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 @@ -579,12 +579,38 @@ ERROR: option "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 | Messages | 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 - messages must be boolean +CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, messages = foo); +ERROR: messages requires a Boolean value +-- ok +CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, messages = 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 | Messages | 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 (messages = 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 | Messages | 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 cfeebaf9302..c7d5476fe7c 100644 --- a/src/test/regress/sql/subscription.sql +++ b/src/test/regress/sql/subscription.sql @@ -439,6 +439,23 @@ ALTER SUBSCRIPTION regress_testsub SET (max_retention_duration = 0); ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE); DROP SUBSCRIPTION regress_testsub; +-- fail - messages must be boolean +CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, messages = foo); + +-- ok +CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, messages = true); + +\dRs+ + +-- ok +ALTER SUBSCRIPTION regress_testsub SET (messages = 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 5d432074c2c..2a53289e9f6 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -1684,6 +1684,8 @@ LogicalRepBeginData LogicalRepCommitData LogicalRepCommitPreparedTxnData LogicalRepCtxStruct +LogicalRepMessageData +LogicalRepMessageHandle_hook_type LogicalRepMsgType LogicalRepPartMapEntry LogicalRepPreparedTxnData -- 2.55.0