Add pg_recvlogical —-endpos=LSN

Allow pg_recvlogical to specify an ending LSN, complementing
the existing -—startpos=LSN option.

Craig Ringer, reviewed by Euler Taveira and Naoki Okano

diff --git a/doc/src/sgml/ref/pg_recvlogical.sgml b/doc/src/sgml/ref/pg_recvlogical.sgml
index b35881f2b9eca9e614a1ad4f940609b54f1727d1..d066ce87016e379ef935d137bd149aed71d74469 100644 (file)
--- a/doc/src/sgml/ref/pg_recvlogical.sgml
+++ b/doc/src/sgml/ref/pg_recvlogical.sgml
@@ -38,6 +38,14 @@ PostgreSQL documentation
    constraints as <xref linkend="app-pgreceivexlog">, plus those for logical
    replication (see <xref linkend="logicaldecoding">).
   </para>
+
+  <para>
+   <command>pg_recvlogical</> has no equivalent to the logical decoding
+   SQL interface's peek and get modes. It sends replay confirmations for
+   data lazily as it receives it and on clean exit. To examine pending data on
+    a slot without consuming it, use
+   <link linkend="functions-replication"><function>pg_logical_slot_peek_changes</></>.
+  </para>
  </refsect1>
 
  <refsect1>
@@ -154,6 +162,32 @@ PostgreSQL documentation
       </listitem>
      </varlistentry>
 
+     <varlistentry>
+      <term><option>-E <replaceable>lsn</replaceable></option></term>
+      <term><option>--endpos=<replaceable>lsn</replaceable></option></term>
+      <listitem>
+       <para>
+        In <option>--start</option> mode, automatically stop replication
+        and exit with normal exit status 0 when receiving reaches the
+        specified LSN.  If specified when not in <option>--start</option>
+        mode, an error is raised.
+       </para>
+
+       <para>
+        If there's a record with LSN exactly equal to <replaceable>lsn</>,
+        the record will be output.
+       </para>
+
+       <para>
+        The <option>--endpos</option> option is not aware of transaction
+        boundaries and may truncate output partway through a transaction.
+        Any partially output transaction will not be consumed and will be
+        replayed again when the slot is next read from. Individual messages
+        are never truncated.
+       </para>
+      </listitem>
+     </varlistentry>
+
      <varlistentry>
       <term><option>--if-not-exists</option></term>
       <listitem>

diff --git a/src/bin/pg_basebackup/pg_recvlogical.c b/src/bin/pg_basebackup/pg_recvlogical.c
index 49ed2abe55822d7b1e8484cbc572ed9ef5484ac7..658e2ba91f56f29d46841335199f003adb4e122b 100644 (file)
--- a/src/bin/pg_basebackup/pg_recvlogical.c
+++ b/src/bin/pg_basebackup/pg_recvlogical.c
@@ -40,6 +40,7 @@ static int    noloop = 0;
 static int     standby_message_timeout = 10 * 1000;            /* 10 sec = default */
 static int     fsync_interval = 10 * 1000; /* 10 sec = default */
 static XLogRecPtr startpos = InvalidXLogRecPtr;
+static XLogRecPtr endpos = InvalidXLogRecPtr;
 static bool do_create_slot = false;
 static bool slot_exists_ok = false;
 static bool do_start_slot = false;
@@ -63,6 +64,9 @@ static XLogRecPtr output_fsync_lsn = InvalidXLogRecPtr;
 static void usage(void);
 static void StreamLogicalLog(void);
 static void disconnect_and_exit(int code);
+static bool flushAndSendFeedback(PGconn *conn, TimestampTz *now);
+static void prepareToTerminate(PGconn *conn, XLogRecPtr endpos,
+                                  bool keepalive, XLogRecPtr lsn);
 
 static void
 usage(void)
@@ -81,6 +85,7 @@ usage(void)
                         "                         time between fsyncs to the output file (default: %d)\n"), (fsync_interval / 1000));
        printf(_("      --if-not-exists    do not error if slot already exists when creating a slot\n"));
        printf(_("  -I, --startpos=LSN     where in an existing slot should the streaming start\n"));
+       printf(_("  -E, --endpos=LSN       exit after receiving the specified LSN\n"));
        printf(_("  -n, --no-loop          do not loop on connection lost\n"));
        printf(_("  -o, --option=NAME[=VALUE]\n"
                         "                         pass option NAME with optional value VALUE to the\n"
@@ -281,6 +286,7 @@ StreamLogicalLog(void)
                int                     bytes_written;
                int64           now;
                int                     hdr_len;
+               XLogRecPtr      cur_record_lsn = InvalidXLogRecPtr;
 
                if (copybuf != NULL)
                {
@@ -454,6 +460,7 @@ StreamLogicalLog(void)
                        int                     pos;
                        bool            replyRequested;
                        XLogRecPtr      walEnd;
+                       bool            endposReached = false;
 
                        /*
                         * Parse the keepalive message, enclosed in the CopyData message.
@@ -476,18 +483,32 @@ StreamLogicalLog(void)
                        }
                        replyRequested = copybuf[pos];
 
-                       /* If the server requested an immediate reply, send one. */
-                       if (replyRequested)
+                       if (endpos != InvalidXLogRecPtr && walEnd >= endpos)
                        {
-                               /* fsync data, so we send a recent flush pointer */
-                               if (!OutputFsync(now))
-                                       goto error;
+                               /*
+                                * If there's nothing to read on the socket until a keepalive
+                                * we know that the server has nothing to send us; and if
+                                * walEnd has passed endpos, we know nothing else can have
+                                * committed before endpos.  So we can bail out now.
+                                */
+                               endposReached = true;
+                       }
 
-                               now = feGetCurrentTimestamp();
-                               if (!sendFeedback(conn, now, true, false))
+                       /* Send a reply, if necessary */
+                       if (replyRequested || endposReached)
+                       {
+                               if (!flushAndSendFeedback(conn, &now))
                                        goto error;
                                last_status = now;
                        }
+
+                       if (endposReached)
+                       {
+                               prepareToTerminate(conn, endpos, true, InvalidXLogRecPtr);
+                               time_to_abort = true;
+                               break;
+                       }
+
                        continue;
                }
                else if (copybuf[0] != 'w')
@@ -497,7 +518,6 @@ StreamLogicalLog(void)
                        goto error;
                }
 
-
                /*
                 * Read the header of the XLogData message, enclosed in the CopyData
                 * message. We only need the WAL location field (dataStart), the rest
@@ -515,12 +535,23 @@ StreamLogicalLog(void)
                }
 
                /* Extract WAL location for this block */
-               {
-                       XLogRecPtr      temp = fe_recvint64(&copybuf[1]);
+               cur_record_lsn = fe_recvint64(&copybuf[1]);
 
-                       output_written_lsn = Max(temp, output_written_lsn);
+               if (endpos != InvalidXLogRecPtr && cur_record_lsn > endpos)
+               {
+                       /*
+                        * We've read past our endpoint, so prepare to go away being
+                        * cautious about what happens to our output data.
+                        */
+                       if (!flushAndSendFeedback(conn, &now))
+                               goto error;
+                       prepareToTerminate(conn, endpos, false, cur_record_lsn);
+                       time_to_abort = true;
+                       break;
                }
 
+               output_written_lsn = Max(cur_record_lsn, output_written_lsn);
+
                bytes_left = r - hdr_len;
                bytes_written = 0;
 
@@ -557,10 +588,29 @@ StreamLogicalLog(void)
                                        strerror(errno));
                        goto error;
                }
+
+               if (endpos != InvalidXLogRecPtr && cur_record_lsn == endpos)
+               {
+                       /* endpos was exactly the record we just processed, we're done */
+                       if (!flushAndSendFeedback(conn, &now))
+                               goto error;
+                       prepareToTerminate(conn, endpos, false, cur_record_lsn);
+                       time_to_abort = true;
+                       break;
+               }
        }
 
        res = PQgetResult(conn);
-       if (PQresultStatus(res) != PGRES_COMMAND_OK)
+       if (PQresultStatus(res) == PGRES_COPY_OUT)
+       {
+               /*
+                * We're doing a client-initiated clean exit and have sent CopyDone to
+                * the server. We've already sent replay confirmation and fsync'd so
+                * we can just clean up the connection now.
+                */
+               goto error;
+       }
+       else if (PQresultStatus(res) != PGRES_COMMAND_OK)
        {
                fprintf(stderr,
                                _("%s: unexpected termination of replication stream: %s"),
@@ -638,6 +688,7 @@ main(int argc, char **argv)
                {"password", no_argument, NULL, 'W'},
 /* replication options */
                {"startpos", required_argument, NULL, 'I'},
+               {"endpos", required_argument, NULL, 'E'},
                {"option", required_argument, NULL, 'o'},
                {"plugin", required_argument, NULL, 'P'},
                {"status-interval", required_argument, NULL, 's'},
@@ -673,7 +724,7 @@ main(int argc, char **argv)
                }
        }
 
-       while ((c = getopt_long(argc, argv, "f:F:nvd:h:p:U:wWI:o:P:s:S:",
+       while ((c = getopt_long(argc, argv, "f:F:nvd:h:p:U:wWI:E:o:P:s:S:",
                                                        long_options, &option_index)) != -1)
        {
                switch (c)
@@ -733,6 +784,16 @@ main(int argc, char **argv)
                                }
                                startpos = ((uint64) hi) << 32 | lo;
                                break;
+                       case 'E':
+                               if (sscanf(optarg, "%X/%X", &hi, &lo) != 2)
+                               {
+                                       fprintf(stderr,
+                                                       _("%s: could not parse end position \"%s\"\n"),
+                                                       progname, optarg);
+                                       exit(1);
+                               }
+                               endpos = ((uint64) hi) << 32 | lo;
+                               break;
                        case 'o':
                                {
                                        char       *data = pg_strdup(optarg);
@@ -857,6 +918,16 @@ main(int argc, char **argv)
                exit(1);
        }
 
+       if (endpos != InvalidXLogRecPtr && !do_start_slot)
+       {
+               fprintf(stderr,
+                               _("%s: --endpos may only be specified with --start\n"),
+                               progname);
+               fprintf(stderr, _("Try \"%s --help\" for more information.\n"),
+                               progname);
+               exit(1);
+       }
+
 #ifndef WIN32
        pqsignal(SIGINT, sigint_handler);
        pqsignal(SIGHUP, sighup_handler);
@@ -923,8 +994,8 @@ main(int argc, char **argv)
                if (time_to_abort)
                {
                        /*
-                        * We've been Ctrl-C'ed. That's not an error, so exit without an
-                        * errorcode.
+                        * We've been Ctrl-C'ed or reached an exit limit condition. That's
+                        * not an error, so exit without an errorcode.
                         */
                        disconnect_and_exit(0);
                }
@@ -943,3 +1014,47 @@ main(int argc, char **argv)
                }
        }
 }
+
+/*
+ * Fsync our output data, and send a feedback message to the server.  Returns
+ * true if successful, false otherwise.
+ *
+ * If successful, *now is updated to the current timestamp just before sending
+ * feedback.
+ */
+static bool
+flushAndSendFeedback(PGconn *conn, TimestampTz *now)
+{
+       /* flush data to disk, so that we send a recent flush pointer */
+       if (!OutputFsync(*now))
+               return false;
+       *now = feGetCurrentTimestamp();
+       if (!sendFeedback(conn, *now, true, false))
+               return false;
+
+       return true;
+}
+
+/*
+ * Try to inform the server about of upcoming demise, but don't wait around or
+ * retry on failure.
+ */
+static void
+prepareToTerminate(PGconn *conn, XLogRecPtr endpos, bool keepalive, XLogRecPtr lsn)
+{
+       (void) PQputCopyEnd(conn, NULL);
+       (void) PQflush(conn);
+
+       if (verbose)
+       {
+               if (keepalive)
+                       fprintf(stderr, "%s: endpos %X/%X reached by keepalive\n",
+                                       progname,
+                                       (uint32) (endpos >> 32), (uint32) endpos);
+               else
+                       fprintf(stderr, "%s: endpos %X/%X reached by record at %X/%X\n",
+                                       progname, (uint32) (endpos >> 32), (uint32) (endpos),
+                                       (uint32) (lsn >> 32), (uint32) lsn);
+
+       }
+}