On Thu, Oct 5, 2023 at 4:24 PM Amit Kapila <amit.kapila16@gmail.com> wrote:
>
> > > Today, I discussed this problem with Andres at PGConf NYC and he
> > > suggested as following. To verify, if there is any pending unexpected
> > > WAL after shutdown, we can have an API like
> > > pg_logical_replication_slot_advance() which will simply process
> > > records without actually sending anything downstream.
+1 for this approach. It looks neat.
I think we also need to add TAP tests to generate decodable WAL
records (RUNNING_XACT, CHECKPOINT_ONLINE, XLOG_FPI_FOR_HINT,
XLOG_SWITCH, XLOG_PARAMETER_CHANGE, XLOG_HEAP2_PRUNE) during
pg_upgrade as described here
https://www.postgresql.org/message-id/TYAPR01MB58660273EACEFC5BF256B133F50DA%40TYAPR01MB5866.jpnprd01.prod.outlook.com.
Basically, these were the exceptional WAL records that may be
generated by pg_upgrade, so having tests for them is good.
> > So I assume in each lower-level decode function (e.g. heap_decode() )
> > we will add the check that if we are checking the WAL for an upgrade
> > then from that level we will return true or false based on whether the
> > WAL is decodable or not. Is my understanding correct?
> >
>
> Yes, this is one way to achive but I think this will require changing return value of many APIs. Can we somehow just
getthis via LogicalDecodingContext or some other way at the caller by allowing to set some variable at required places?
+1 for adding the required flags to the decoding context similar to
fast_forward.
Another way without adding any new variables is to pass the WAL record
to LogicalDecodingProcessRecord, and upon return check the reorder
buffer if there's any decoded change generated for the xid associated
with the WAL record. If any decoded change related to the WAL record
xid is found, then that's the end for the new function. Here's what I
think [1], haven't tested it.
[1]
change_found = false;
end_of_wal = false;
ctx = CreateDecodingContext();
XLogBeginRead(ctx->reader, MyReplicationSlot->data.restart_lsn);
while(!end_of_wal || !change_found)
{
XLogRecord *record;
TransactionId xid;
ReorderBufferTXN *txn;
record = XLogReadRecord(ctx->reader, &errm);
if (record)
LogicalDecodingProcessRecord(ctx, ctx->reader);
xid = XLogRecGetXid(record);
txn = ReorderBufferTXNByXid(ctx->reorder, xid, false, NULL,
InvalidXLogRecPtr,
false);
if (txn != NULL)
{
change_found = true;
break;
}
CHECK_FOR_INTERRUPTS();
}
--
Bharath Rupireddy
PostgreSQL Contributors Team
RDS Open Source Databases
Amazon Web Services: https://aws.amazon.com
