Re: [PATCH] Logical decoding timeline following take II

On 2016-10-24 17:49:13 +0200, Petr Jelinek wrote:
> + * Determine which timeline to read an xlog page from and set the
> + * XLogReaderState's state->currTLI to that timeline ID.

"XLogReaderState's state->currTLI" - the state's a bit redundant.

> + * The caller must also make sure it doesn't read past the current redo pointer
> + * so it doesn't fail to notice that the current timeline became historical.
> + */

Not sure what that means? The redo pointer usually menas the "logical
start" of the last checkpoint, but I don't see how thta could be meant
here?

> +static void
> +XLogReadDetermineTimeline(XLogReaderState *state, XLogRecPtr wantPage, uint32 wantLength)
> +{
> + const XLogRecPtr lastReadPage = state->readSegNo * XLogSegSize + state->readOff;
> +
> + elog(DEBUG4, "Determining timeline for read at %X/%X+%X",
> + (uint32)(wantPage>>32), (uint32)wantPage, wantLength);

Doing this on every single read from a page seems quite verbose. It's
also (like most or all the following debug elogs) violating the casing
prescribed by the message style guidelines.

> + /*
> + * If the desired page is currently read in and valid, we have nothing to do.
> + *
> + * The caller should've ensured that it didn't previously advance readOff
> + * past the valid limit of this timeline, so it doesn't matter if the current
> + * TLI has since become historical.
> + */
> + if (lastReadPage == wantPage &&
> + state->readLen != 0 &&
> + lastReadPage + state->readLen >= wantPage + Min(wantLength,XLOG_BLCKSZ-1))
> + {
> + elog(DEBUG4, "Wanted data already valid"); //XXX
> + return;
> + }

With that kind of comment/XXX present, this surely can't be ready for
committer?

> + /*
> + * If we're reading from the current timeline, it hasn't become historical
> + * and the page we're reading is after the last page read, we can again
> + * just carry on. (Seeking backwards requires a check to make sure the older
> + * page isn't on a prior timeline).
> + */

How can ThisTimeLineID become historical?

> + if (state->currTLI == ThisTimeLineID && wantPage >= lastReadPage)
> + {
> + Assert(state->currTLIValidUntil == InvalidXLogRecPtr);
> + elog(DEBUG4, "On current timeline");
> + return;
> + }

Also, is it actually ok to rely on ThisTimeLineID here? That's IIRC not
maintained outside of the startup process, until recovery ended (cf it
being set in InitXLOGAccess() called via RecoveryInProgress()).

> /*
> - * 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.
> */

I guess you mean cascading as "replicating logically from a physical
standby"? I don't think it's good to use cascading here, because that's
usually thought to mean doing physical SR from a standby...

> diff --git a/src/backend/replication/logical/logicalfuncs.c b/src/backend/replication/logical/logicalfuncs.c
> index 318726e..4315fb3 100644
> --- a/src/backend/replication/logical/logicalfuncs.c
> +++ b/src/backend/replication/logical/logicalfuncs.c
> @@ -234,12 +234,6 @@ pg_logical_slot_get_changes_guts(FunctionCallInfo fcinfo, bool confirm, bool bin
> rsinfo->setResult = p->tupstore;
> rsinfo->setDesc = p->tupdesc;
>
> - /* compute the current end-of-wal */
> - if (!RecoveryInProgress())
> - end_of_wal = GetFlushRecPtr();
> - else
> - end_of_wal = GetXLogReplayRecPtr(NULL);
> -
> ReplicationSlotAcquire(NameStr(*name));
>
> PG_TRY();
> @@ -279,6 +273,12 @@ pg_logical_slot_get_changes_guts(FunctionCallInfo fcinfo, bool confirm, bool bin
> /* invalidate non-timetravel entries */
> InvalidateSystemCaches();
>
> + if (!RecoveryInProgress())
> + end_of_wal = GetFlushRecPtr();
> + else
> + end_of_wal = GetXLogReplayRecPtr(NULL);
> +
> + /* Decode until we run out of records */
> while ((startptr != InvalidXLogRecPtr && startptr < end_of_wal) ||
> (ctx->reader->EndRecPtr != InvalidXLogRecPtr && ctx->reader->EndRecPtr < end_of_wal))
> {

That seems like a pretty random change?

> diff --git a/src/include/access/xlogreader.h b/src/include/access/xlogreader.h
> index deaa7f5..caff9a6 100644
> --- a/src/include/access/xlogreader.h
> +++ b/src/include/access/xlogreader.h
> @@ -27,6 +27,10 @@
>
> #include "access/xlogrecord.h"
>
> +#ifndef FRONTEND
> +#include "nodes/pg_list.h"
> +#endif

Why is that needed?
> diff --git a/src/test/modules/test_slot_timelines/README b/src/test/modules/test_slot_timelines/README
> new file mode 100644
> index 0000000..585f02f
> --- /dev/null
> +++ b/src/test/modules/test_slot_timelines/README
> @@ -0,0 +1,19 @@
> +A test module for logical decoding failover and timeline following.
> +
> +This module provides a minimal way to maintain logical slots on replicas
> +that mirror the state on the master. It doesn't make decoding possible,
> +just tracking slot state so that a decoding client that's using the master
> +can follow a physical failover to the standby. The master doesn't know
> +about the slots on the standby, they're synced by a client that connects
> +to both.
> +
> +This is intentionally not part of the test_decoding module because that's meant
> +to serve as example code, where this module exercises internal server features
> +by unsafely exposing internal state to SQL. It's not the right way to do
> +failover, it's just a simple way to test it from the perl TAP framework to
> +prove the feature works.
> +
> +In a practical implementation of this approach a bgworker on the master would
> +monitor slot positions and relay them to a bgworker on the standby that applies
> +the position updates without exposing slot internals to SQL. That's too complex
> +for this test framework though.

I still don't want to have this code in postgres, it's just an example
for doing something bad. Lets get proper feedback control in, and go
from there.

> diff --git a/src/test/modules/test_slot_timelines/expected/load_extension_1.out b/src/test/modules/test_slot_timelines/expected/load_extension_1.out
> new file mode 100644
> index 0000000..0db21e4
> --- /dev/null
> +++ b/src/test/modules/test_slot_timelines/expected/load_extension_1.out
> @@ -0,0 +1,7 @@
> +CREATE EXTENSION test_slot_timelines;
> +SELECT test_slot_timelines_create_logical_slot('test_slot', 'test_decoding');
> +ERROR: replication slots can only be used if max_replication_slots > 0
> +SELECT test_slot_timelines_advance_logical_slot('test_slot', txid_current()::text::xid, txid_current()::text::xid, pg_current_xlog_location(), pg_current_xlog_location());
> +ERROR: replication slots can only be used if max_replication_slots > 0
> +SELECT pg_drop_replication_slot('test_slot');
> +ERROR: replication slots can only be used if max_replication_slots > 0
> diff --git a/src/test/modules/test_slot_timelines/sql/load_extension.sql b/src/test/modules/test_slot_timelines/sql/load_extension.sql
> new file mode 100644
> index 0000000..2440355

It doesn't seem worthwhlie to maintain a test for this - why do we run
the tests with those settings in the first place?

Greetings,

Andres Freund