0001-Cache-standby-slot-index.patch.txt

From 3b24e04a225eb3bc82950432727662bdcb41f511 Mon Sep 17 00:00:00 2001
From: Hou Zhijie <houzj.fnst@cn.fujitsu.com>
Date: Fri, 8 Mar 2024 09:09:26 +0800
Subject: [PATCH] Cache standby slot index

---
 src/backend/replication/slot.c | 47 ++++++++++++++++++++++++++++++++--
 1 file changed, 45 insertions(+), 2 deletions(-)

diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index 30db34ece8..8736d9a87b 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -93,10 +93,17 @@ typedef struct
 	/* Number of slot names in the slot_names[] */
 	int			nslotnames;
 
+	/* The offset of the slot index array */
+	int			slot_index_offset;
+
 	/*
 	 * slot_names contains 'nslotnames' consecutive null-terminated C strings.
 	 */
 	char		slot_names[FLEXIBLE_ARRAY_MEMBER];
+
+	/*
+	 * After all the C strings, there is a size 'nslotnames' slot index array.
+	 */
 } StandbySlotNamesConfigData;
 
 /*
@@ -156,6 +163,14 @@ static StandbySlotNamesConfigData *standby_slot_names_config;
  */
 static XLogRecPtr ss_oldest_flush_lsn = InvalidXLogRecPtr;
 
+/*
+ * This array caches the position of each standby slot in
+ * ReplicationSlotCtl->replication_slots. The position will be updated if it
+ * becomes outdated due to a slot drop. See StandbySlotsHaveCaughtup for
+ * details.
+ */
+static int *standby_slot_index = NULL;
+
 static void ReplicationSlotShmemExit(int code, Datum arg);
 static void ReplicationSlotDropPtr(ReplicationSlot *slot);
 
@@ -2453,6 +2468,7 @@ check_standby_slot_names(char **newval, void **extra, GucSource source)
 	char	   *ptr;
 	List	   *elemlist;
 	int			size;
+	int			index_offset;
 	bool		ok;
 	StandbySlotNamesConfigData *config;
 
@@ -2477,11 +2493,15 @@ check_standby_slot_names(char **newval, void **extra, GucSource source)
 	foreach_ptr(char, slot_name, elemlist)
 		size += strlen(slot_name) + 1;
 
+	index_offset = size;
+	size += list_length(elemlist) * sizeof(int);
+
 	/* GUC extra value must be guc_malloc'd, not palloc'd */
 	config = (StandbySlotNamesConfigData *) guc_malloc(LOG, size);
 
 	/* Transform the data into StandbySlotNamesConfigData */
 	config->nslotnames = list_length(elemlist);
+	config->slot_index_offset = index_offset;
 
 	ptr = config->slot_names;
 	foreach_ptr(char, slot_name, elemlist)
@@ -2490,6 +2510,9 @@ check_standby_slot_names(char **newval, void **extra, GucSource source)
 		ptr += strlen(slot_name) + 1;
 	}
 
+	/* Initialize the index array */
+	memset(ptr, -1, list_length(elemlist) * sizeof(int));
+
 	*extra = (void *) config;
 
 	pfree(rawname);
@@ -2510,6 +2533,12 @@ assign_standby_slot_names(const char *newval, void *extra)
 	ss_oldest_flush_lsn = InvalidXLogRecPtr;
 
 	standby_slot_names_config = (StandbySlotNamesConfigData *) extra;
+
+	if (extra)
+	{
+		char *ptr = (char *) extra + standby_slot_names_config->slot_index_offset;
+		standby_slot_index = (int *) ptr;
+	}
 }
 
 /*
@@ -2590,9 +2619,21 @@ StandbySlotsHaveCaughtup(XLogRecPtr wait_for_lsn, int elevel)
 		XLogRecPtr	restart_lsn;
 		bool		invalidated;
 		bool		inactive;
-		ReplicationSlot *slot;
+		ReplicationSlot *slot = NULL;
+		int			slot_index = standby_slot_index[i];
 
-		slot = SearchNamedReplicationSlot(name, false);
+		/* Get the replication slot using the standby slot index */
+		if (slot_index >= 0)
+			slot = &ReplicationSlotCtl->replication_slots[slot_index];
+
+		/*
+		 * If the slot index has not been initialized, or if the slot found by
+		 * the index does not match, which means the original slot was dropped,
+		 * then we search for the named slot among all replication slots and
+		 * update the index if the slot is found.
+		 */
+		if (!slot || !slot->in_use || strcmp(name, NameStr(slot->data.name)) != 0)
+			slot = SearchNamedReplicationSlot(name, false);
 
 		if (!slot)
 		{
@@ -2616,6 +2657,8 @@ StandbySlotsHaveCaughtup(XLogRecPtr wait_for_lsn, int elevel)
 			break;
 		}
 
+		standby_slot_index[i] = ReplicationSlotIndex(slot);
+
 		if (SlotIsLogical(slot))
 		{
 			/*
-- 
2.30.0.windows.2