Add 'active_in' column to pg_replication_slots.

Right now it is visible whether a replication slot is active in any
session, but not in which.  Adding the active_in column, containing the
pid of the backend having acquired the slot, makes it much easier to
associate pg_replication_slots entries with the corresponding
pg_stat_replication/pg_stat_activity row.

This should have been done from the start, but I (Andres) dropped the
ball there somehow.

Author: Craig Ringer, revised by me Discussion:
CAMsr+YFKgZca5_7_ouaMWxA5PneJC9LNViPzpDHusaPhU9pA7g@mail.gmail.com

diff --git a/contrib/test_decoding/expected/ddl.out b/contrib/test_decoding/expected/ddl.out
index 780120d7314ec1edc12a44c767c6f0de95e6fbf4..838483ee5c0132561359117fb9a678d42a509bba 100644 (file)
--- a/contrib/test_decoding/expected/ddl.out
+++ b/contrib/test_decoding/expected/ddl.out
@@ -603,7 +603,7 @@ SELECT pg_drop_replication_slot('regression_slot');
 
 /* check that the slot is gone */
 SELECT * FROM pg_replication_slots;
- slot_name | plugin | slot_type | datoid | database | active | xmin | catalog_xmin | restart_lsn 
------------+--------+-----------+--------+----------+--------+------+--------------+-------------
+ slot_name | plugin | slot_type | datoid | database | active | active_in | xmin | catalog_xmin | restart_lsn 
+-----------+--------+-----------+--------+----------+--------+-----------+------+--------------+-------------
 (0 rows)
 

diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml
index d0b78f27827cd5b1dd883af248657f33b802d013..98847354bd0525e428ca678f8b5032d35eefb8f5 100644 (file)
--- a/doc/src/sgml/catalogs.sgml
+++ b/doc/src/sgml/catalogs.sgml
@@ -5400,6 +5400,16 @@
       <entry>True if this slot is currently actively being used</entry>
      </row>
 
+     <row>
+      <entry><structfield>active_in</structfield></entry>
+      <entry><type>integer</type></entry>
+      <entry></entry>
+      <entry>The process ID of the session using this slot if the slot
+       is currently actively being used. <literal>NULL</literal> if
+       inactive.
+      </entry>
+     </row>
+
      <row>
       <entry><structfield>xmin</structfield></entry>
       <entry><type>xid</type></entry>

diff --git a/doc/src/sgml/logicaldecoding.sgml b/doc/src/sgml/logicaldecoding.sgml
index 3650567852656d7ea91c17e492d0dabfbed8f2aa..0810a2d1f97b045dfd0931dec5feda500ab4d045 100644 (file)
--- a/doc/src/sgml/logicaldecoding.sgml
+++ b/doc/src/sgml/logicaldecoding.sgml
@@ -62,10 +62,10 @@ postgres=# SELECT * FROM pg_create_logical_replication_slot('regression_slot', '
  regression_slot | 0/16B1970
 (1 row)
 
-postgres=# SELECT * FROM pg_replication_slots;
-    slot_name    |    plugin     | slot_type | datoid | database | active |  xmin  | catalog_xmin | restart_lsn
------------------+---------------+-----------+--------+----------+--------+--------+--------------+-------------
- regression_slot | test_decoding | logical   |  12052 | postgres | f      |        |          684 | 0/16A4408
+postgres=# SELECT slot_name, plugin, slot_type, database, active, restart_lsn FROM pg_replication_slots;
+    slot_name    |    plugin     | slot_type | database | active | restart_lsn
+-----------------+---------------+-----------+----------+--------+-------------
+ regression_slot | test_decoding | logical   | postgres | f      | 0/16A4408
 (1 row)
 
 postgres=# -- There are no changes to see yet

diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index a4fd88fa98d7edc9674240d00f6de06e7f600a1b..75ff231de5587c01e7470d7d2dab7d9b947e283c 100644 (file)
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -665,6 +665,7 @@ CREATE VIEW pg_replication_slots AS
             L.datoid,
             D.datname AS database,
             L.active,
+            L.active_in,
             L.xmin,
             L.catalog_xmin,
             L.restart_lsn

diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index 16ea80bc3a0467e5f97e06a0128b522e1a98f87b..fa1f07b3f3e9a36a8830c2bd6f5ef00bbd902e01 100644 (file)
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -262,7 +262,7 @@ ReplicationSlotCreate(const char *name, bool db_specific,
         * be doing that.  So it's safe to initialize the slot.
         */
        Assert(!slot->in_use);
-       Assert(!slot->active);
+       Assert(slot->active_pid == 0);
        slot->data.persistency = persistency;
        slot->data.xmin = InvalidTransactionId;
        slot->effective_xmin = InvalidTransactionId;
@@ -291,8 +291,8 @@ ReplicationSlotCreate(const char *name, bool db_specific,
                volatile ReplicationSlot *vslot = slot;
 
                SpinLockAcquire(&slot->mutex);
-               Assert(!vslot->active);
-               vslot->active = true;
+               Assert(vslot->active_pid == 0);
+               vslot->active_pid = MyProcPid;
                SpinLockRelease(&slot->mutex);
                MyReplicationSlot = slot;
        }
@@ -314,7 +314,7 @@ ReplicationSlotAcquire(const char *name)
 {
        ReplicationSlot *slot = NULL;
        int                     i;
-       bool            active = false;
+       int                     active_pid = 0;
 
        Assert(MyReplicationSlot == NULL);
 
@@ -331,8 +331,9 @@ ReplicationSlotAcquire(const char *name)
                        volatile ReplicationSlot *vslot = s;
 
                        SpinLockAcquire(&s->mutex);
-                       active = vslot->active;
-                       vslot->active = true;
+                       active_pid = vslot->active_pid;
+                       if (active_pid == 0)
+                               vslot->active_pid = MyProcPid;
                        SpinLockRelease(&s->mutex);
                        slot = s;
                        break;
@@ -345,10 +346,11 @@ ReplicationSlotAcquire(const char *name)
                ereport(ERROR,
                                (errcode(ERRCODE_UNDEFINED_OBJECT),
                                 errmsg("replication slot \"%s\" does not exist", name)));
-       if (active)
+       if (active_pid != 0)
                ereport(ERROR,
                                (errcode(ERRCODE_OBJECT_IN_USE),
-                                errmsg("replication slot \"%s\" is already active", name)));
+                                errmsg("replication slot \"%s\" is already active for pid %d",
+                                               name, active_pid)));
 
        /* We made this slot active, so it's ours now. */
        MyReplicationSlot = slot;
@@ -363,7 +365,7 @@ ReplicationSlotRelease(void)
 {
        ReplicationSlot *slot = MyReplicationSlot;
 
-       Assert(slot != NULL && slot->active);
+       Assert(slot != NULL && slot->active_pid != 0);
 
        if (slot->data.persistency == RS_EPHEMERAL)
        {
@@ -380,7 +382,7 @@ ReplicationSlotRelease(void)
                volatile ReplicationSlot *vslot = slot;
 
                SpinLockAcquire(&slot->mutex);
-               vslot->active = false;
+               vslot->active_pid = 0;
                SpinLockRelease(&slot->mutex);
        }
 
@@ -460,7 +462,7 @@ ReplicationSlotDropAcquired(void)
                bool            fail_softly = slot->data.persistency == RS_EPHEMERAL;
 
                SpinLockAcquire(&slot->mutex);
-               vslot->active = false;
+               vslot->active_pid = 0;
                SpinLockRelease(&slot->mutex);
 
                ereport(fail_softly ? WARNING : ERROR,
@@ -477,7 +479,7 @@ ReplicationSlotDropAcquired(void)
         * scanning the array.
         */
        LWLockAcquire(ReplicationSlotControlLock, LW_EXCLUSIVE);
-       slot->active = false;
+       slot->active_pid = 0;
        slot->in_use = false;
        LWLockRelease(ReplicationSlotControlLock);
 
@@ -749,7 +751,7 @@ ReplicationSlotsCountDBSlots(Oid dboid, int *nslots, int *nactive)
                /* count slots with spinlock held */
                SpinLockAcquire(&s->mutex);
                (*nslots)++;
-               if (s->active)
+               if (s->active_pid != 0)
                        (*nactive)++;
                SpinLockRelease(&s->mutex);
        }
@@ -1227,7 +1229,7 @@ RestoreSlotFromDisk(const char *name)
                slot->candidate_restart_valid = InvalidXLogRecPtr;
 
                slot->in_use = true;
-               slot->active = false;
+               slot->active_pid = 0;
 
                restored = true;
                break;

diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c
index f31925d862b6749d7c4196a8ff1e11d092b10138..3d9aadbd839909df51752e5a11b7a45209eb2fa1 100644 (file)
--- a/src/backend/replication/slotfuncs.c
+++ b/src/backend/replication/slotfuncs.c
@@ -158,7 +158,7 @@ pg_drop_replication_slot(PG_FUNCTION_ARGS)
 Datum
 pg_get_replication_slots(PG_FUNCTION_ARGS)
 {
-#define PG_GET_REPLICATION_SLOTS_COLS 8
+#define PG_GET_REPLICATION_SLOTS_COLS 9
        ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
        TupleDesc       tupdesc;
        Tuplestorestate *tupstore;
@@ -206,7 +206,7 @@ pg_get_replication_slots(PG_FUNCTION_ARGS)
                TransactionId xmin;
                TransactionId catalog_xmin;
                XLogRecPtr      restart_lsn;
-               bool            active;
+               pid_t           active_pid;
                Oid                     database;
                NameData        slot_name;
                NameData        plugin;
@@ -227,7 +227,7 @@ pg_get_replication_slots(PG_FUNCTION_ARGS)
                        namecpy(&slot_name, &slot->data.name);
                        namecpy(&plugin, &slot->data.plugin);
 
-                       active = slot->active;
+                       active_pid = slot->active_pid;
                }
                SpinLockRelease(&slot->mutex);
 
@@ -251,7 +251,12 @@ pg_get_replication_slots(PG_FUNCTION_ARGS)
                else
                        values[i++] = database;
 
-               values[i++] = BoolGetDatum(active);
+               values[i++] = BoolGetDatum(active_pid != 0);
+
+               if (active_pid != 0)
+                       values[i++] = Int32GetDatum(active_pid);
+               else
+                       nulls[i++] = true;
 
                if (xmin != InvalidTransactionId)
                        values[i++] = TransactionIdGetDatum(xmin);

diff --git a/src/include/catalog/pg_proc.h b/src/include/catalog/pg_proc.h
index 619d9969cab74d9142112117963269f8ed155ac9..de18a5ad3ca4983f6141c75c6425f3df2d9ffae0 100644 (file)
--- a/src/include/catalog/pg_proc.h
+++ b/src/include/catalog/pg_proc.h
@@ -5106,7 +5106,7 @@ DATA(insert OID = 3779 (  pg_create_physical_replication_slot PGNSP PGUID 12 1 0
 DESCR("create a physical replication slot");
 DATA(insert OID = 3780 (  pg_drop_replication_slot PGNSP PGUID 12 1 0 0 0 f f f f f f v 1 0 2278 "19" _null_ _null_ _null_ _null_ pg_drop_replication_slot _null_ _null_ _null_ ));
 DESCR("drop a replication slot");
-DATA(insert OID = 3781 (  pg_get_replication_slots     PGNSP PGUID 12 1 10 0 0 f f f f f t s 0 0 2249 "" "{19,19,25,26,16,28,28,3220}" "{o,o,o,o,o,o,o,o}" "{slot_name,plugin,slot_type,datoid,active,xmin,catalog_xmin,restart_lsn}" _null_ pg_get_replication_slots _null_ _null_ _null_ ));
+DATA(insert OID = 3781 (  pg_get_replication_slots     PGNSP PGUID 12 1 10 0 0 f f f f f t s 0 0 2249 "" "{19,19,25,26,16,23,28,28,3220}" "{o,o,o,o,o,o,o,o,o}" "{slot_name,plugin,slot_type,datoid,active,active_in,xmin,catalog_xmin,restart_lsn}" _null_ pg_get_replication_slots _null_ _null_ _null_ ));
 DESCR("information about replication slots currently in use");
 DATA(insert OID = 3786 (  pg_create_logical_replication_slot PGNSP PGUID 12 1 0 0 0 f f f f f f v 2 0 2249 "19 19" "{19,19,25,3220}" "{i,i,o,o}" "{slot_name,plugin,slot_name,xlog_position}" _null_ pg_create_logical_replication_slot _null_ _null_ _null_ ));
 DESCR("set up a logical replication slot");

diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h
index a4001360c43bcf7210a6231e6444b3a7c3424c6d..78cff07abf801db7544fc7e30f68680246833e4b 100644 (file)
--- a/src/include/replication/slot.h
+++ b/src/include/replication/slot.h
@@ -84,8 +84,8 @@ typedef struct ReplicationSlot
        /* is this slot defined */
        bool            in_use;
 
-       /* is somebody streaming out changes for this slot */
-       bool            active;
+       /* Who is streaming out changes for this slot? 0 in unused slots. */
+       pid_t           active_pid;
 
        /* any outstanding modifications? */
        bool            just_dirtied;

diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index 71fa44a6beaa359642f9067efb97c6fcdf87d996..1577ff0e297f4e4f1df55a98bd7797f373f21fc9 100644 (file)
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1396,10 +1396,11 @@ pg_replication_slots| SELECT l.slot_name,
     l.datoid,
     d.datname AS database,
     l.active,
+    l.active_in,
     l.xmin,
     l.catalog_xmin,
     l.restart_lsn
-   FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, active, xmin, catalog_xmin, restart_lsn)
+   FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, active, active_in, xmin, catalog_xmin, restart_lsn)
      LEFT JOIN pg_database d ON ((l.datoid = d.oid)));
 pg_roles| SELECT pg_authid.rolname,
     pg_authid.rolsuper,