From 65be663680fbde9812392a4fa739633060625f82 Mon Sep 17 00:00:00 2001 From: Bharath Rupireddy 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 + + + + pg_alter_replication_slot + + pg_alter_replication_slot ( slot_name name, inactive_timeout integer ) + void + + + Alters the given property of a replication slot + named slot_name. Same as replication protocol + command ALTER_REPLICATION_SLOT, except that 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. + + + 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