Re: Timeline following for logical slots

Hi

On 2016-03-14 20:10:58 -0300, Alvaro Herrera wrote:
> diff --git a/src/backend/access/transam/xlogreader.c b/src/backend/access/transam/xlogreader.c
> index fcb0872..7b60b8f 100644
> --- a/src/backend/access/transam/xlogreader.c
> +++ b/src/backend/access/transam/xlogreader.c
> @@ -10,9 +10,11 @@
>   *
>   * NOTES
>   *           See xlogreader.h for more notes on this facility.
> + *
> + *           This file is compiled as both front-end and backend code, so it
> + *           may not use ereport, server-defined static variables, etc.
>   *-------------------------------------------------------------------------
>   */
> -

Huh?

> @@ -135,6 +142,10 @@ XLogReaderFree(XLogReaderState *state)
>       pfree(state->errormsg_buf);
>       if (state->readRecordBuf)
>               pfree(state->readRecordBuf);
> +#ifndef FRONTEND
> +     if (state->timelineHistory)
> +             list_free_deep(state->timelineHistory);
> +#endif

Hm. So we don't support timelines following for frontend code, although
it'd be rather helpful for pg_xlogdump. And possibly pg_rewind.

>       pfree(state->readBuf);
>       pfree(state);
>  }
> @@ -208,10 +219,11 @@ XLogReadRecord(XLogReaderState *state, XLogRecPtr RecPtr, char **errormsg)
>
>       if (RecPtr == InvalidXLogRecPtr)
>       {
> +             /* No explicit start point; read the record after the one we just read */
>               RecPtr = state->EndRecPtr;
>
>               if (state->ReadRecPtr == InvalidXLogRecPtr)
> -                     randAccess = true;
> +                     randAccess = true;      /* allow readPageTLI to go backwards */

randAccess is doing more than that, so I'm doubtful that comment is an
improvment.

> @@ -466,9 +482,7 @@ err:
>        * Invalidate the xlog page we've cached. We might read from a different
>        * source after failure.
>        */
> -     state->readSegNo = 0;
> -     state->readOff = 0;
> -     state->readLen = 0;
> +     XLogReaderInvalCache(state);

I don't think that "cache" is the right way to describe this.

>  #include <unistd.h>
>
> -#include "miscadmin.h"
> -

spurious change imo.

> -             if (sendFile < 0 || !XLByteInSeg(recptr, sendSegNo))
> +             /* Do we need to open a new xlog segment? */
> +             if (sendFile < 0 || !XLByteInSeg(recptr, sendSegNo) ||
> +                     sendTLI != tli)
>               {

s/open a new/open a different/?  New imo has connotations that we don't
really want here.

>               /* Need to seek in the file? */
>               if (sendOff != startoff)
>               {
>                       if (lseek(sendFile, (off_t) startoff, SEEK_SET) < 0)
> -                     {
> -                             char            path[MAXPGPATH];
> -
> -                             XLogFilePath(path, tli, sendSegNo);
> -
>                               ereport(ERROR,
>                                               (errcode_for_file_access(),
>                                 errmsg("could not seek in log segment %s to offset %u: %m",
> -                                              path, startoff)));
> -                     }
> +                                              XLogFileNameP(tli, sendSegNo), startoff)));
>                       sendOff = startoff;
>               }

Not a serious issue, more a general remark: I'm doubtful that going for
palloc in error situations is good practice. This will be allocated in
the current memory context; without access to the emergency error
reserves.

> +static void
> +XLogReadDetermineTimeline(XLogReaderState *state)
> +{
> +     /* Read the history on first time through */
> +     if (state->timelineHistory == NIL)
> +             state->timelineHistory = readTimeLineHistory(ThisTimeLineID);
> +
> +     /*
> +      * Are we reading the record immediately following the one we read last
> +      * time?  If not, then don't use the cached timeline info.
> +      */
> +     if (state->currRecPtr != state->EndRecPtr)
> +     {
> +             state->currTLI = 0;
> +             state->currTLIValidUntil = InvalidXLogRecPtr;
> +     }


Hm. So we grow essentially a second version of the last end position and
the randAccess stuff in XLogReadRecord().

> +     if (state->currTLI == 0)
> +     {
> +             /*
> +              * Something changed; work out what timeline this record is on. We
> +              * might read it from the segment on this TLI or, if the segment is
> +              * also contained by newer timelines, the copy from a newer TLI.
> +              */
> +             state->currTLI = tliOfPointInHistory(state->currRecPtr,
> +                                                                                      state->timelineHistory);
> +
> +             /*
> +              * Look for the most recent timeline that's on the same xlog segment
> +              * as this record, since that's the only one we can assume is still
> +              * readable.
> +              */
> +             while (state->currTLI != ThisTimeLineID &&
> +                        state->currTLIValidUntil == InvalidXLogRecPtr)
> +             {
> +                     XLogRecPtr      tliSwitch;
> +                     TimeLineID      nextTLI;
> +
> +                     tliSwitch = tliSwitchPoint(state->currTLI, state->timelineHistory,
> +                                                                        &nextTLI);
> +
> +                     /* round ValidUntil down to start of seg containing the switch */
> +                     state->currTLIValidUntil =
> +                             ((tliSwitch / XLogSegSize) * XLogSegSize);
> +
> +                     if (state->currRecPtr >= state->currTLIValidUntil)
> +                     {
> +                             /*
> +                              * The new currTLI ends on this WAL segment so check the next
> +                              * TLI to see if it's the last one on the segment.
> +                              *
> +                              * If that's the current TLI we'll stop searching.

I don't really understand how we're stopping searching here?

> +                              */
> +                             state->currTLI = nextTLI;
> +                             state->currTLIValidUntil = InvalidXLogRecPtr;
> +                     }
> +             }
> +}


XLogReadDetermineTimeline() doesn't sit quite right with me, I do wonder
whether there's not a simpler way to write this.

> +/*
> + * XLogPageReadCB callback for reading local xlog files
>   *
>   * Public because it would likely be very helpful for someone writing another
>   * output method outside walsender, e.g. in a bgworker.
>   *
> - * TODO: The walsender has it's own version of this, but it relies on the
> + * TODO: The walsender has its own version of this, but it relies on the
>   * walsender's latch being set whenever WAL is flushed. No such infrastructure
>   * exists for normal backends, so we have to do a check/sleep/repeat style of
>   * loop for now.
> @@ -754,46 +897,88 @@ int
>  read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr,
>       int reqLen, XLogRecPtr targetRecPtr, char *cur_page, TimeLineID *pageTLI)
>  {
> -     XLogRecPtr      flushptr,
> +     XLogRecPtr      read_upto,
>                               loc;
>       int                     count;
>
>       loc = targetPagePtr + reqLen;
> +
> +     /* Make sure enough xlog is available... */
>       while (1)
>       {
>               /*
> -              * TODO: we're going to have to do something more intelligent about
> -              * timelines on standbys. Use readTimeLineHistory() and
> -              * tliOfPointInHistory() to get the proper LSN? For now we'll catch
> -              * that case earlier, but the code and TODO is left in here for when
> -              * that changes.
> +              * Check which timeline to get the record from.
> +              *
> +              * We have to do it each time through the loop because if we're in
> +              * recovery as a cascading standby, the current timeline might've
> +              * become historical.
>                */
> -             if (!RecoveryInProgress())
> +             XLogReadDetermineTimeline(state);
> +
> +             if (state->currTLI == ThisTimeLineID)
>               {
> -                     *pageTLI = ThisTimeLineID;
> -                     flushptr = GetFlushRecPtr();
> +                     /*
> +                      * We're reading from the current timeline so we might have to
> +                      * wait for the desired record to be generated (or, for a standby,
> +                      * received & replayed)
> +                      */
> +                     if (!RecoveryInProgress())
> +                     {
> +                             *pageTLI = ThisTimeLineID;
> +                             read_upto = GetFlushRecPtr();
> +                     }
> +                     else
> +                             read_upto = GetXLogReplayRecPtr(pageTLI);
> +
> +                     if (loc <= read_upto)
> +                             break;
> +
> +                     CHECK_FOR_INTERRUPTS();
> +                     pg_usleep(1000L);
>               }
>               else
> -                     flushptr = GetXLogReplayRecPtr(pageTLI);
> +             {
> +                     /*
> +                      * We're on a historical timeline, so limit reading to the switch
> +                      * point where we moved to the next timeline.
> +                      */
> +                     read_upto = state->currTLIValidUntil;

Hm. Is it ok to not check GetFlushRecPtr/GetXLogReplayRecPtr() here? If
so, how come?

> -     /* more than one block available */
> -     if (targetPagePtr + XLOG_BLCKSZ <= flushptr)
> +     if (targetPagePtr + XLOG_BLCKSZ <= read_upto)
> +     {
> +             /*
> +              * more than one block available; read only that block, have caller
> +              * come back if they need more.
> +              */
>               count = XLOG_BLCKSZ;
> -     /* not enough data there */
> -     else if (targetPagePtr + reqLen > flushptr)
> +     }
> +     else if (targetPagePtr + reqLen > read_upto)
> +     {
> +             /* not enough data there */
>               return -1;
> -     /* part of the page available */
> +     }
>       else
> -             count = flushptr - targetPagePtr;
> +     {
> +             /* enough bytes available to satisfy the request */
> +             count = read_upto - targetPagePtr;
> +     }
>
> -     XLogRead(cur_page, *pageTLI, targetPagePtr, XLOG_BLCKSZ);
> +     XLogRead(cur_page, *pageTLI, targetPagePtr, count);

When are we reading less than a page? That should afaik never be required.

> +             /*
> +              * We start reading xlog from the restart lsn, even though in
> +              * CreateDecodingContext we set the snapshot builder up using the
> +              * slot's candidate_restart_lsn. This means we might read xlog we
> +              * don't actually decode rows from, but the snapshot builder might
> +              * need it to get to a consistent point. The point we start returning
> +              * data to *users* at is the candidate restart lsn from the decoding
> +              * context.
> +              */

Uh? Where are we using candidate_restart_lsn that way?

> @@ -299,6 +312,18 @@ pg_logical_slot_get_changes_guts(FunctionCallInfo fcinfo, bool confirm, bool bin
>                       CHECK_FOR_INTERRUPTS();
>               }
>
> +             /* Make sure timeline lookups use the start of the next record */
> +             startptr = ctx->reader->EndRecPtr;

Huh? startptr isn't used after this, so I'm not sure what this even
means?

> +             /*
> +              * The XLogReader will read a page past the valid end of WAL because
> +              * it doesn't know about timelines. When we switch timelines and ask
> +              * it for the first page on the new timeline it will think it has it
> +              * cached, but it'll have the old partial page and say it can't find
> +              * the next record. So flush the cache.
> +              */
> +             XLogReaderInvalCache(ctx->reader);
> +

dito.

> + * Create a new logical slot, with invalid LSN and xid, directly. This does not
> + * use the snapshot builder or logical decoding machinery. It's only intended
> + * for creating a slot on a replica that mirrors the state of a slot on an
> + * upstream master.
> + *
> + * You should immediately decoding_failover_advance_logical_slot(...) it
> + * after creation.
> + */

Uh. I doubt we want this, even if it's formally located in
src/test/modules.  These comments make it appear not to be only intended
for that, and I have serious doubts about the validity of the concept as
is.