From 3a81649cf5d8421d47e6b901e4b3bfb7159aa20b Mon Sep 17 00:00:00 2001 From: Masahiko Sawada Date: Tue, 15 Jul 2025 00:07:26 -0700 Subject: [PATCH v3 2/3] Add pg_[de]activate_logical_decoding() SQL functions. Author: Reviewed-by: Discussion: https://postgr.es/m/ Backpatch-through: --- src/backend/access/transam/xlogrecovery.c | 2 +- src/backend/commands/subscriptioncmds.c | 2 +- src/backend/replication/logical/logicalctl.c | 125 ++++++++++++++++++ .../replication/logical/reorderbuffer.c | 2 +- src/backend/replication/logical/slotsync.c | 2 +- src/backend/replication/slot.c | 52 +++++++- src/backend/replication/slotfuncs.c | 8 +- src/backend/replication/walsender.c | 4 +- src/include/catalog/pg_proc.dat | 10 ++ src/include/replication/slot.h | 15 ++- 10 files changed, 208 insertions(+), 14 deletions(-) diff --git a/src/backend/access/transam/xlogrecovery.c b/src/backend/access/transam/xlogrecovery.c index 23878b2dd91..e8f3ba00caa 100644 --- a/src/backend/access/transam/xlogrecovery.c +++ b/src/backend/access/transam/xlogrecovery.c @@ -4760,7 +4760,7 @@ bool check_primary_slot_name(char **newval, void **extra, GucSource source) { if (*newval && strcmp(*newval, "") != 0 && - !ReplicationSlotValidateName(*newval, WARNING)) + !ReplicationSlotValidateName(*newval, false, WARNING)) return false; return true; diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index e23b0de7242..bb28f6c95e2 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -210,7 +210,7 @@ parse_subscription_options(ParseState *pstate, List *stmt_options, if (strcmp(opts->slot_name, "none") == 0) opts->slot_name = NULL; else - ReplicationSlotValidateName(opts->slot_name, ERROR); + ReplicationSlotValidateName(opts->slot_name, false, ERROR); } else if (IsSet(supported_opts, SUBOPT_COPY_DATA) && strcmp(defel->defname, "copy_data") == 0) diff --git a/src/backend/replication/logical/logicalctl.c b/src/backend/replication/logical/logicalctl.c index 320a43b0574..f4b7dc072ea 100644 --- a/src/backend/replication/logical/logicalctl.c +++ b/src/backend/replication/logical/logicalctl.c @@ -38,6 +38,9 @@ LogicalDecodingCtlData *LogicalDecodingCtl = NULL; */ bool XLogLogicalInfo = false; +PG_FUNCTION_INFO_V1(pg_activate_logical_decoding); +PG_FUNCTION_INFO_V1(pg_deactivate_logical_decoding); + Size LogicalDecodingCtlShmemSize(void) { @@ -403,3 +406,125 @@ UpdateLogicalDecodingStatusEndOfRecovery(void) WaitForProcSignalBarrier( EmitProcSignalBarrier(PROCSIGNAL_BARRIER_UPDATE_XLOG_LOGICAL_INFO)); } + +/* + * SQL function to activate the logical decoding when wal_level = 'replica'. + * This function essentially creates an empty logical slot with the name + * LOGICAL_DECODING_ACTIVATION_SLOT to keep the logical decoding active + * until users disable it by calling pg_deactivate_logical_decoding() SQL + * function. + */ +Datum +pg_activate_logical_decoding(PG_FUNCTION_ARGS) +{ + CheckSlotPermissions(); + CheckSlotRequirements(); + + if (RecoveryInProgress()) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("cannot activate logical decoding on standbys"))); + + if (wal_level >= WAL_LEVEL_LOGICAL) + { + ereport(NOTICE, + (errmsg("skipping logical decoding activation"), + errdetail("logical decoding is already activated because \"wal_level\" >= \"logical\"."))); + + PG_RETURN_VOID(); + } + + if (SearchNamedReplicationSlot(LOGICAL_DECODING_ACTIVATION_SLOT, true)) + { + ereport(NOTICE, + (errmsg("skipping logical decoding activation"), + errdetail("logical decoding is already activated using pg_activate_logical_decoding()."))); + + PG_RETURN_VOID(); + } + + if (IsLogicalDecodingEnabled()) + { + uint32 nslots = pg_atomic_read_u32(&ReplicationSlotCtl->n_inuse_logical_slots); + + /* + * The logical decoding is already activated since we have at least + * one logical slot, but we don't have + * LOGICAL_DECODING_ACTIVATION_SLOT. We let users know it's already + * enabled but proceed with creating the slot. + */ + ereport(NOTICE, + (errmsg_plural("logical decoding is already activated because there is %u in-use logical slot", + "logical decoding is already activated because there are %u in-use logical slots", + nslots, nslots))); + } + + /* + * Create an empty logical slot with the reserved name, and enable the + * logical decoding. + */ + ReplicationSlotCreate(LOGICAL_DECODING_ACTIVATION_SLOT, true, + RS_PERSISTENT, false, false, false, true); + EnsureLogicalDecodingEnabled(); + Assert(IsLogicalDecodingEnabled()); + + ReplicationSlotRelease(); + + PG_RETURN_VOID(); +} + +/* + * Deactivate the logical decoding if it has been enabled using + * pg_activate_logical_decoding() SQL function. + */ +Datum +pg_deactivate_logical_decoding(PG_FUNCTION_ARGS) +{ + if (RecoveryInProgress()) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("cannot deactivate logical decoding on standbys"))); + + if (wal_level >= WAL_LEVEL_LOGICAL) + { + ereport(NOTICE, + (errmsg("skipping logical decoding deactivation"), + errdetail("logical decoding is activated because \"wal_level\" >= \"logical\"."))); + + PG_RETURN_VOID(); + } + + if (!IsLogicalDecodingEnabled()) + { + ereport(NOTICE, + (errmsg("skipping logical decoding deactivation"), + errdetail("logical decoding is already deactivated."))); + + PG_RETURN_VOID(); + } + + if (!SearchNamedReplicationSlot(LOGICAL_DECODING_ACTIVATION_SLOT, true)) + { + /* + * The logical decoding is enabled but there is no + * LOGICAL_DECODING_ACTIVATION_SLOT slot, meaning that there is at + * least one logical slot. The only way to disable it in this case is + * to drop all logical slot. + */ + ereport(NOTICE, + (errmsg("skipping logical decoding deactivation"), + errdetail("pg_activate_logical_decoding() has not called."), + errhint("Drop all logical replication slots to disable logical decoding"))); + + PG_RETURN_VOID(); + } + + DropLogicalDecodingActivationSlot(); + DisableLogicalDecodingIfNecessary(); + + if (IsLogicalDecodingEnabled()) + ereport(NOTICE, + (errmsg("logical replication is still active until all logical slots are dropped"))); + + PG_RETURN_VOID(); +} diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c index 7b4e8629553..1e1af094bab 100644 --- a/src/backend/replication/logical/reorderbuffer.c +++ b/src/backend/replication/logical/reorderbuffer.c @@ -4917,7 +4917,7 @@ StartupReorderBuffer(void) continue; /* if it cannot be a slot, skip the directory */ - if (!ReplicationSlotValidateName(logical_de->d_name, DEBUG2)) + if (!ReplicationSlotValidateName(logical_de->d_name, false, DEBUG2)) continue; /* diff --git a/src/backend/replication/logical/slotsync.c b/src/backend/replication/logical/slotsync.c index f3ed58b9001..7841a03d67f 100644 --- a/src/backend/replication/logical/slotsync.c +++ b/src/backend/replication/logical/slotsync.c @@ -762,7 +762,7 @@ synchronize_one_slot(RemoteSlot *remote_slot, Oid remote_dbid) ReplicationSlotCreate(remote_slot->name, true, RS_TEMPORARY, remote_slot->two_phase, remote_slot->failover, - true); + true, false); EnsureLogicalDecodingEnabled(); diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c index 671b7fa2083..5ec8e3b4aab 100644 --- a/src/backend/replication/slot.c +++ b/src/backend/replication/slot.c @@ -264,10 +264,14 @@ ReplicationSlotShmemExit(int code, Datum arg) * Slot names may consist out of [a-z0-9_]{1,NAMEDATALEN-1} which should allow * the name to be used as a directory name on every supported OS. * + * An error will be reported for a reserved replication slot name if + * allow_reserved_name is set to false. + * * Returns whether the directory name is valid or not if elevel < ERROR. */ bool -ReplicationSlotValidateName(const char *name, int elevel) +ReplicationSlotValidateName(const char *name, bool allow_reserved_name, + int elevel) { const char *cp; @@ -303,6 +307,20 @@ ReplicationSlotValidateName(const char *name, int elevel) return false; } } + + if (!allow_reserved_name && + (strcmp(name, LOGICAL_DECODING_ACTIVATION_SLOT) == 0)) + { + ereport(elevel, + errcode(ERRCODE_RESERVED_NAME), + errmsg("replication slot name \"%s\" is reserved", + name), + errdetail("The name \"%s\" is reserved for the conflict detection slot.", + LOGICAL_DECODING_ACTIVATION_SLOT)); + + return false; + } + return true; } @@ -326,14 +344,16 @@ ReplicationSlotValidateName(const char *name, int elevel) void ReplicationSlotCreate(const char *name, bool db_specific, ReplicationSlotPersistency persistency, - bool two_phase, bool failover, bool synced) + bool two_phase, bool failover, bool synced, + bool skip_name_validation) { ReplicationSlot *slot = NULL; int i; Assert(MyReplicationSlot == NULL); - ReplicationSlotValidateName(name, ERROR); + if (!skip_name_validation) + ReplicationSlotValidateName(name, true, ERROR); if (failover) { @@ -570,6 +590,12 @@ ReplicationSlotAcquire(const char *name, bool nowait, bool error_if_invalid) Assert(name != NULL); + /* Any operations are not permitted on the special-purpose slot */ + if (strcmp(name, LOGICAL_DECODING_ACTIVATION_SLOT) == 0) + ereport(ERROR, + (errmsg("cannot acquire \"%s\" slot", + LOGICAL_DECODING_ACTIVATION_SLOT))); + retry: Assert(MyReplicationSlot == NULL); @@ -748,8 +774,11 @@ ReplicationSlotRelease(void) SpinLockRelease(&slot->mutex); ConditionVariableBroadcast(&slot->active_cv); } - else + else if (strcmp(NameStr(slot->data.name), LOGICAL_DECODING_ACTIVATION_SLOT) != 0) + { + /* Set inactive_since value unless the slot is not reserved one */ ReplicationSlotSetInactiveSince(slot, now, true); + } MyReplicationSlot = NULL; @@ -839,6 +868,21 @@ ReplicationSlotDrop(const char *name, bool nowait) ReplicationSlotDropAcquired(); } +/* + * Drop the (reserved) LOGICAL_DECODING_ACTIVATION_SLOT slot. + */ +void +DropLogicalDecodingActivationSlot(void) +{ + ReplicationSlot *slot; + + LWLockAcquire(ReplicationSlotControlLock, LW_SHARED); + slot = SearchNamedReplicationSlot(LOGICAL_DECODING_ACTIVATION_SLOT, false); + LWLockRelease(ReplicationSlotControlLock); /* avoid deadlock */ + + ReplicationSlotDropPtr(slot); +} + /* * Change the definition of the slot identified by the specified name. */ diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c index 6744a7979d7..99c36b9a525 100644 --- a/src/backend/replication/slotfuncs.c +++ b/src/backend/replication/slotfuncs.c @@ -42,7 +42,7 @@ create_physical_replication_slot(char *name, bool immediately_reserve, /* acquire replication slot, this will check for conflicting names */ ReplicationSlotCreate(name, false, temporary ? RS_TEMPORARY : RS_PERSISTENT, false, - false, false); + false, false, false); if (immediately_reserve) { @@ -135,7 +135,11 @@ create_logical_replication_slot(char *name, char *plugin, */ ReplicationSlotCreate(name, true, temporary ? RS_TEMPORARY : RS_EPHEMERAL, two_phase, - failover, false); + failover, false, false); + + EnsureLogicalDecodingEnabled(); + CheckLogicalDecodingRequirements(); + EnsureLogicalDecodingEnabled(); CheckLogicalDecodingRequirements(); diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index 7158c638836..dea9445adce 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -1199,7 +1199,7 @@ CreateReplicationSlot(CreateReplicationSlotCmd *cmd) { ReplicationSlotCreate(cmd->slotname, false, cmd->temporary ? RS_TEMPORARY : RS_PERSISTENT, - false, false, false); + false, false, false, false); if (reserve_wal) { @@ -1231,7 +1231,7 @@ CreateReplicationSlot(CreateReplicationSlotCmd *cmd) */ ReplicationSlotCreate(cmd->slotname, true, cmd->temporary ? RS_TEMPORARY : RS_EPHEMERAL, - two_phase, failover, false); + two_phase, failover, false, false); /* * Do options check early so that we can bail before calling the diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat index 1fc19146f46..2d3686ae867 100644 --- a/src/include/catalog/pg_proc.dat +++ b/src/include/catalog/pg_proc.dat @@ -12572,4 +12572,14 @@ proargnames => '{pid,io_id,io_generation,state,operation,off,length,target,handle_data_len,raw_result,result,target_desc,f_sync,f_localmem,f_buffered}', prosrc => 'pg_get_aios' }, +# Logical decoding activation/deactivation functions +{ oid => '8562', descr => 'activate logical decoding', + proname => 'pg_activate_logical_decoding', prorettype => 'void', + provolatile => 'v', proparallel => 'u', proargtypes => '', + prosrc => 'pg_activate_logical_decoding' }, +{ oid => '8563', descr => 'deactivate logical decoding', + proname => 'pg_deactivate_logical_decoding', prorettype => 'void', + provolatile => 'v', proparallel => 'u', proargtypes => '', + prosrc => 'pg_deactivate_logical_decoding' }, + ] diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h index a983e025d63..9ea45266f19 100644 --- a/src/include/replication/slot.h +++ b/src/include/replication/slot.h @@ -21,6 +21,14 @@ /* directory to store replication slot data in */ #define PG_REPLSLOT_DIR "pg_replslot" +/* + * The reserved name for a replication slot used to activate the logical + * decoding even when wal_level = 'replica'. This slot is created and + * dropped only via dedicated functions such as + * pg_activate_logical_decoding(). + */ +#define LOGICAL_DECODING_ACTIVATION_SLOT "pg_logical_decoding_activation" + /* * Behaviour of replication slots, upon release or crash. * @@ -279,7 +287,7 @@ extern void ReplicationSlotsShmemInit(void); extern void ReplicationSlotCreate(const char *name, bool db_specific, ReplicationSlotPersistency persistency, bool two_phase, bool failover, - bool synced); + bool synced, bool skip_name_validation); extern void ReplicationSlotPersist(void); extern void ReplicationSlotDrop(const char *name, bool nowait); extern void ReplicationSlotDropAcquired(void); @@ -295,7 +303,8 @@ extern void ReplicationSlotMarkDirty(void); /* misc stuff */ extern void ReplicationSlotInitialize(void); -extern bool ReplicationSlotValidateName(const char *name, int elevel); +extern bool ReplicationSlotValidateName(const char *name, bool allow_reserved_name, + int elevel); extern void ReplicationSlotReserveWal(void); extern void ReplicationSlotsComputeRequiredXmin(bool already_locked); extern void ReplicationSlotsComputeRequiredLSN(void); @@ -325,4 +334,6 @@ extern bool SlotExistsInSyncStandbySlots(const char *slot_name); extern bool StandbySlotsHaveCaughtup(XLogRecPtr wait_for_lsn, int elevel); extern void WaitForStandbyConfirmation(XLogRecPtr wait_for_lsn); +extern void DropLogicalDecodingActivationSlot(void); + #endif /* SLOT_H */ -- 2.43.5