v18-0003-Introduce-new-SQL-funtion-pg_alter_replication_s.patch

From 65be663680fbde9812392a4fa739633060625f82 Mon Sep 17 00:00:00 2001
From: Bharath Rupireddy <bharath.rupireddyforpostgres@gmail.com>
Date: Sun, 24 Mar 2024 07:18:17 +0000
Subject: [PATCH v18 3/5] Introduce new SQL funtion pg_alter_replication_slot

This commit adds a new function pg_alter_replication_slot to alter
the given property of a replication slot. It is similar to
replication protocol command ALTER_REPLICATION_SLOT, except that
for now it allows only inactive_timeout property to be set. The
reason for disallowing failover property to be altered via this
function is to avoid inconsistency with the catalog
pg_subscription on the logical subscriber. Because, the subscriber
won't know the altered value of its replication slot on the
publisher.

Author: Bharath Rupireddy
Reviewed-by: Bertrand Drouvot, Amit Kapila
Discussion: https://www.postgresql.org/message-id/CALj2ACW4aUe-_uFQOjdWCEN-xXoLGhmvRFnL8SNw_TZ5nJe+aw@mail.gmail.com
---
 contrib/test_decoding/expected/slot.out   | 44 ++++++++++++++-
 contrib/test_decoding/sql/slot.sql        | 10 ++++
 doc/src/sgml/func.sgml                    | 21 ++++++++
 src/backend/replication/slot.c            | 22 ++++----
 src/backend/replication/slotfuncs.c       | 66 ++++++++++++++++++++++-
 src/bin/pg_upgrade/t/003_logical_slots.pl | 14 +++--
 src/include/catalog/pg_proc.dat           |  5 ++
 src/include/replication/slot.h            |  2 +
 8 files changed, 167 insertions(+), 17 deletions(-)

diff --git a/contrib/test_decoding/expected/slot.out b/contrib/test_decoding/expected/slot.out
index 6771520afb..5b8dbf6f52 100644
--- a/contrib/test_decoding/expected/slot.out
+++ b/contrib/test_decoding/expected/slot.out
@@ -496,13 +496,27 @@ SELECT 'copy' FROM pg_copy_physical_replication_slot(src_slot_name := 'it_phy_sl
  copy
 (1 row)
 
+-- Test alter physical slot with inactive_timeout option set.
+SELECT 'init' FROM pg_create_physical_replication_slot(slot_name := 'it_phy_slot4');
+ ?column? 
+----------
+ init
+(1 row)
+
+SELECT 'alter' FROM pg_alter_replication_slot(slot_name := 'it_phy_slot4', inactive_timeout := 900);
+ ?column? 
+----------
+ alter
+(1 row)
+
 SELECT slot_name, slot_type, inactive_timeout FROM pg_replication_slots ORDER BY 1;
   slot_name   | slot_type | inactive_timeout 
 --------------+-----------+------------------
  it_phy_slot1 | physical  |              300
  it_phy_slot2 | physical  |                0
  it_phy_slot3 | physical  |              300
-(3 rows)
+ it_phy_slot4 | physical  |              900
+(4 rows)
 
 SELECT pg_drop_replication_slot('it_phy_slot1');
  pg_drop_replication_slot 
@@ -522,6 +536,12 @@ SELECT pg_drop_replication_slot('it_phy_slot3');
  
 (1 row)
 
+SELECT pg_drop_replication_slot('it_phy_slot4');
+ pg_drop_replication_slot 
+--------------------------
+ 
+(1 row)
+
 -- Test inactive_timeout option of logical slots.
 SELECT 'init' FROM pg_create_logical_replication_slot(slot_name := 'it_log_slot1', plugin := 'test_decoding', inactive_timeout := 600);
  ?column? 
@@ -542,13 +562,27 @@ SELECT 'copy' FROM pg_copy_logical_replication_slot(src_slot_name := 'it_log_slo
  copy
 (1 row)
 
+-- Test alter logical slot with inactive_timeout option set.
+SELECT 'init' FROM pg_create_logical_replication_slot(slot_name := 'it_log_slot4', plugin := 'test_decoding');
+ ?column? 
+----------
+ init
+(1 row)
+
+SELECT 'alter' FROM pg_alter_replication_slot(slot_name := 'it_log_slot4', inactive_timeout := 900);
+ ?column? 
+----------
+ alter
+(1 row)
+
 SELECT slot_name, slot_type, inactive_timeout FROM pg_replication_slots ORDER BY 1;
   slot_name   | slot_type | inactive_timeout 
 --------------+-----------+------------------
  it_log_slot1 | logical   |              600
  it_log_slot2 | logical   |                0
  it_log_slot3 | logical   |              600
-(3 rows)
+ it_log_slot4 | logical   |              900
+(4 rows)
 
 SELECT pg_drop_replication_slot('it_log_slot1');
  pg_drop_replication_slot 
@@ -568,3 +602,9 @@ SELECT pg_drop_replication_slot('it_log_slot3');
  
 (1 row)
 
+SELECT pg_drop_replication_slot('it_log_slot4');
+ pg_drop_replication_slot 
+--------------------------
+ 
+(1 row)
+
diff --git a/contrib/test_decoding/sql/slot.sql b/contrib/test_decoding/sql/slot.sql
index 443e91da07..6785714cc7 100644
--- a/contrib/test_decoding/sql/slot.sql
+++ b/contrib/test_decoding/sql/slot.sql
@@ -206,11 +206,16 @@ SELECT 'init' FROM pg_create_physical_replication_slot(slot_name := 'it_phy_slot
 -- Copy physical slot with inactive_timeout option set.
 SELECT 'copy' FROM pg_copy_physical_replication_slot(src_slot_name := 'it_phy_slot1', dst_slot_name := 'it_phy_slot3');
 
+-- Test alter physical slot with inactive_timeout option set.
+SELECT 'init' FROM pg_create_physical_replication_slot(slot_name := 'it_phy_slot4');
+SELECT 'alter' FROM pg_alter_replication_slot(slot_name := 'it_phy_slot4', inactive_timeout := 900);
+
 SELECT slot_name, slot_type, inactive_timeout FROM pg_replication_slots ORDER BY 1;
 
 SELECT pg_drop_replication_slot('it_phy_slot1');
 SELECT pg_drop_replication_slot('it_phy_slot2');
 SELECT pg_drop_replication_slot('it_phy_slot3');
+SELECT pg_drop_replication_slot('it_phy_slot4');
 
 -- Test inactive_timeout option of logical slots.
 SELECT 'init' FROM pg_create_logical_replication_slot(slot_name := 'it_log_slot1', plugin := 'test_decoding', inactive_timeout := 600);
@@ -219,8 +224,13 @@ SELECT 'init' FROM pg_create_logical_replication_slot(slot_name := 'it_log_slot2
 -- Copy logical slot with inactive_timeout option set.
 SELECT 'copy' FROM pg_copy_logical_replication_slot(src_slot_name := 'it_log_slot1', dst_slot_name := 'it_log_slot3');
 
+-- Test alter logical slot with inactive_timeout option set.
+SELECT 'init' FROM pg_create_logical_replication_slot(slot_name := 'it_log_slot4', plugin := 'test_decoding');
+SELECT 'alter' FROM pg_alter_replication_slot(slot_name := 'it_log_slot4', inactive_timeout := 900);
+
 SELECT slot_name, slot_type, inactive_timeout FROM pg_replication_slots ORDER BY 1;
 
 SELECT pg_drop_replication_slot('it_log_slot1');
 SELECT pg_drop_replication_slot('it_log_slot2');
 SELECT pg_drop_replication_slot('it_log_slot3');
+SELECT pg_drop_replication_slot('it_log_slot4');
diff --git a/doc/src/sgml/func.sgml b/doc/src/sgml/func.sgml
index afaafa35ad..22c8e0d39c 100644
--- a/doc/src/sgml/func.sgml
+++ b/doc/src/sgml/func.sgml
@@ -28829,6 +28829,27 @@ postgres=# SELECT '0/0'::pg_lsn + pd.segment_number * ps.setting::int + :offset
       </entry>
       </row>
 
+      <row>
+       <entry role="func_table_entry"><para role="func_signature">
+        <indexterm>
+         <primary>pg_alter_replication_slot</primary>
+        </indexterm>
+        <function>pg_alter_replication_slot</function> ( <parameter>slot_name</parameter> <type>name</type>, <parameter>inactive_timeout</parameter> <type>integer</type> )
+        <returnvalue>void</returnvalue>
+       </para>
+       <para>
+        Alters the given property of a replication slot
+        named <parameter>slot_name</parameter>. Same as replication protocol
+        command <literal>ALTER_REPLICATION_SLOT</literal>, except that it
+        allows only <parameter>inactive_timeout</parameter> property to be set.
+        The reason for disallowing <parameter>failover</parameter> property to
+        be altered via this function is to avoid inconsistency with the catalog
+        <structname>pg_subscription</structname> on the logical subscriber.
+        Because, the subscriber won't know the altered value of its
+        replication slot on the publisher.
+       </para></entry>
+      </row>
+
      </tbody>
     </tgroup>
    </table>
diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index 852a657e97..3287aa2860 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -162,7 +162,6 @@ static void ReplicationSlotDropPtr(ReplicationSlot *slot);
 /* internal persistency functions */
 static void RestoreSlotFromDisk(const char *name);
 static void CreateSlotOnDisk(ReplicationSlot *slot);
-static void SaveSlotToPath(ReplicationSlot *slot, const char *dir, int elevel);
 
 /*
  * Report shared-memory space needed by ReplicationSlotsShmemInit.
@@ -870,6 +869,7 @@ ReplicationSlotAlter(const char *name, bool failover)
 	ReplicationSlotRelease();
 }
 
+
 /*
  * Permanently drop the currently acquired replication slot.
  */
@@ -1005,7 +1005,7 @@ ReplicationSlotSave(void)
 	Assert(MyReplicationSlot != NULL);
 
 	sprintf(path, "pg_replslot/%s", NameStr(MyReplicationSlot->data.name));
-	SaveSlotToPath(MyReplicationSlot, path, ERROR);
+	ReplicationSlotSaveToPath(MyReplicationSlot, path, ERROR);
 }
 
 /*
@@ -1868,7 +1868,10 @@ CheckPointReplicationSlots(bool is_shutdown)
 		if (!s->in_use)
 			continue;
 
-		/* save the slot to disk, locking is handled in SaveSlotToPath() */
+		/*
+		 * Save the slot to disk, locking is handled in
+		 * ReplicationSlotSaveToPath.
+		 */
 		sprintf(path, "pg_replslot/%s", NameStr(s->data.name));
 
 		/*
@@ -1894,7 +1897,7 @@ CheckPointReplicationSlots(bool is_shutdown)
 			SpinLockRelease(&s->mutex);
 		}
 
-		SaveSlotToPath(s, path, LOG);
+		ReplicationSlotSaveToPath(s, path, LOG);
 	}
 	LWLockRelease(ReplicationSlotAllocationLock);
 }
@@ -1973,8 +1976,9 @@ CreateSlotOnDisk(ReplicationSlot *slot)
 
 	/*
 	 * No need to take out the io_in_progress_lock, nobody else can see this
-	 * slot yet, so nobody else will write. We're reusing SaveSlotToPath which
-	 * takes out the lock, if we'd take the lock here, we'd deadlock.
+	 * slot yet, so nobody else will write. We're reusing
+	 * ReplicationSlotSaveToPath which takes out the lock, if we'd take the
+	 * lock here, we'd deadlock.
 	 */
 
 	sprintf(path, "pg_replslot/%s", NameStr(slot->data.name));
@@ -2000,7 +2004,7 @@ CreateSlotOnDisk(ReplicationSlot *slot)
 
 	/* Write the actual state file. */
 	slot->dirty = true;			/* signal that we really need to write */
-	SaveSlotToPath(slot, tmppath, ERROR);
+	ReplicationSlotSaveToPath(slot, tmppath, ERROR);
 
 	/* Rename the directory into place. */
 	if (rename(tmppath, path) != 0)
@@ -2025,8 +2029,8 @@ CreateSlotOnDisk(ReplicationSlot *slot)
 /*
  * Shared functionality between saving and creating a replication slot.
  */
-static void
-SaveSlotToPath(ReplicationSlot *slot, const char *dir, int elevel)
+void
+ReplicationSlotSaveToPath(ReplicationSlot *slot, const char *dir, int elevel)
 {
 	char		tmppath[MAXPGPATH];
 	char		path[MAXPGPATH];
diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c
index fb79401c50..dba80ac1bb 100644
--- a/src/backend/replication/slotfuncs.c
+++ b/src/backend/replication/slotfuncs.c
@@ -229,7 +229,6 @@ pg_create_logical_replication_slot(PG_FUNCTION_ARGS)
 	PG_RETURN_DATUM(result);
 }
 
-
 /*
  * SQL function for dropping a replication slot.
  */
@@ -1038,3 +1037,68 @@ pg_sync_replication_slots(PG_FUNCTION_ARGS)
 
 	PG_RETURN_VOID();
 }
+
+/*
+ * SQL function for altering given properties of a replication slot.
+ */
+Datum
+pg_alter_replication_slot(PG_FUNCTION_ARGS)
+{
+	Name		name = PG_GETARG_NAME(0);
+	int			inactive_timeout = PG_GETARG_INT32(1);
+	ReplicationSlot *slot;
+	char		path[MAXPGPATH];
+
+	CheckSlotPermissions();
+
+	CheckSlotRequirements();
+
+	LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
+
+	/* Check if the slot exits with the given name. */
+	slot = SearchNamedReplicationSlot(NameStr(*name), false);
+
+	if (!slot)
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+				 errmsg("replication slot \"%s\" does not exist",
+						NameStr(*name))));
+
+	/*
+	 * Do not allow users to set inactive_timeout for temporary slots because
+	 * temporary, slots will not be saved to the disk.
+	 */
+	if (slot->data.persistency == RS_TEMPORARY)
+		ereport(ERROR,
+				errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+				errmsg("cannot set inactive_timeout for a temporary replication slot"));
+
+	LWLockRelease(ReplicationSlotControlLock);
+
+	if (inactive_timeout < 0)
+		ereport(ERROR,
+				(errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE),
+				 errmsg("\"inactive_timeout\" must not be negative")));
+
+	/*
+	 * We need to briefly prevent any other backend from acquiring the slot
+	 * while we set the property. Without holding the ControlLock exclusively,
+	 * a concurrent ReplicationSlotAcquire() could acquire the slot as well.
+	 */
+	LWLockAcquire(ReplicationSlotControlLock, LW_EXCLUSIVE);
+
+	SpinLockAcquire(&slot->mutex);
+	slot->data.inactive_timeout = inactive_timeout;
+
+	/* Make sure the invalidated state persists across server restart */
+	slot->just_dirtied = true;
+	slot->dirty = true;
+	SpinLockRelease(&slot->mutex);
+
+	sprintf(path, "pg_replslot/%s", NameStr(slot->data.name));
+	ReplicationSlotSaveToPath(slot, path, ERROR);
+
+	LWLockRelease(ReplicationSlotControlLock);
+
+	PG_RETURN_VOID();
+}
diff --git a/src/bin/pg_upgrade/t/003_logical_slots.pl b/src/bin/pg_upgrade/t/003_logical_slots.pl
index 6e82d2cb7b..b79db24f42 100644
--- a/src/bin/pg_upgrade/t/003_logical_slots.pl
+++ b/src/bin/pg_upgrade/t/003_logical_slots.pl
@@ -153,17 +153,14 @@ like(
 # TEST: Successful upgrade
 
 # Preparations for the subsequent test:
-# 1. Setup logical replication (first, cleanup slots from the previous tests,
-# and then create slot for this test with inactive_timeout set).
+# 1. Setup logical replication (first, cleanup slots from the previous tests)
 my $old_connstr = $oldpub->connstr . ' dbname=postgres';
 
-my $inactive_timeout = 3600;
 $oldpub->start;
 $oldpub->safe_psql(
 	'postgres', qq[
 	SELECT * FROM pg_drop_replication_slot('test_slot1');
 	SELECT * FROM pg_drop_replication_slot('test_slot2');
-	SELECT pg_create_logical_replication_slot(slot_name := 'regress_sub', plugin := 'pgoutput', inactive_timeout := $inactive_timeout);
 	CREATE PUBLICATION regress_pub FOR ALL TABLES;
 ]);
 
@@ -175,7 +172,7 @@ $sub->start;
 $sub->safe_psql(
 	'postgres', qq[
 	CREATE TABLE tbl (a int);
-	CREATE SUBSCRIPTION regress_sub CONNECTION '$old_connstr' PUBLICATION regress_pub WITH (slot_name = 'regress_sub', create_slot = false, two_phase = 'true', failover = 'true')
+	CREATE SUBSCRIPTION regress_sub CONNECTION '$old_connstr' PUBLICATION regress_pub WITH (two_phase = 'true', failover = 'true')
 ]);
 $sub->wait_for_subscription_sync($oldpub, 'regress_sub');
 
@@ -185,6 +182,13 @@ my $twophase_query =
 $sub->poll_query_until('postgres', $twophase_query)
   or die "Timed out while waiting for subscriber to enable twophase";
 
+# Alter slot to set inactive_timeout
+my $inactive_timeout = 3600;
+$oldpub->safe_psql(
+	'postgres', qq[
+	SELECT pg_alter_replication_slot(slot_name := 'regress_sub', inactive_timeout := $inactive_timeout);
+]);
+
 # 2. Temporarily disable the subscription
 $sub->safe_psql('postgres', "ALTER SUBSCRIPTION regress_sub DISABLE");
 $oldpub->stop;
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index a09da44b6a..9a8134aa46 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -11222,6 +11222,11 @@
   proname => 'pg_sync_replication_slots', provolatile => 'v', proparallel => 'u',
   prorettype => 'void', proargtypes => '',
   prosrc => 'pg_sync_replication_slots' },
+{ oid => '9039', descr => 'alter given properties of a replication slot',
+  proname => 'pg_alter_replication_slot', provolatile => 'v', proparallel => 'u',
+  prorettype => 'void', proargtypes => 'name int4',
+  proargnames => '{slot_name,inactive_timeout}',
+  prosrc => 'pg_alter_replication_slot' },
 
 # event triggers
 { oid => '3566', descr => 'list objects dropped by the current command',
diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h
index 24623cfdc1..915edf7617 100644
--- a/src/include/replication/slot.h
+++ b/src/include/replication/slot.h
@@ -252,6 +252,8 @@ extern void ReplicationSlotAcquire(const char *name, bool nowait);
 extern void ReplicationSlotRelease(void);
 extern void ReplicationSlotCleanup(void);
 extern void ReplicationSlotSave(void);
+extern void ReplicationSlotSaveToPath(ReplicationSlot *slot, const char *dir,
+									  int elevel);
 extern void ReplicationSlotMarkDirty(void);
 
 /* misc stuff */
-- 
2.34.1