From c5adedcec21f4523243af2c0e5c0d63ef2e54b23 Mon Sep 17 00:00:00 2001
From: Antonin Houska <ah@cybertec.at>
Date: Wed, 26 Feb 2025 09:17:21 +0100
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                | 107 ++++++++++++++----
 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 -
 12 files changed, 216 insertions(+), 46 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 4ab5df9213..4e6e92c4db 100644
--- a/src/backend/access/transam/parallel.c
+++ b/src/backend/access/transam/parallel.c
@@ -97,6 +97,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;
@@ -351,6 +352,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);
@@ -1546,6 +1548,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 0bfd329847..0e4f234440 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"
@@ -137,6 +138,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
@@ -648,6 +655,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));
@@ -662,6 +670,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
@@ -691,20 +725,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.
@@ -729,6 +749,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
@@ -2243,6 +2311,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 799fc739e1..4a8a43a6da 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 a5790d77b5..910ff9fa91 100644
--- a/src/backend/commands/cluster.c
+++ b/src/backend/commands/cluster.c
@@ -1298,7 +1298,7 @@ copy_table_data(Relation NewHeap, Relation OldHeap, Relation OldIndex,
 	 *
 	 * In the REPACK CONCURRENTLY case, the lock does not help because we need
 	 * to release it temporarily at some point. Instead, we expect VACUUM /
-	 * CLUSTER to skip tables which are present in RepackedRelsHash.
+	 * CLUSTER to skip tables which are present in repackedRels->hashtable.
 	 */
 	if (OldHeap->rd_rel->reltoastrelid && !concurrent)
 		LockRelationOid(OldHeap->rd_rel->reltoastrelid, AccessExclusiveLock);
@@ -2312,7 +2312,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. */
 static int maxRepackedRels = 0;
@@ -2320,28 +2329,44 @@ static int maxRepackedRels = 0;
 Size
 RepackShmemSize(void)
 {
+	Size	result;
+
+	result = sizeof(RepackedRels);
+
 	/*
 	 * A replication slot is needed for the processing, so use this GUC to
 	 * allocate memory for the hashtable.
 	 */
 	maxRepackedRels = max_replication_slots;
 
-	return hash_estimate_size(maxRepackedRels, sizeof(RepackedRel));
+	result += hash_estimate_size(maxRepackedRels, 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",
-									 maxRepackedRels,
-									 maxRepackedRels,
-									 &info,
-									 HASH_ELEM | HASH_BLOBS);
+	repackedRels->hashtable = ShmemInitHash("Repacked Relations Hash",
+											maxRepackedRels,
+											maxRepackedRels,
+											&info,
+											HASH_ELEM | HASH_BLOBS);
 }
 
 /*
@@ -2373,12 +2398,13 @@ begin_concurrent_repack(Relation *rel_p, Relation *index_p,
 	RelReopenInfo	rri[2];
 	int		nrel;
 	static bool before_shmem_exit_callback_setup = false;
+	uint32	nrels PG_USED_FOR_ASSERTS_ONLY;
 
 	relid = RelationGetRelid(rel);
 
 	/*
-	 * 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)
 	{
@@ -2393,7 +2419,7 @@ begin_concurrent_repack(Relation *rel_p, Relation *index_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)
 	{
 		/*
@@ -2411,6 +2437,10 @@ begin_concurrent_repack(Relation *rel_p, Relation *index_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 < maxRepackedRels);
+
 	/*
 	 * Even if the insertion of TOAST relid should fail below, the caller has
 	 * to do cleanup.
@@ -2438,7 +2468,8 @@ begin_concurrent_repack(Relation *rel_p, Relation *index_p,
 	{
 		key.relid = toastrelid;
 		entry = (RepackedRel *)
-			hash_search(RepackedRelsHash, &key, HASH_ENTER_NULL, &found);
+			hash_search(repackedRels->hashtable, &key, HASH_ENTER_NULL,
+						&found);
 		if (found)
 			/*
 			 * If we could enter the main fork the TOAST should succeed
@@ -2452,6 +2483,10 @@ begin_concurrent_repack(Relation *rel_p, Relation *index_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 < maxRepackedRels);
+
 		Assert(!OidIsValid(repacked_rel_toast));
 		repacked_rel_toast = toastrelid;
 	}
@@ -2531,6 +2566,7 @@ end_concurrent_repack(bool error)
 	RepackedRel	*entry = NULL, *entry_toast = NULL;
 	Oid		relid = repacked_rel;
 	Oid		toastrelid = repacked_rel_toast;
+	uint32	nrels PG_USED_FOR_ASSERTS_ONLY;
 
 	/* Remove the relation from the hash if we managed to insert one. */
 	if (OidIsValid(repacked_rel))
@@ -2539,23 +2575,32 @@ 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);
 
 		/*
 		 * By clearing this variable we also disable
 		 * 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);
 	}
 
 	/* Remove the TOAST relation if there is one. */
 	if (OidIsValid(repacked_rel_toast))
 	{
 		key.relid = repacked_rel_toast;
-		entry_toast = hash_search(RepackedRelsHash, &key, HASH_REMOVE,
+		entry_toast = hash_search(repackedRels->hashtable, &key, HASH_REMOVE,
 								  NULL);
 
 		repacked_rel_toast = InvalidOid;
+
+		/* Decrement the number of relations. */
+		nrels = pg_atomic_fetch_sub_u32(&repackedRels->nrels, 1);
+		Assert(nrels > 0);
 	}
 	LWLockRelease(RepackedRelsLock);
 
@@ -2621,7 +2666,7 @@ end_concurrent_repack(bool error)
 }
 
 /*
- * A wrapper to call end_concurrent_repack() as a before_shmem_exit callback.
+ * A wrapper to call end_concurrent_cluster() as a before_shmem_exit callback.
  */
 static void
 cluster_before_shmem_exit_callback(int code, Datum arg)
@@ -2632,24 +2677,48 @@ 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)
 {
 	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);
+}
+
 /*
  * Check if REPACK CONCURRENTLY is already running for given relation, and if
  * so, raise ERROR. The problem is that cluster_rel() needs to release its
@@ -2944,8 +3013,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 8ea846bfc3..e5790d3fe8 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 5acb4508f8..413bcc1add 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 d313099c02..a325bb1d16 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 f5600bf4f6..3ed3066b36 100644
--- a/src/include/commands/cluster.h
+++ b/src/include/commands/cluster.h
@@ -173,6 +173,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 check_for_concurrent_repack(Oid relid, LOCKMODE lockmode);
 
 extern void repack(ParseState *pstate, RepackStmt *stmt, bool isTopLevel);
diff --git a/src/include/utils/rel.h b/src/include/utils/rel.h
index 741b29226d..81f5348a4f 100644
--- a/src/include/utils/rel.h
+++ b/src/include/utils/rel.h
@@ -709,12 +709,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 405d0811b4..4f6c0ca3a8 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 c8f264bc6c..0000000000
--- 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 0e3c47ba99..716e5619aa 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': {
-- 
2.43.5

