From 172cb418d25ecd5a4aa5f66a889867f56355a30d Mon Sep 17 00:00:00 2001 From: Masahiko Sawada Date: Tue, 3 Jun 2025 12:52:22 -0700 Subject: [PATCH v6] Enable logical decoding dynamically based on logical slot presence. Previously logical decoding required wal_level to be set to 'logical' at server start.This commit adds functionality to automatically control logical decoding based on logical replication slot presence. The newly introduced module logicalctl.c allows logical decoding to be dynamically activated when needed, even with wal_level set to 'replica'. When the first logical replication is created, the system automatically increases the effective WAL level to maintain logical-level WAL records. Conversely, when the last logical slot is dropped, it decreases back to 'replica' WAL level. A new read-only parameter effective_wal_level is introduced to monitor the actual WAL level in effect. This parameter reflects the current operational WAL level, which may differ from the configured wal_level setting. XXX Bump PG_CONTROL_VERSION as it adds a new field to CheckPoint struct. Reviewed-by: Discussion: https://postgr.es/m/ --- doc/src/sgml/config.sgml | 43 ++ doc/src/sgml/logical-replication.sgml | 4 +- doc/src/sgml/logicaldecoding.sgml | 31 +- doc/src/sgml/ref/pg_createsubscriber.sgml | 12 +- src/backend/access/heap/heapam.c | 8 +- src/backend/access/rmgrdesc/xlogdesc.c | 13 +- src/backend/access/transam/xact.c | 24 +- src/backend/access/transam/xlog.c | 105 ++- src/backend/access/transam/xlogrecovery.c | 2 +- src/backend/commands/publicationcmds.c | 7 +- src/backend/commands/tablecmds.c | 2 +- src/backend/replication/logical/Makefile | 1 + src/backend/replication/logical/decode.c | 51 +- src/backend/replication/logical/logical.c | 24 +- src/backend/replication/logical/logicalctl.c | 612 ++++++++++++++++++ src/backend/replication/logical/meson.build | 1 + src/backend/replication/logical/slotsync.c | 38 +- src/backend/replication/slot.c | 84 ++- src/backend/replication/slotfuncs.c | 7 + src/backend/replication/walsender.c | 7 + src/backend/storage/ipc/ipci.c | 3 + src/backend/storage/ipc/procsignal.c | 4 + src/backend/storage/ipc/standby.c | 7 +- .../utils/activity/wait_event_names.txt | 3 + src/backend/utils/cache/inval.c | 8 +- src/backend/utils/init/postinit.c | 4 + src/backend/utils/misc/guc_tables.c | 11 + src/bin/pg_basebackup/pg_createsubscriber.c | 6 +- src/bin/pg_upgrade/check.c | 6 +- src/include/access/xlog.h | 5 +- src/include/catalog/pg_control.h | 2 + src/include/replication/logicalctl.h | 66 ++ src/include/replication/slot.h | 1 + src/include/replication/slotsync.h | 2 +- src/include/storage/lwlocklist.h | 1 + src/include/storage/procsignal.h | 2 + src/include/utils/guc_hooks.h | 1 + src/test/recovery/meson.build | 3 +- .../t/035_standby_logical_decoding.pl | 2 +- .../recovery/t/049_effective_wal_level.pl | 213 ++++++ src/test/subscription/t/001_rep_changes.pl | 2 +- src/tools/pgindent/typedefs.list | 1 + 42 files changed, 1294 insertions(+), 135 deletions(-) create mode 100644 src/backend/replication/logical/logicalctl.c create mode 100644 src/include/replication/logicalctl.h create mode 100644 src/test/recovery/t/049_effective_wal_level.pl diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml index 20ccb2d6b54..50ed3fbc198 100644 --- a/doc/src/sgml/config.sgml +++ b/doc/src/sgml/config.sgml @@ -3034,6 +3034,17 @@ include_dir 'conf.d' many UPDATE and DELETE statements are executed. + + It is important to note that when wal_level is set to + replica the effective WAL level can automatically change + based on the presence of + logical replication slots. The system automatically increases the + effective WAL level to logical when creating the first + logical replication slot, and decreases it back to replica + when dropping the last logical replication slot. The current effective WAL + level can be monitored through + parameter. + In releases prior to 9.6, this parameter also allowed the values archive and hot_standby. @@ -11746,6 +11757,38 @@ dynamic_library_path = '/usr/local/lib/postgresql:$libdir' + + effective_wal_level (enum) + + effective_wal_level configuration parameter + + + + + Reports the actual WAL logging level currently in effect in the + system. This parameter shares the same set of values as + , but reflects the operational WAL + level rather than the configured setting. For descriptions of + possible values, refer to the wal_level + parameter documentation. + + + The effective WAL level can differ from the configured + wal_level in certain situations. For example, + when wal_level is set to replica + and the system has one or more logical replication slots, + effective_wal_level will show logical + to indicate that the system is maintaining WAL records at + logical level equivalent. + + + On standby servers, effective_wal_level matches + the value of effective_wal_level from the most + upstream server in the replication chain. + + + + huge_pages_status (enum) diff --git a/doc/src/sgml/logical-replication.sgml b/doc/src/sgml/logical-replication.sgml index fcac55aefe6..367157e9d91 100644 --- a/doc/src/sgml/logical-replication.sgml +++ b/doc/src/sgml/logical-replication.sgml @@ -2359,7 +2359,7 @@ CONTEXT: processing remote data for replication origin "pg_16395" during "INSER wal_level must be - set to logical. + set to replica or logical. @@ -2480,7 +2480,7 @@ CONTEXT: processing remote data for replication origin "pg_16395" during "INSER The new cluster must have wal_level as - logical. + replica or logical. diff --git a/doc/src/sgml/logicaldecoding.sgml b/doc/src/sgml/logicaldecoding.sgml index 593f784b69d..96242a589f6 100644 --- a/doc/src/sgml/logicaldecoding.sgml +++ b/doc/src/sgml/logicaldecoding.sgml @@ -47,7 +47,7 @@ Before you can use logical decoding, you must set - to logical and + to replica or higher and to at least 1. Then, you should connect to the target database (in the example below, postgres) as a superuser. @@ -257,6 +257,32 @@ postgres=# select * from pg_logical_slot_get_changes('regression_slot', NULL, NU log, which describe changes on a storage level, into an application-specific form such as a stream of tuples or SQL statements. + + + Logical decoding becomes available in two conditions: + + + + + When is set to logical. + + + + + When is set to replica + and at least one logical replication slot exists on the system. + + + + + When wal_level is set to replica, + logical decoding is automatically activated upon creation of the first + logical replication slot. This activation process involves several steps + and requires waiting for any concurrent transactions to finish, ensureing + system-wide conistency. Conversely, when the last logical replication slot + is dropped from a system with wal_level set to + replica, logical decoding is automatically disabled. + @@ -328,8 +354,7 @@ postgres=# select * from pg_logical_slot_get_changes('regression_slot', NULL, NU that could be needed by the logical decoding on the standby (as it does not know about the catalog_xmin on the standby). Existing logical slots on standby also get invalidated if - wal_level on the primary is reduced to less than - logical. + logical decoding becomes disabled on the primary. This is done as soon as the standby detects such a change in the WAL stream. It means that, for walsenders that are lagging (if any), some WAL records up to the wal_level parameter change on the primary won't be diff --git a/doc/src/sgml/ref/pg_createsubscriber.sgml b/doc/src/sgml/ref/pg_createsubscriber.sgml index bb9cc72576c..5b44710639a 100644 --- a/doc/src/sgml/ref/pg_createsubscriber.sgml +++ b/doc/src/sgml/ref/pg_createsubscriber.sgml @@ -379,12 +379,12 @@ PostgreSQL documentation The source server must accept connections from the target server. The source server must not be in recovery. The source server must have as logical. The source server - must have configured to a value - greater than or equal to the number of specified databases plus existing - replication slots. The source server must have configured to a value greater than or equal - to the number of specified databases and existing WAL sender processes. + linkend="guc-wal-level"/> as replica or logical. + The source server must have + configured to a value greater than or equal to the number of specified + databases plus existing replication slots. The source server must have + configured to a value greater than or + equal to the number of specified databases and existing WAL sender processes. diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c index 0dcd6ee817e..3b733d22e19 100644 --- a/src/backend/access/heap/heapam.c +++ b/src/backend/access/heap/heapam.c @@ -8873,8 +8873,8 @@ log_heap_update(Relation reln, Buffer oldbuf, * * Skip this if we're taking a full-page image of the new page, as we * don't include the new tuple in the WAL record in that case. Also - * disable if wal_level='logical', as logical decoding needs to be able to - * read the new tuple in whole from the WAL record alone. + * disable if logical decoding is enabled, as logical decoding needs to be + * able to read the new tuple in whole from the WAL record alone. */ if (oldbuf == newbuf && !need_tuple_data && !XLogCheckBufferNeedsBackup(newbuf)) @@ -9046,8 +9046,8 @@ log_heap_update(Relation reln, Buffer oldbuf, /* * Perform XLogInsert of an XLOG_HEAP2_NEW_CID record * - * This is only used in wal_level >= WAL_LEVEL_LOGICAL, and only for catalog - * tuples. + * This is only used when XLogLogicalInfoActive() is true, and only for + * catalog tuples. */ static XLogRecPtr log_heap_new_cid(Relation relation, HeapTuple tup) diff --git a/src/backend/access/rmgrdesc/xlogdesc.c b/src/backend/access/rmgrdesc/xlogdesc.c index cd6c2a2f650..ffaf6ac01e7 100644 --- a/src/backend/access/rmgrdesc/xlogdesc.c +++ b/src/backend/access/rmgrdesc/xlogdesc.c @@ -66,7 +66,7 @@ xlog_desc(StringInfo buf, XLogReaderState *record) CheckPoint *checkpoint = (CheckPoint *) rec; appendStringInfo(buf, "redo %X/%08X; " - "tli %u; prev tli %u; fpw %s; wal_level %s; xid %u:%u; oid %u; multi %u; offset %u; " + "tli %u; prev tli %u; fpw %s; wal_level %s; logical decoding %s; xid %u:%u; oid %u; multi %u; offset %u; " "oldest xid %u in DB %u; oldest multi %u in DB %u; " "oldest/newest commit timestamp xid: %u/%u; " "oldest running xid %u; %s", @@ -75,6 +75,7 @@ xlog_desc(StringInfo buf, XLogReaderState *record) checkpoint->PrevTimeLineID, checkpoint->fullPageWrites ? "true" : "false", get_wal_level_string(checkpoint->wal_level), + checkpoint->logicalDecodingEnabled ? "true" : "false", EpochFromFullTransactionId(checkpoint->nextXid), XidFromFullTransactionId(checkpoint->nextXid), checkpoint->nextOid, @@ -167,6 +168,13 @@ xlog_desc(StringInfo buf, XLogReaderState *record) memcpy(&wal_level, rec, sizeof(int)); appendStringInfo(buf, "wal_level %s", get_wal_level_string(wal_level)); } + else if (info == XLOG_LOGICAL_DECODING_STATUS_CHANGE) + { + bool enabled; + + memcpy(&enabled, rec, sizeof(bool)); + appendStringInfo(buf, enabled ? "true" : "false"); + } } const char * @@ -218,6 +226,9 @@ xlog_identify(uint8 info) case XLOG_CHECKPOINT_REDO: id = "CHECKPOINT_REDO"; break; + case XLOG_LOGICAL_DECODING_STATUS_CHANGE: + id = "LOGICAL_DECODING_STATUS_CHANGE"; + break; } return id; diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c index b46e7e9c2a6..cbb4c82ff53 100644 --- a/src/backend/access/transam/xact.c +++ b/src/backend/access/transam/xact.c @@ -551,9 +551,9 @@ MarkCurrentTransactionIdLoggedIfAny(void) * operation in a subtransaction. We require that for logical decoding, see * LogicalDecodingProcessRecord. * - * This returns true if wal_level >= logical and we are inside a valid - * subtransaction, for which the assignment was not yet written to any WAL - * record. + * This returns true if XLogLogicalInfoActive() is true and we are inside + * a valid subtransaction, for which the assignment was not yet written to + * any WAL record. */ bool IsSubxactTopXidLogPending(void) @@ -562,7 +562,7 @@ IsSubxactTopXidLogPending(void) if (CurrentTransactionState->topXidLogged) return false; - /* wal_level has to be logical */ + /* effective WAL level has to be logical */ if (!XLogLogicalInfoActive()) return false; @@ -681,14 +681,14 @@ AssignTransactionId(TransactionState s) } /* - * When wal_level=logical, guarantee that a subtransaction's xid can only - * be seen in the WAL stream if its toplevel xid has been logged before. - * If necessary we log an xact_assignment record with fewer than - * PGPROC_MAX_CACHED_SUBXIDS. Note that it is fine if didLogXid isn't set - * for a transaction even though it appears in a WAL record, we just might - * superfluously log something. That can happen when an xid is included - * somewhere inside a wal record, but not in XLogRecord->xl_xid, like in - * xl_standby_locks. + * When XLogLogicalInfoActive() is true, guarantee that a subtransaction's + * xid can only be seen in the WAL stream if its toplevel xid has been + * logged before. If necessary we log an xact_assignment record with fewer + * than PGPROC_MAX_CACHED_SUBXIDS. Note that it is fine if didLogXid isn't + * set for a transaction even though it appears in a WAL record, we just + * might superfluously log something. That can happen when an xid is + * included somewhere inside a wal record, but not in XLogRecord->xl_xid, + * like in xl_standby_locks. */ if (isSubXact && XLogLogicalInfoActive() && !TopTransactionStateData.didLogXid) diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index b0891998b24..e2dc7319486 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -78,7 +78,9 @@ #include "postmaster/walsummarizer.h" #include "postmaster/walwriter.h" #include "replication/origin.h" +#include "replication/logicalctl.h" #include "replication/slot.h" +#include "replication/slotsync.h" #include "replication/snapbuild.h" #include "replication/walreceiver.h" #include "replication/walsender.h" @@ -142,6 +144,7 @@ bool XLOG_DEBUG = false; #endif int wal_segment_size = DEFAULT_XLOG_SEG_SIZE; +int effective_wal_level = WAL_LEVEL_REPLICA; /* * Number of WAL insertion locks to use. A higher value allows more insertions @@ -5001,6 +5004,41 @@ show_in_hot_standby(void) return RecoveryInProgress() ? "on" : "off"; } +/* + * GUC show_hook for effective_wal_level + */ +const char * +show_effective_wal_level(void) +{ + char *str; + + if (wal_level == WAL_LEVEL_MINIMAL) + return "minimal"; + + /* + * During the recovery, we don't synchronously update the XLogLogicalInfo + * so need to check the shared state. + */ + if (RecoveryInProgress()) + return IsXLogLogicalInfoEnabled() ? "logical" : "replica"; + + if (wal_level == WAL_LEVEL_REPLICA) + { + /* + * With wal_level='replica', XLogLogicalInfo indicates the actual WAL + * level. + */ + if (IsXLogLogicalInfoEnabled()) + str = "logical"; + else + str = "replica"; + } + else + str = "logical"; + + return str; +} + /* * Read the control file, set respective GUCs. * @@ -5253,6 +5291,7 @@ BootStrapXLOG(uint32 data_checksum_version) checkPoint.ThisTimeLineID = BootstrapTimeLineID; checkPoint.PrevTimeLineID = BootstrapTimeLineID; checkPoint.fullPageWrites = fullPageWrites; + checkPoint.logicalDecodingEnabled = IsLogicalDecodingEnabled(); checkPoint.wal_level = wal_level; checkPoint.nextXid = FullTransactionIdFromEpochAndXid(0, FirstNormalTransactionId); @@ -5774,6 +5813,12 @@ StartupXLOG(void) */ StartupReplicationSlots(); + /* + * Startup the logical decoding status with the last status stored in the + * checkpoint record. + */ + StartupLogicalDecodingStatus(checkPoint.logicalDecodingEnabled); + /* * Startup logical state, needs to be setup now so we have proper data * during crash recovery. @@ -6303,6 +6348,12 @@ StartupXLOG(void) Insert->fullPageWrites = lastFullPageWrites; UpdateFullPageWrites(); + /* + * Update logical decoding status in shared memory and write an + * XLOG_LOGICAL_DECODING_STATUS_CHANGE, if necessary. + */ + UpdateLogicalDecodingStatusEndOfRecovery(); + /* * Emit checkpoint or end-of-recovery record in XLOG, if required. */ @@ -7179,6 +7230,7 @@ CreateCheckPoint(int flags) checkPoint.fullPageWrites = Insert->fullPageWrites; checkPoint.wal_level = wal_level; + checkPoint.logicalDecodingEnabled = IsLogicalDecodingEnabled(); if (shutdown) { @@ -8668,21 +8720,6 @@ xlog_redo(XLogReaderState *record) /* Update our copy of the parameters in pg_control */ memcpy(&xlrec, XLogRecGetData(record), sizeof(xl_parameter_change)); - /* - * Invalidate logical slots if we are in hot standby and the primary - * does not have a WAL level sufficient for logical decoding. No need - * to search for potentially conflicting logically slots if standby is - * running with wal_level lower than logical, because in that case, we - * would have either disallowed creation of logical slots or - * invalidated existing ones. - */ - if (InRecovery && InHotStandby && - xlrec.wal_level < WAL_LEVEL_LOGICAL && - wal_level >= WAL_LEVEL_LOGICAL) - InvalidateObsoleteReplicationSlots(RS_INVAL_WAL_LEVEL, - 0, InvalidOid, - InvalidTransactionId); - LWLockAcquire(ControlFileLock, LW_EXCLUSIVE); ControlFile->MaxConnections = xlrec.MaxConnections; ControlFile->max_worker_processes = xlrec.max_worker_processes; @@ -8750,6 +8787,44 @@ xlog_redo(XLogReaderState *record) { /* nothing to do here, just for informational purposes */ } + else if (info == XLOG_LOGICAL_DECODING_STATUS_CHANGE) + { + bool logical_decoding; + + /* Update the status on shared memory */ + memcpy(&logical_decoding, XLogRecGetData(record), sizeof(bool)); + UpdateLogicalDecodingStatus(logical_decoding, true); + + if (InRecovery && InHotStandby) + { + if (!logical_decoding) + { + /* + * Invalidate logical slots if we are in hot standby and the + * primary disabled the logical decoding. + */ + InvalidateObsoleteReplicationSlots(RS_INVAL_WAL_LEVEL, + 0, InvalidOid, + InvalidTransactionId); + + /* + * Stop the slotsync worker if it's running. We don't disable + * the slot sync functionality here as we might enable logical + * decoding again during recovery. + */ + if (sync_replication_slots) + ShutDownSlotSync(false); + } + else if (sync_replication_slots) + { + /* + * If logical decoding becomes enabled, wake up postmaster to + * launch the slotsync worker. + */ + kill(PostmasterPid, SIGUSR1); + } + } + } } /* diff --git a/src/backend/access/transam/xlogrecovery.c b/src/backend/access/transam/xlogrecovery.c index e8f3ba00caa..4dc5e39325c 100644 --- a/src/backend/access/transam/xlogrecovery.c +++ b/src/backend/access/transam/xlogrecovery.c @@ -1491,7 +1491,7 @@ FinishWalRecovery(void) * 'synced' column as true after promotion as it may provide useful * information about the slot origin. */ - ShutDownSlotSync(); + ShutDownSlotSync(true); /* * We are now done reading the xlog from stream. Turn off streaming diff --git a/src/backend/commands/publicationcmds.c b/src/backend/commands/publicationcmds.c index 1bf7eaae5b3..f21fddd2722 100644 --- a/src/backend/commands/publicationcmds.c +++ b/src/backend/commands/publicationcmds.c @@ -38,6 +38,7 @@ #include "parser/parse_clause.h" #include "parser/parse_collate.h" #include "parser/parse_relation.h" +#include "replication/logicalctl.h" #include "rewrite/rewriteHandler.h" #include "storage/lmgr.h" #include "utils/acl.h" @@ -960,11 +961,11 @@ CreatePublication(ParseState *pstate, CreatePublicationStmt *stmt) InvokeObjectPostCreateHook(PublicationRelationId, puboid, 0); - if (wal_level != WAL_LEVEL_LOGICAL) + if (!IsLogicalDecodingEnabled()) ereport(WARNING, (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), - errmsg("\"wal_level\" is insufficient to publish logical changes"), - errhint("Set \"wal_level\" to \"logical\" before creating subscriptions."))); + errmsg("logical decoding needs to be enabled to publish logical changes"), + errhint("Set \"wal_level\" to \"logical\" or create a logical replication slot with \"replica\" \"wal_level\" before creating subscriptions."))); return myself; } diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c index cb811520c29..0da2c244144 100644 --- a/src/backend/commands/tablecmds.c +++ b/src/backend/commands/tablecmds.c @@ -2294,7 +2294,7 @@ ExecuteTruncateGuts(List *explicit_rels, xl_heap_truncate xlrec; int i = 0; - /* should only get here if wal_level >= logical */ + /* should only get here if effective WAL level is 'logical' */ Assert(XLogLogicalInfoActive()); logrelids = palloc(list_length(relids_logged) * sizeof(Oid)); diff --git a/src/backend/replication/logical/Makefile b/src/backend/replication/logical/Makefile index 1e08bbbd4eb..50ec127e9ef 100644 --- a/src/backend/replication/logical/Makefile +++ b/src/backend/replication/logical/Makefile @@ -20,6 +20,7 @@ OBJS = \ decode.o \ launcher.o \ logical.o \ + logicalctl.o \ logicalfuncs.o \ message.o \ origin.o \ diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c index cc03f0706e9..23955f29065 100644 --- a/src/backend/replication/logical/decode.c +++ b/src/backend/replication/logical/decode.c @@ -150,44 +150,43 @@ xlog_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf) */ break; case XLOG_PARAMETER_CHANGE: + + /* + * Even if wal_level on the primary got decreased to 'replica' it + * doesn't necessarily mean to disable the logical decoding as + * long as we have at least one logical slot. So we don't check + * the logical decoding availability here but do in + * XLOG_LOGICAL_DECODING_STATUS_CHANGE case. + */ + break; + case XLOG_NOOP: + case XLOG_NEXTOID: + case XLOG_SWITCH: + case XLOG_BACKUP_END: + case XLOG_RESTORE_POINT: + case XLOG_FPW_CHANGE: + case XLOG_FPI_FOR_HINT: + case XLOG_FPI: + case XLOG_OVERWRITE_CONTRECORD: + case XLOG_CHECKPOINT_REDO: + break; + case XLOG_LOGICAL_DECODING_STATUS_CHANGE: { - xl_parameter_change *xlrec = - (xl_parameter_change *) XLogRecGetData(buf->record); + bool *logical_decoding = (bool *) XLogRecGetData(buf->record); - /* - * If wal_level on the primary is reduced to less than - * logical, we want to prevent existing logical slots from - * being used. Existing logical slots on the standby get - * invalidated when this WAL record is replayed; and further, - * slot creation fails when wal_level is not sufficient; but - * all these operations are not synchronized, so a logical - * slot may creep in while the wal_level is being reduced. - * Hence this extra check. - */ - if (xlrec->wal_level < WAL_LEVEL_LOGICAL) + if (!(*logical_decoding)) { /* * This can occur only on a standby, as a primary would - * not allow to restart after changing wal_level < logical + * not allow to restart after changing wal_level < replica * if there is pre-existing logical slot. */ Assert(RecoveryInProgress()); ereport(ERROR, (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), - errmsg("logical decoding on standby requires \"wal_level\" >= \"logical\" on the primary"))); + errmsg("logical decoding must be enabled on the primary"))); } - break; } - case XLOG_NOOP: - case XLOG_NEXTOID: - case XLOG_SWITCH: - case XLOG_BACKUP_END: - case XLOG_RESTORE_POINT: - case XLOG_FPW_CHANGE: - case XLOG_FPI_FOR_HINT: - case XLOG_FPI: - case XLOG_OVERWRITE_CONTRECORD: - case XLOG_CHECKPOINT_REDO: break; default: elog(ERROR, "unexpected RM_XLOG_ID record type: %u", info); diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c index 7e363a7c05b..8fa22ba55b5 100644 --- a/src/backend/replication/logical/logical.c +++ b/src/backend/replication/logical/logical.c @@ -36,6 +36,7 @@ #include "pgstat.h" #include "replication/decode.h" #include "replication/logical.h" +#include "replication/logicalctl.h" #include "replication/reorderbuffer.h" #include "replication/slotsync.h" #include "replication/snapbuild.h" @@ -117,31 +118,18 @@ CheckLogicalDecodingRequirements(void) * needs the same check. */ - if (wal_level < WAL_LEVEL_LOGICAL) + /* CheckSlotRequirements() has already checked that wal_level >= 'replica' */ + + if (RecoveryInProgress() && !IsLogicalDecodingEnabled()) ereport(ERROR, (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), - errmsg("logical decoding requires \"wal_level\" >= \"logical\""))); + errmsg("logical decoding needs to be enabled on the primary"), + errhint("Set \"wal_level\" >= \"logical\" or create at least one logical slot on the primary."))); if (MyDatabaseId == InvalidOid) ereport(ERROR, (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), errmsg("logical decoding requires a database connection"))); - - if (RecoveryInProgress()) - { - /* - * This check may have race conditions, but whenever - * XLOG_PARAMETER_CHANGE indicates that wal_level has changed, we - * verify that there are no existing logical replication slots. And to - * avoid races around creating a new slot, - * CheckLogicalDecodingRequirements() is called once before creating - * the slot, and once when logical decoding is initially starting up. - */ - if (GetActiveWalLevelOnStandby() < WAL_LEVEL_LOGICAL) - ereport(ERROR, - (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), - errmsg("logical decoding on standby requires \"wal_level\" >= \"logical\" on the primary"))); - } } /* diff --git a/src/backend/replication/logical/logicalctl.c b/src/backend/replication/logical/logicalctl.c new file mode 100644 index 00000000000..74fc9d19551 --- /dev/null +++ b/src/backend/replication/logical/logicalctl.c @@ -0,0 +1,612 @@ +/*------------------------------------------------------------------------- + * logicalctl.c + * Functionality to control logical decoding status online. + * + * This module enables dynamic control of logical decoding availability. + * Logical decoding becomes active under two conditions: when the wal_level + * parameter is set to 'logical', or when at least one logical replication + * slot exists with wal_level set to 'replica'. The system disables logical + * decoding when neither condition is met. + * + * The modules maintains separate controls of two aspects: writing information + * required by logical decoding to WAL records and utilizing logical decoding + * itself. The activation process involves several steps, beginning with + * maintaining logical decoding in a disabled state while incrementing the + * effective WAL level to its 'logical' equivalent. This changes is reflected + * in the read-only effective_wal_level parameter. The process includes + * necessary synchronization to ensure all process adapt to the new effective + * WAL level before logical decoding is fully enabled. Deactivation follows a + * similarly careful, multi-step process in the reverse order. + * + * Portions Copyright (c) 1996-2025, PostgreSQL Global Development Group + * Portions Copyright (c) 1994, Regents of the University of California + * + * IDENTIFICATION + * src/backend/replication/logical/logicalctl.c + * + *------------------------------------------------------------------------- + */ +#include "postgres.h" + +#include "access/xloginsert.h" +#include "catalog/pg_control.h" +#include "miscadmin.h" +#include "storage/procarray.h" +#include "storage/ipc.h" +#include "storage/lmgr.h" +#include "replication/logicalctl.h" +#include "replication/slot.h" +#include "utils/wait_event.h" +#include "utils/wait_event_types.h" + +LogicalDecodingCtlData *LogicalDecodingCtl = NULL; + +/* + * A process local cache of LogicalDecodingCtl->xlog_logical_info. This is + * initialized at process startup time, and could be updated when absorbing + * the process barrier signal in ProcessBarrierUpdateXLogLogicalInfo(). + */ +bool XLogLogicalInfo = false; + +static void update_xlog_logical_info(void); +static void abort_logical_decoding_activation(int code, Datum arg); +static bool read_logical_decoding_status_transition(void); +static bool start_logical_decoding_status_change(bool new_status); + +Size +LogicalDecodingCtlShmemSize(void) +{ + return sizeof(LogicalDecodingCtlData); +} + +void +LogicalDecodingCtlShmemInit(void) +{ + bool found; + + LogicalDecodingCtl = ShmemInitStruct("Logical information control", + LogicalDecodingCtlShmemSize(), + &found); + + if (!found) + { + LogicalDecodingCtl->transition_in_progress = false; + LogicalDecodingCtl->delay_status_change = false; + LogicalDecodingCtl->logical_decoding_enabled = false; + ConditionVariableInit(&LogicalDecodingCtl->transition_cv); + pg_atomic_init_flag(&LogicalDecodingCtl->xlog_logical_info); + } +} + +/* + * Initialize logical decoding status on shmem at server startup. This + * must be called ONCE during postmaster or standalone-backend startup, + * before initializing replication slots. + */ +void +StartupLogicalDecodingStatus(bool last_status) +{ + /* Logical decoding is always disabled when 'minimal' WAL level */ + if (wal_level == WAL_LEVEL_MINIMAL) + return; + + /* + * Set the initial logical decoding status based on the last status. If + * logical decoding was enabled before the last shutdown, it remains + * enabled as we might have set wal_level='logical' or have a few logical + * slots. + */ + if (last_status) + { + pg_atomic_test_set_flag(&(LogicalDecodingCtl->xlog_logical_info)); + LogicalDecodingCtl->logical_decoding_enabled = true; + } +} + +/* + * Update the XLogLogicalInfo cache. + */ +static inline void +update_xlog_logical_info(void) +{ + XLogLogicalInfo = IsXLogLogicalInfoEnabled(); +} + +/* + * Initialize XLogLogicalInfo backend-private cache. This routine is called + * during process initialization. + */ +void +InitializeProcessXLogLogicalInfo(void) +{ + update_xlog_logical_info(); +} + +/* + * This routine is called when we are ordered to update XLogLogicalInfo + * by a ProcSignalBarrier. + */ +bool +ProcessBarrierUpdateXLogLogicalInfo(void) +{ + update_xlog_logical_info(); + return true; +} + +/* + * Check the shared memory state and return true if logical decoding is + * enabled on the system. + */ +bool +IsLogicalDecodingEnabled(void) +{ + bool enabled; + + LWLockAcquire(LogicalDecodingControlLock, LW_SHARED); + enabled = LogicalDecodingCtl->logical_decoding_enabled; + LWLockRelease(LogicalDecodingControlLock); + + return enabled; +} + +/* + * Check the shared memory state and return true if logical information WAL + * logging is enabled. + */ +bool +IsXLogLogicalInfoEnabled(void) +{ + return !pg_atomic_unlocked_test_flag(&(LogicalDecodingCtl->xlog_logical_info)); +} + +/* + * Enable or disable both status of logical info WAL logging and logical decoding + * on shared memory. + + * Note that this function updates the global flags without the state transition + * process. EnsureLogicalDecodingEnabled() and DisableLogicalDecodingIfNecessary() + * should be used instead if there could be concurrent processes doing writes + * or logical decoding. + */ +void +UpdateLogicalDecodingStatus(bool new_status, bool need_lock) +{ + if (need_lock) + LWLockAcquire(LogicalDecodingControlLock, LW_EXCLUSIVE); + + LogicalDecodingCtl->logical_decoding_enabled = new_status; + + if (new_status) + pg_atomic_test_set_flag(&(LogicalDecodingCtl->xlog_logical_info)); + else + pg_atomic_clear_flag(&(LogicalDecodingCtl->xlog_logical_info)); + + if (need_lock) + LWLockRelease(LogicalDecodingControlLock); + + elog(DEBUG1, "update logical decoding status to %d", new_status); +} + +/* + * A PG_ENSURE_ERROR_CLEANUP callback for activating logical decoding. + */ +static void +abort_logical_decoding_activation(int code, Datum arg) +{ + Assert(LogicalDecodingCtl->transition_in_progress); + + elog(DEBUG1, "aborting the process of logical decoding activation"); + + pg_atomic_clear_flag(&(LogicalDecodingCtl->xlog_logical_info)); + + /* + * We don't need to wait for all processes to disable xlog_logical_info + * locally as it's always safe to write logical information to WAL + * records, even when not strictly required. + */ + EmitProcSignalBarrier(PROCSIGNAL_BARRIER_UPDATE_XLOG_LOGICAL_INFO); + + LWLockAcquire(LogicalDecodingControlLock, LW_EXCLUSIVE); + LogicalDecodingCtl->logical_decoding_enabled = false; + LogicalDecodingCtl->transition_in_progress = false; + LWLockRelease(LogicalDecodingControlLock); + + /* Let waiters know the WAL level change completed */ + ConditionVariableBroadcast(&LogicalDecodingCtl->transition_cv); +} + +/* + * A helper routine for start_logical_decoding_status_change() to read the + * current logical decoding status possibly with a wait in case the + * transaction is in progress. This function returns to the caller while + * holding LogicalDecodingControlLock in an exclusive mode. + */ +static bool +read_logical_decoding_status_transition(void) +{ +retry: + LWLockAcquire(LogicalDecodingControlLock, LW_EXCLUSIVE); + + /* + * We check the current status to see if we need to change it. If a status + * change is in-progress, we need to wait for completion. + */ + if (LogicalDecodingCtl->transition_in_progress) + { + /* Release the lock and wait for someone to complete the transition */ + LWLockRelease(LogicalDecodingControlLock); + ConditionVariableSleep(&LogicalDecodingCtl->transition_cv, + WAIT_EVENT_LOGICAL_DECODING_STATUS_CHANGE); + + goto retry; + } + + /* Keep holding LogicalDecodingControlLock */ + return LogicalDecodingCtl->logical_decoding_enabled; +} + +/* + * This function does several kinds of preparation works required to start + * the process of logical decoding status change. If the status change is + * required, it ensures we can change logical decoding status, set + * LogicalDecodingCtl->transition_in_progress on and returns true. + * Otherwise, if it's not required or not allowed (e.g., during recovery), + * it returns false. + */ +static bool +start_logical_decoding_status_change(bool new_status) +{ +retry: + + /* + * On the primary with 'logical' WAL level, we can skip logical decoding + * status change as it's always enabled. On standbys, we need to check the + * status on shared memory propagated from the primary and might handle + * status change delay. + */ + if (!RecoveryInProgress() && wal_level == WAL_LEVEL_LOGICAL) + return false; + + /* Return if no need to change the status */ + if (read_logical_decoding_status_transition() == new_status) + { + LWLockRelease(LogicalDecodingControlLock); + return false; + } + + if (LogicalDecodingCtl->transition_in_progress) + { + LWLockRelease(LogicalDecodingControlLock); + + /* Wait for someone to complete the transition */ + ConditionVariableSleep(&LogicalDecodingCtl->transition_cv, + WAIT_EVENT_LOGICAL_DECODING_STATUS_CHANGE); + + goto retry; + } + + /* Return if we don't need to change the status */ + if (LogicalDecodingCtl->logical_decoding_enabled == new_status) + { + LWLockRelease(LogicalDecodingControlLock); + return false; + } + + /* + * When attempting to disable logical decoding, if there is at least one + * logical slots we cannot disable it. + */ + if (!new_status && CheckLogicalSlotExists()) + { + LWLockRelease(LogicalDecodingControlLock); + return false; + } + + /* + * We need to change the status but need to check if it's allowed. We + * always are allowed to change logical decoding status if we're not in + * the recovery. During recovery, there is a race condition with the + * startup process's end-of-recovery action; after the startup process + * updates logical decoding status at the end of recovery, it's possible + * that other processes try to enable or disable logical decoding status + * before the recovery completes but are unable to write WAL records. + * Therefore, if the startup process has done its end-of-recovery work, we + * need to wait for the recovery to finish. + */ + if (RecoveryInProgress()) + { + bool delay_status_change; + + delay_status_change = LogicalDecodingCtl->delay_status_change; + LWLockRelease(LogicalDecodingControlLock); + + /* + * If we're in recovery and the startup process is still taking + * responsibility to update the status, we cannot change. + */ + if (!delay_status_change) + return false; + + elog(DEBUG1, + "waiting for recovery completion to change logical decoding status"); + + /* + * The startup process already updated logical decoding status at the + * end of recovery but it might not be allowed to write WAL records + * yet. Wait for the recovery to complete and check the status again. + */ + while (RecoveryInProgress()) + { + pgstat_report_wait_start(WAIT_EVENT_LOGICAL_DECODING_STATUS_CHANGE_DELAY); + pg_usleep(100000L); /* wait for 100 msec */ + pgstat_report_wait_end(); + } + + /* + * The status might have changed while waiting for the recovery + * completion. So retry from the beginning. + */ + goto retry; + } + + /* Mark the state transition is in-progress */ + LogicalDecodingCtl->transition_in_progress = true; + + LWLockRelease(LogicalDecodingControlLock); + + return true; +} + +/* + * Enable logical decoding if disabled. + * + * Note that there is no interlock between logical decoding activation + * and slot creation. To ensure enabling logical decoding the caller + * needs to call this function after creating a logical slot without + * initializing its logical decoding context. + */ +void +EnsureLogicalDecodingEnabled(void) +{ + if (wal_level == WAL_LEVEL_MINIMAL) + return; + + /* Prepare and start the activation process if it's disabled */ + if (!start_logical_decoding_status_change(true)) + return; + + /* + * Ensure we reset the activation process if we cancelled or errored out + * below + */ + PG_ENSURE_ERROR_CLEANUP(abort_logical_decoding_activation, (Datum) 0); + { + RunningTransactions running; + + /* + * Set logical info WAL logging on the shmem. All process starts after + * this point will include the information required by logical + * decoding to WAL records. + */ + pg_atomic_test_set_flag(&(LogicalDecodingCtl->xlog_logical_info)); + + /* + * Order all running processes to reflect the xlog_logical_info + * update, and wait. This ensures that all running processes have + * enabled logical information WAL logging. + */ + WaitForProcSignalBarrier( + EmitProcSignalBarrier(PROCSIGNAL_BARRIER_UPDATE_XLOG_LOGICAL_INFO)); + + /* + * While all processes are using the new status, there could be some + * transactions that might have started with the old status. So wait + * for the running transactions to complete so that logical decoding + * doesn't include transactions that wrote WAL with insufficient + * information. + */ + running = GetRunningTransactionData(); + LWLockRelease(ProcArrayLock); + LWLockRelease(XidGenLock); + + for (int i = 0; i < running->xcnt; i++) + { + TransactionId xid = running->xids[i]; + + if (TransactionIdIsCurrentTransactionId(xid)) + continue; + + XactLockTableWait(xid, NULL, NULL, XLTW_None); + } + } + PG_END_ENSURE_ERROR_CLEANUP(abort_logical_decoding_activation, (Datum) 0); + + START_CRIT_SECTION(); + + /* + * Here, we can ensure that all running transactions are using the new + * xlog_logical_info value, writing logical information to WAL records. So + * now enable logical decoding globally. + * + * It's always safe to write logical information to WAL records even when + * not strictly required. So we first enable it and write the WAL record + * below. + */ + LWLockAcquire(LogicalDecodingControlLock, LW_EXCLUSIVE); + LogicalDecodingCtl->logical_decoding_enabled = true; + LogicalDecodingCtl->transition_in_progress = false; + LWLockRelease(LogicalDecodingControlLock); + + { + XLogRecPtr recptr; + bool logical_decoding = true; + + XLogBeginInsert(); + XLogRegisterData(&logical_decoding, sizeof(bool)); + recptr = XLogInsert(RM_XLOG_ID, XLOG_LOGICAL_DECODING_STATUS_CHANGE); + XLogFlush(recptr); + } + + END_CRIT_SECTION(); + + ereport(LOG, + (errmsg("logical decoding is enabled upon creating a new logical replication slot"))); + + /* Let waiters know the work finished */ + ConditionVariableBroadcast(&LogicalDecodingCtl->transition_cv); +} + +/* + * Disable logical decoding if enabled. + * + * This function expects to be called after dropping a possibly-last logical + * replication slot. Logical decoding can be disabled only when wal_level is set + * to 'replica' and there is no logical replication slot on the system. + * + * XXX: This function could write a WAL record in order to tell the standbys + * know logical decoding got disabled. However, we need to note that this + * function could be called during process exits (e.g., by ReplicationSlotCleanup() + * via before_shmem_exit callbacks), which looks something that we want to + * avoid. + */ +void +DisableLogicalDecodingIfNecessary(void) +{ + bool recoveryInProgress = RecoveryInProgress(); + + if (wal_level == WAL_LEVEL_MINIMAL) + return; + + /* Prepare and start the deactivation process if it's enabled */ + if (!start_logical_decoding_status_change(false)) + return; + + /* + * We don't need PG_ENSURE_ERROR_CLEANUP() to abort the deactivation + * process since we can expect all operations below not to throw ERROR or + * FATAL. + */ + + START_CRIT_SECTION(); + + /* + * When disabling logical decoding, we need to disable logical decoding + * first and disable logical information WAL logging in order to ensure + * that no logical decoding processes WAL records with insufficient + * information. + */ + LWLockAcquire(LogicalDecodingControlLock, LW_EXCLUSIVE); + LogicalDecodingCtl->logical_decoding_enabled = false; + LWLockRelease(LogicalDecodingControlLock); + + /* Write the WAL to disable logical decoding on standbys too */ + if (XLogStandbyInfoActive() && !recoveryInProgress) + { + bool logical_decoding = false; + XLogRecPtr recptr; + + XLogBeginInsert(); + XLogRegisterData(&logical_decoding, sizeof(bool)); + recptr = XLogInsert(RM_XLOG_ID, XLOG_LOGICAL_DECODING_STATUS_CHANGE); + XLogFlush(recptr); + } + + /* Now disable logical information WAL logging */ + pg_atomic_clear_flag(&(LogicalDecodingCtl->xlog_logical_info)); + + /* + * Order all running processes to reflect the xlog_logical_info update. + * Unlike when enabling logical decoding, we don't need to wait for all + * processes to complete it in this case. We already disabled logical + * decoding and it's always safe to write logical information to WAL + * records, even when not strictly required. Therefore, we don't need to + * wait for all running transactions to finish either. + */ + EmitProcSignalBarrier(PROCSIGNAL_BARRIER_UPDATE_XLOG_LOGICAL_INFO); + + /* Complete the transition */ + LWLockAcquire(LogicalDecodingControlLock, LW_EXCLUSIVE); + LogicalDecodingCtl->transition_in_progress = false; + LWLockRelease(LogicalDecodingControlLock); + + END_CRIT_SECTION(); + + ereport(LOG, + (errmsg("logical decoding is disabled because all logical replication slots are removed"))); + + /* Let waiters know the work finished */ + ConditionVariableBroadcast(&LogicalDecodingCtl->transition_cv); +} + +/* + * Update logical decoding status at end of the recovery. This function + * must be called before accepting writes. + */ +void +UpdateLogicalDecodingStatusEndOfRecovery(void) +{ + bool new_status = false; + bool need_wal = false; + + Assert(RecoveryInProgress()); + Assert(!LogicalDecodingCtl->transition_in_progress); + + /* With 'minimal' WAL level, logical decoding is always disabled */ + if (wal_level == WAL_LEVEL_MINIMAL) + return; + + LWLockAcquire(LogicalDecodingControlLock, LW_EXCLUSIVE); + + /* + * We can use logical decoding if we're using 'logical' WAL level or there + * is at least one logical replication slot. + */ + if (wal_level == WAL_LEVEL_LOGICAL || CheckLogicalSlotExists()) + new_status = true; + + if (LogicalDecodingCtl->logical_decoding_enabled != new_status) + need_wal = true; + + /* + * Update shmem flags. We don't need to care about the order of setting + * global flag and writing the WAL record this case since writes are not + * allowed yet. + */ + UpdateLogicalDecodingStatus(new_status, false); + + /* + * We disallow any logical decoding status change until we actually + * completes the recovery, i.e., RecoveryInProgress() returns false. This + * is necessary to deal with the race condition that could happen after + * this point; processes are able to create or drop logical replication + * slots and tries to enable or disable logical decoding accordingly, but + * they are not allowed to write any WAL records until the recovery + * completes. + */ + LogicalDecodingCtl->delay_status_change = true; + + LWLockRelease(LogicalDecodingControlLock); + + if (need_wal) + { + XLogRecPtr recptr; + + Assert(XLogStandbyInfoActive()); + + XLogBeginInsert(); + XLogRegisterData(&new_status, sizeof(bool)); + recptr = XLogInsert(RM_XLOG_ID, XLOG_LOGICAL_DECODING_STATUS_CHANGE); + XLogFlush(recptr); + } + + /* + * Ensure all running processes have the updated status. We don't need to + * wait for running transactions to finish as we don't accept any writes + * yet. We need the wait even if we've not updated the status above as the + * status have been turned on and off during recovery, having running + * processes have different status on their local caches. + */ + if (IsUnderPostmaster) + WaitForProcSignalBarrier( + EmitProcSignalBarrier(PROCSIGNAL_BARRIER_UPDATE_XLOG_LOGICAL_INFO)); +} diff --git a/src/backend/replication/logical/meson.build b/src/backend/replication/logical/meson.build index 6f19614c79d..19c7130b961 100644 --- a/src/backend/replication/logical/meson.build +++ b/src/backend/replication/logical/meson.build @@ -6,6 +6,7 @@ backend_sources += files( 'decode.c', 'launcher.c', 'logical.c', + 'logicalctl.c', 'logicalfuncs.c', 'message.c', 'origin.c', diff --git a/src/backend/replication/logical/slotsync.c b/src/backend/replication/logical/slotsync.c index 2f0c08b8fbd..4038a317f61 100644 --- a/src/backend/replication/logical/slotsync.c +++ b/src/backend/replication/logical/slotsync.c @@ -57,6 +57,7 @@ #include "pgstat.h" #include "postmaster/interrupt.h" #include "replication/logical.h" +#include "replication/logicalctl.h" #include "replication/slotsync.h" #include "replication/snapbuild.h" #include "storage/ipc.h" @@ -422,12 +423,12 @@ local_sync_slot_required(ReplicationSlot *local_slot, List *remote_slots) * drop and recreate such slots as long as these are not consumable on the * standby (which is the case currently). * - * Note: Change of 'wal_level' on the primary server to a level lower than - * logical may also result in slot invalidation and removal on the standby. - * This is because such 'wal_level' change is only possible if the logical - * slots are removed on the primary server, so it's expected to see the - * slots being invalidated and removed on the standby too (and re-created - * if they are re-created on the primary server). + * Note: Disabling logical decoding on the primary server may also result in + * in slot invalidation and removal on the standby. This is because logical + * decoding is disabled only if the logical slots are removed on the primary + * server (and setting 'wal_level' lower than 'logical'), so it's expected + * to see the slots being invalidated and removed on the standby too (and + * re-created if they are re-created on the primary server). */ static void drop_local_obsolete_slots(List *remote_slot_list) @@ -1058,15 +1059,20 @@ bool ValidateSlotSyncParams(int elevel) { /* - * Logical slot sync/creation requires wal_level >= logical. + * Logical slot sync/creation requires to the logical decoding to be + * enabled. * * Since altering the wal_level requires a server restart, so error out in * this case regardless of elevel provided by caller. */ - if (wal_level < WAL_LEVEL_LOGICAL) - ereport(ERROR, - errcode(ERRCODE_INVALID_PARAMETER_VALUE), - errmsg("replication slot synchronization requires \"wal_level\" >= \"logical\"")); + if (!IsLogicalDecodingEnabled()) + { + ereport(elevel, + errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("replication slot synchronization requires logical decoding to be enabled"), + errhint("Set \"wal_level\" >= \"logical\" or create at least one logical slot on the primary ")); + return false; + } /* * A physical replication slot(primary_slot_name) is required on the @@ -1144,7 +1150,7 @@ slotsync_reread_config(void) if (old_sync_replication_slots != sync_replication_slots) { ereport(LOG, - /* translator: %s is a GUC variable name */ + /* translator: %s is a GUC variable name */ errmsg("replication slot synchronization worker will shut down because \"%s\" is disabled", "sync_replication_slots")); proc_exit(0); } @@ -1575,15 +1581,19 @@ update_synced_slots_inactive_since(void) * This function sends signal to shutdown slot sync worker, if required. It * also waits till the slot sync worker has exited or * pg_sync_replication_slots() has finished. + * + * If permanent is true, it also sets stopSignaled, disabling the slot sync + * functionality until the next server restart. */ void -ShutDownSlotSync(void) +ShutDownSlotSync(bool permanent) { pid_t worker_pid; SpinLockAcquire(&SlotSyncCtx->mutex); - SlotSyncCtx->stopSignaled = true; + if (permanent) + SlotSyncCtx->stopSignaled = true; /* * Return if neither the slot sync worker is running nor the function diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c index 8605776ad86..4c0ba9939f8 100644 --- a/src/backend/replication/slot.c +++ b/src/backend/replication/slot.c @@ -48,6 +48,7 @@ #include "pgstat.h" #include "postmaster/interrupt.h" #include "replication/logicallauncher.h" +#include "replication/logicalctl.h" #include "replication/slotsync.h" #include "replication/slot.h" #include "replication/walsender_private.h" @@ -732,16 +733,15 @@ ReplicationSlotRelease(void) { ReplicationSlot *slot = MyReplicationSlot; char *slotname = NULL; /* keep compiler quiet */ - bool is_logical = false; /* keep compiler quiet */ + bool is_logical; TimestampTz now = 0; Assert(slot != NULL && slot->active_pid != 0); + is_logical = SlotIsLogical(slot); + if (am_walsender) - { slotname = pstrdup(NameStr(slot->data.name)); - is_logical = SlotIsLogical(slot); - } if (slot->data.persistency == RS_EPHEMERAL) { @@ -751,6 +751,9 @@ ReplicationSlotRelease(void) * data. */ ReplicationSlotDropAcquired(); + + if (is_logical) + DisableLogicalDecodingIfNecessary(); } /* @@ -820,10 +823,13 @@ void ReplicationSlotCleanup(bool synced_only) { int i; + bool dropped_logical = false; + int nlogicalslots; Assert(MyReplicationSlot == NULL); restart: + nlogicalslots = 0; LWLockAcquire(ReplicationSlotControlLock, LW_SHARED); for (i = 0; i < max_replication_slots; i++) { @@ -832,6 +838,9 @@ restart: if (!s->in_use) continue; + if (SlotIsLogical(s)) + nlogicalslots++; + SpinLockAcquire(&s->mutex); if ((s->active_pid == MyProcPid && (!synced_only || s->data.synced))) @@ -840,6 +849,9 @@ restart: SpinLockRelease(&s->mutex); LWLockRelease(ReplicationSlotControlLock); /* avoid deadlock */ + if (SlotIsLogical(s)) + dropped_logical = true; + ReplicationSlotDropPtr(s); ConditionVariableBroadcast(&s->active_cv); @@ -850,6 +862,9 @@ restart: } LWLockRelease(ReplicationSlotControlLock); + + if (dropped_logical && nlogicalslots == 0) + DisableLogicalDecodingIfNecessary(); } /* @@ -858,6 +873,8 @@ restart: void ReplicationSlotDrop(const char *name, bool nowait) { + bool is_logical; + Assert(MyReplicationSlot == NULL); ReplicationSlotAcquire(name, nowait, false); @@ -872,7 +889,12 @@ ReplicationSlotDrop(const char *name, bool nowait) errmsg("cannot drop replication slot \"%s\"", name), errdetail("This replication slot is being synchronized from the primary server.")); + is_logical = SlotIsLogical(MyReplicationSlot); + ReplicationSlotDropAcquired(); + + if (is_logical) + DisableLogicalDecodingIfNecessary(); } /* @@ -1408,11 +1430,14 @@ void ReplicationSlotsDropDBSlots(Oid dboid) { int i; + int nlogicalslots; + bool dropped = false; if (max_replication_slots <= 0) return; restart: + nlogicalslots = 0; LWLockAcquire(ReplicationSlotControlLock, LW_SHARED); for (i = 0; i < max_replication_slots; i++) { @@ -1436,6 +1461,8 @@ restart: /* NB: intentionally including invalidated slots */ + nlogicalslots++; + /* acquire slot, so ReplicationSlotDropAcquired can be reused */ SpinLockAcquire(&s->mutex); /* can't change while ReplicationSlotControlLock is held */ @@ -1486,11 +1513,49 @@ restart: */ LWLockRelease(ReplicationSlotControlLock); ReplicationSlotDropAcquired(); + dropped = true; goto restart; } LWLockRelease(ReplicationSlotControlLock); + + if (dropped && nlogicalslots == 0) + DisableLogicalDecodingIfNecessary(); } +/* + * Returns if there is at least in-use logical replication slot. + */ +bool +CheckLogicalSlotExists(void) +{ + bool found = false; + + if (max_replication_slots <= 0) + return false; + + LWLockAcquire(ReplicationSlotControlLock, LW_SHARED); + for (int i = 0; i < max_replication_slots; i++) + { + ReplicationSlot *s; + + s = &ReplicationSlotCtl->replication_slots[i]; + + /* cannot change while ReplicationSlotCtlLock is held */ + if (!s->in_use) + continue; + + /* NB: intentionally counting invalidated slots */ + + if (SlotIsLogical(s)) + { + found = true; + break; + } + } + LWLockRelease(ReplicationSlotControlLock); + + return found; +} /* * Check whether the server's configuration supports using replication @@ -1651,7 +1716,7 @@ ReportSlotInvalidation(ReplicationSlotInvalidationCause cause, break; case RS_INVAL_WAL_LEVEL: - appendStringInfoString(&err_detail, _("Logical decoding on standby requires \"wal_level\" >= \"logical\" on the primary server.")); + appendStringInfoString(&err_detail, _("Logical decoding on standby requires \"wal_level\" >= \"logical\" or to create at least one logical slot on the primary server.")); break; case RS_INVAL_IDLE_TIMEOUT: @@ -2041,7 +2106,8 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes, * - RS_INVAL_WAL_REMOVED: requires a LSN older than the given segment * - RS_INVAL_HORIZON: requires a snapshot <= the given horizon in the given * db; dboid may be InvalidOid for shared relations - * - RS_INVAL_WAL_LEVEL: is logical and wal_level is insufficient + * - RS_INVAL_WAL_LEVEL: is logical and logical decoding is disabled due to + * insufficient WAL level. * - RS_INVAL_IDLE_TIMEOUT: has been idle longer than the configured * "idle_replication_slot_timeout" duration. * @@ -2622,12 +2688,12 @@ RestoreSlotFromDisk(const char *name) */ if (cp.slotdata.database != InvalidOid) { - if (wal_level < WAL_LEVEL_LOGICAL) + if (wal_level < WAL_LEVEL_REPLICA) ereport(FATAL, (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), - errmsg("logical replication slot \"%s\" exists, but \"wal_level\" < \"logical\"", + errmsg("logical replication slot \"%s\" exists, but \"wal_level\" < \"replica\"", NameStr(cp.slotdata.name)), - errhint("Change \"wal_level\" to be \"logical\" or higher."))); + errhint("Change \"wal_level\" to be \"replica\" or higher."))); /* * In standby mode, the hot standby must be enabled. This check is diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c index 69f4c6157c5..f852e258d95 100644 --- a/src/backend/replication/slotfuncs.c +++ b/src/backend/replication/slotfuncs.c @@ -18,6 +18,7 @@ #include "access/xlogutils.h" #include "funcapi.h" #include "replication/logical.h" +#include "replication/logicalctl.h" #include "replication/slot.h" #include "replication/slotsync.h" #include "utils/builtins.h" @@ -136,6 +137,12 @@ create_logical_replication_slot(char *name, char *plugin, temporary ? RS_TEMPORARY : RS_EPHEMERAL, two_phase, failover, false); + /* + * Ensure the logical decoding is enabled before initializing the logical + * decoding context. + */ + EnsureLogicalDecodingEnabled(); + /* * Create logical decoding context to find start point or, if we don't * need it, to 1) bump slot's restart_lsn and xmin 2) check plugin sanity. diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index ee911394a23..249b5ba11ff 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -72,6 +72,7 @@ #include "postmaster/interrupt.h" #include "replication/decode.h" #include "replication/logical.h" +#include "replication/logicalctl.h" #include "replication/slotsync.h" #include "replication/slot.h" #include "replication/snapbuild.h" @@ -1282,6 +1283,12 @@ CreateReplicationSlot(CreateReplicationSlotCmd *cmd) need_full_snapshot = true; } + /* + * Ensure the logical decoding is enabled before initializing the + * logical decoding context. + */ + EnsureLogicalDecodingEnabled(); + ctx = CreateInitDecodingContext(cmd->plugin, NIL, need_full_snapshot, InvalidXLogRecPtr, XL_ROUTINE(.page_read = logical_read_xlog_page, diff --git a/src/backend/storage/ipc/ipci.c b/src/backend/storage/ipc/ipci.c index 2fa045e6b0f..f1ef837755c 100644 --- a/src/backend/storage/ipc/ipci.c +++ b/src/backend/storage/ipc/ipci.c @@ -31,6 +31,7 @@ #include "postmaster/bgworker_internals.h" #include "postmaster/bgwriter.h" #include "postmaster/walsummarizer.h" +#include "replication/logicalctl.h" #include "replication/logicallauncher.h" #include "replication/origin.h" #include "replication/slot.h" @@ -150,6 +151,7 @@ CalculateShmemSize(int *num_semaphores) size = add_size(size, InjectionPointShmemSize()); size = add_size(size, SlotSyncShmemSize()); size = add_size(size, AioShmemSize()); + size = add_size(size, LogicalDecodingCtlShmemSize()); /* include additional requested shmem from preload libraries */ size = add_size(size, total_addin_request); @@ -343,6 +345,7 @@ CreateOrAttachShmemStructs(void) WaitEventCustomShmemInit(); InjectionPointShmemInit(); AioShmemInit(); + LogicalDecodingCtlShmemInit(); } /* diff --git a/src/backend/storage/ipc/procsignal.c b/src/backend/storage/ipc/procsignal.c index a9bb540b55a..0c3788cb836 100644 --- a/src/backend/storage/ipc/procsignal.c +++ b/src/backend/storage/ipc/procsignal.c @@ -22,6 +22,7 @@ #include "miscadmin.h" #include "pgstat.h" #include "port/pg_bitutils.h" +#include "replication/logicalctl.h" #include "replication/logicalworker.h" #include "replication/walsender.h" #include "storage/condition_variable.h" @@ -576,6 +577,9 @@ ProcessProcSignalBarrier(void) case PROCSIGNAL_BARRIER_SMGRRELEASE: processed = ProcessBarrierSmgrRelease(); break; + case PROCSIGNAL_BARRIER_UPDATE_XLOG_LOGICAL_INFO: + processed = ProcessBarrierUpdateXLogLogicalInfo(); + break; } /* diff --git a/src/backend/storage/ipc/standby.c b/src/backend/storage/ipc/standby.c index 4222bdab078..a65ce9cd02b 100644 --- a/src/backend/storage/ipc/standby.c +++ b/src/backend/storage/ipc/standby.c @@ -24,6 +24,7 @@ #include "access/xlogutils.h" #include "miscadmin.h" #include "pgstat.h" +#include "replication/logicalctl.h" #include "replication/slot.h" #include "storage/bufmgr.h" #include "storage/proc.h" @@ -499,7 +500,7 @@ ResolveRecoveryConflictWithSnapshot(TransactionId snapshotConflictHorizon, * seems OK, given that this kind of conflict should not normally be * reached, e.g. due to using a physical replication slot. */ - if (wal_level >= WAL_LEVEL_LOGICAL && isCatalogRel) + if (IsLogicalDecodingEnabled() && isCatalogRel) InvalidateObsoleteReplicationSlots(RS_INVAL_HORIZON, 0, locator.dbOid, snapshotConflictHorizon); } @@ -1325,13 +1326,13 @@ LogStandbySnapshot(void) * record. Fortunately this routine isn't executed frequently, and it's * only a shared lock. */ - if (wal_level < WAL_LEVEL_LOGICAL) + if (!IsLogicalDecodingEnabled()) LWLockRelease(ProcArrayLock); recptr = LogCurrentRunningXacts(running); /* Release lock if we kept it longer ... */ - if (wal_level >= WAL_LEVEL_LOGICAL) + if (IsLogicalDecodingEnabled()) LWLockRelease(ProcArrayLock); /* GetRunningTransactionData() acquired XidGenLock, we must release it */ diff --git a/src/backend/utils/activity/wait_event_names.txt b/src/backend/utils/activity/wait_event_names.txt index 0be307d2ca0..b35c6dd9159 100644 --- a/src/backend/utils/activity/wait_event_names.txt +++ b/src/backend/utils/activity/wait_event_names.txt @@ -133,6 +133,8 @@ HASH_GROW_BUCKETS_ELECT "Waiting to elect a Parallel Hash participant to allocat HASH_GROW_BUCKETS_REALLOCATE "Waiting for an elected Parallel Hash participant to finish allocating more buckets." HASH_GROW_BUCKETS_REINSERT "Waiting for other Parallel Hash participants to finish inserting tuples into new buckets." LOGICAL_APPLY_SEND_DATA "Waiting for a logical replication leader apply process to send data to a parallel apply process." +LOGICAL_DECODING_STATUS_CHANGE "Waiting for logical decoding status change." +LOGICAL_DECODING_STATUS_CHANGE_DELAY "Waiting for recovery to complete to change logical decoding status." LOGICAL_PARALLEL_APPLY_STATE_CHANGE "Waiting for a logical replication parallel apply process to change state." LOGICAL_SYNC_DATA "Waiting for a logical replication remote server to send data for initial table synchronization." LOGICAL_SYNC_STATE_CHANGE "Waiting for a logical replication remote server to change state." @@ -352,6 +354,7 @@ DSMRegistry "Waiting to read or update the dynamic shared memory registry." InjectionPoint "Waiting to read or update information related to injection points." SerialControl "Waiting to read or update shared pg_serial state." AioWorkerSubmissionQueue "Waiting to access AIO worker submission queue." +LogicalDecodingControl "Waiting to access logical decoding status information." # # END OF PREDEFINED LWLOCKS (DO NOT CHANGE THIS LINE) diff --git a/src/backend/utils/cache/inval.c b/src/backend/utils/cache/inval.c index 02505c88b8e..6ea31f74d4c 100644 --- a/src/backend/utils/cache/inval.c +++ b/src/backend/utils/cache/inval.c @@ -98,9 +98,9 @@ * likewise send the invalidation immediately, before ending the change's * critical section. This includes inplace heap updates, relmap, and smgr. * - * When wal_level=logical, write invalidations into WAL at each command end to - * support the decoding of the in-progress transactions. See - * CommandEndInvalidationMessages. + * When effective WAL level is 'logical', write invalidations into WAL at + * each command end to support the decoding of the in-progress transactions. + * See CommandEndInvalidationMessages. * * Portions Copyright (c) 1996-2025, PostgreSQL Global Development Group * Portions Copyright (c) 1994, Regents of the University of California @@ -1419,7 +1419,7 @@ CommandEndInvalidationMessages(void) ProcessInvalidationMessages(&transInvalInfo->ii.CurrentCmdInvalidMsgs, LocalExecuteInvalidationMessage); - /* WAL Log per-command invalidation messages for wal_level=logical */ + /* WAL Log per-command invalidation messages for logical decoding */ if (XLogLogicalInfoActive()) LogLogicalInvalidations(); diff --git a/src/backend/utils/init/postinit.c b/src/backend/utils/init/postinit.c index 641e535a73c..1aa5c16a462 100644 --- a/src/backend/utils/init/postinit.c +++ b/src/backend/utils/init/postinit.c @@ -40,6 +40,7 @@ #include "pgstat.h" #include "postmaster/autovacuum.h" #include "postmaster/postmaster.h" +#include "replication/logicalctl.h" #include "replication/slot.h" #include "replication/slotsync.h" #include "replication/walsender.h" @@ -657,6 +658,9 @@ BaseInit(void) /* Initialize lock manager's local structs */ InitLockManagerAccess(); + /* Initialize logical info WAL logging state */ + InitializeProcessXLogLogicalInfo(); + /* * Initialize replication slots after pgstat. The exit hook might need to * drop ephemeral slots, which in turn triggers stats reporting. diff --git a/src/backend/utils/misc/guc_tables.c b/src/backend/utils/misc/guc_tables.c index d14b1678e7f..5ac58aa8806 100644 --- a/src/backend/utils/misc/guc_tables.c +++ b/src/backend/utils/misc/guc_tables.c @@ -5234,6 +5234,17 @@ struct config_enum ConfigureNamesEnum[] = NULL, NULL, NULL }, + { + {"effective_wal_level", PGC_INTERNAL, PRESET_OPTIONS, + gettext_noop("Show the effective WAL level."), + NULL, + GUC_NOT_IN_SAMPLE | GUC_DISALLOW_IN_FILE + }, + &effective_wal_level, + WAL_LEVEL_REPLICA, wal_level_options, + NULL, NULL, show_effective_wal_level + }, + { {"dynamic_shared_memory_type", PGC_POSTMASTER, RESOURCES_MEM, gettext_noop("Selects the dynamic shared memory implementation used."), diff --git a/src/bin/pg_basebackup/pg_createsubscriber.c b/src/bin/pg_basebackup/pg_createsubscriber.c index 3986882f042..527d1e75eb6 100644 --- a/src/bin/pg_basebackup/pg_createsubscriber.c +++ b/src/bin/pg_basebackup/pg_createsubscriber.c @@ -900,7 +900,7 @@ check_publisher(const struct LogicalRepInfo *dbinfo) * Since these parameters are not a requirement for physical replication, * we should check it to make sure it won't fail. * - * - wal_level = logical + * - wal_level >= replica * - max_replication_slots >= current + number of dbs to be converted * - max_wal_senders >= current + number of dbs to be converted * - max_slot_wal_keep_size = -1 (to prevent deletion of required WAL files) @@ -944,9 +944,9 @@ check_publisher(const struct LogicalRepInfo *dbinfo) disconnect_database(conn, false); - if (strcmp(wal_level, "logical") != 0) + if (strcmp(wal_level, "minimal") == 0) { - pg_log_error("publisher requires \"wal_level\" >= \"logical\""); + pg_log_error("publisher requires \"wal_level\" >= \"replica\""); failed = true; } diff --git a/src/bin/pg_upgrade/check.c b/src/bin/pg_upgrade/check.c index 5e6403f0773..18707ea2da9 100644 --- a/src/bin/pg_upgrade/check.c +++ b/src/bin/pg_upgrade/check.c @@ -2126,11 +2126,7 @@ check_new_cluster_replication_slots(void) wal_level = PQgetvalue(res, 0, 0); - if (nslots_on_old > 0 && strcmp(wal_level, "logical") != 0) - pg_fatal("\"wal_level\" must be \"logical\" but is set to \"%s\"", - wal_level); - - if (old_cluster.sub_retain_dead_tuples && + if ((nslots_on_old > 0 || old_cluster.sub_retain_dead_tuples) && strcmp(wal_level, "minimal") == 0) pg_fatal("\"wal_level\" must be \"replica\" or \"logical\" but is set to \"%s\"", wal_level); diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h index d12798be3d8..586baba7c61 100644 --- a/src/include/access/xlog.h +++ b/src/include/access/xlog.h @@ -94,6 +94,9 @@ typedef enum RecoveryState } RecoveryState; extern PGDLLIMPORT int wal_level; +extern PGDLLEXPORT int effective_wal_level; + +extern PGDLLEXPORT bool XLogLogicalInfo; /* Is WAL archiving enabled (always or only while server is running normally)? */ #define XLogArchivingActive() \ @@ -123,7 +126,7 @@ extern PGDLLIMPORT int wal_level; #define XLogStandbyInfoActive() (wal_level >= WAL_LEVEL_REPLICA) /* Do we need to WAL-log information required only for logical replication? */ -#define XLogLogicalInfoActive() (wal_level >= WAL_LEVEL_LOGICAL) +#define XLogLogicalInfoActive() (wal_level >= WAL_LEVEL_LOGICAL || XLogLogicalInfo) #ifdef WAL_DEBUG extern PGDLLIMPORT bool XLOG_DEBUG; diff --git a/src/include/catalog/pg_control.h b/src/include/catalog/pg_control.h index 63e834a6ce4..d62f6188d83 100644 --- a/src/include/catalog/pg_control.h +++ b/src/include/catalog/pg_control.h @@ -41,6 +41,7 @@ typedef struct CheckPoint * timeline (equals ThisTimeLineID otherwise) */ bool fullPageWrites; /* current full_page_writes */ int wal_level; /* current wal_level */ + bool logicalDecodingEnabled; /* current logical decoding status */ FullTransactionId nextXid; /* next free transaction ID */ Oid nextOid; /* next free OID */ MultiXactId nextMulti; /* next free MultiXactId */ @@ -80,6 +81,7 @@ typedef struct CheckPoint /* 0xC0 is used in Postgres 9.5-11 */ #define XLOG_OVERWRITE_CONTRECORD 0xD0 #define XLOG_CHECKPOINT_REDO 0xE0 +#define XLOG_LOGICAL_DECODING_STATUS_CHANGE 0xF0 /* diff --git a/src/include/replication/logicalctl.h b/src/include/replication/logicalctl.h new file mode 100644 index 00000000000..8020fb6e567 --- /dev/null +++ b/src/include/replication/logicalctl.h @@ -0,0 +1,66 @@ +/*------------------------------------------------------------------------- + * + * logicalctl.h + * Definitions for logical decoding status control facility. + * + * Portions Copyright (c) 2013-2025, PostgreSQL Global Development Group + * + * IDENTIFICATION + * src/include/replication/logicalctl.h + * + *------------------------------------------------------------------------- + */ +#ifndef LOGICALCTL_H +#define LOGICALCTL_H + +#include "port/atomics.h" +#include "storage/condition_variable.h" + +/* + * Struct for controlling the logical decoding status. + * + * This struct is protected by LogicalDecodingControlLock. + */ +typedef struct LogicalDecodingCtlData +{ + /* True if logical decoding is available in the system */ + bool logical_decoding_enabled; + + /* True while the logical decoding status is being changed */ + bool transition_in_progress; + + /* + * This flag is set to true by the startup process during recovery, to + * delay any logical decoding status change attempts until the recovery + * actually completes. See comments in + * start_logical_decoding_status_change() for details. + */ + bool delay_status_change; + + /* Condition variable signaled when a transition completes */ + ConditionVariable transition_cv; + + /* + * This is the authoritative value used by the all process to determine + * whether to write additional information required by logical decoding to + * WAL. Since this information could be checked frequently, each process + * caches this value in XLogLogicalInfo for better performance. + */ + pg_atomic_flag xlog_logical_info; +} LogicalDecodingCtlData; +extern LogicalDecodingCtlData *LogicalDecodingCtl; + +extern Size LogicalDecodingCtlShmemSize(void); +extern void LogicalDecodingCtlShmemInit(void); +extern void StartupLogicalDecodingStatus(bool status_in_control_file); +extern void UpdateNumberOfLogicalSlots(bool incr); +extern void InitializeProcessXLogLogicalInfo(void); +extern bool ProcessBarrierUpdateXLogLogicalInfo(void); +extern bool IsLogicalDecodingEnabled(void); +extern bool IsXLogLogicalInfoEnabled(void); +extern void EnsureLogicalDecodingEnabled(void); +extern void DisableLogicalDecodingIfNecessary(void); +extern void UpdateLogicalDecodingStatus(bool new_status, bool need_lock); +extern void UpdateLogicalDecodingStatusEndOfRecovery(void); + +#endif diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h index e8fc342d1a9..2082e307897 100644 --- a/src/include/replication/slot.h +++ b/src/include/replication/slot.h @@ -326,6 +326,7 @@ extern void ReplicationSlotsComputeRequiredXmin(bool already_locked); extern void ReplicationSlotsComputeRequiredLSN(void); extern XLogRecPtr ReplicationSlotsComputeLogicalRestartLSN(void); extern bool ReplicationSlotsCountDBSlots(Oid dboid, int *nslots, int *nactive); +extern bool CheckLogicalSlotExists(void); extern void ReplicationSlotsDropDBSlots(Oid dboid); extern bool InvalidateObsoleteReplicationSlots(uint32 possible_causes, XLogSegNo oldestSegno, diff --git a/src/include/replication/slotsync.h b/src/include/replication/slotsync.h index 16b721463dd..cfb10a53e9f 100644 --- a/src/include/replication/slotsync.h +++ b/src/include/replication/slotsync.h @@ -28,7 +28,7 @@ extern bool ValidateSlotSyncParams(int elevel); pg_noreturn extern void ReplSlotSyncWorkerMain(const void *startup_data, size_t startup_data_len); -extern void ShutDownSlotSync(void); +extern void ShutDownSlotSync(bool permanent); extern bool SlotSyncWorkerCanRestart(void); extern bool IsSyncingReplicationSlots(void); extern Size SlotSyncShmemSize(void); diff --git a/src/include/storage/lwlocklist.h b/src/include/storage/lwlocklist.h index 208d2e3a8ed..da4ab882c6c 100644 --- a/src/include/storage/lwlocklist.h +++ b/src/include/storage/lwlocklist.h @@ -85,6 +85,7 @@ PG_LWLOCK(50, DSMRegistry) PG_LWLOCK(51, InjectionPoint) PG_LWLOCK(52, SerialControl) PG_LWLOCK(53, AioWorkerSubmissionQueue) +PG_LWLOCK(54, LogicalDecodingControl) /* * There also exist several built-in LWLock tranches. As with the predefined diff --git a/src/include/storage/procsignal.h b/src/include/storage/procsignal.h index afeeb1ca019..8e428f298c6 100644 --- a/src/include/storage/procsignal.h +++ b/src/include/storage/procsignal.h @@ -54,6 +54,8 @@ typedef enum typedef enum { PROCSIGNAL_BARRIER_SMGRRELEASE, /* ask smgr to close files */ + PROCSIGNAL_BARRIER_UPDATE_XLOG_LOGICAL_INFO, /* ask to update + * XLogLogicalInfo */ } ProcSignalBarrierType; /* diff --git a/src/include/utils/guc_hooks.h b/src/include/utils/guc_hooks.h index 82ac8646a8d..fbe0b1e2e3d 100644 --- a/src/include/utils/guc_hooks.h +++ b/src/include/utils/guc_hooks.h @@ -61,6 +61,7 @@ extern bool check_default_text_search_config(char **newval, void **extra, GucSou extern void assign_default_text_search_config(const char *newval, void *extra); extern bool check_default_with_oids(bool *newval, void **extra, GucSource source); +extern const char *show_effective_wal_level(void); extern bool check_huge_page_size(int *newval, void **extra, GucSource source); extern void assign_io_method(int newval, void *extra); extern bool check_io_max_concurrency(int *newval, void **extra, GucSource source); diff --git a/src/test/recovery/meson.build b/src/test/recovery/meson.build index 52993c32dbb..170363859fa 100644 --- a/src/test/recovery/meson.build +++ b/src/test/recovery/meson.build @@ -56,7 +56,8 @@ tests += { 't/045_archive_restartpoint.pl', 't/046_checkpoint_logical_slot.pl', 't/047_checkpoint_physical_slot.pl', - 't/048_vacuum_horizon_floor.pl' + 't/048_vacuum_horizon_floor.pl', + 't/049_effective_wal_level.pl' ], }, } diff --git a/src/test/recovery/t/035_standby_logical_decoding.pl b/src/test/recovery/t/035_standby_logical_decoding.pl index 921813483e3..344e6b2f5c9 100644 --- a/src/test/recovery/t/035_standby_logical_decoding.pl +++ b/src/test/recovery/t/035_standby_logical_decoding.pl @@ -876,7 +876,7 @@ $handle = make_slot_active($node_standby, 'wal_level_', 0, \$stdout, \$stderr); # We are not able to read from the slot as it requires wal_level >= logical on the primary server check_pg_recvlogical_stderr($handle, - "logical decoding on standby requires \"wal_level\" >= \"logical\" on the primary" + "logical decoding needs to be enabled on the primary" ); # Restore primary wal_level diff --git a/src/test/recovery/t/049_effective_wal_level.pl b/src/test/recovery/t/049_effective_wal_level.pl new file mode 100644 index 00000000000..e8a209070c7 --- /dev/null +++ b/src/test/recovery/t/049_effective_wal_level.pl @@ -0,0 +1,213 @@ + +# Copyright (c) 2024-2025, PostgreSQL Global Development Group + +use strict; +use warnings FATAL => 'all'; +use PostgreSQL::Test::Cluster; +use PostgreSQL::Test::Utils; +use Test::More; + +# Check both wal_level and effective_wal_level values on the given node +# are expected. +sub test_wal_level +{ + my ($node, $expected, $msg) = @_; + + is( $node->safe_psql( + 'postgres', + qq[select current_setting('wal_level'), current_setting('effective_wal_level');]), + "$expected", + "$msg"); +} + +# Initialize the primary server with wal_level = 'replica' +my $primary = PostgreSQL::Test::Cluster->new('primary'); +$primary->init(allows_streaming => 1); +$primary->append_conf('postgresql.conf', "log_min_messages = debug1"); +$primary->start(); + +# Check both wal_level and effective_wal_level values. +test_wal_level($primary, "replica|replica", + "wal_level and effective_wal_level starts with the same value 'replica'"); + +# Create a physical slot. +$primary->safe_psql('postgres', + qq[select pg_create_physical_replication_slot('test_phy_slot', false, false)] +); +test_wal_level($primary, "replica|replica", + "effective_wal_level doesn't change with a new physical slot"); + +# Create a new logical slot, enabling the logical decoding. +$primary->safe_psql('postgres', + qq[select pg_create_logical_replication_slot('test_slot', 'pgoutput')]); + +# effective_wal_level must be bumped to 'logical' +test_wal_level($primary, "replica|logical", + "effective_wal_level bumped to logical upon logical slot creation"); + +# restart the server and check again. +$primary->restart(); +test_wal_level($primary, "replica|logical", + "effective_wal_level becomes logical during startup"); + +# Take backup during the effective_wal_level being 'logical'. +$primary->backup('my_backup'); + +# Initialize standby1 node from the backup 'my_backup'. Note that the +# backup was taken during the logical decoding being enabled on the +# primary because of one logical slot, but replication slots are not +# included in the backup. +my $standby1 = PostgreSQL::Test::Cluster->new('standby1'); +$standby1->init_from_backup($primary, 'my_backup', has_streaming => 1); +$standby1->set_standby_mode(); +$standby1->start; + +# Check if the standby's effective_wal_level should be 'logical' in spite +# of wal_level being 'replica'. +test_wal_level($standby1, "replica|logical", + "effective_wal_level='logical' on standby"); + +# Promote the standby1 node that doesn't have any logical slot. So +# the logical decoding must be disabled at promotion. +$standby1->promote; +test_wal_level($standby1, "replica|replica", + "effective_wal_level got decrased to 'replica' during promotion"); +$standby1->stop; + +# Initialize standby2 ndoe form the backup 'my_backup'. +my $standby2 = PostgreSQL::Test::Cluster->new('standby2'); +$standby2->init_from_backup($primary, 'my_backup', has_streaming => 1); +$standby2->set_standby_mode(); +$standby2->start; + +# Create a logical slot on the standby, which should be succeeded +# as the primary enables it. +$standby2->create_logical_slot_on_standby($primary, 'standby2_slot', + 'postgres'); + +# Promote the standby2 node that has one logical slot. So the logical decoding +# keeps enabled even after the promotion. +$standby2->promote; +test_wal_level($standby2, "replica|logical", + "effective_wal_level keeps 'logical' even after the promotion"); +$standby2->safe_psql('postgres', + qq[select pg_create_logical_replication_slot('standby2_slot2', 'pgoutput')] +); +$standby2->stop; + +# Initialize standby3 and starts it with wal_level = 'logical'. +my $standby3 = PostgreSQL::Test::Cluster->new('standby3'); +$standby3->init_from_backup($primary, 'my_backup', has_streaming => 1); +$standby3->set_standby_mode(); +$standby3->append_conf('postgresql.conf', qq[wal_level = 'logical']); +$standby3->start(); +$standby3->backup('my_backup3'); + +# Initialize cascade standby and starts with wal_level = 'replica'. +my $cascade = PostgreSQL::Test::Cluster->new('cascade'); +$cascade->init_from_backup($standby3, 'my_backup3', has_streaming => 1); +$cascade->adjust_conf('postgresql.conf', 'wal_level', 'replica'); +$cascade->set_standby_mode(); +$cascade->start(); + +# Regardless of their wal_level values, effective_wal_level values on the +# standby and the cascaded standby depend on the primary's value, 'logical'. +test_wal_level($standby3, "logical|logical", + "check wal_level and effective_wal_level on standby"); +test_wal_level($cascade, "replica|logical", + "check wal_level and effective_wal_level on cascaded standby"); + +# Drop the primary's last logical slot, disabling the logical decoding on +# all nodes. +$primary->safe_psql('postgres', + qq[select pg_drop_replication_slot('test_slot')]); + +$primary->wait_for_replay_catchup($standby3); +$standby3->wait_for_replay_catchup($cascade, $primary); + +test_wal_level($primary, "replica|replica", + "effective_wal_level got decreased to 'replica' on primary"); +test_wal_level($standby3, "logical|replica", + "effective_wal_level got decreased to 'replica' on standby"); +test_wal_level($cascade, "replica|replica", + "effective_wal_level got decreased to 'logical' on standby"); + +# Promote standby3. It enables the logical decoding at promotion as it uses +# 'logical' WAL level. +$standby3->promote; +$standby3->wait_for_replay_catchup($cascade); + +test_wal_level($cascade, "replica|logical", + "effective_wal_level got increased to 'logical' on standby"); + +$standby3->stop; +$cascade->stop; + +# Initialize standby4 and starts it with wal_level = 'logical'. +my $standby4 = PostgreSQL::Test::Cluster->new('standby4'); +$standby4->init_from_backup($primary, 'my_backup', has_streaming => 1); +$standby4->set_standby_mode(); +$standby4->append_conf('postgresql.conf', qq[wal_level = 'logical']); +$standby4->start; + +$primary->wait_for_replay_catchup($standby4); + +# Create logical slots on both nodes. +$primary->safe_psql('postgres', + qq[select pg_create_logical_replication_slot('test_slot', 'pgoutput')]); +$standby4->create_logical_slot_on_standby($primary, 'standby4_slot', + 'postgres'); + +# Drop the logical slot from the primary, disabling the logical decoding on the +# primary. Which leads to invalidate the logical slot on the standby due to +# 'wal_level_insufficient'. +$primary->safe_psql('postgres', + qq[select pg_drop_replication_slot('test_slot')]); +test_wal_level($primary, "replica|replica", + "logical decoding is disabled on the primary"); +$standby4->poll_query_until( + 'postgres', qq[ +select invalidation_reason = 'wal_level_insufficient' from pg_replication_slots where slot_name = 'standby4_slot' + ]); + +# Restart the server to check if the slot is successfully restored during +# startup. +$standby4->restart; + +# Check if the logical decoding is not enabled on the standby4. +test_wal_level($standby4, "logical|replica", + "standby's effective_wal_level got decreased to 'replica'"); +$standby4->safe_psql('postgres', + qq[select pg_drop_replication_slot('standby4_slot')]); + +# Restart the primary with setting wal_level = 'logical' and create a new logical +# slot. +$primary->append_conf('postgresql.conf', qq[wal_level = 'logical']); +$primary->restart; +$primary->safe_psql('postgres', + qq[select pg_create_logical_replication_slot('test_slot', 'pgoutput')]); + +# The logical decoding should be enabled on both nodes. +$primary->wait_for_replay_catchup($standby4); +test_wal_level($primary, "logical|logical", + "check WAL levels on the primary node"); +test_wal_level($standby4, "logical|logical", + "standby's effective_wal_level got increased to 'logical' again"); + +# Set wal_level to 'replica' and restart the primary. Since one logical slot +# is still present on the primary, the logical decoding is not disabled even +# if wal_level got decreased to 'replica'. +$primary->adjust_conf('postgresql.conf', 'wal_level', 'replica'); +$primary->restart; +$primary->wait_for_replay_catchup($standby4); + +# Check if the logical decoding is still enabled on the both nodes +test_wal_level($primary, "replica|logical", + "logical decoding is still enabled on the primary"); +test_wal_level($standby4, "logical|logical", + "logical decoding is still enabled on the standby"); + +$standby4->stop; +$primary->stop; + +done_testing(); diff --git a/src/test/subscription/t/001_rep_changes.pl b/src/test/subscription/t/001_rep_changes.pl index 916fdb48b3b..51f102f0c9f 100644 --- a/src/test/subscription/t/001_rep_changes.pl +++ b/src/test/subscription/t/001_rep_changes.pl @@ -589,7 +589,7 @@ CREATE PUBLICATION tap_pub2 FOR TABLE skip_wal; ROLLBACK; }); ok( $reterr =~ - m/WARNING: "wal_level" is insufficient to publish logical changes/, + m/WARNING: logical decoding needs to be enabled to publish logical changes/, 'CREATE PUBLICATION while "wal_level=minimal"'); done_testing(); diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index 3daba26b237..5f0132fe2e9 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -1612,6 +1612,7 @@ LogicalDecodeStreamStopCB LogicalDecodeStreamTruncateCB LogicalDecodeTruncateCB LogicalDecodingContext +LogicalDecodingCtlData LogicalErrorCallbackState LogicalOutputPluginInit LogicalOutputPluginWriterPrepareWrite -- 2.47.3