decode.h File Reference

#include "access/xlogreader.h"
#include "access/xlogrecord.h"
#include "replication/logical.h"
#include "replication/reorderbuffer.h"

Include dependency graph for decode.h:

This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Data Structures
struct	XLogRecordBuffer

Typedefs
typedef struct XLogRecordBuffer	XLogRecordBuffer

Functions
void	xlog_decode (LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
void	xlog2_decode (LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
void	heap_decode (LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
void	heap2_decode (LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
void	xact_decode (LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
void	standby_decode (LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
void	logicalmsg_decode (LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
void	LogicalDecodingProcessRecord (LogicalDecodingContext *ctx, XLogReaderState *record)
bool	change_useless_for_repack (XLogRecordBuffer *buf)

Typedef Documentation

XLogRecordBuffer

typedef struct XLogRecordBuffer XLogRecordBuffer

Function Documentation

change_useless_for_repack()

bool change_useless_for_repack ( XLogRecordBuffer *  buf )

extern

Definition at line 500 of file repack_worker.c.

{
    XLogReaderState *r = buf->record;
    RelFileLocator locator;
 
    /* TOAST locator should not be set unless the main is. */
    Assert(!OidIsValid(repacked_rel_toast_locator.relNumber) ||
           OidIsValid(repacked_rel_locator.relNumber));
 
    /*
     * Backends not involved in REPACK (CONCURRENTLY) should not do the
     * filtering.
     */
    if (!OidIsValid(repacked_rel_locator.relNumber))
        return false;
 
    /*
     * If the record does not contain the block 0, it's probably not INSERT /
     * UPDATE / DELETE. In any case, we do not have enough information to
     * filter the change out.
     */
    if (!XLogRecGetBlockTagExtended(r, 0, &locator, NULL, NULL, NULL))
        return false;
 
    /*
     * Decode the change if it belongs to the table we are repacking, or if it
     * belongs to its TOAST relation.
     */
    if (RelFileLocatorEquals(locator, repacked_rel_locator))
        return false;
    if (OidIsValid(repacked_rel_toast_locator.relNumber) &&
        RelFileLocatorEquals(locator, repacked_rel_toast_locator))
        return false;
 
    /* Filter out changes of other tables. */
    return true;
}

References Assert, buf, fb(), OidIsValid, XLogReaderState::record, RelFileLocatorEquals, RelFileLocator::relNumber, repacked_rel_locator, repacked_rel_toast_locator, and XLogRecGetBlockTagExtended().

Referenced by heap2_decode(), and heap_decode().

heap2_decode()

void heap2_decode ( LogicalDecodingContext *  ctx, XLogRecordBuffer *  buf  )

extern

Definition at line 422 of file decode.c.

{
    uint8       info = XLogRecGetInfo(buf->record) & XLOG_HEAP_OPMASK;
    TransactionId xid = XLogRecGetXid(buf->record);
    SnapBuild  *builder = ctx->snapshot_builder;
 
    ReorderBufferProcessXid(ctx->reorder, xid, buf->origptr);
 
    /*
     * If we don't have snapshot or we are just fast-forwarding, there is no
     * point in decoding data changes. However, it's crucial to build the base
     * snapshot during fast-forward mode (as is done in
     * SnapBuildProcessChange()) because we require the snapshot's xmin when
     * determining the candidate catalog_xmin for the replication slot. See
     * SnapBuildProcessRunningXacts().
     */
    if (SnapBuildCurrentState(builder) < SNAPBUILD_FULL_SNAPSHOT)
        return;
 
    switch (info)
    {
        case XLOG_HEAP2_MULTI_INSERT:
            if (SnapBuildProcessChange(builder, xid, buf->origptr) &&
                !ctx->fast_forward &&
                !change_useless_for_repack(buf))
                DecodeMultiInsert(ctx, buf);
            break;
        case XLOG_HEAP2_NEW_CID:
            if (!ctx->fast_forward)
            {
                xl_heap_new_cid *xlrec;
 
                xlrec = (xl_heap_new_cid *) XLogRecGetData(buf->record);
                SnapBuildProcessNewCid(builder, xid, buf->origptr, xlrec);
            }
            break;
        case XLOG_HEAP2_REWRITE:
 
            /*
             * Although these records only exist to serve the needs of logical
             * decoding, all the work happens as part of crash or archive
             * recovery, so we don't need to do anything here.
             */
            break;
 
            /*
             * Everything else here is just low level physical stuff we're not
             * interested in.
             */
        case XLOG_HEAP2_PRUNE_ON_ACCESS:
        case XLOG_HEAP2_PRUNE_VACUUM_SCAN:
        case XLOG_HEAP2_PRUNE_VACUUM_CLEANUP:
        case XLOG_HEAP2_LOCK_UPDATED:
            break;
        default:
            elog(ERROR, "unexpected RM_HEAP2_ID record type: %u", info);
    }
}

References buf, change_useless_for_repack(), DecodeMultiInsert(), elog, ERROR, LogicalDecodingContext::fast_forward, fb(), LogicalDecodingContext::reorder, ReorderBufferProcessXid(), SNAPBUILD_FULL_SNAPSHOT, SnapBuildCurrentState(), SnapBuildProcessChange(), SnapBuildProcessNewCid(), LogicalDecodingContext::snapshot_builder, XLOG_HEAP2_LOCK_UPDATED, XLOG_HEAP2_MULTI_INSERT, XLOG_HEAP2_NEW_CID, XLOG_HEAP2_PRUNE_ON_ACCESS, XLOG_HEAP2_PRUNE_VACUUM_CLEANUP, XLOG_HEAP2_PRUNE_VACUUM_SCAN, XLOG_HEAP2_REWRITE, XLOG_HEAP_OPMASK, XLogRecGetData, XLogRecGetInfo, and XLogRecGetXid.

heap_decode()

void heap_decode ( LogicalDecodingContext *  ctx, XLogRecordBuffer *  buf  )

extern

Definition at line 485 of file decode.c.

{
    uint8       info = XLogRecGetInfo(buf->record) & XLOG_HEAP_OPMASK;
    TransactionId xid = XLogRecGetXid(buf->record);
    SnapBuild  *builder = ctx->snapshot_builder;
 
    ReorderBufferProcessXid(ctx->reorder, xid, buf->origptr);
 
    /*
     * If we don't have snapshot or we are just fast-forwarding, there is no
     * point in decoding data changes. However, it's crucial to build the base
     * snapshot during fast-forward mode (as is done in
     * SnapBuildProcessChange()) because we require the snapshot's xmin when
     * determining the candidate catalog_xmin for the replication slot. See
     * SnapBuildProcessRunningXacts().
     */
    if (SnapBuildCurrentState(builder) < SNAPBUILD_FULL_SNAPSHOT)
        return;
 
    switch (info)
    {
        case XLOG_HEAP_INSERT:
            if (SnapBuildProcessChange(builder, xid, buf->origptr) &&
                !ctx->fast_forward &&
                !change_useless_for_repack(buf))
                DecodeInsert(ctx, buf);
            break;
 
            /*
             * Treat HOT update as normal updates. There is no useful
             * information in the fact that we could make it a HOT update
             * locally and the WAL layout is compatible.
             */
        case XLOG_HEAP_HOT_UPDATE:
        case XLOG_HEAP_UPDATE:
            if (SnapBuildProcessChange(builder, xid, buf->origptr) &&
                !ctx->fast_forward &&
                !change_useless_for_repack(buf))
                DecodeUpdate(ctx, buf);
            break;
 
        case XLOG_HEAP_DELETE:
            if (SnapBuildProcessChange(builder, xid, buf->origptr) &&
                !ctx->fast_forward &&
                !change_useless_for_repack(buf))
                DecodeDelete(ctx, buf);
            break;
 
        case XLOG_HEAP_TRUNCATE:
            if (SnapBuildProcessChange(builder, xid, buf->origptr) &&
                !ctx->fast_forward &&
                !change_useless_for_repack(buf))
                DecodeTruncate(ctx, buf);
            break;
 
        case XLOG_HEAP_INPLACE:
 
            /*
             * Inplace updates are only ever performed on catalog tuples and
             * can, per definition, not change tuple visibility.  Since we
             * also don't decode catalog tuples, we're not interested in the
             * record's contents.
             */
            break;
 
        case XLOG_HEAP_CONFIRM:
            if (SnapBuildProcessChange(builder, xid, buf->origptr) &&
                !ctx->fast_forward &&
                !change_useless_for_repack(buf))
                DecodeSpecConfirm(ctx, buf);
            break;
 
        case XLOG_HEAP_LOCK:
            /* we don't care about row level locks for now */
            break;
 
        default:
            elog(ERROR, "unexpected RM_HEAP_ID record type: %u", info);
            break;
    }
}

References buf, change_useless_for_repack(), DecodeDelete(), DecodeInsert(), DecodeSpecConfirm(), DecodeTruncate(), DecodeUpdate(), elog, ERROR, LogicalDecodingContext::fast_forward, LogicalDecodingContext::reorder, ReorderBufferProcessXid(), SNAPBUILD_FULL_SNAPSHOT, SnapBuildCurrentState(), SnapBuildProcessChange(), LogicalDecodingContext::snapshot_builder, XLOG_HEAP_CONFIRM, XLOG_HEAP_DELETE, XLOG_HEAP_HOT_UPDATE, XLOG_HEAP_INPLACE, XLOG_HEAP_INSERT, XLOG_HEAP_LOCK, XLOG_HEAP_OPMASK, XLOG_HEAP_TRUNCATE, XLOG_HEAP_UPDATE, XLogRecGetInfo, and XLogRecGetXid.

LogicalDecodingProcessRecord()

void LogicalDecodingProcessRecord ( LogicalDecodingContext *  ctx, XLogReaderState *  record  )

extern

Definition at line 89 of file decode.c.

{
    XLogRecordBuffer buf;
    TransactionId txid;
    RmgrData    rmgr;
 
    buf.origptr = ctx->reader->ReadRecPtr;
    buf.endptr = ctx->reader->EndRecPtr;
    buf.record = record;
 
    txid = XLogRecGetTopXid(record);
 
    /*
     * If the top-level xid is valid, we need to assign the subxact to the
     * top-level xact. We need to do this for all records, hence we do it
     * before the switch.
     */
    if (TransactionIdIsValid(txid))
    {
        ReorderBufferAssignChild(ctx->reorder,
                                 txid,
                                 XLogRecGetXid(record),
                                 buf.origptr);
    }
 
    rmgr = GetRmgr(XLogRecGetRmid(record));
 
    if (rmgr.rm_decode != NULL)
        rmgr.rm_decode(ctx, &buf);
    else
    {
        /* just deal with xid, and done */
        ReorderBufferProcessXid(ctx->reorder, XLogRecGetXid(record),
                                buf.origptr);
    }
}

References buf, XLogReaderState::EndRecPtr, fb(), GetRmgr(), LogicalDecodingContext::reader, XLogReaderState::ReadRecPtr, LogicalDecodingContext::reorder, ReorderBufferAssignChild(), ReorderBufferProcessXid(), TransactionIdIsValid, XLogRecGetRmid, XLogRecGetTopXid, and XLogRecGetXid.

Referenced by decode_concurrent_changes(), DecodingContextFindStartpoint(), LogicalReplicationSlotCheckPendingWal(), LogicalSlotAdvanceAndCheckSnapState(), pg_logical_slot_get_changes_guts(), and XLogSendLogical().

logicalmsg_decode()

void logicalmsg_decode ( LogicalDecodingContext *  ctx, XLogRecordBuffer *  buf  )

extern

Definition at line 607 of file decode.c.

{
    SnapBuild  *builder = ctx->snapshot_builder;
    XLogReaderState *r = buf->record;
    TransactionId xid = XLogRecGetXid(r);
    uint8       info = XLogRecGetInfo(r) & ~XLR_INFO_MASK;
    ReplOriginId origin_id = XLogRecGetOrigin(r);
    Snapshot    snapshot = NULL;
    xl_logical_message *message;
 
    if (info != XLOG_LOGICAL_MESSAGE)
        elog(ERROR, "unexpected RM_LOGICALMSG_ID record type: %u", info);
 
    ReorderBufferProcessXid(ctx->reorder, XLogRecGetXid(r), buf->origptr);
 
    /* If we don't have snapshot, there is no point in decoding messages */
    if (SnapBuildCurrentState(builder) < SNAPBUILD_FULL_SNAPSHOT)
        return;
 
    message = (xl_logical_message *) XLogRecGetData(r);
 
    if (message->dbId != ctx->slot->data.database ||
        FilterByOrigin(ctx, origin_id))
        return;
 
    if (message->transactional &&
        !SnapBuildProcessChange(builder, xid, buf->origptr))
        return;
    else if (!message->transactional &&
             (SnapBuildCurrentState(builder) != SNAPBUILD_CONSISTENT ||
              SnapBuildXactNeedsSkip(builder, buf->origptr)))
        return;
 
    /*
     * We also skip decoding in fast_forward mode. This check must be last
     * because we don't want to set the processing_required flag unless we
     * have a decodable message.
     */
    if (ctx->fast_forward)
    {
        /*
         * We need to set processing_required flag to notify the message's
         * existence to the caller. Usually, the flag is set when either the
         * COMMIT or ABORT records are decoded, but this must be turned on
         * here because the non-transactional logical message is decoded
         * without waiting for these records.
         */
        if (!message->transactional)
            ctx->processing_required = true;
 
        return;
    }
 
    /*
     * If this is a non-transactional change, get the snapshot we're expected
     * to use. We only get here when the snapshot is consistent, and the
     * change is not meant to be skipped.
     *
     * For transactional changes we don't need a snapshot, we'll use the
     * regular snapshot maintained by ReorderBuffer. We just leave it NULL.
     */
    if (!message->transactional)
        snapshot = SnapBuildGetOrBuildSnapshot(builder);
 
    ReorderBufferQueueMessage(ctx->reorder, xid, snapshot, buf->endptr,
                              message->transactional,
                              message->message, /* first part of message is
                                                 * prefix */
                              message->message_size,
                              message->message + message->prefix_size);
}

References buf, ReplicationSlot::data, ReplicationSlotPersistentData::database, xl_logical_message::dbId, elog, ERROR, LogicalDecodingContext::fast_forward, fb(), FilterByOrigin(), xl_logical_message::message, xl_logical_message::message_size, xl_logical_message::prefix_size, LogicalDecodingContext::processing_required, XLogReaderState::record, LogicalDecodingContext::reorder, ReorderBufferProcessXid(), ReorderBufferQueueMessage(), LogicalDecodingContext::slot, SNAPBUILD_CONSISTENT, SNAPBUILD_FULL_SNAPSHOT, SnapBuildCurrentState(), SnapBuildGetOrBuildSnapshot(), SnapBuildProcessChange(), SnapBuildXactNeedsSkip(), LogicalDecodingContext::snapshot_builder, xl_logical_message::transactional, XLOG_LOGICAL_MESSAGE, XLogRecGetData, XLogRecGetInfo, XLogRecGetOrigin, and XLogRecGetXid.

standby_decode()

void standby_decode ( LogicalDecodingContext *  ctx, XLogRecordBuffer *  buf  )

extern

Definition at line 371 of file decode.c.

{
    SnapBuild  *builder = ctx->snapshot_builder;
    XLogReaderState *r = buf->record;
    uint8       info = XLogRecGetInfo(r) & ~XLR_INFO_MASK;
 
    ReorderBufferProcessXid(ctx->reorder, XLogRecGetXid(r), buf->origptr);
 
    switch (info)
    {
        case XLOG_RUNNING_XACTS:
            {
                xl_running_xacts *running = (xl_running_xacts *) XLogRecGetData(r);
 
                /*
                 * Update this decoder's idea of transactions currently
                 * running.  In doing so we will determine whether we have
                 * reached consistent status.
                 */
                SnapBuildProcessRunningXacts(builder, buf->origptr, running);
 
                /*
                 * Abort all transactions that we keep track of, that are
                 * older than the record's oldestRunningXid. This is the most
                 * convenient spot for doing so since, in contrast to shutdown
                 * or end-of-recovery checkpoints, we have information about
                 * all running transactions which includes prepared ones,
                 * while shutdown checkpoints just know that no non-prepared
                 * transactions are in progress.
                 */
                ReorderBufferAbortOld(ctx->reorder, running->oldestRunningXid);
            }
            break;
        case XLOG_STANDBY_LOCK:
            break;
        case XLOG_INVALIDATIONS:
 
            /*
             * We are processing the invalidations at the command level via
             * XLOG_XACT_INVALIDATIONS.  So we don't need to do anything here.
             */
            break;
        default:
            elog(ERROR, "unexpected RM_STANDBY_ID record type: %u", info);
    }
}

References buf, elog, ERROR, fb(), xl_running_xacts::oldestRunningXid, XLogReaderState::record, LogicalDecodingContext::reorder, ReorderBufferAbortOld(), ReorderBufferProcessXid(), SnapBuildProcessRunningXacts(), LogicalDecodingContext::snapshot_builder, XLOG_INVALIDATIONS, XLOG_RUNNING_XACTS, XLOG_STANDBY_LOCK, XLogRecGetData, XLogRecGetInfo, and XLogRecGetXid.

xact_decode()

void xact_decode ( LogicalDecodingContext *  ctx, XLogRecordBuffer *  buf  )

extern

Definition at line 213 of file decode.c.

{
    SnapBuild  *builder = ctx->snapshot_builder;
    ReorderBuffer *reorder = ctx->reorder;
    XLogReaderState *r = buf->record;
    uint8       info = XLogRecGetInfo(r) & XLOG_XACT_OPMASK;
 
    /*
     * If the snapshot isn't yet fully built, we cannot decode anything, so
     * bail out.
     */
    if (SnapBuildCurrentState(builder) < SNAPBUILD_FULL_SNAPSHOT)
        return;
 
    switch (info)
    {
        case XLOG_XACT_COMMIT:
        case XLOG_XACT_COMMIT_PREPARED:
            {
                xl_xact_commit *xlrec;
                xl_xact_parsed_commit parsed;
                TransactionId xid;
                bool        two_phase = false;
 
                xlrec = (xl_xact_commit *) XLogRecGetData(r);
                ParseCommitRecord(XLogRecGetInfo(buf->record), xlrec, &parsed);
 
                if (!TransactionIdIsValid(parsed.twophase_xid))
                    xid = XLogRecGetXid(r);
                else
                    xid = parsed.twophase_xid;
 
                /*
                 * We would like to process the transaction in a two-phase
                 * manner iff output plugin supports two-phase commits and
                 * doesn't filter the transaction at prepare time.
                 */
                if (info == XLOG_XACT_COMMIT_PREPARED)
                    two_phase = !(FilterPrepare(ctx, xid,
                                                parsed.twophase_gid));
 
                DecodeCommit(ctx, buf, &parsed, xid, two_phase);
                break;
            }
        case XLOG_XACT_ABORT:
        case XLOG_XACT_ABORT_PREPARED:
            {
                xl_xact_abort *xlrec;
                xl_xact_parsed_abort parsed;
                TransactionId xid;
                bool        two_phase = false;
 
                xlrec = (xl_xact_abort *) XLogRecGetData(r);
                ParseAbortRecord(XLogRecGetInfo(buf->record), xlrec, &parsed);
 
                if (!TransactionIdIsValid(parsed.twophase_xid))
                    xid = XLogRecGetXid(r);
                else
                    xid = parsed.twophase_xid;
 
                /*
                 * We would like to process the transaction in a two-phase
                 * manner iff output plugin supports two-phase commits and
                 * doesn't filter the transaction at prepare time.
                 */
                if (info == XLOG_XACT_ABORT_PREPARED)
                    two_phase = !(FilterPrepare(ctx, xid,
                                                parsed.twophase_gid));
 
                DecodeAbort(ctx, buf, &parsed, xid, two_phase);
                break;
            }
        case XLOG_XACT_ASSIGNMENT:
 
            /*
             * We assign subxact to the toplevel xact while processing each
             * record if required.  So, we don't need to do anything here. See
             * LogicalDecodingProcessRecord.
             */
            break;
        case XLOG_XACT_INVALIDATIONS:
            {
                TransactionId xid;
                xl_xact_invals *invals;
 
                xid = XLogRecGetXid(r);
                invals = (xl_xact_invals *) XLogRecGetData(r);
 
                /*
                 * Execute the invalidations for xid-less transactions,
                 * otherwise, accumulate them so that they can be processed at
                 * the commit time.
                 */
                if (TransactionIdIsValid(xid))
                {
                    if (!ctx->fast_forward)
                        ReorderBufferAddInvalidations(reorder, xid,
                                                      buf->origptr,
                                                      invals->nmsgs,
                                                      invals->msgs);
                    ReorderBufferXidSetCatalogChanges(ctx->reorder, xid,
                                                      buf->origptr);
                }
                else if (!ctx->fast_forward)
                    ReorderBufferImmediateInvalidation(ctx->reorder,
                                                       invals->nmsgs,
                                                       invals->msgs);
 
                break;
            }
        case XLOG_XACT_PREPARE:
            {
                xl_xact_parsed_prepare parsed;
                xl_xact_prepare *xlrec;
 
                /* ok, parse it */
                xlrec = (xl_xact_prepare *) XLogRecGetData(r);
                ParsePrepareRecord(XLogRecGetInfo(buf->record),
                                   xlrec, &parsed);
 
                /*
                 * We would like to process the transaction in a two-phase
                 * manner iff output plugin supports two-phase commits and
                 * doesn't filter the transaction at prepare time.
                 */
                if (FilterPrepare(ctx, parsed.twophase_xid,
                                  parsed.twophase_gid))
                {
                    ReorderBufferProcessXid(reorder, parsed.twophase_xid,
                                            buf->origptr);
                    break;
                }
 
                /*
                 * Note that if the prepared transaction has locked [user]
                 * catalog tables exclusively then decoding prepare can block
                 * till the main transaction is committed because it needs to
                 * lock the catalog tables.
                 *
                 * XXX Now, this can even lead to a deadlock if the prepare
                 * transaction is waiting to get it logically replicated for
                 * distributed 2PC. This can be avoided by disallowing
                 * preparing transactions that have locked [user] catalog
                 * tables exclusively but as of now, we ask users not to do
                 * such an operation.
                 */
                DecodePrepare(ctx, buf, &parsed);
                break;
            }
        default:
            elog(ERROR, "unexpected RM_XACT_ID record type: %u", info);
    }
}

References buf, DecodeAbort(), DecodeCommit(), DecodePrepare(), elog, ERROR, LogicalDecodingContext::fast_forward, fb(), FilterPrepare(), ParseAbortRecord(), ParseCommitRecord(), ParsePrepareRecord(), XLogReaderState::record, LogicalDecodingContext::reorder, ReorderBufferAddInvalidations(), ReorderBufferImmediateInvalidation(), ReorderBufferProcessXid(), ReorderBufferXidSetCatalogChanges(), SNAPBUILD_FULL_SNAPSHOT, SnapBuildCurrentState(), LogicalDecodingContext::snapshot_builder, TransactionIdIsValid, two_phase, XLOG_XACT_ABORT, XLOG_XACT_ABORT_PREPARED, XLOG_XACT_ASSIGNMENT, XLOG_XACT_COMMIT, XLOG_XACT_COMMIT_PREPARED, XLOG_XACT_INVALIDATIONS, XLOG_XACT_OPMASK, XLOG_XACT_PREPARE, XLogRecGetData, XLogRecGetInfo, and XLogRecGetXid.

xlog2_decode()

void xlog2_decode ( LogicalDecodingContext *  ctx, XLogRecordBuffer *  buf  )

extern

Definition at line 194 of file decode.c.

{
    uint8       info = XLogRecGetInfo(buf->record) & ~XLR_INFO_MASK;
 
    ReorderBufferProcessXid(ctx->reorder, XLogRecGetXid(buf->record), buf->origptr);
 
    switch (info)
    {
        case XLOG2_CHECKSUMS:
            break;
        default:
            elog(ERROR, "unexpected RM_XLOG2_ID record type: %u", info);
    }
}

References buf, elog, ERROR, fb(), LogicalDecodingContext::reorder, ReorderBufferProcessXid(), XLOG2_CHECKSUMS, XLogRecGetInfo, and XLogRecGetXid.

xlog_decode()

void xlog_decode ( LogicalDecodingContext *  ctx, XLogRecordBuffer *  buf  )

extern

Definition at line 130 of file decode.c.

{
    SnapBuild  *builder = ctx->snapshot_builder;
    uint8       info = XLogRecGetInfo(buf->record) & ~XLR_INFO_MASK;
 
    ReorderBufferProcessXid(ctx->reorder, XLogRecGetXid(buf->record),
                            buf->origptr);
 
    switch (info)
    {
            /* this is also used in END_OF_RECOVERY checkpoints */
        case XLOG_CHECKPOINT_SHUTDOWN:
        case XLOG_END_OF_RECOVERY:
            SnapBuildSerializationPoint(builder, buf->origptr);
 
            break;
        case XLOG_CHECKPOINT_ONLINE:
 
            /*
             * a RUNNING_XACTS record will have been logged near to this, we
             * can restart from there.
             */
            break;
        case XLOG_LOGICAL_DECODING_STATUS_CHANGE:
            {
                bool        logical_decoding;
 
                memcpy(&logical_decoding, XLogRecGetData(buf->record), sizeof(bool));
 
                /*
                 * Error out as we should not decode this WAL record.
                 *
                 * Logical decoding is disabled, and existing logical slots on
                 * the standby are invalidated when this WAL record is
                 * replayed. No logical decoder can process this WAL record
                 * until replay completes, and by then the slots are already
                 * invalidated. Furthermore, no new logical slots can be
                 * created while logical decoding is disabled. This cannot
                 * occur even on primary either, since it will not restart
                 * with wal_level < replica if any logical slots exist.
                 */
                elog(ERROR, "unexpected logical decoding status change %d",
                     logical_decoding);
 
                break;
            }
        case XLOG_NOOP:
        case XLOG_NEXTOID:
        case XLOG_SWITCH:
        case XLOG_BACKUP_END:
        case XLOG_PARAMETER_CHANGE:
        case XLOG_RESTORE_POINT:
        case XLOG_FPW_CHANGE:
        case XLOG_FPI_FOR_HINT:
        case XLOG_FPI:
        case XLOG_OVERWRITE_CONTRECORD:
        case XLOG_CHECKPOINT_REDO:
            break;
        default:
            elog(ERROR, "unexpected RM_XLOG_ID record type: %u", info);
    }
}

References buf, elog, ERROR, fb(), memcpy(), LogicalDecodingContext::reorder, ReorderBufferProcessXid(), SnapBuildSerializationPoint(), LogicalDecodingContext::snapshot_builder, XLOG_BACKUP_END, XLOG_CHECKPOINT_ONLINE, XLOG_CHECKPOINT_REDO, XLOG_CHECKPOINT_SHUTDOWN, XLOG_END_OF_RECOVERY, XLOG_FPI, XLOG_FPI_FOR_HINT, XLOG_FPW_CHANGE, XLOG_LOGICAL_DECODING_STATUS_CHANGE, XLOG_NEXTOID, XLOG_NOOP, XLOG_OVERWRITE_CONTRECORD, XLOG_PARAMETER_CHANGE, XLOG_RESTORE_POINT, XLOG_SWITCH, XLogRecGetData, XLogRecGetInfo, and XLogRecGetXid.