Re: Minimal logical decoding on standbys

On Wed, 10 Jul 2019 at 08:44, Andres Freund <andres@anarazel.de> wrote:
>
> Hi,
>
> Thanks for the new version! Looks like we're making progress towards
> something committable here.
>
> I think it'd be good to split the patch into a few pieces. I'd maybe do
> that like:
> 1) WAL format changes (plus required other changes)
> 2) Recovery conflicts with slots
> 3) logical decoding on standby
> 4) tests

All right. Will do that in the next patch set. For now, I have quickly
done the below changes in a single patch again (attached), in order to
get early comments if any.

>
>
> > @@ -589,6 +590,7 @@ gistXLogPageReuse(Relation rel, BlockNumber blkno, TransactionId latestRemovedXi
> >        */
> >
> >       /* XLOG stuff */
> > +     xlrec_reuse.onCatalogTable = RelationIsAccessibleInLogicalDecoding(rel);
> >       xlrec_reuse.node = rel->rd_node;
> >       xlrec_reuse.block = blkno;
> >       xlrec_reuse.latestRemovedXid = latestRemovedXid;
>
> Hm. I think we otherwise only ever use
> RelationIsAccessibleInLogicalDecoding() on tables, not on indexes.  And
> while I think this would mostly work for builtin catalog tables, it
> won't work for "user catalog tables" as RelationIsUsedAsCatalogTable()
> won't perform any useful checks for indexes.
>
> So I think we either need to look up the table, or pass it down.

Done. Passed down the heap rel.

>
>
> > diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c
> > index d768b9b..10b7857 100644
> > --- a/src/backend/access/heap/heapam.c
> > +++ b/src/backend/access/heap/heapam.c
> > @@ -7149,12 +7149,13 @@ heap_compute_xid_horizon_for_tuples(Relation rel,
> >   * see comments for vacuum_log_cleanup_info().
> >   */
> >  XLogRecPtr
> > -log_heap_cleanup_info(RelFileNode rnode, TransactionId latestRemovedXid)
> > +log_heap_cleanup_info(Relation rel, TransactionId latestRemovedXid)
> >  {
> >       xl_heap_cleanup_info xlrec;
> >       XLogRecPtr      recptr;
> >
> > -     xlrec.node = rnode;
> > +     xlrec.onCatalogTable = RelationIsAccessibleInLogicalDecoding(rel);
> > +     xlrec.node = rel->rd_node;
> >       xlrec.latestRemovedXid = latestRemovedXid;
> >
> >       XLogBeginInsert();
> > @@ -7190,6 +7191,7 @@ log_heap_clean(Relation reln, Buffer buffer,
> >       /* Caller should not call me on a non-WAL-logged relation */
> >       Assert(RelationNeedsWAL(reln));
> >
> > +     xlrec.onCatalogTable = RelationIsAccessibleInLogicalDecoding(reln);
>
> It'd probably be a good idea to add a comment to
> RelationIsUsedAsCatalogTable() that it better never invoke anything
> performing catalog accesses. Otherwise there's quite the danger with
> recursion (some operation doing RelationIsAccessibleInLogicalDecoding(),
> that then accessing the catalog, which in turn could again need to
> perform said operation, loop).

Added comments in RelationIsUsedAsCatalogTable() as well as
RelationIsAccessibleInLogicalDecoding() :

 * RelationIsAccessibleInLogicalDecoding
 * True if we need to log enough information to have access via
 * decoding snapshot.
 * This definition should not invoke anything that performs catalog
 * access. Otherwise, e.g. logging a WAL entry for catalog relation may
 * invoke this function, which will in turn do catalog access, which may
 * in turn cause another similar WAL entry to be logged, leading to
 * infinite recursion.

> >  /* Entry in pending-list of TIDs we need to revisit */
> > @@ -502,6 +503,7 @@ vacuumRedirectAndPlaceholder(Relation index, Buffer buffer)
> >       OffsetNumber itemnos[MaxIndexTuplesPerPage];
> >       spgxlogVacuumRedirect xlrec;
> >
> > +     xlrec.onCatalogTable = get_rel_logical_catalog(index->rd_index->indrelid);
> >       xlrec.nToPlaceholder = 0;
> >       xlrec.newestRedirectXid = InvalidTransactionId;
>
> We should document that it is safe to do catalog acceses here, because
> spgist is never used to back catalogs. Otherwise there would be an a
> endless recursion danger here.

Comments added.

>
> Did you check how hard it we to just pass down the heap relation?

It does look hard. Check my comments in an earlier reply, that I have
pasted below :

> This one seems harder, but I'm not actually sure why we make it so
> hard. It seems like we just ought to add the table to IndexVacuumInfo.

This means we have to add heapRel assignment wherever we initialize
IndexVacuumInfo structure, namely in lazy_vacuum_index(),
lazy_cleanup_index(), validate_index(), analyze_rel(), and make sure
these functions have a heap rel handle. Do you think we should do this
as part of this patch ?

>
>
> >  /*
> > + * Get the wal_level from the control file.
> > + */
> > +WalLevel
> > +GetActiveWalLevel(void)
> > +{
> > +     return ControlFile->wal_level;
> > +}
>
> What does "Active" mean here? I assume it's supposed to indicate that it
> could be different than what's configured in postgresql.conf, for a
> replica? If so, that should be mentioned.

Done. Here are the new comments :
 * Get the wal_level from the control file. For a standby, this value should be
 * considered as its active wal_level, because it may be different from what
 * was originally configured on standby.

>
>
> > +/*
> >   * Initialization of shared memory for XLOG
> >   */
> >  Size
> > @@ -9843,6 +9852,19 @@ xlog_redo(XLogReaderState *record)
> >               /* Update our copy of the parameters in pg_control */
> >               memcpy(&xlrec, XLogRecGetData(record), sizeof(xl_parameter_change));
> >
> > +             /*
> > +              * Drop logical slots if we are in hot standby and master does not have
> > +              * logical data.
>
> nitpick: s/master/the primary/ (mostly adding the "the", but I
> personally also prefer primary over master)
>
> s/logical data/a WAL level sufficient for logical decoding/
>
>
> > Don't bother to search for the slots if standby is
> > +              * running with wal_level lower than logical, because in that case,
> > +              * we would have either disallowed creation of logical slots or dropped
> > +              * existing ones.
>
> s/Don't bother/No need/
> s/slots/potentially conflicting logically slots/

Done.

>
> > +             if (InRecovery && InHotStandby &&
> > +                     xlrec.wal_level < WAL_LEVEL_LOGICAL &&
> > +                     wal_level >= WAL_LEVEL_LOGICAL)
> > +                     ResolveRecoveryConflictWithLogicalSlots(InvalidOid, InvalidTransactionId,
> > +                             gettext_noop("Logical decoding on standby requires wal_level >= logical on master."));
>
>
>
> > diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c
> > index 151c3ef..c1bd028 100644
> > --- a/src/backend/replication/logical/decode.c
> > +++ b/src/backend/replication/logical/decode.c
> > @@ -190,11 +190,23 @@ DecodeXLogOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
> >                        * can restart from there.
> >                        */
> >                       break;
> > +             case XLOG_PARAMETER_CHANGE:
> > +             {
> > +                     xl_parameter_change *xlrec =
> > +                             (xl_parameter_change *) XLogRecGetData(buf->record);
> > +                     /* Cannot proceed if master itself does not have logical data */
>
> This needs an explanation as to how this is reachable...

Done. Here are the comments :
 * If wal_level on primary is reduced to less than logical, then we
 * want to prevent existing logical slots from being used.
 * Existing logical slot on standby gets dropped when this WAL
 * record is replayed; and further, slot creation fails when the
 * wal level is not sufficient; but all these operations are not
 * synchronized, so a logical slot may creep in while the wal_level
 * is being reduced.  Hence this extra check.

>
>
> > +                     if (xlrec->wal_level < WAL_LEVEL_LOGICAL)
> > +                             ereport(ERROR,
> > +                                             (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
> > +                                              errmsg("logical decoding on standby requires "
> > +                                                             "wal_level >= logical on master")));
> > +                     break;
>
> Hm, this strikes me as a not quite good enough error message (same in
> other copies of the message). Perhaps something roughly like "could not
> continue with logical decoding, the primary's wal level is now too low
> (%u)"?

Haven't changed this. There is another reply from Robert. I think what
you want to emphasize is that we can't *continue*. I am not sure why
user can't infer that the "logical decoding could not continue" when
we say "logical decoding requires wal_level >= ...."

>
>
> >       if (RecoveryInProgress())
> > -             ereport(ERROR,
> > -                             (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
> > -                              errmsg("logical decoding cannot be used while in recovery")));
> > +     {
> > +             /*
> > +              * This check may have race conditions, but whenever
> > +              * XLOG_PARAMETER_CHANGE indicates that wal_level has changed, we
> > +              * verify that there are no existing logical replication slots. And to
> > +              * avoid races around creating a new slot,
> > +              * CheckLogicalDecodingRequirements() is called once before creating
> > +              * the slot, and once when logical decoding is initially starting up.
> > +              */
> > +             if (GetActiveWalLevel() < WAL_LEVEL_LOGICAL)
> > +                     ereport(ERROR,
> > +                                     (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
> > +                                      errmsg("logical decoding on standby requires "
> > +                                                     "wal_level >= logical on master")));
> > +     }
> >  }
> >
> >  /*
> > @@ -241,6 +240,8 @@ CreateInitDecodingContext(char *plugin,
> >       LogicalDecodingContext *ctx;
> >       MemoryContext old_context;
> >
> > +     CheckLogicalDecodingRequirements();
> > +
>
> This should reference the above explanation.

Done.

>
>
>
>
> >  /*
> > + * Permanently drop a conflicting replication slot. If it's already active by
> > + * another backend, send it a recovery conflict signal, and then try again.
> > + */
> > +static void
> > +ReplicationSlotDropConflicting(ReplicationSlot *slot)
>
>
> > +void
> > +ResolveRecoveryConflictWithLogicalSlots(Oid dboid, TransactionId xid,
> > +                                                                             char *conflict_reason)
> > +{
> > +                     /*
> > +                      * Build the conflict_str which will look like :
> > +                      * "Slot conflicted with xid horizon which was being increased
> > +                      * to 9012 (slot xmin: 1234, slot catalog_xmin: 5678)."
> > +                      */
> > +                     initStringInfo(&conflict_xmins);
> > +                     if (TransactionIdIsValid(slot_xmin) &&
> > +                             TransactionIdPrecedesOrEquals(slot_xmin, xid))
> > +                     {
> > +                             appendStringInfo(&conflict_xmins, "slot xmin: %d", slot_xmin);
> > +                     }
> > +                     if (TransactionIdIsValid(slot_catalog_xmin) &&
> > +                             TransactionIdPrecedesOrEquals(slot_catalog_xmin, xid))
> > +                             appendStringInfo(&conflict_xmins, "%sslot catalog_xmin: %d",
> > +                                                              conflict_xmins.len > 0 ? ", " : "",
> > +                                                              slot_catalog_xmin);
> > +
> > +                     if (conflict_xmins.len > 0)
> > +                     {
> > +                             initStringInfo(&conflict_str);
> > +                             appendStringInfo(&conflict_str, "%s %d (%s).",
> > +                                                              conflict_sentence, xid, conflict_xmins.data);
> > +                             found_conflict = true;
> > +                             conflict_reason = conflict_str.data;
> > +                     }
> > +             }
>
>
> I think this is going to be a nightmare for translators, no?

For translators, I think the .po files will have the required text,
because I have used gettext_noop() for both conflict_sentence and the
passed in conflict_reason parameter. And the "dropped conflicting
slot." is passed to ereport() as usual.  The rest portion of errdetail
is not language specific. E.g. "slot" remains "slot".

> I'm not clear as to why any of this is needed?

The conflict can happen for either xmin or catalog_xmin or both, right
? The purpose of the above is to show only conflicting xmin out of the
two.

>
>
>
> > +                     /* ReplicationSlotDropPtr() would acquire the lock below */
> > +                     LWLockRelease(ReplicationSlotControlLock);
>
> "would acquire"? I think it *does* acquire, right?

Yes, Changed to "will".

>
>
>
> > @@ -2879,6 +2882,25 @@ RecoveryConflictInterrupt(ProcSignalReason reason)
> >                       case PROCSIG_RECOVERY_CONFLICT_LOCK:
> >                       case PROCSIG_RECOVERY_CONFLICT_TABLESPACE:
> >                       case PROCSIG_RECOVERY_CONFLICT_SNAPSHOT:
> > +                     case PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT:
> > +                             /*
> > +                              * For conflicts that require a logical slot to be dropped, the
> > +                              * requirement is for the signal receiver to release the slot,
> > +                              * so that it could be dropped by the signal sender. So for
> > +                              * normal backends, the transaction should be aborted, just
> > +                              * like for other recovery conflicts. But if it's walsender on
> > +                              * standby, then it has to be killed so as to release an
> > +                              * acquired logical slot.
> > +                              */
> > +                             if (am_cascading_walsender &&
> > +                                     reason == PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT &&
> > +                                     MyReplicationSlot && SlotIsLogical(MyReplicationSlot))
> > +                             {
> > +                                     RecoveryConflictPending = true;
> > +                                     QueryCancelPending = true;
> > +                                     InterruptPending = true;
> > +                                     break;
> > +                             }
>
> Huh, I'm not following as to why that's needed for walsenders?

For normal backends, we ignore this signal if we aren't in a
transaction (block). But for walsender, there is no transaction, but
we cannot ignore the signal. This is because walsender can keep a
logical slot acquired when it was spawned by "pg_recvlogical --start".
So we can't ignore the signal. So the only way that we can make it
release the acquired slot is to kill it.

>
>
> > @@ -1499,6 +1499,7 @@ pg_stat_get_db_conflict_all(PG_FUNCTION_ARGS)
> >                                                 dbentry->n_conflict_tablespace +
> >                                                 dbentry->n_conflict_lock +
> >                                                 dbentry->n_conflict_snapshot +
> > +                                               dbentry->n_conflict_logicalslot +
> >                                                 dbentry->n_conflict_bufferpin +
> >                                                 dbentry->n_conflict_startup_deadlock);
>
> I think this probably needs adjustments in a few more places,
> e.g. monitoring.sgml...

Oops, yeah, to search for similar additions, I had looked for
"conflict_snapshot" using cscope. I should have done the same using
"git grep".
Done now.

-- 
Thanks,
-Amit Khandekar
EnterpriseDB Corporation
The Postgres Database Company