From 6e2f23e1f8c22b03bbfb19fbc1ae510286cb9801 Mon Sep 17 00:00:00 2001 From: Antonin Houska Date: Tue, 1 Apr 2025 13:48:57 +0200 Subject: [PATCH 8/9] Enable logical decoding transiently, only for REPACK CONCURRENTLY. As REPACK CONCURRENTLY uses logical decoding, it requires wal_level to be set to 'logical', while 'replica' is the default value. If logical replication is not used, users will probably be reluctant to set the GUC to 'logical' because it can affect server performance (by writing additional information to WAL) and because it cannot be changed to 'logical' only for the time REPACK CONCURRENTLY is running: change of this GUC requires server restart to take effect. This patch teaches postgres backend to recognize whether it should consider wal_level='logical' "locally" for particular transaction, even if the wal_level GUC is actually set to 'replica'. Also it ensures that the logical decoding specific information is added to WAL only for the tables which are currently being processed by REPACK CONCURRENTLY. If the logical decoding is enabled this way, only temporary replication slots should be created. The problem of permanent slot is that it is restored during server restart, and the restore fails if wal_level is not "globally" 'logical'. There is an independent work in progres to enable logical decoding transiently [1]. ISTM that this is too "heavyweight" solution for our problem. And I think that these two approaches are not mutually exclusive: once [1] is committed, we only need to adjust the XLogLogicalInfoActive() macro. [1] https://www.postgresql.org/message-id/CAD21AoCVLeLYq09pQPaWs%2BJwdni5FuJ8v2jgq-u9_uFbcp6UbA%40mail.gmail.com --- src/backend/access/transam/parallel.c | 8 ++ src/backend/access/transam/xact.c | 106 +++++++++++++++--- src/backend/access/transam/xlog.c | 1 + src/backend/commands/cluster.c | 94 +++++++++++++--- src/backend/replication/logical/logical.c | 9 +- src/backend/storage/ipc/standby.c | 4 +- src/include/access/xlog.h | 15 ++- src/include/commands/cluster.h | 1 + src/include/utils/rel.h | 6 +- src/test/modules/injection_points/Makefile | 1 - .../modules/injection_points/logical.conf | 1 - src/test/modules/injection_points/meson.build | 3 - src/tools/pgindent/typedefs.list | 1 + 13 files changed, 206 insertions(+), 44 deletions(-) delete mode 100644 src/test/modules/injection_points/logical.conf diff --git a/src/backend/access/transam/parallel.c b/src/backend/access/transam/parallel.c index 94db1ec3012..a33318ea7bd 100644 --- a/src/backend/access/transam/parallel.c +++ b/src/backend/access/transam/parallel.c @@ -98,6 +98,7 @@ typedef struct FixedParallelState TimestampTz xact_ts; TimestampTz stmt_ts; SerializableXactHandle serializable_xact_handle; + int wal_level_transient; /* Mutex protects remaining fields. */ slock_t mutex; @@ -355,6 +356,7 @@ InitializeParallelDSM(ParallelContext *pcxt) fps->xact_ts = GetCurrentTransactionStartTimestamp(); fps->stmt_ts = GetCurrentStatementStartTimestamp(); fps->serializable_xact_handle = ShareSerializableXact(); + fps->wal_level_transient = wal_level_transient; SpinLockInit(&fps->mutex); fps->last_xlog_end = 0; shm_toc_insert(pcxt->toc, PARALLEL_KEY_FIXED, fps); @@ -1550,6 +1552,12 @@ ParallelWorkerMain(Datum main_arg) /* Attach to the leader's serializable transaction, if SERIALIZABLE. */ AttachSerializableXact(fps->serializable_xact_handle); + /* + * Restore the information whether this worker should behave as if + * wal_level was WAL_LEVEL_LOGICAL.. + */ + wal_level_transient = fps->wal_level_transient; + /* * We've initialized all of our state now; nothing should change * hereafter. diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c index 3db4cac030e..608dc5c79bb 100644 --- a/src/backend/access/transam/xact.c +++ b/src/backend/access/transam/xact.c @@ -36,6 +36,7 @@ #include "catalog/pg_enum.h" #include "catalog/storage.h" #include "commands/async.h" +#include "commands/cluster.h" #include "commands/tablecmds.h" #include "commands/trigger.h" #include "common/pg_prng.h" @@ -138,6 +139,12 @@ static TransactionId *ParallelCurrentXids; static int nRepackCurrentXids = 0; static TransactionId *RepackCurrentXids = NULL; +/* + * Have we determined the value of wal_level_transient for the current + * transaction? + */ +static bool wal_level_transient_checked = false; + /* * Miscellaneous flag bits to record events which occur on the top level * transaction. These flags are only persisted in MyXactFlags and are intended @@ -650,6 +657,7 @@ AssignTransactionId(TransactionState s) bool isSubXact = (s->parent != NULL); ResourceOwner currentOwner; bool log_unknown_top = false; + bool set_wal_level_transient = false; /* Assert that caller didn't screw up */ Assert(!FullTransactionIdIsValid(s->fullTransactionId)); @@ -664,6 +672,32 @@ AssignTransactionId(TransactionState s) (errcode(ERRCODE_INVALID_TRANSACTION_STATE), errmsg("cannot assign transaction IDs during a parallel operation"))); + /* + * The first call (i.e. the first write) in the transaction tree + * determines whether the whole transaction assumes logical decoding or + * not. + */ + if (!wal_level_transient_checked) + { + Assert(wal_level_transient == WAL_LEVEL_MINIMAL); + + /* + * Do not repeat the check when calling this function for parent + * transactions. + */ + wal_level_transient_checked = true; + + /* + * Remember that the actual check is needed. We cannot do it until the + * top-level transaction has its XID assigned, see comments below. + * + * There is no use case for overriding MINIMAL, and LOGICAL cannot be + * overridden as such. + */ + if (wal_level == WAL_LEVEL_REPLICA) + set_wal_level_transient = true; + } + /* * Ensure parent(s) have XIDs, so that a child always has an XID later * than its parent. Mustn't recurse here, or we might get a stack @@ -693,20 +727,6 @@ AssignTransactionId(TransactionState s) pfree(parents); } - /* - * 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. - */ - if (isSubXact && XLogLogicalInfoActive() && - !TopTransactionStateData.didLogXid) - log_unknown_top = true; - /* * Generate a new FullTransactionId and record its xid in PGPROC and * pg_subtrans. @@ -731,6 +751,54 @@ AssignTransactionId(TransactionState s) if (!isSubXact) RegisterPredicateLockingXid(XidFromFullTransactionId(s->fullTransactionId)); + /* + * Check if this transaction should consider wal_level=logical. + * + * Sometimes we need to turn on the logical decoding transiently although + * wal_level=WAL_LEVEL_REPLICA. Currently we do so when at least one table + * is being clustered concurrently, i.e. when we should assume that + * changes done by this transaction will be decoded. In such a case we + * adjust the value of XLogLogicalInfoActive() by setting + * wal_level_transient to LOGICAL. + * + * It's important not to do this check until the XID of the top-level + * transaction is in ProcGlobal: if the decoding becomes mandatory right + * after the check, our transaction will fail to write the necessary + * information to WAL. However, if the top-level transaction is already in + * ProcGlobal, its XID is guaranteed to appear in the xl_running_xacts + * record and therefore the snapshot builder will not try to decode the + * transaction (because it assumes it could have missed the initial part + * of the transaction). + * + * On the other hand, if the decoding became mandatory between the actual + * XID assignment and now, the transaction will WAL the decoding specific + * information unnecessarily. Let's assume that such race conditions do + * not happen too often. + */ + if (set_wal_level_transient) + { + /* + * Check for the operation that enables the logical decoding + * transiently. + */ + if (is_concurrent_repack_in_progress(InvalidOid)) + wal_level_transient = WAL_LEVEL_LOGICAL; + } + + /* + * 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. + */ + if (isSubXact && XLogLogicalInfoActive() && + !TopTransactionStateData.didLogXid) + log_unknown_top = true; + /* * Acquire lock on the transaction XID. (We assume this cannot block.) We * have to ensure that the lock is assigned to the transaction's own @@ -2245,6 +2313,16 @@ StartTransaction(void) if (TransactionTimeout > 0) enable_timeout_after(TRANSACTION_TIMEOUT, TransactionTimeout); + /* + * wal_level_transient can override wal_level for individual transactions, + * which effectively enables logical decoding for them. At the moment we + * don't know if this transaction will write any data changes to be + * decoded. Should it do, AssignTransactionId() will check if the decoding + * needs to be considered. + */ + wal_level_transient = WAL_LEVEL_MINIMAL; + wal_level_transient_checked = false; + ShowTransactionState("StartTransaction"); } diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index fc30a52d496..ba758deefb4 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -129,6 +129,7 @@ bool wal_recycle = true; bool log_checkpoints = true; int wal_sync_method = DEFAULT_WAL_SYNC_METHOD; int wal_level = WAL_LEVEL_REPLICA; +int wal_level_transient = WAL_LEVEL_MINIMAL; int CommitDelay = 0; /* precommit delay in microseconds */ int CommitSiblings = 5; /* # concurrent xacts needed to sleep */ int wal_retrieve_retry_interval = 5000; diff --git a/src/backend/commands/cluster.c b/src/backend/commands/cluster.c index 085649716ac..c2201b046bc 100644 --- a/src/backend/commands/cluster.c +++ b/src/backend/commands/cluster.c @@ -2204,7 +2204,16 @@ typedef struct RepackedRel Oid dbid; } RepackedRel; -static HTAB *RepackedRelsHash = NULL; +typedef struct RepackedRels +{ + /* Hashtable of RepackedRel elements. */ + HTAB *hashtable; + + /* The number of elements in the hashtable.. */ + pg_atomic_uint32 nrels; +} RepackedRels; + +static RepackedRels *repackedRels = NULL; /* * Maximum number of entries in the hashtable. @@ -2217,22 +2226,38 @@ static HTAB *RepackedRelsHash = NULL; Size RepackShmemSize(void) { - return hash_estimate_size(MAX_REPACKED_RELS, sizeof(RepackedRel)); + Size result; + + result = sizeof(RepackedRels); + + result += hash_estimate_size(MAX_REPACKED_RELS, sizeof(RepackedRel)); + return result; } void RepackShmemInit(void) { + bool found; HASHCTL info; + repackedRels = ShmemInitStruct("Repacked Relations", + sizeof(RepackedRels), + &found); + if (!IsUnderPostmaster) + { + Assert(!found); + pg_atomic_init_u32(&repackedRels->nrels, 0); + } + else + Assert(found); + info.keysize = sizeof(RepackedRel); info.entrysize = info.keysize; - - RepackedRelsHash = ShmemInitHash("Repacked Relations", - MAX_REPACKED_RELS, - MAX_REPACKED_RELS, - &info, - HASH_ELEM | HASH_BLOBS); + repackedRels->hashtable = ShmemInitHash("Repacked Relations Hash", + MAX_REPACKED_RELS, + MAX_REPACKED_RELS, + &info, + HASH_ELEM | HASH_BLOBS); } /* @@ -2267,13 +2292,14 @@ begin_concurrent_repack(Relation rel, Relation *index_p, bool *entered_p) *entry; bool found; static bool before_shmem_exit_callback_setup = false; + uint32 nrels PG_USED_FOR_ASSERTS_ONLY; relid = RelationGetRelid(rel); index = index_p ? *index_p : NULL; /* - * Make sure that we do not leave an entry in RepackedRelsHash if exiting - * due to FATAL. + * Make sure that we do not leave an entry in repackedRels->hashtable if + * exiting due to FATAL. */ if (!before_shmem_exit_callback_setup) { @@ -2288,7 +2314,7 @@ begin_concurrent_repack(Relation rel, Relation *index_p, bool *entered_p) *entered_p = false; LWLockAcquire(RepackedRelsLock, LW_EXCLUSIVE); entry = (RepackedRel *) - hash_search(RepackedRelsHash, &key, HASH_ENTER_NULL, &found); + hash_search(repackedRels->hashtable, &key, HASH_ENTER_NULL, &found); if (found) { /* @@ -2306,9 +2332,13 @@ begin_concurrent_repack(Relation rel, Relation *index_p, bool *entered_p) (errmsg("too many requests for REPACK CONCURRENTLY at a time")), (errhint("Please consider increasing the \"max_replication_slots\" configuration parameter."))); + /* Increment the number of relations. */ + nrels = pg_atomic_fetch_add_u32(&repackedRels->nrels, 1); + Assert(nrels < MAX_REPACKED_RELS); + /* - * Even if anything fails below, the caller has to do cleanup in the - * shared memory. + * Even if the insertion of TOAST relid should fail below, the caller has + * to do cleanup. */ *entered_p = true; @@ -2390,6 +2420,7 @@ end_concurrent_repack(bool error) RepackedRel key; RepackedRel *entry = NULL; Oid relid = repacked_rel; + uint32 nrels PG_USED_FOR_ASSERTS_ONLY; /* Remove the relation from the hash if we managed to insert one. */ if (OidIsValid(repacked_rel)) @@ -2398,7 +2429,8 @@ end_concurrent_repack(bool error) key.relid = repacked_rel; key.dbid = MyDatabaseId; LWLockAcquire(RepackedRelsLock, LW_EXCLUSIVE); - entry = hash_search(RepackedRelsHash, &key, HASH_REMOVE, NULL); + entry = hash_search(repackedRels->hashtable, &key, HASH_REMOVE, + NULL); LWLockRelease(RepackedRelsLock); /* @@ -2427,6 +2459,10 @@ end_concurrent_repack(bool error) * cluster_before_shmem_exit_callback(). */ repacked_rel = InvalidOid; + + /* Decrement the number of relations. */ + nrels = pg_atomic_fetch_sub_u32(&repackedRels->nrels, 1); + Assert(nrels > 0); } /* @@ -2479,6 +2515,8 @@ cluster_before_shmem_exit_callback(int code, Datum arg) /* * Check if relation is currently being processed by REPACK CONCURRENTLY. + * + * If relid is InvalidOid, check if any relation is being processed. */ bool is_concurrent_repack_in_progress(Oid relid) @@ -2486,18 +2524,40 @@ is_concurrent_repack_in_progress(Oid relid) RepackedRel key, *entry; + /* + * If the caller is interested whether any relation is being repacked, + * just use the counter. + */ + if (!OidIsValid(relid)) + { + if (pg_atomic_read_u32(&repackedRels->nrels) > 0) + return true; + else + return false; + } + + /* For particular relation we need to search in the hashtable. */ memset(&key, 0, sizeof(key)); key.relid = relid; key.dbid = MyDatabaseId; LWLockAcquire(RepackedRelsLock, LW_SHARED); entry = (RepackedRel *) - hash_search(RepackedRelsHash, &key, HASH_FIND, NULL); + hash_search(repackedRels->hashtable, &key, HASH_FIND, NULL); LWLockRelease(RepackedRelsLock); return entry != NULL; } +/* + * Is this backend performing REPACK CONCURRENTLY? + */ +bool +is_concurrent_repack_run_by_me(void) +{ + return OidIsValid(repacked_rel); +} + /* * This function is much like pg_create_logical_replication_slot() except that * the new slot is neither released (if anyone else could read changes from @@ -2525,8 +2585,8 @@ setup_logical_decoding(Oid relid, const char *slotname, TupleDesc tupdesc) * useful for us. * * Regarding the value of need_full_snapshot, we pass false because the - * table we are processing is present in RepackedRelsHash and therefore, - * regarding logical decoding, treated like a catalog. + * table we are processing is present in repackedRels->hashtable and + * therefore, regarding logical decoding, treated like a catalog. */ ctx = CreateInitDecodingContext(REPL_PLUGIN_NAME, NIL, diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c index a8d2e024d34..4909432d585 100644 --- a/src/backend/replication/logical/logical.c +++ b/src/backend/replication/logical/logical.c @@ -30,6 +30,7 @@ #include "access/xact.h" #include "access/xlogutils.h" +#include "commands/cluster.h" #include "fmgr.h" #include "miscadmin.h" #include "pgstat.h" @@ -112,10 +113,12 @@ CheckLogicalDecodingRequirements(void) /* * NB: Adding a new requirement likely means that RestoreSlotFromDisk() - * needs the same check. + * needs the same check. (Except that only temporary slots should be + * created for REPACK CONCURRENTLY, which effectively raises wal_level to + * LOGICAL.) */ - - if (wal_level < WAL_LEVEL_LOGICAL) + if ((wal_level < WAL_LEVEL_LOGICAL && !is_concurrent_repack_run_by_me()) + || wal_level < WAL_LEVEL_REPLICA) ereport(ERROR, (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), errmsg("logical decoding requires \"wal_level\" >= \"logical\""))); diff --git a/src/backend/storage/ipc/standby.c b/src/backend/storage/ipc/standby.c index 5acb4508f85..413bcc1addb 100644 --- a/src/backend/storage/ipc/standby.c +++ b/src/backend/storage/ipc/standby.c @@ -1313,13 +1313,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 (!XLogLogicalInfoActive()) LWLockRelease(ProcArrayLock); recptr = LogCurrentRunningXacts(running); /* Release lock if we kept it longer ... */ - if (wal_level >= WAL_LEVEL_LOGICAL) + if (XLogLogicalInfoActive()) LWLockRelease(ProcArrayLock); /* GetRunningTransactionData() acquired XidGenLock, we must release it */ diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h index d313099c027..a325bb1d16b 100644 --- a/src/include/access/xlog.h +++ b/src/include/access/xlog.h @@ -95,6 +95,12 @@ typedef enum RecoveryState extern PGDLLIMPORT int wal_level; +/* + * wal_level_transient overrides wal_level if logical decoding needs to be + * enabled transiently. + */ +extern PGDLLIMPORT int wal_level_transient; + /* Is WAL archiving enabled (always or only while server is running normally)? */ #define XLogArchivingActive() \ (AssertMacro(XLogArchiveMode == ARCHIVE_MODE_OFF || wal_level >= WAL_LEVEL_REPLICA), XLogArchiveMode > ARCHIVE_MODE_OFF) @@ -122,8 +128,13 @@ extern PGDLLIMPORT int wal_level; /* Do we need to WAL-log information required only for Hot Standby and logical replication? */ #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) +/* + * Do we need to WAL-log information required only for logical replication? + * + * wal_level_transient overrides wal_level if logical decoding needs to be + * active transiently. + */ +#define XLogLogicalInfoActive() (Max(wal_level, wal_level_transient) == WAL_LEVEL_LOGICAL) #ifdef WAL_DEBUG extern PGDLLIMPORT bool XLOG_DEBUG; diff --git a/src/include/commands/cluster.h b/src/include/commands/cluster.h index be283c70fce..0267357a261 100644 --- a/src/include/commands/cluster.h +++ b/src/include/commands/cluster.h @@ -172,6 +172,7 @@ extern void finish_heap_swap(Oid OIDOldHeap, Oid OIDNewHeap, extern Size RepackShmemSize(void); extern void RepackShmemInit(void); extern bool is_concurrent_repack_in_progress(Oid relid); +extern bool is_concurrent_repack_run_by_me(void); extern void repack(ParseState *pstate, RepackStmt *stmt, bool isTopLevel); #endif /* CLUSTER_H */ diff --git a/src/include/utils/rel.h b/src/include/utils/rel.h index 372065fc570..fcbad5c1720 100644 --- a/src/include/utils/rel.h +++ b/src/include/utils/rel.h @@ -710,12 +710,16 @@ RelationCloseSmgr(Relation relation) * it would complicate decoding slightly for little gain). Note that we *do* * log information for user defined catalog tables since they presumably are * interesting to the user... + * + * If particular relations require that, the logical decoding can be active + * even if wal_level is REPLICA. Do not log other relations in that case. */ #define RelationIsLogicallyLogged(relation) \ (XLogLogicalInfoActive() && \ RelationNeedsWAL(relation) && \ (relation)->rd_rel->relkind != RELKIND_FOREIGN_TABLE && \ - !IsCatalogRelation(relation)) + !IsCatalogRelation(relation) && \ + (wal_level == WAL_LEVEL_LOGICAL || (relation)->rd_repack_concurrent)) /* routines in utils/cache/relcache.c */ extern void RelationIncrementReferenceCount(Relation rel); diff --git a/src/test/modules/injection_points/Makefile b/src/test/modules/injection_points/Makefile index 405d0811b4f..4f6c0ca3a8a 100644 --- a/src/test/modules/injection_points/Makefile +++ b/src/test/modules/injection_points/Makefile @@ -15,7 +15,6 @@ REGRESS = injection_points hashagg reindex_conc REGRESS_OPTS = --dlpath=$(top_builddir)/src/test/regress ISOLATION = basic inplace syscache-update-pruned repack -ISOLATION_OPTS = --temp-config $(top_srcdir)/src/test/modules/injection_points/logical.conf TAP_TESTS = 1 diff --git a/src/test/modules/injection_points/logical.conf b/src/test/modules/injection_points/logical.conf deleted file mode 100644 index c8f264bc6cb..00000000000 --- a/src/test/modules/injection_points/logical.conf +++ /dev/null @@ -1 +0,0 @@ -wal_level = logical \ No newline at end of file diff --git a/src/test/modules/injection_points/meson.build b/src/test/modules/injection_points/meson.build index 0e3c47ba999..716e5619aa7 100644 --- a/src/test/modules/injection_points/meson.build +++ b/src/test/modules/injection_points/meson.build @@ -50,9 +50,6 @@ tests += { 'syscache-update-pruned', ], 'runningcheck': false, # see syscache-update-pruned - # 'repack' requires wal_level = 'logical'. - 'regress_args': ['--temp-config', files('logical.conf')], - }, 'tap': { 'env': { diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index e1e3e619c4b..b3be8572132 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -2511,6 +2511,7 @@ ReorderBufferUpdateProgressTxnCB ReorderTuple RepOriginId RepackedRel +RepackedRels RepackDecodingState RepackStmt ReparameterizeForeignPathByChild_function -- 2.43.5