Add replication command READ_REPLICATION_SLOT

The command is supported for physical slots for now, and returns the
type of slot, its restart_lsn and its restart_tli.

This will be useful for an upcoming patch related to pg_receivewal, to
allow the tool to be able to stream from the position of a slot, rather
than the last WAL position flushed by the backend (as reported by
IDENTIFY_SYSTEM) if the archive directory is found as empty, which would
be an advantage in the case of switching to a different archive
locations with the same slot used to avoid holes in WAL segment
archives.

Author: Ronan Dunklau
Reviewed-by: Kyotaro Horiguchi, Michael Paquier, Bharath Rupireddy
Discussion: https://postgr.es/m/18708360.4lzOvYHigE@aivenronan

diff --git a/doc/src/sgml/protocol.sgml b/doc/src/sgml/protocol.sgml
index b95cc88599a1850a46999c402129d2b661262565..132436c6e6842956d8987297b5a0c62f90f8ee37 100644 (file)
--- a/doc/src/sgml/protocol.sgml
+++ b/doc/src/sgml/protocol.sgml
@@ -2067,6 +2067,54 @@ The commands accepted in replication mode are:
     </listitem>
   </varlistentry>
 
+  <varlistentry>
+    <term><literal>READ_REPLICATION_SLOT</literal> <replaceable class="parameter">slot_name</replaceable>
+      <indexterm><primary>READ_REPLICATION_SLOT</primary></indexterm>
+    </term>
+    <listitem>
+     <para>
+      Read some information associated to a replication slot. Returns a tuple
+      with <literal>NULL</literal> values if the replication slot does not
+      exist. This command is currently only supported for physical replication
+      slots.
+     </para>
+     <para>
+      In response to this command, the server will return a one-row result set,
+      containing the following fields:
+      <variablelist>
+       <varlistentry>
+        <term><literal>slot_type</literal> (<type>text</type>)</term>
+        <listitem>
+         <para>
+          The replication slot's type, either <literal>physical</literal> or
+          <literal>NULL</literal>.
+         </para>
+        </listitem>
+       </varlistentry>
+
+       <varlistentry>
+        <term><literal>restart_lsn</literal> (<type>text</type>)</term>
+        <listitem>
+         <para>
+          The replication slot's <literal>restart_lsn</literal>.
+         </para>
+        </listitem>
+       </varlistentry>
+
+       <varlistentry>
+        <term><literal>restart_tli</literal> (<type>int8</type>)</term>
+        <listitem>
+         <para>
+          The timeline ID associated to <literal>restart_lsn</literal>,
+          following the current timeline history.
+         </para>
+        </listitem>
+       </varlistentry>
+      </variablelist>
+     </para>
+    </listitem>
+  </varlistentry>
+
   <varlistentry>
     <term><literal>START_REPLICATION</literal> [ <literal>SLOT</literal> <replaceable class="parameter">slot_name</replaceable> ] [ <literal>PHYSICAL</literal> ] <replaceable class="parameter">XXX/XXX</replaceable> [ <literal>TIMELINE</literal> <replaceable class="parameter">tli</replaceable> ]
      <indexterm><primary>START_REPLICATION</primary></indexterm>

diff --git a/src/backend/replication/repl_gram.y b/src/backend/replication/repl_gram.y
index 126380e2df77d3f29e391c0db0d68f9d38bf5bfe..dcb1108579e0ac931a935b21b95e697d9aee2180 100644 (file)
--- a/src/backend/replication/repl_gram.y
+++ b/src/backend/replication/repl_gram.y
@@ -64,6 +64,7 @@ static SQLCmd *make_sqlcmd(void);
 /* Keyword tokens. */
 %token K_BASE_BACKUP
 %token K_IDENTIFY_SYSTEM
+%token K_READ_REPLICATION_SLOT
 %token K_SHOW
 %token K_START_REPLICATION
 %token K_CREATE_REPLICATION_SLOT
@@ -94,7 +95,7 @@ static SQLCmd *make_sqlcmd(void);
 %type <node>   command
 %type <node>   base_backup start_replication start_logical_replication
                                create_replication_slot drop_replication_slot identify_system
-                               timeline_history show sql_cmd
+                               read_replication_slot timeline_history show sql_cmd
 %type <list>   base_backup_legacy_opt_list generic_option_list
 %type <defelt> base_backup_legacy_opt generic_option
 %type <uintval>        opt_timeline
@@ -125,6 +126,7 @@ command:
                        | start_logical_replication
                        | create_replication_slot
                        | drop_replication_slot
+                       | read_replication_slot
                        | timeline_history
                        | show
                        | sql_cmd
@@ -140,6 +142,18 @@ identify_system:
                                }
                        ;
 
+/*
+ * READ_REPLICATION_SLOT %s
+ */
+read_replication_slot:
+                       K_READ_REPLICATION_SLOT var_name
+                               {
+                                       ReadReplicationSlotCmd *n = makeNode(ReadReplicationSlotCmd);
+                                       n->slotname = $2;
+                                       $$ = (Node *) n;
+                               }
+                       ;
+
 /*
  * SHOW setting
  */

diff --git a/src/backend/replication/repl_scanner.l b/src/backend/replication/repl_scanner.l
index c038a636c3839baf25f8fc2ad6586b74b47b3fff..1b599c255e8fbcb5d2e7439d2adbf081eec7de7a 100644 (file)
--- a/src/backend/replication/repl_scanner.l
+++ b/src/backend/replication/repl_scanner.l
@@ -85,6 +85,7 @@ identifier            {ident_start}{ident_cont}*
 BASE_BACKUP                    { return K_BASE_BACKUP; }
 FAST                   { return K_FAST; }
 IDENTIFY_SYSTEM                { return K_IDENTIFY_SYSTEM; }
+READ_REPLICATION_SLOT  { return K_READ_REPLICATION_SLOT; }
 SHOW           { return K_SHOW; }
 LABEL                  { return K_LABEL; }
 NOWAIT                 { return K_NOWAIT; }

diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index b811a5c0ef2b77e89eb6e5ca91bafb7e4ea42c1e..d8f5f113c02ef341a459cecf02c0d9c018ecd82a 100644 (file)
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -232,6 +232,7 @@ static void XLogSendLogical(void);
 static void WalSndDone(WalSndSendDataCallback send_data);
 static XLogRecPtr GetStandbyFlushRecPtr(void);
 static void IdentifySystem(void);
+static void ReadReplicationSlot(ReadReplicationSlotCmd *cmd);
 static void CreateReplicationSlot(CreateReplicationSlotCmd *cmd);
 static void DropReplicationSlot(DropReplicationSlotCmd *cmd);
 static void StartReplication(StartReplicationCmd *cmd);
@@ -457,6 +458,104 @@ IdentifySystem(void)
        end_tup_output(tstate);
 }
 
+/* Handle READ_REPLICATION_SLOT command */
+static void
+ReadReplicationSlot(ReadReplicationSlotCmd *cmd)
+{
+#define READ_REPLICATION_SLOT_COLS 3
+       ReplicationSlot *slot;
+       DestReceiver *dest;
+       TupOutputState *tstate;
+       TupleDesc       tupdesc;
+       Datum           values[READ_REPLICATION_SLOT_COLS];
+       bool            nulls[READ_REPLICATION_SLOT_COLS];
+
+       tupdesc = CreateTemplateTupleDesc(READ_REPLICATION_SLOT_COLS);
+       TupleDescInitBuiltinEntry(tupdesc, (AttrNumber) 1, "slot_type",
+                                                         TEXTOID, -1, 0);
+       TupleDescInitBuiltinEntry(tupdesc, (AttrNumber) 2, "restart_lsn",
+                                                         TEXTOID, -1, 0);
+       /* TimeLineID is unsigned, so int4 is not wide enough. */
+       TupleDescInitBuiltinEntry(tupdesc, (AttrNumber) 3, "restart_tli",
+                                                         INT8OID, -1, 0);
+
+       MemSet(values, 0, READ_REPLICATION_SLOT_COLS * sizeof(Datum));
+       MemSet(nulls, true, READ_REPLICATION_SLOT_COLS * sizeof(bool));
+
+       LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
+       slot = SearchNamedReplicationSlot(cmd->slotname, false);
+       if (slot == NULL || !slot->in_use)
+       {
+               LWLockRelease(ReplicationSlotControlLock);
+       }
+       else
+       {
+               ReplicationSlot slot_contents;
+               int                     i = 0;
+
+               /* Copy slot contents while holding spinlock */
+               SpinLockAcquire(&slot->mutex);
+               slot_contents = *slot;
+               SpinLockRelease(&slot->mutex);
+               LWLockRelease(ReplicationSlotControlLock);
+
+               if (OidIsValid(slot_contents.data.database))
+                       ereport(ERROR,
+                                       errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+                                       errmsg("cannot use \"%s\" with logical replication slot \"%s\"",
+                                                  "READ_REPLICATION_SLOT",
+                                                  NameStr(slot_contents.data.name)));
+
+               /* slot type */
+               values[i] = CStringGetTextDatum("physical");
+               nulls[i] = false;
+               i++;
+
+               /* start LSN */
+               if (!XLogRecPtrIsInvalid(slot_contents.data.restart_lsn))
+               {
+                       char            xloc[64];
+
+                       snprintf(xloc, sizeof(xloc), "%X/%X",
+                                        LSN_FORMAT_ARGS(slot_contents.data.restart_lsn));
+                       values[i] = CStringGetTextDatum(xloc);
+                       nulls[i] = false;
+               }
+               i++;
+
+               /* timeline this WAL was produced on */
+               if (!XLogRecPtrIsInvalid(slot_contents.data.restart_lsn))
+               {
+                       TimeLineID      slots_position_timeline;
+                       TimeLineID      current_timeline;
+                       List       *timeline_history = NIL;
+
+                       /*
+                        * While in recovery, use as timeline the currently-replaying one
+                        * to get the LSN position's history.
+                        */
+                       if (RecoveryInProgress())
+                               (void) GetXLogReplayRecPtr(&current_timeline);
+                       else
+                               current_timeline = ThisTimeLineID;
+
+                       timeline_history = readTimeLineHistory(current_timeline);
+                       slots_position_timeline = tliOfPointInHistory(slot_contents.data.restart_lsn,
+                                                                                                                 timeline_history);
+                       values[i] = Int64GetDatum((int64) slots_position_timeline);
+                       nulls[i] = false;
+               }
+               i++;
+
+               Assert(i == READ_REPLICATION_SLOT_COLS);
+       }
+
+       dest = CreateDestReceiver(DestRemoteSimple);
+       tstate = begin_tup_output_tupdesc(dest, tupdesc, &TTSOpsVirtual);
+       do_tup_output(tstate, values, nulls);
+       end_tup_output(tstate);
+}
+
 
 /*
  * Handle TIMELINE_HISTORY command.
@@ -1622,6 +1721,13 @@ exec_replication_command(const char *cmd_string)
                        EndReplicationCommand(cmdtag);
                        break;
 
+               case T_ReadReplicationSlotCmd:
+                       cmdtag = "READ_REPLICATION_SLOT";
+                       set_ps_display(cmdtag);
+                       ReadReplicationSlot((ReadReplicationSlotCmd *) cmd_node);
+                       EndReplicationCommand(cmdtag);
+                       break;
+
                case T_BaseBackupCmd:
                        cmdtag = "BASE_BACKUP";
                        set_ps_display(cmdtag);

diff --git a/src/include/nodes/nodes.h b/src/include/nodes/nodes.h
index e0057daa0616a3cbf35a6aa1afedbbe877e5fd36..541e9861bab388b05de1dc7654822d4c75701282 100644 (file)
--- a/src/include/nodes/nodes.h
+++ b/src/include/nodes/nodes.h
@@ -496,6 +496,7 @@ typedef enum NodeTag
        T_BaseBackupCmd,
        T_CreateReplicationSlotCmd,
        T_DropReplicationSlotCmd,
+       T_ReadReplicationSlotCmd,
        T_StartReplicationCmd,
        T_TimeLineHistoryCmd,
        T_SQLCmd,

diff --git a/src/include/nodes/replnodes.h b/src/include/nodes/replnodes.h
index faa3a251f26d51b7a0f06ad1217a7cbae64f4198..a746fafc1216c6f853bcfa941b8d79a1f7ffb0b8 100644 (file)
--- a/src/include/nodes/replnodes.h
+++ b/src/include/nodes/replnodes.h
@@ -87,6 +87,17 @@ typedef struct StartReplicationCmd
 } StartReplicationCmd;
 
 
+/* ----------------------
+ *             READ_REPLICATION_SLOT command
+ * ----------------------
+ */
+typedef struct ReadReplicationSlotCmd
+{
+       NodeTag         type;
+       char       *slotname;
+} ReadReplicationSlotCmd;
+
+
 /* ----------------------
  *             TIMELINE_HISTORY command
  * ----------------------

diff --git a/src/test/recovery/t/001_stream_rep.pl b/src/test/recovery/t/001_stream_rep.pl
index bc62ec66bce9da80d4d5bec87ee9809b92951c38..b3dc68d54bc827ee61d4b1f185e60d0ecf09bd74 100644 (file)
--- a/src/test/recovery/t/001_stream_rep.pl
+++ b/src/test/recovery/t/001_stream_rep.pl
@@ -6,7 +6,7 @@ use strict;
 use warnings;
 use PostgreSQL::Test::Cluster;
 use PostgreSQL::Test::Utils;
-use Test::More tests => 49;
+use Test::More tests => 53;
 
 # Initialize primary node
 my $node_primary = PostgreSQL::Test::Cluster->new('primary');
@@ -254,6 +254,36 @@ ok( $ret == 0,
        "SHOW with superuser-settable parameter, replication role and logical replication"
 );
 
+note "testing READ_REPLICATION_SLOT command for replication connection";
+
+my $slotname = 'test_read_replication_slot_physical';
+
+($ret, $stdout, $stderr) = $node_primary->psql(
+       'postgres',
+       'READ_REPLICATION_SLOT non_existent_slot;',
+       extra_params => [ '-d', $connstr_rep ]);
+ok($ret == 0, "READ_REPLICATION_SLOT exit code 0 on success");
+like($stdout, qr/^||$/,
+       "READ_REPLICATION_SLOT returns NULL values if slot does not exist");
+
+$node_primary->psql(
+       'postgres',
+       "CREATE_REPLICATION_SLOT $slotname PHYSICAL RESERVE_WAL;",
+       extra_params => [ '-d', $connstr_rep ]);
+
+($ret, $stdout, $stderr) = $node_primary->psql(
+       'postgres',
+       "READ_REPLICATION_SLOT $slotname;",
+       extra_params => [ '-d', $connstr_rep ]);
+ok($ret == 0, "READ_REPLICATION_SLOT success with existing slot");
+like($stdout, qr/^physical\|[^|]*\|1$/,
+       "READ_REPLICATION_SLOT returns tuple with slot information");
+
+$node_primary->psql(
+       'postgres',
+       "DROP_REPLICATION_SLOT $slotname;",
+       extra_params => [ '-d', $connstr_rep ]);
+
 note "switching to physical replication slot";
 
 # Switch to using a physical replication slot. We can do this without a new

diff --git a/src/test/recovery/t/006_logical_decoding.pl b/src/test/recovery/t/006_logical_decoding.pl
index 1655298bf5ba1a6d1cce0492ff39c184f45b6c26..46ea5a3866ef69240e416a9a2ddec8543fd36b51 100644 (file)
--- a/src/test/recovery/t/006_logical_decoding.pl
+++ b/src/test/recovery/t/006_logical_decoding.pl
@@ -10,7 +10,7 @@ use strict;
 use warnings;
 use PostgreSQL::Test::Cluster;
 use PostgreSQL::Test::Utils;
-use Test::More tests => 14;
+use Test::More tests => 15;
 use Config;
 
 # Initialize primary node
@@ -39,6 +39,15 @@ ok( $stderr =~
          m/replication slot "test_slot" was not created in this database/,
        "Logical decoding correctly fails to start");
 
+($result, $stdout, $stderr) = $node_primary->psql(
+       'template1',
+       qq[READ_REPLICATION_SLOT test_slot;],
+       replication => 'database');
+like(
+       $stderr,
+       qr/cannot use "READ_REPLICATION_SLOT" with logical replication slot "test_slot"/,
+       'READ_REPLICATION_SLOT not supported for logical slots');
+
 # Check case of walsender not using a database connection.  Logical
 # decoding should not be allowed.
 ($result, $stdout, $stderr) = $node_primary->psql(

diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 1c7bac0578c7893da9693fd256a09982bae1ef58..40fbcddd2012e664e084521ef31784522fbd33bc 100644 (file)
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -2129,6 +2129,7 @@ ReadBufferMode
 ReadBytePtrType
 ReadExtraTocPtrType
 ReadFunc
+ReadReplicationSlotCmd
 ReassignOwnedStmt
 RecheckForeignScan_function
 RecordCacheEntry