From 12bcc34a6a47bdd04aa994f8a790155cf45b4fb6 Mon Sep 17 00:00:00 2001 From: Masahiko Sawada Date: Mon, 8 Jun 2026 11:25:47 -0700 Subject: [PATCH v2] Add a hook for handling logical messages on subscribers. pgoutput can emit logical decoding messages (those written with pg_logical_emit_message()), but the built-in subscriber never requested them from the publisher, and the apply worker had only a placeholder that discarded the MESSAGE message type. Add a "message" subscription option (default false) that asks the publisher to send these messages, and dispatch each received message to a new LogicalRepMessageHandle_hook. This lets an extension act on messages on the subscriber while still relying on the built-in apply worker for streaming, conflict handling, and progress tracking, instead of writing a separate logical decoding client. When the option is off or no handler is installed, behavior is unchanged. Logical messages are a general-purpose, in-band side channel, so the possible consumers are varied: replicating DDL by emitting deparsed statements on the publisher and executing them on the subscriber; change data capture and auditing, where a message carries context (the acting user, request id, or business reason) that the row data does not; and cache invalidation or event notification, where the message is delivered in commit order relative to the data it refers to. A transactional message is applied as a step of the remote transaction it belongs to and commits atomically with it; a non-transactional message is committed as an independent unit without advancing replication progress. Either way a message may be delivered more than once (for example after an apply worker restart), so handlers must be idempotent. Transactional messages also honor transaction streaming and ALTER SUBSCRIPTION ... SKIP. XXX Bump catalog version for the new pg_subscription.submessage column. Discussion: https://www.postgresql.org/message-id/CAD21AoCTNGiddikkUcDKj5QLnsg-51bpr-o6L-GTHWZL4ZFYtQ%40mail.gmail.com --- doc/src/sgml/catalogs.sgml | 11 + doc/src/sgml/func/func-admin.sgml | 7 +- doc/src/sgml/ref/alter_subscription.sgml | 5 +- doc/src/sgml/ref/create_subscription.sgml | 40 ++++ src/backend/catalog/pg_subscription.c | 1 + src/backend/catalog/system_views.sql | 3 +- src/backend/commands/subscriptioncmds.c | 25 +- .../libpqwalreceiver/libpqwalreceiver.c | 4 + src/backend/replication/logical/proto.c | 29 +++ src/backend/replication/logical/worker.c | 59 ++++- src/bin/pg_dump/pg_dump.c | 16 +- src/bin/pg_dump/pg_dump.h | 1 + src/bin/pg_dump/t/002_pg_dump.pl | 4 +- src/bin/psql/describe.c | 7 +- src/include/catalog/pg_subscription.h | 5 + src/include/replication/logicalproto.h | 13 ++ src/include/replication/walreceiver.h | 1 + src/include/replication/worker_internal.h | 19 ++ 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 | 220 ++++++++++++++++++ .../test_logicalmsg_hooks.c | 104 +++++++++ src/test/regress/expected/subscription.out | 202 +++++++++------- src/test/regress/sql/subscription.sql | 17 ++ src/tools/pgindent/typedefs.list | 1 + 28 files changed, 742 insertions(+), 106 deletions(-) create mode 100644 src/test/modules/test_logicalmsg_hooks/.gitignore create mode 100644 src/test/modules/test_logicalmsg_hooks/Makefile create mode 100644 src/test/modules/test_logicalmsg_hooks/meson.build create mode 100644 src/test/modules/test_logicalmsg_hooks/t/001_basic.pl create mode 100644 src/test/modules/test_logicalmsg_hooks/test_logicalmsg_hooks.c diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml index 6066c4784f4..50c57519acd 100644 --- a/doc/src/sgml/catalogs.sgml +++ b/doc/src/sgml/catalogs.sgml @@ -8702,6 +8702,17 @@ SCRAM-SHA-256$<iteration count>:&l + + + submessage bool + + + If true, the subscription requests the publisher to send logical + decoding messages, which are passed to a handler registered by an + extension on the subscriber. + + + submaxretention int4 diff --git a/doc/src/sgml/func/func-admin.sgml b/doc/src/sgml/func/func-admin.sgml index 0eae1c1f616..3d42500f33a 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 + subscriber whose subscription has + message + enabled, where they are passed to a handler registered by an extension + on the subscriber. diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml index 0f81af5608b..b41c0ca4c05 100644 --- a/doc/src/sgml/ref/alter_subscription.sgml +++ b/doc/src/sgml/ref/alter_subscription.sgml @@ -309,8 +309,9 @@ ALTER SUBSCRIPTION name RENAME TO < two_phase, retain_dead_tuples, max_retention_duration, - wal_receiver_timeout, and - conflict_log_destination. + wal_receiver_timeout, + conflict_log_destination, and + message. Only a superuser can set password_required = false. diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml index 81fbf3487a4..afac5ad2442 100644 --- a/doc/src/sgml/ref/create_subscription.sgml +++ b/doc/src/sgml/ref/create_subscription.sgml @@ -478,6 +478,46 @@ CREATE SUBSCRIPTION subscription_name + + message (boolean) + + + Specifies whether the subscription requests the publisher to send + logical decoding messages, that is, those emitted with + pg_logical_emit_message. + The default is false. + + + Received messages are only acted upon if an extension installed on + the subscriber has registered a handler for them; otherwise they are + received and discarded. A message may be passed to the handler more + than once, for example after an apply worker restart, so a handler + must be prepared to process the same message repeatedly. + + + A transactional message is applied as part of the remote transaction + that emitted it, and is therefore delivered in commit order with + respect to other transactions, not in the order in which it was written. + Note that + streaming + and SKIP + apply to transactional messages just as they do to other changes. A + non-transactional message is delivered as soon as it is decoded, even + if the transaction that emitted it later aborted. + + + Messages are not restricted to the tables or schemas of the + subscribed publications: every message emitted in the publisher + database is sent, subject only to + origin + filtering. Message contents are supplied by whoever calls + pg_logical_emit_message on the publisher, which + requires no special privileges, so a handler should treat them as + untrusted input. + + + + origin (string) diff --git a/src/backend/catalog/pg_subscription.c b/src/backend/catalog/pg_subscription.c index 5ff61edb989..3618161ea96 100644 --- a/src/backend/catalog/pg_subscription.c +++ b/src/backend/catalog/pg_subscription.c @@ -139,6 +139,7 @@ GetSubscription(Oid subid, bool missing_ok, bool conninfo_needed, sub->maxretention = subform->submaxretention; sub->retentionactive = subform->subretentionactive; sub->conflictlogrelid = subform->subconflictlogrelid; + sub->message = subform->submessage; if (conninfo_needed) { diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql index 090281a03dd..b726f6e71a8 100644 --- a/src/backend/catalog/system_views.sql +++ b/src/backend/catalog/system_views.sql @@ -1539,7 +1539,8 @@ GRANT SELECT (oid, subdbid, subskiplsn, subname, subowner, subenabled, subpasswordrequired, subrunasowner, subfailover, subretaindeadtuples, submaxretention, subretentionactive, subserver, subconflictlogrelid, subconflictlogdest, subslotname, - subsynccommit, subwalrcvtimeout, subpublications, suborigin) + subsynccommit, subwalrcvtimeout, subpublications, suborigin, + submessage) 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 67f5699b2c7..c13e4c84f85 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -83,6 +83,7 @@ #define SUBOPT_LSN 0x00020000 #define SUBOPT_ORIGIN 0x00040000 #define SUBOPT_CONFLICT_LOG_DEST 0x00080000 +#define SUBOPT_MESSAGE 0x00100000 /* check if the 'val' has 'bits' set */ #define IsSet(val, bits) (((val) & (bits)) == (bits)) @@ -109,6 +110,7 @@ typedef struct SubOpts bool runasowner; bool failover; bool retaindeadtuples; + bool message; int32 maxretention; char *origin; ConflictLogDest conflictlogdest; @@ -202,6 +204,8 @@ parse_subscription_options(ParseState *pstate, List *stmt_options, opts->failover = false; if (IsSet(supported_opts, SUBOPT_RETAIN_DEAD_TUPLES)) opts->retaindeadtuples = false; + if (IsSet(supported_opts, SUBOPT_MESSAGE)) + opts->message = false; if (IsSet(supported_opts, SUBOPT_MAX_RETENTION_DURATION)) opts->maxretention = 0; if (IsSet(supported_opts, SUBOPT_ORIGIN)) @@ -360,6 +364,15 @@ parse_subscription_options(ParseState *pstate, List *stmt_options, opts->specified_opts |= SUBOPT_RETAIN_DEAD_TUPLES; opts->retaindeadtuples = defGetBoolean(defel); } + else if (IsSet(supported_opts, SUBOPT_MESSAGE) && + strcmp(defel->defname, "message") == 0) + { + if (IsSet(opts->specified_opts, SUBOPT_MESSAGE)) + errorConflictingDefElem(defel, pstate); + + opts->specified_opts |= SUBOPT_MESSAGE; + opts->message = defGetBoolean(defel); + } else if (IsSet(supported_opts, SUBOPT_MAX_RETENTION_DURATION) && strcmp(defel->defname, "max_retention_duration") == 0) { @@ -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_MESSAGE); parse_subscription_options(pstate, stmt->options, supported_opts, &opts); /* @@ -850,6 +863,7 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt, values[Anum_pg_subscription_subfailover - 1] = BoolGetDatum(opts.failover); values[Anum_pg_subscription_subretaindeadtuples - 1] = BoolGetDatum(opts.retaindeadtuples); + values[Anum_pg_subscription_submessage - 1] = BoolGetDatum(opts.message); values[Anum_pg_subscription_submaxretention - 1] = Int32GetDatum(opts.maxretention); values[Anum_pg_subscription_subretentionactive - 1] = @@ -1667,7 +1681,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, SUBOPT_MAX_RETENTION_DURATION | SUBOPT_WAL_RECEIVER_TIMEOUT | SUBOPT_ORIGIN | - SUBOPT_CONFLICT_LOG_DEST); + SUBOPT_CONFLICT_LOG_DEST | + SUBOPT_MESSAGE); break; case ALTER_SUBSCRIPTION_ENABLED: @@ -2054,6 +2069,12 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, } } + if (IsSet(opts.specified_opts, SUBOPT_MESSAGE)) + { + values[Anum_pg_subscription_submessage - 1] = BoolGetDatum(opts.message); + replaces[Anum_pg_subscription_submessage - 1] = true; + } + update_tuple = true; break; } diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c index 029990d9fce..e345b80690f 100644 --- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c +++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c @@ -631,6 +631,10 @@ libpqrcv_startstreaming(WalReceiverConn *conn, appendQuotedLiteral(&cmd, options->proto.logical.origin); } + if (options->proto.logical.messages && + PQserverVersion(conn->streamConn) >= 140000) + appendStringInfo(&cmd, ", messages 'on'"); + pubnames = options->proto.logical.publication_names; pubnames_str = stringlist_to_identifierstr(pubnames); appendStringInfoString(&cmd, ", publication_names "); diff --git a/src/backend/replication/logical/proto.c b/src/backend/replication/logical/proto.c index 86ad97cd937..973a67606be 100644 --- a/src/backend/replication/logical/proto.c +++ b/src/backend/replication/logical/proto.c @@ -660,6 +660,35 @@ 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); + + 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 2548d3feb54..7d5706e1598 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -549,6 +549,9 @@ typedef struct ApplySubXactData static ApplySubXactData subxact_data = {0, 0, InvalidTransactionId, NULL}; +/* Hook for plugins to get control in handling logical decoding messages */ +LogicalRepMessageHandle_hook_type LogicalRepMessageHandle_hook = NULL; + static inline void subxact_filename(char *path, Oid subid, TransactionId xid); static inline void changes_filename(char *path, Oid subid, TransactionId xid); @@ -1694,6 +1697,51 @@ 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; + + if (!LogicalRepMessageHandle_hook) + return; + + /* + * 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. Logical messages are handled only + * the (parallel) apply workers + */ + if (am_tablesync_worker() || am_sequencesync_worker()) + return; + + if (is_skipping_changes() || + handle_streamed_transaction(LOGICAL_REP_MSG_MESSAGE, s)) + return; + + logicalrep_read_message(s, &msg); + + begin_replication_step(); + + (*LogicalRepMessageHandle_hook) (&msg); + + end_replication_step(); + + /* + * A transactional message is applied as a step of the remote transaction + * that emitted it, and is committed together with it when applying the + * commit message. A non-transactional message belongs to no remote + * transaction, so commit it here. + */ + if (!msg.transactional) + CommitTransactionCommand(); +} + /* * Initialize fileset (if not already done). * @@ -3846,12 +3894,7 @@ apply_dispatch(StringInfo s) break; case LOGICAL_REP_MSG_MESSAGE: - - /* - * Logical replication does not use generic logical messages yet. - * Although, it could be used by other applications that use this - * output plugin. - */ + apply_handle_message(s); break; case LOGICAL_REP_MSG_STREAM_START: @@ -5128,7 +5171,8 @@ maybe_reread_subscription(void) newsub->passwordrequired != MySubscription->passwordrequired || strcmp(newsub->origin, MySubscription->origin) != 0 || newsub->owner != MySubscription->owner || - !equal(newsub->publications, MySubscription->publications)) + !equal(newsub->publications, MySubscription->publications) || + newsub->message != MySubscription->message) { if (am_parallel_apply_worker()) ereport(LOG, @@ -5614,6 +5658,7 @@ set_stream_options(WalRcvStreamOptions *options, options->proto.logical.twophase = false; options->proto.logical.origin = pstrdup(MySubscription->origin); + options->proto.logical.messages = MySubscription->message; } /* diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c index 4948e6d80c7..015770e0b5d 100644 --- a/src/bin/pg_dump/pg_dump.c +++ b/src/bin/pg_dump/pg_dump.c @@ -5138,6 +5138,7 @@ getSubscriptions(Archive *fout) int i_subfailover; int i_subretaindeadtuples; int i_submaxretention; + int i_submessage; int i, ntups; @@ -5236,9 +5237,14 @@ getSubscriptions(Archive *fout) " '-1' AS subwalrcvtimeout,\n"); if (fout->remoteVersion >= 190000) - appendPQExpBufferStr(query, " fs.srvname AS subservername\n"); + appendPQExpBufferStr(query, " fs.srvname AS subservername,\n"); else - appendPQExpBufferStr(query, " NULL AS subservername\n"); + appendPQExpBufferStr(query, " NULL AS subservername,\n"); + + if (fout->remoteVersion >= 200000) + appendPQExpBufferStr(query, " s.submessage\n"); + else + appendPQExpBufferStr(query, "false AS submessage\n"); appendPQExpBufferStr(query, "FROM pg_subscription s\n"); @@ -5287,6 +5293,7 @@ getSubscriptions(Archive *fout) i_subpublications = PQfnumber(res, "subpublications"); i_suborigin = PQfnumber(res, "suborigin"); i_suboriginremotelsn = PQfnumber(res, "suboriginremotelsn"); + i_submessage = PQfnumber(res, "submessage"); subinfo = pg_malloc_array(SubscriptionInfo, ntups); @@ -5344,6 +5351,8 @@ getSubscriptions(Archive *fout) else subinfo[i].suboriginremotelsn = pg_strdup(PQgetvalue(res, i, i_suboriginremotelsn)); + subinfo[i].submessage = + (strcmp(PQgetvalue(res, i, i_submessage), "t") == 0); /* Decide whether we want to dump it */ selectDumpableObject(&(subinfo[i].dobj), fout); @@ -5596,6 +5605,9 @@ dumpSubscription(Archive *fout, const SubscriptionInfo *subinfo) if (subinfo->subretaindeadtuples) appendPQExpBufferStr(query, ", retain_dead_tuples = true"); + if (subinfo->submessage) + appendPQExpBufferStr(query, ", message = true"); + if (subinfo->submaxretention) appendPQExpBuffer(query, ", max_retention_duration = %d", subinfo->submaxretention); diff --git a/src/bin/pg_dump/pg_dump.h b/src/bin/pg_dump/pg_dump.h index 5a6726d8b12..30bd1575a26 100644 --- a/src/bin/pg_dump/pg_dump.h +++ b/src/bin/pg_dump/pg_dump.h @@ -721,6 +721,7 @@ typedef struct _SubscriptionInfo bool subrunasowner; bool subfailover; bool subretaindeadtuples; + bool submessage; int submaxretention; char *subservername; char *subconninfo; diff --git a/src/bin/pg_dump/t/002_pg_dump.pl b/src/bin/pg_dump/t/002_pg_dump.pl index 9258948b583..019ba9350cd 100644 --- a/src/bin/pg_dump/t/002_pg_dump.pl +++ b/src/bin/pg_dump/t/002_pg_dump.pl @@ -3317,9 +3317,9 @@ my %tests = ( create_order => 50, create_sql => 'CREATE SUBSCRIPTION sub3 CONNECTION \'dbname=doesnotexist\' PUBLICATION pub1 - WITH (connect = false, origin = any, streaming = on);', + WITH (connect = false, origin = any, streaming = on, message = on);', regexp => qr/^ - \QCREATE SUBSCRIPTION sub3 CONNECTION 'dbname=doesnotexist' PUBLICATION pub1 WITH (connect = false, slot_name = 'sub3', streaming = on);\E + \QCREATE SUBSCRIPTION sub3 CONNECTION 'dbname=doesnotexist' PUBLICATION pub1 WITH (connect = false, slot_name = 'sub3', streaming = on, message = on);\E /xm, like => { %full_runs, section_post_data => 1, }, unlike => { diff --git a/src/bin/psql/describe.c b/src/bin/psql/describe.c index ad9c8affb4f..cdb6bbb2184 100644 --- a/src/bin/psql/describe.c +++ b/src/bin/psql/describe.c @@ -6894,7 +6894,7 @@ describeSubscriptions(const char *pattern, bool verbose) printQueryOpt myopt = pset.popt; static const bool translate_columns[] = {false, false, false, false, false, false, false, false, false, false, false, false, false, false, - false, false, false, false, false, false, false}; + false, false, false, false, false, false, false, false}; initPQExpBuffer(&buf); @@ -6972,6 +6972,11 @@ describeSubscriptions(const char *pattern, bool verbose) gettext_noop("Retention active")); } + if (pset.sversion >= 200000) + appendPQExpBuffer(&buf, + ", submessage AS \"%s\"\n", + gettext_noop("Message")); + appendPQExpBuffer(&buf, ", subsynccommit AS \"%s\"\n" ", subconninfo AS \"%s\"\n", diff --git a/src/include/catalog/pg_subscription.h b/src/include/catalog/pg_subscription.h index 65ce8e145fb..1194e006715 100644 --- a/src/include/catalog/pg_subscription.h +++ b/src/include/catalog/pg_subscription.h @@ -83,6 +83,9 @@ CATALOG(pg_subscription,6100,SubscriptionRelationId) BKI_SHARED_RELATION BKI_ROW bool subretaindeadtuples; /* True if dead tuples useful for * conflict detection are retained */ + bool submessage; /* True if the subscription wants to receive + * logical messages. */ + int32 submaxretention; /* The maximum duration (in milliseconds) * for which information useful for * conflict detection can be retained */ @@ -165,6 +168,8 @@ typedef struct Subscription * to be synchronized to the standbys. */ bool retaindeadtuples; /* True if dead tuples useful for conflict * detection are retained */ + bool message; /* True if the subscription wants to receive + * logical messages */ int32 maxretention; /* The maximum duration (in milliseconds) for * which information useful for conflict * detection can be retained */ 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/walreceiver.h b/src/include/replication/walreceiver.h index 5cfe90f9989..eca847c4a8f 100644 --- a/src/include/replication/walreceiver.h +++ b/src/include/replication/walreceiver.h @@ -188,6 +188,7 @@ typedef struct * prepare time */ char *origin; /* Only publish data originating from the * specified origin */ + bool messages; /* Logical message */ } logical; } proto; } WalRcvStreamOptions; diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h index 745b7d9e969..6bbefe5b8f8 100644 --- a/src/include/replication/worker_internal.h +++ b/src/include/replication/worker_internal.h @@ -18,6 +18,7 @@ #include "miscadmin.h" #include "replication/logicalrelation.h" #include "replication/walreceiver.h" +#include "replication/logicalproto.h" #include "storage/buffile.h" #include "storage/fileset.h" #include "storage/shm_mq.h" @@ -255,6 +256,24 @@ extern PGDLLIMPORT bool InitializingApplyWorker; extern PGDLLIMPORT List *table_states_not_ready; +/* + * Hook for extensions to handle logical decoding messages (see + * pg_logical_emit_message()) received from the publisher. Called only when + * the subscription has the "message" option enabled. + * + * The handler runs in the apply worker, inside a transaction, with an active + * snapshot pushed. For a transactional message the handler's work is part of + * the remote transaction and commits with it; a non-transactional message is + * committed as an independent unit. The same message may be delivered more + * than once, so the handler must be idempotent. + * + * 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. + */ +typedef void (*LogicalRepMessageHandle_hook_type) (LogicalRepMessageData *msg); +extern PGDLLIMPORT LogicalRepMessageHandle_hook_type LogicalRepMessageHandle_hook; + extern void logicalrep_worker_attach(int slot); extern LogicalRepWorker *logicalrep_worker_find(LogicalRepWorkerType wtype, Oid subid, Oid relid, diff --git a/src/test/modules/Makefile b/src/test/modules/Makefile index 098bb8142ae..e8a7f50eb38 100644 --- a/src/test/modules/Makefile +++ b/src/test/modules/Makefile @@ -36,6 +36,7 @@ SUBDIRS = \ test_integerset \ test_json_parser \ test_lfind \ + test_logicalmsg_hooks \ test_lwlock_tranches \ test_misc \ test_oat_hooks \ diff --git a/src/test/modules/meson.build b/src/test/modules/meson.build index 4bca42bb370..5f6c0cae77a 100644 --- a/src/test/modules/meson.build +++ b/src/test/modules/meson.build @@ -37,6 +37,7 @@ subdir('test_int128') subdir('test_integerset') subdir('test_json_parser') subdir('test_lfind') +subdir('test_logicalmsg_hooks') subdir('test_lwlock_tranches') subdir('test_misc') subdir('test_oat_hooks') diff --git a/src/test/modules/test_logicalmsg_hooks/.gitignore b/src/test/modules/test_logicalmsg_hooks/.gitignore new file mode 100644 index 00000000000..5dcb3ff9723 --- /dev/null +++ b/src/test/modules/test_logicalmsg_hooks/.gitignore @@ -0,0 +1,4 @@ +# Generated subdirectories +/log/ +/results/ +/tmp_check/ diff --git a/src/test/modules/test_logicalmsg_hooks/Makefile b/src/test/modules/test_logicalmsg_hooks/Makefile new file mode 100644 index 00000000000..af129009b8a --- /dev/null +++ b/src/test/modules/test_logicalmsg_hooks/Makefile @@ -0,0 +1,20 @@ +# src/test/modules/test_logicalmsg_hooks/Makefile + +MODULE_big = test_logicalmsg_hooks +OBJS = \ + $(WIN32RES) \ + test_logicalmsg_hooks.o +PGFILEDESC = "test_logicalmsg_hooks - test logical replication message hooks" + +TAP_TESTS = 1 + +ifdef USE_PGXS +PG_CONFIG = pg_config +PGXS := $(shell $(PG_CONFIG) --pgxs) +include $(PGXS) +else +subdir = src/test/modules/test_logicalmsg_hooks +top_builddir = ../../../.. +include $(top_builddir)/src/Makefile.global +include $(top_srcdir)/contrib/contrib-global.mk +endif diff --git a/src/test/modules/test_logicalmsg_hooks/meson.build b/src/test/modules/test_logicalmsg_hooks/meson.build new file mode 100644 index 00000000000..8084d82e47b --- /dev/null +++ b/src/test/modules/test_logicalmsg_hooks/meson.build @@ -0,0 +1,28 @@ +# Copyright (c) 2022-2026, PostgreSQL Global Development Group + +test_logicalmsg_hooks_sources = files( + 'test_logicalmsg_hooks.c', +) + +if host_system == 'windows' + test_logicalmsg_hooks_sources += rc_lib_gen.process(win32ver_rc, extra_args: [ + '--NAME', 'test_logicalmsg_hooks', + '--FILEDESC', 'test_logicalmsg_hooks - test logical message hooks',]) +endif + +test_logicalmsg_hooks = shared_module('test_logicalmsg_hooks', + test_logicalmsg_hooks_sources, + kwargs: pg_test_mod_args, +) +test_install_libs += test_logicalmsg_hooks + +tests += { + 'name': 'test_logicalmsg_hooks', + 'sd': meson.current_source_dir(), + 'bd': meson.current_build_dir(), + 'tap': { + 'tests': [ + 't/001_basic.pl', + ], + }, +} diff --git a/src/test/modules/test_logicalmsg_hooks/t/001_basic.pl b/src/test/modules/test_logicalmsg_hooks/t/001_basic.pl new file mode 100644 index 00000000000..0d7c7d90530 --- /dev/null +++ b/src/test/modules/test_logicalmsg_hooks/t/001_basic.pl @@ -0,0 +1,220 @@ +# 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; +}); + +# The log table has to exist before any message arrives; the hook silently +# skips recording while it is missing. +$sub->safe_psql( + 'postgres', qq{ +CREATE TABLE test (a int primary key); +CREATE TABLE test_logicalmsg_log ( + lsn pg_lsn, + transactional boolean, + prefix text, + message_size int, + message text +); +}); + +my $pub_connstr = $pub->connstr . ' dbname=postgres'; + +$sub->safe_psql( + 'postgres', qq{ +CREATE SUBSCRIPTION sub CONNECTION '$pub_connstr' PUBLICATION pub WITH (message = true, disable_on_error = true) +}); + +$sub->wait_for_subscription_sync($pub, 'sub'); + +# Number of messages recorded under the given prefix. +sub logged_count +{ + my ($prefix) = @_; + + return $sub->safe_psql('postgres', + "SELECT count(*) FROM test_logicalmsg_log WHERE prefix = '$prefix'"); +} + +# Verify that a transactional message is passed to the hook, and that the +# message data given to the hook is correct. +$pub->safe_psql('postgres', + q{SELECT pg_logical_emit_message(true, 'basic', 'transactional message')} +); +$pub->wait_for_catchup('sub'); + +is( $sub->safe_psql( + 'postgres', + q{SELECT transactional, message_size, message FROM test_logicalmsg_log + WHERE prefix = 'basic'}), + 't|21|transactional message', + "transactional message is passed to the hook"); + +# Verify that a non-transactional message is applied even if the transaction +# that emitted it rolls back. +$pub->safe_psql( + 'postgres', q{ +BEGIN; +SELECT pg_logical_emit_message(false, 'nontransactional', 'emitted then rolled back'); +ROLLBACK; +}); + +# Commit another message to flush WAL, as ROLLBACK doesn't flush. +$pub->safe_psql('postgres', + q{SELECT pg_logical_emit_message(true, 'flush', 'flush')}); +$pub->wait_for_catchup('sub'); + +is( $sub->safe_psql( + 'postgres', + q{SELECT transactional, message FROM test_logicalmsg_log + WHERE prefix = 'nontransactional'}), + 'f|emitted then rolled back', + "non-transactional message of an aborted transaction is applied"); + +# Verify that a transactional message emitted in a subtransaction that is +# rolled back is not passed to the hook. +$pub->safe_psql( + 'postgres', q{ +BEGIN; +SAVEPOINT sp; +SELECT pg_logical_emit_message(true, 'subxact_aborted', 'not delivered'); +ROLLBACK TO SAVEPOINT sp; +SELECT pg_logical_emit_message(true, 'subxact_committed', 'delivered'); +COMMIT; +}); +$pub->wait_for_catchup('sub'); + +is(logged_count('subxact_aborted'), + 0, "message of an aborted subtransaction is not applied"); +is(logged_count('subxact_committed'), + 1, "message of a committed subtransaction is applied"); + +# Verify that transactional messages are applied in streaming transactions, +# both when the leader apply worker serializes the transaction to a file and +# when it hands the transaction to a parallel apply worker. +foreach my $mode ('on', 'parallel') +{ + $sub->safe_psql('postgres', + "ALTER SUBSCRIPTION sub SET (streaming = $mode)"); + + # Reconnect so that the new streaming mode takes effect immediately. + $sub->safe_psql('postgres', 'ALTER SUBSCRIPTION sub DISABLE'); + $sub->safe_psql('postgres', 'ALTER SUBSCRIPTION sub ENABLE'); + + $pub->safe_psql('postgres', + q{SELECT pg_stat_reset_replication_slot('sub')}); + + $pub->safe_psql( + 'postgres', qq{ +BEGIN; +INSERT INTO test SELECT generate_series(1000, 6000); +SELECT pg_logical_emit_message(true, 'streaming_$mode', 'streamed message'); +COMMIT; +}); + $pub->wait_for_catchup('sub'); + + # Make sure the transaction really was streamed, so that this test cannot + # silently degrade into a non-streaming one. + $pub->poll_query_until('postgres', + q{SELECT stream_txns > 0 FROM pg_stat_replication_slots WHERE slot_name = 'sub'} + ) or die "transaction was not streamed"; + + is(logged_count("streaming_$mode"), + 1, + "message in a streamed transaction is applied (streaming = $mode)"); + + $pub->safe_psql('postgres', 'TRUNCATE test'); + $pub->wait_for_catchup('sub'); +} + +$sub->safe_psql('postgres', 'ALTER SUBSCRIPTION sub SET (streaming = off)'); + +# Verify that ALTER SUBSCRIPTION ... SKIP skips a transactional message along +# with the rest of its transaction. The message is emitted before the +# conflicting change, so the hook runs and records the message on the first +# attempt; that work must be rolled back with the failed transaction, and must +# not be redone once the transaction is skipped. +$pub->safe_psql('postgres', 'INSERT INTO test VALUES (1)'); +$pub->wait_for_catchup('sub'); + +my $log_location = -s $sub->logfile; + +$pub->safe_psql( + 'postgres', q{ +BEGIN; +SELECT pg_logical_emit_message(true, 'skipped', 'rolled back, then skipped'); +INSERT INTO test VALUES (1); +COMMIT; +}); + +# Wait until the conflict disables the subscription. +$sub->poll_query_until('postgres', + q{SELECT subenabled = FALSE FROM pg_subscription WHERE subname = 'sub'}); + +# The hook did run before the transaction failed. +ok( $sub->log_contains( + qr/LOG[^\n]+received message: [^\n]+prefix: skipped/, + $log_location), + "hook is called for a message in a transaction that later fails"); + +# ... but its work was rolled back with the transaction. +is(logged_count('skipped'), 0, + "work done by the hook is rolled back with the remote transaction"); + +# Get the finish LSN of the failed transaction. +my $contents = slurp_file($sub->logfile, $log_location); +$contents =~ + qr/conflict detected on relation "public.test".*\n.*DETAIL:.*Could not apply remote change.*\n.*Key already exists in unique index "test_pkey", modified in transaction \d+: key .*, local row .*\n.*CONTEXT:.* for replication target relation "public.test" in transaction \d+, finished at ([[:xdigit:]]+\/[[:xdigit:]]+)/m + or die "could not get error-LSN"; +my $lsn = $1; + +$log_location = -s $sub->logfile; + +# Set skip LSN and re-enable the subscription. +$sub->safe_psql('postgres', qq{ALTER SUBSCRIPTION sub SKIP (lsn = '$lsn')}); +$sub->safe_psql('postgres', 'ALTER SUBSCRIPTION sub ENABLE'); + +# Wait for the failed transaction to be skipped. +$sub->poll_query_until('postgres', + q{SELECT subskiplsn = '0/0' FROM pg_subscription WHERE subname = 'sub'}); + +ok( !$sub->log_contains( + qr/LOG[^\n]+received message: [^\n]+prefix: skipped/, + $log_location), + "hook is not called for a message in a skipped transaction"); +is(logged_count('skipped'), 0, + "message in a skipped transaction is not applied"); + +$sub->stop; +$pub->stop; +done_testing(); diff --git a/src/test/modules/test_logicalmsg_hooks/test_logicalmsg_hooks.c b/src/test/modules/test_logicalmsg_hooks/test_logicalmsg_hooks.c new file mode 100644 index 00000000000..1e3120f4ee1 --- /dev/null +++ b/src/test/modules/test_logicalmsg_hooks/test_logicalmsg_hooks.c @@ -0,0 +1,104 @@ +/*-------------------------------------------------------------------------- + * + * 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 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 "nodes/makefuncs.h" +#include "replication/worker_internal.h" +#include "utils/builtins.h" +#include "utils/pg_lsn.h" + +PG_MODULE_MAGIC; + +#define LOG_SCHEMA "public" +#define LOG_TABLE "test_logicalmsg_log" + +static void test_logical_message_handler(LogicalRepMessageData *msg); + +/* + * Module load callback + */ +void +_PG_init(void) +{ + LogicalRepMessageHandle_hook = &test_logical_message_handler; +} + +static void +test_logical_message_handler(LogicalRepMessageData *msg) +{ + RangeVar *rv; + Oid argtypes[5] = {LSNOID, BOOLOID, TEXTOID, INT4OID, TEXTOID}; + Datum values[5]; + int ret; + + ereport(LOG, + (errmsg("received message: LSN %X/%08X, prefix: %s, message: %s, transactional: %d", + LSN_FORMAT_ARGS(msg->lsn), + msg->prefix, msg->message, msg->transactional))); + + /* + * The log table is created by the test, not by this module, so it may not + * exist yet. Skip recording rather than throwing an error, which would + * put the apply worker into a restart loop and make the test hang instead + * of fail. No lock is needed for this probe; the INSERT below takes the + * proper locks. + */ + rv = makeRangeVar(LOG_SCHEMA, LOG_TABLE, -1); + if (!OidIsValid(RangeVarGetRelid(rv, NoLock, true))) + return; + + /* + * 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. + */ + values[0] = LSNGetDatum(msg->lsn); + values[1] = BoolGetDatum(msg->transactional); + values[2] = CStringGetTextDatum(msg->prefix); + values[3] = Int32GetDatum((int32) msg->message_size); + values[4] = PointerGetDatum(cstring_to_text_with_len(msg->message, + msg->message_size)); + + SPI_connect(); + + /* + * The apply worker runs with an empty search_path, so the table name must + * be schema-qualified. + */ + ret = SPI_execute_with_args("INSERT INTO " LOG_SCHEMA "." LOG_TABLE + " (lsn, transactional, prefix, message_size, message)" + " VALUES ($1, $2, $3, $4, $5)", + 5, argtypes, values, NULL, false, 0); + if (ret != SPI_OK_INSERT) + elog(ERROR, "could not record logical message: SPI returned %d", ret); + + SPI_finish(); +} diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out index 259db747334..83642fc03ee 100644 --- a/src/test/regress/expected/subscription.out +++ b/src/test/regress/expected/subscription.out @@ -143,18 +143,18 @@ CREATE SUBSCRIPTION regress_testsub4 CONNECTION 'dbname=regress_doesnotexist' PU WARNING: subscription was created, but is not connected HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications. \dRs+ regress_testsub4 - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description -------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- - regress_testsub4 | regress_subscription_user | f | {testpub} | f | parallel | d | f | none | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub4 | regress_subscription_user | f | {testpub} | f | parallel | d | f | none | t | f | f | | f | 0 | f | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | (1 row) ALTER SUBSCRIPTION regress_testsub4 SET (origin = any); \dRs+ regress_testsub4 - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description -------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- - regress_testsub4 | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub4 | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | (1 row) DROP SUBSCRIPTION regress_testsub3; @@ -258,10 +258,10 @@ ALTER SUBSCRIPTION regress_testsub CONNECTION 'foobar'; ERROR: invalid connection string syntax: missing "=" after "foobar" in connection info string \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------------- - regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | test subscription + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+------------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | test subscription (1 row) ALTER SUBSCRIPTION regress_testsub SET PUBLICATION testpub2, testpub3 WITH (refresh = false); @@ -270,10 +270,10 @@ ALTER SUBSCRIPTION regress_testsub SET (slot_name = 'newname'); ALTER SUBSCRIPTION regress_testsub SET (password_required = false); ALTER SUBSCRIPTION regress_testsub SET (run_as_owner = true); \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+------------------------------+------------------+------------+------------------- - regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | f | t | f | | f | 0 | f | off | dbname=regress_doesnotexist2 | -1 | 0/00000000 | test subscription + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+------------------------------+------------------+------------+------------------- + regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | f | t | f | | f | 0 | f | f | off | dbname=regress_doesnotexist2 | -1 | 0/00000000 | test subscription (1 row) ALTER SUBSCRIPTION regress_testsub SET (password_required = true); @@ -289,10 +289,10 @@ ERROR: unrecognized subscription parameter: "create_slot" -- ok ALTER SUBSCRIPTION regress_testsub SKIP (lsn = '0/12345'); \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+------------------------------+------------------+------------+------------------- - regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist2 | -1 | 0/00012345 | test subscription + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+------------------------------+------------------+------------+------------------- + regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | t | f | f | | f | 0 | f | f | off | dbname=regress_doesnotexist2 | -1 | 0/00012345 | test subscription (1 row) -- ok - with lsn = NONE @@ -301,10 +301,10 @@ ALTER SUBSCRIPTION regress_testsub SKIP (lsn = NONE); ALTER SUBSCRIPTION regress_testsub SKIP (lsn = '0/0'); ERROR: invalid WAL location (LSN): 0/0 \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+------------------------------+------------------+------------+------------------- - regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist2 | -1 | 0/00000000 | test subscription + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+------------------------------+------------------+------------+------------------- + regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | t | f | f | | f | 0 | f | f | off | dbname=regress_doesnotexist2 | -1 | 0/00000000 | test subscription (1 row) BEGIN; @@ -340,10 +340,10 @@ ALTER SUBSCRIPTION regress_testsub_foo SET (wal_receiver_timeout = '80s'); ALTER SUBSCRIPTION regress_testsub_foo SET (wal_receiver_timeout = 'foobar'); ERROR: invalid value for parameter "wal_receiver_timeout": "foobar" \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ----------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+------------------------------+------------------+------------+------------------- - regress_testsub_foo | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | t | f | f | | f | 0 | f | local | dbname=regress_doesnotexist2 | 80s | 0/00000000 | test subscription + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +---------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+------------------------------+------------------+------------+------------------- + regress_testsub_foo | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | t | f | f | | f | 0 | f | f | local | dbname=regress_doesnotexist2 | 80s | 0/00000000 | test subscription (1 row) -- rename back to keep the rest simple @@ -372,19 +372,19 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB WARNING: subscription was created, but is not connected HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications. \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- - regress_testsub | regress_subscription_user | f | {testpub} | t | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub | regress_subscription_user | f | {testpub} | t | parallel | d | f | any | t | f | f | | f | 0 | f | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | (1 row) ALTER SUBSCRIPTION regress_testsub SET (binary = false); ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE); \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- - regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | (1 row) DROP SUBSCRIPTION regress_testsub; @@ -396,27 +396,27 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB WARNING: subscription was created, but is not connected HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications. \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- - regress_testsub | regress_subscription_user | f | {testpub} | f | on | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | on | d | f | any | t | f | f | | f | 0 | f | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | (1 row) ALTER SUBSCRIPTION regress_testsub SET (streaming = parallel); \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- - regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | (1 row) ALTER SUBSCRIPTION regress_testsub SET (streaming = false); ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE); \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- - regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | | f | 0 | f | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | (1 row) -- fail - publication already exists @@ -431,10 +431,10 @@ ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub2 WITH (refr ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub2 WITH (refresh = false); ERROR: publication "testpub1" is already in subscription "regress_testsub" \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+-----------------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- - regress_testsub | regress_subscription_user | f | {testpub,testpub1,testpub2} | f | off | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+-----------------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub | regress_subscription_user | f | {testpub,testpub1,testpub2} | f | off | d | f | any | t | f | f | | f | 0 | f | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | (1 row) -- fail - publication used more than once @@ -449,10 +449,10 @@ ERROR: publication "testpub3" is not in subscription "regress_testsub" -- ok - delete publications ALTER SUBSCRIPTION regress_testsub DROP PUBLICATION testpub1, testpub2 WITH (refresh = false); \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- - regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | | f | 0 | f | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | (1 row) DROP SUBSCRIPTION regress_testsub; @@ -488,19 +488,19 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB WARNING: subscription was created, but is not connected HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications. \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- - regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | p | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | p | f | any | t | f | f | | f | 0 | f | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | (1 row) -- we can alter streaming when two_phase enabled ALTER SUBSCRIPTION regress_testsub SET (streaming = true); \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- - regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | t | f | f | | f | 0 | f | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | (1 row) ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE); @@ -510,10 +510,10 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB WARNING: subscription was created, but is not connected HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications. \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- - regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | t | f | f | | f | 0 | f | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | (1 row) ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE); @@ -526,18 +526,18 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB WARNING: subscription was created, but is not connected HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications. \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- - regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | (1 row) ALTER SUBSCRIPTION regress_testsub SET (disable_on_error = true); \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- - regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | t | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | t | any | t | f | f | | f | 0 | f | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | (1 row) ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE); @@ -550,10 +550,10 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB WARNING: subscription was created, but is not connected HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications. \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- - regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | (1 row) ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE); @@ -570,10 +570,10 @@ NOTICE: max_retention_duration is ineffective when retain_dead_tuples is disabl WARNING: subscription was created, but is not connected HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications. \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- - regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 1000 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 1000 | f | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | (1 row) -- fail - max_retention_duration must be non-negative @@ -582,12 +582,38 @@ ERROR: max_retention_duration cannot be negative -- ok ALTER SUBSCRIPTION regress_testsub SET (max_retention_duration = 0); \dRs+ - List of subscriptions - Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description ------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+------------- - regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | +(1 row) + +ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE); +DROP SUBSCRIPTION regress_testsub; +-- fail - message must be boolean +CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, message = foo); +ERROR: message requires a Boolean value +-- ok +CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, message = true); +WARNING: subscription was created, but is not connected +HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications. +\dRs+ + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | t | off | dbname=regress_doesnotexist | -1 | 0/00000000 | +(1 row) + +-- ok +ALTER SUBSCRIPTION regress_testsub SET (message = false); +\dRs+ + List of subscriptions + Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Message | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description +-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+---------+--------------------+-----------------------------+------------------+------------+------------- + regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | (1 row) +-- cleanup ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE); DROP SUBSCRIPTION regress_testsub; -- let's do some tests with pg_create_subscription rather than superuser diff --git a/src/test/regress/sql/subscription.sql b/src/test/regress/sql/subscription.sql index 7718c742974..55682f255d1 100644 --- a/src/test/regress/sql/subscription.sql +++ b/src/test/regress/sql/subscription.sql @@ -441,6 +441,23 @@ ALTER SUBSCRIPTION regress_testsub SET (max_retention_duration = 0); ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE); DROP SUBSCRIPTION regress_testsub; +-- fail - message must be boolean +CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, message = foo); + +-- ok +CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, message = true); + +\dRs+ + +-- ok +ALTER SUBSCRIPTION regress_testsub SET (message = false); + +\dRs+ + +-- cleanup +ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE); +DROP SUBSCRIPTION regress_testsub; + -- let's do some tests with pg_create_subscription rather than superuser SET SESSION AUTHORIZATION regress_subscription_user3; diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index 85d989f395d..0c1ac8f6b64 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -1707,6 +1707,7 @@ LogicalRepBeginData LogicalRepCommitData LogicalRepCommitPreparedTxnData LogicalRepCtxStruct +LogicalRepMessageData LogicalRepMsgType LogicalRepPartMapEntry LogicalRepPreparedTxnData -- 2.55.0