Disable logical decoding after REPACK (CONCURRENTLY)

REPACK (CONCURRENTLY) uses a temporary logical replication slot, which
is dropped once done, but it wasn't calling RequestDisableLogicalDecoding(),
leaving effective_wal_level stuck at 'logical'.

Fix by adding a Boolean flag to ReplicationSlotDropAcquired() to have it
request to disable logical decoding, and passing it as true on REPACK.
Other callers of that function preserve their existing behavior.

Author: Imran Zaheer <imran.zhir@gmail.com>
Reviewed-by: shveta malik <shveta.malik@gmail.com>
Discussion: https://postgr.es/m/CA+UBfaktds57dw2M8BEv_kS-=ixph3w+3MxKixtaDQMi_k7Ybg@mail.gmail.com

diff --git a/src/backend/commands/repack_worker.c b/src/backend/commands/repack_worker.c
index b84041372b844fa68b20c520b208bc816beca76e..4f82eb46bec479fb37445083192f5ebf0bf9cfd6 100644 (file)
--- a/src/backend/commands/repack_worker.c
+++ b/src/backend/commands/repack_worker.c
@@ -323,7 +323,7 @@ repack_cleanup_logical_decoding(LogicalDecodingContext *ctx)
                ExecDropSingleTupleTableSlot(dstate->slot);
 
        FreeDecodingContext(ctx);
-       ReplicationSlotDropAcquired();
+       ReplicationSlotDropAcquired(true);
 }
 
 /*

diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index 50051dea8c73a3dc880b311b1af5b2c0d846af91..313e31ff2e38950c1ed306987f04d8f2afcdf7bc 100644 (file)
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -1406,7 +1406,8 @@ ApplyLauncherMain(Datum main_arg)
                if (MyReplicationSlot)
                {
                        if (!retain_dead_tuples)
-                               ReplicationSlotDropAcquired();
+                               /* XXX unclear why we don't request logical decoding disable */
+                               ReplicationSlotDropAcquired(false);
                        else if (can_update_xmin)
                                update_conflict_slot_xmin(xmin);
                }

diff --git a/src/backend/replication/logical/slotsync.c b/src/backend/replication/logical/slotsync.c
index 9855c941f9d4344d2c31bf6b8afbf3255f1932f7..93f41be32af2107628177e62240437c41ef5baac 100644 (file)
--- a/src/backend/replication/logical/slotsync.c
+++ b/src/backend/replication/logical/slotsync.c
@@ -557,7 +557,7 @@ drop_local_obsolete_slots(List *remote_slot_list)
                         * database drop by the startup process and the creation of a new
                         * slot by the user. This new user-created slot may end up using
                         * the same shared memory as that of 'local_slot'. Thus check if
-                        * local_slot is still the synced one before performing actual
+                        * local_slot is still the synced one before performing the actual
                         * drop.
                         */
                        SpinLockAcquire(&local_slot->mutex);
@@ -566,8 +566,14 @@ drop_local_obsolete_slots(List *remote_slot_list)
 
                        if (synced_slot)
                        {
+                               /*
+                                * Now acquire and drop the slot.  Note we purposely don't
+                                * request logical decoding to be disabled here: since this is
+                                * a standby, which derives its logical decoding state from
+                                * the primary, it would be wrong to do so.
+                                */
                                ReplicationSlotAcquire(NameStr(local_slot->data.name), true, false);
-                               ReplicationSlotDropAcquired();
+                               ReplicationSlotDropAcquired(false);
                        }
 
                        UnlockSharedObject(DatabaseRelationId, local_slot->data.database,

diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index c0c9f514f7b909311d20a3c7d454bd7a7c904bcb..f012e99c9d7d5ef4f7e17fbd4491989de5d83424 100644 (file)
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -783,19 +783,10 @@ ReplicationSlotRelease(void)
        if (slot->data.persistency == RS_EPHEMERAL)
        {
                /*
-                * Delete the slot. There is no !PANIC case where this is allowed to
-                * fail, all that may happen is an incomplete cleanup of the on-disk
-                * data.
+                * If slot is ephemeral, we drop it upon release, and request logical
+                * decoding be disabled.
                 */
-               ReplicationSlotDropAcquired();
-
-               /*
-                * Request to disable logical decoding, even though this slot may not
-                * have been the last logical slot. The checkpointer will verify if
-                * logical decoding should actually be disabled.
-                */
-               if (is_logical)
-                       RequestDisableLogicalDecoding();
+               ReplicationSlotDropAcquired(is_logical);
        }
 
        /*
@@ -914,15 +905,13 @@ restart:
 }
 
 /*
- * Permanently drop replication slot identified by the passed in name.
+ * Permanently drop the replication slot identified by the passed-in name.
+ *
+ * If this is a logical slot, request that logical decoding be disabled.
  */
 void
 ReplicationSlotDrop(const char *name, bool nowait)
 {
-       bool            is_logical;
-
-       Assert(MyReplicationSlot == NULL);
-
        ReplicationSlotAcquire(name, nowait, false);
 
        /*
@@ -935,12 +924,7 @@ ReplicationSlotDrop(const char *name, bool nowait)
                                errmsg("cannot drop replication slot \"%s\"", name),
                                errdetail("This replication slot is being synchronized from the primary server."));
 
-       is_logical = SlotIsLogical(MyReplicationSlot);
-
-       ReplicationSlotDropAcquired();
-
-       if (is_logical)
-               RequestDisableLogicalDecoding();
+       ReplicationSlotDropAcquired(SlotIsLogical(MyReplicationSlot));
 }
 
 /*
@@ -1037,18 +1021,28 @@ ReplicationSlotAlter(const char *name, const bool *failover,
 
 /*
  * Permanently drop the currently acquired replication slot.
+ *
+ * If caller requests it, have checkpointer attempt to disable logical
+ * decoding.  Obviously, this should only be done if the slot is logical.
  */
 void
-ReplicationSlotDropAcquired(void)
+ReplicationSlotDropAcquired(bool try_disable)
 {
-       ReplicationSlot *slot = MyReplicationSlot;
+       ReplicationSlot *slot;
 
        Assert(MyReplicationSlot != NULL);
+       slot = MyReplicationSlot;
+
+       /* Can only disable logical decoding if slot is logical */
+       Assert(!try_disable || SlotIsLogical(slot));
 
        /* slot isn't acquired anymore */
        MyReplicationSlot = NULL;
 
        ReplicationSlotDropPtr(slot);
+
+       if (try_disable)
+               RequestDisableLogicalDecoding();
 }
 
 /*
@@ -1511,7 +1505,7 @@ ReplicationSlotsCountDBSlots(Oid dboid, int *nslots, int *nactive)
  * This routine isn't as efficient as it could be - but we don't drop
  * databases often, especially databases with lots of slots.
  *
- * If it drops the last logical slot in the cluster, it requests to disable
+ * If the last logical slot in the cluster is dropped, request to disable
  * logical decoding.
  */
 void
@@ -1606,7 +1600,7 @@ restart:
                 * beginning each time we release the lock.
                 */
                LWLockRelease(ReplicationSlotControlLock);
-               ReplicationSlotDropAcquired();
+               ReplicationSlotDropAcquired(false);
                dropped = true;
                goto restart;
        }

diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h
index 77c8d0975b6826e244e83cda68885bc7096f31d6..9b29444cbca4a2cfb600851a9c373cf30ccf1748 100644 (file)
--- a/src/include/replication/slot.h
+++ b/src/include/replication/slot.h
@@ -335,7 +335,7 @@ extern void ReplicationSlotCreate(const char *name, bool db_specific,
                                                                  bool synced);
 extern void ReplicationSlotPersist(void);
 extern void ReplicationSlotDrop(const char *name, bool nowait);
-extern void ReplicationSlotDropAcquired(void);
+extern void ReplicationSlotDropAcquired(bool try_disable);
 extern void ReplicationSlotAlter(const char *name, const bool *failover,
                                                                 const bool *two_phase);
 

diff --git a/src/test/recovery/t/051_effective_wal_level.pl b/src/test/recovery/t/051_effective_wal_level.pl
index c4c2662f72b4b267535fb21570c91722a37e945a..9341e11d58ad55c04d7aefd08654cb9a6ea7cdc6 100644 (file)
--- a/src/test/recovery/t/051_effective_wal_level.pl
+++ b/src/test/recovery/t/051_effective_wal_level.pl
@@ -400,6 +400,20 @@ select pg_cancel_backend(pid) from pg_stat_activity where query ~ 'slot_canceled
                "the activation process aborted");
 }
 
+# Test that logical decoding is disabled after repack
+$primary->safe_psql('postgres', qq[create table foo(a int primary key)]);
+$primary->safe_psql('postgres', qq[repack (concurrently) foo;]);
+ok( $primary->log_contains(
+               "logical decoding is enabled upon creating a new logical replication slot"
+       ),
+       "logical decoding enabled by repack");
+
+# Wait for the checkpointer to disable logical decoding.
+wait_for_logical_decoding_disabled($primary);
+test_wal_level($primary, "replica|replica",
+       "logical decoding disabled after repack"
+);
+
 $primary->stop;
 
 done_testing();