decode.c File Reference

#include "postgres.h"
#include "access/heapam.h"
#include "access/heapam_xlog.h"
#include "access/transam.h"
#include "access/xact.h"
#include "access/xlog_internal.h"
#include "access/xlogutils.h"
#include "access/xlogreader.h"
#include "access/xlogrecord.h"
#include "catalog/pg_control.h"
#include "replication/decode.h"
#include "replication/logical.h"
#include "replication/message.h"
#include "replication/reorderbuffer.h"
#include "replication/origin.h"
#include "replication/snapbuild.h"
#include "storage/standby.h"

Include dependency graph for decode.c:

Go to the source code of this file.

Data Structures
struct	XLogRecordBuffer

Typedefs
typedef struct XLogRecordBuffer	XLogRecordBuffer

Functions
static void	DecodeXLogOp (LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
static void	DecodeHeapOp (LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
static void	DecodeHeap2Op (LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
static void	DecodeXactOp (LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
static void	DecodeStandbyOp (LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
static void	DecodeLogicalMsgOp (LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
static void	DecodeInsert (LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
static void	DecodeUpdate (LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
static void	DecodeDelete (LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
static void	DecodeTruncate (LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
static void	DecodeMultiInsert (LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
static void	DecodeSpecConfirm (LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
static void	DecodeCommit (LogicalDecodingContext *ctx, XLogRecordBuffer *buf, xl_xact_parsed_commit *parsed, TransactionId xid)
static void	DecodeAbort (LogicalDecodingContext *ctx, XLogRecordBuffer *buf, xl_xact_parsed_abort *parsed, TransactionId xid)
static void	DecodeXLogTuple (char *data, Size len, ReorderBufferTupleBuf *tup)
void	LogicalDecodingProcessRecord (LogicalDecodingContext *ctx, XLogReaderState *record)
static bool	FilterByOrigin (LogicalDecodingContext *ctx, RepOriginId origin_id)

Typedef Documentation

XLogRecordBuffer

typedef struct XLogRecordBuffer XLogRecordBuffer

Function Documentation

DecodeAbort()

static void DecodeAbort ( LogicalDecodingContext *  ctx, XLogRecordBuffer *  buf, xl_xact_parsed_abort *  parsed, TransactionId  xid  )

static

Definition at line 646 of file decode.c.

References XLogReaderState::EndRecPtr, i, xl_xact_parsed_abort::nsubxacts, XLogRecordBuffer::record, LogicalDecodingContext::reorder, ReorderBufferAbort(), and xl_xact_parsed_abort::subxacts.

Referenced by DecodeXactOp().

{
    int         i;

    for (i = 0; i < parsed->nsubxacts; i++)
    {
        ReorderBufferAbort(ctx->reorder, parsed->subxacts[i],
                           buf->record->EndRecPtr);
    }

    ReorderBufferAbort(ctx->reorder, xid, buf->record->EndRecPtr);
}

DecodeCommit()

static void DecodeCommit ( LogicalDecodingContext *  ctx, XLogRecordBuffer *  buf, xl_xact_parsed_commit *  parsed, TransactionId  xid  )

static

Definition at line 560 of file decode.c.

References ReplicationSlot::data, ReplicationSlotPersistentData::database, xl_xact_parsed_commit::dbId, XLogRecordBuffer::endptr, LogicalDecodingContext::fast_forward, FilterByOrigin(), i, InvalidOid, InvalidXLogRecPtr, xl_xact_parsed_commit::msgs, xl_xact_parsed_commit::nmsgs, xl_xact_parsed_commit::nsubxacts, xl_xact_parsed_commit::origin_lsn, xl_xact_parsed_commit::origin_timestamp, XLogRecordBuffer::origptr, XLogRecordBuffer::record, LogicalDecodingContext::reorder, ReorderBufferAddInvalidations(), ReorderBufferCommit(), ReorderBufferCommitChild(), ReorderBufferForget(), ReorderBufferXidSetCatalogChanges(), LogicalDecodingContext::slot, SnapBuildCommitTxn(), SnapBuildXactNeedsSkip(), LogicalDecodingContext::snapshot_builder, xl_xact_parsed_commit::subxacts, xl_xact_parsed_commit::xact_time, XACT_XINFO_HAS_ORIGIN, xl_xact_parsed_commit::xinfo, and XLogRecGetOrigin.

Referenced by DecodeXactOp().

{
    XLogRecPtr  origin_lsn = InvalidXLogRecPtr;
    TimestampTz commit_time = parsed->xact_time;
    RepOriginId origin_id = XLogRecGetOrigin(buf->record);
    int         i;

    if (parsed->xinfo & XACT_XINFO_HAS_ORIGIN)
    {
        origin_lsn = parsed->origin_lsn;
        commit_time = parsed->origin_timestamp;
    }

    /*
     * Process invalidation messages, even if we're not interested in the
     * transaction's contents, since the various caches need to always be
     * consistent.
     */
    if (parsed->nmsgs > 0)
    {
        if (!ctx->fast_forward)
            ReorderBufferAddInvalidations(ctx->reorder, xid, buf->origptr,
                                          parsed->nmsgs, parsed->msgs);
        ReorderBufferXidSetCatalogChanges(ctx->reorder, xid, buf->origptr);
    }

    SnapBuildCommitTxn(ctx->snapshot_builder, buf->origptr, xid,
                       parsed->nsubxacts, parsed->subxacts);

    /* ----
     * Check whether we are interested in this specific transaction, and tell
     * the reorderbuffer to forget the content of the (sub-)transactions
     * if not.
     *
     * There can be several reasons we might not be interested in this
     * transaction:
     * 1) We might not be interested in decoding transactions up to this
     *    LSN. This can happen because we previously decoded it and now just
     *    are restarting or if we haven't assembled a consistent snapshot yet.
     * 2) The transaction happened in another database.
     * 3) The output plugin is not interested in the origin.
     * 4) We are doing fast-forwarding
     *
     * We can't just use ReorderBufferAbort() here, because we need to execute
     * the transaction's invalidations.  This currently won't be needed if
     * we're just skipping over the transaction because currently we only do
     * so during startup, to get to the first transaction the client needs. As
     * we have reset the catalog caches before starting to read WAL, and we
     * haven't yet touched any catalogs, there can't be anything to invalidate.
     * But if we're "forgetting" this commit because it's it happened in
     * another database, the invalidations might be important, because they
     * could be for shared catalogs and we might have loaded data into the
     * relevant syscaches.
     * ---
     */
    if (SnapBuildXactNeedsSkip(ctx->snapshot_builder, buf->origptr) ||
        (parsed->dbId != InvalidOid && parsed->dbId != ctx->slot->data.database) ||
        ctx->fast_forward || FilterByOrigin(ctx, origin_id))
    {
        for (i = 0; i < parsed->nsubxacts; i++)
        {
            ReorderBufferForget(ctx->reorder, parsed->subxacts[i], buf->origptr);
        }
        ReorderBufferForget(ctx->reorder, xid, buf->origptr);

        return;
    }

    /* tell the reorderbuffer about the surviving subtransactions */
    for (i = 0; i < parsed->nsubxacts; i++)
    {
        ReorderBufferCommitChild(ctx->reorder, xid, parsed->subxacts[i],
                                 buf->origptr, buf->endptr);
    }

    /* replay actions of all transaction + subtransactions in order */
    ReorderBufferCommit(ctx->reorder, xid, buf->origptr, buf->endptr,
                        commit_time, origin_id, origin_lsn);
}

DecodeDelete()

static void DecodeDelete ( LogicalDecodingContext *  ctx, XLogRecordBuffer *  buf  )

static

Definition at line 783 of file decode.c.

References ReorderBufferChange::action, Assert, ReplicationSlot::data, ReorderBufferChange::data, ReplicationSlotPersistentData::database, RelFileNode::dbNode, DecodeXLogTuple(), FilterByOrigin(), xl_heap_delete::flags, ReorderBufferChange::origin_id, XLogRecordBuffer::origptr, XLogRecordBuffer::record, LogicalDecodingContext::reorder, REORDER_BUFFER_CHANGE_DELETE, ReorderBufferGetChange(), ReorderBufferGetTupleBuf(), ReorderBufferQueueChange(), SizeOfHeapDelete, SizeOfHeapHeader, LogicalDecodingContext::slot, ReorderBufferChange::tp, XLH_DELETE_CONTAINS_OLD, XLH_DELETE_IS_SUPER, XLogRecGetBlockTag(), XLogRecGetData, XLogRecGetDataLen, XLogRecGetOrigin, and XLogRecGetXid.

Referenced by DecodeHeapOp().

{
    XLogReaderState *r = buf->record;
    xl_heap_delete *xlrec;
    ReorderBufferChange *change;
    RelFileNode target_node;

    xlrec = (xl_heap_delete *) XLogRecGetData(r);

    /* only interested in our database */
    XLogRecGetBlockTag(r, 0, &target_node, NULL, NULL);
    if (target_node.dbNode != ctx->slot->data.database)
        return;

    /*
     * Super deletions are irrelevant for logical decoding, it's driven by the
     * confirmation records.
     */
    if (xlrec->flags & XLH_DELETE_IS_SUPER)
        return;

    /* output plugin doesn't look for this origin, no need to queue */
    if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
        return;

    change = ReorderBufferGetChange(ctx->reorder);
    change->action = REORDER_BUFFER_CHANGE_DELETE;
    change->origin_id = XLogRecGetOrigin(r);

    memcpy(&change->data.tp.relnode, &target_node, sizeof(RelFileNode));

    /* old primary key stored */
    if (xlrec->flags & XLH_DELETE_CONTAINS_OLD)
    {
        Size        datalen = XLogRecGetDataLen(r) - SizeOfHeapDelete;
        Size        tuplelen = datalen - SizeOfHeapHeader;

        Assert(XLogRecGetDataLen(r) > (SizeOfHeapDelete + SizeOfHeapHeader));

        change->data.tp.oldtuple =
            ReorderBufferGetTupleBuf(ctx->reorder, tuplelen);

        DecodeXLogTuple((char *) xlrec + SizeOfHeapDelete,
                        datalen, change->data.tp.oldtuple);
    }

    change->data.tp.clear_toast_afterwards = true;

    ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r), buf->origptr, change);
}

DecodeHeap2Op()

static void DecodeHeap2Op ( LogicalDecodingContext *  ctx, XLogRecordBuffer *  buf  )

static

Definition at line 354 of file decode.c.

References DecodeMultiInsert(), elog, ERROR, LogicalDecodingContext::fast_forward, XLogRecordBuffer::origptr, XLogRecordBuffer::record, LogicalDecodingContext::reorder, ReorderBufferProcessXid(), SNAPBUILD_FULL_SNAPSHOT, SnapBuildCurrentState(), SnapBuildProcessChange(), SnapBuildProcessNewCid(), LogicalDecodingContext::snapshot_builder, XLOG_HEAP2_CLEAN, XLOG_HEAP2_CLEANUP_INFO, XLOG_HEAP2_FREEZE_PAGE, XLOG_HEAP2_LOCK_UPDATED, XLOG_HEAP2_MULTI_INSERT, XLOG_HEAP2_NEW_CID, XLOG_HEAP2_REWRITE, XLOG_HEAP2_VISIBLE, XLOG_HEAP_OPMASK, XLogRecGetData, XLogRecGetInfo, and XLogRecGetXid.

Referenced by LogicalDecodingProcessRecord().

{
    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 changes.
     */
    if (SnapBuildCurrentState(builder) < SNAPBUILD_FULL_SNAPSHOT ||
        ctx->fast_forward)
        return;

    switch (info)
    {
        case XLOG_HEAP2_MULTI_INSERT:
            if (!ctx->fast_forward &&
                SnapBuildProcessChange(builder, xid, buf->origptr))
                DecodeMultiInsert(ctx, buf);
            break;
        case XLOG_HEAP2_NEW_CID:
            {
                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_FREEZE_PAGE:
        case XLOG_HEAP2_CLEAN:
        case XLOG_HEAP2_CLEANUP_INFO:
        case XLOG_HEAP2_VISIBLE:
        case XLOG_HEAP2_LOCK_UPDATED:
            break;
        default:
            elog(ERROR, "unexpected RM_HEAP2_ID record type: %u", info);
    }
}

DecodeHeapOp()

static void DecodeHeapOp ( LogicalDecodingContext *  ctx, XLogRecordBuffer *  buf  )

static

Definition at line 414 of file decode.c.

References DecodeDelete(), DecodeInsert(), DecodeSpecConfirm(), DecodeTruncate(), DecodeUpdate(), elog, ERROR, LogicalDecodingContext::fast_forward, XLogRecordBuffer::origptr, XLogRecordBuffer::record, LogicalDecodingContext::reorder, ReorderBufferProcessXid(), ReorderBufferXidSetCatalogChanges(), SNAPBUILD_FULL_SNAPSHOT, SnapBuildCurrentState(), SnapBuildProcessChange(), LogicalDecodingContext::snapshot_builder, TransactionIdIsValid, 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.

Referenced by LogicalDecodingProcessRecord().

{
    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.
     */
    if (SnapBuildCurrentState(builder) < SNAPBUILD_FULL_SNAPSHOT ||
        ctx->fast_forward)
        return;

    switch (info)
    {
        case XLOG_HEAP_INSERT:
            if (SnapBuildProcessChange(builder, xid, buf->origptr))
                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))
                DecodeUpdate(ctx, buf);
            break;

        case XLOG_HEAP_DELETE:
            if (SnapBuildProcessChange(builder, xid, buf->origptr))
                DecodeDelete(ctx, buf);
            break;

        case XLOG_HEAP_TRUNCATE:
            if (SnapBuildProcessChange(builder, xid, buf->origptr))
                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
             * don't decode catalog tuples, we're not interested in the
             * record's contents.
             *
             * In-place updates can be used either by XID-bearing transactions
             * (e.g.  in CREATE INDEX CONCURRENTLY) or by XID-less
             * transactions (e.g.  VACUUM).  In the former case, the commit
             * record will include cache invalidations, so we mark the
             * transaction as catalog modifying here. Currently that's
             * redundant because the commit will do that as well, but once we
             * support decoding in-progress relations, this will be important.
             */
            if (!TransactionIdIsValid(xid))
                break;

            SnapBuildProcessChange(builder, xid, buf->origptr);
            ReorderBufferXidSetCatalogChanges(ctx->reorder, xid, buf->origptr);
            break;

        case XLOG_HEAP_CONFIRM:
            if (SnapBuildProcessChange(builder, xid, buf->origptr))
                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;
    }
}

DecodeInsert()

static void DecodeInsert ( LogicalDecodingContext *  ctx, XLogRecordBuffer *  buf  )

static

Definition at line 666 of file decode.c.

References ReorderBufferChange::action, ReplicationSlot::data, ReorderBufferChange::data, ReplicationSlotPersistentData::database, RelFileNode::dbNode, DecodeXLogTuple(), FilterByOrigin(), xl_heap_insert::flags, ReorderBufferChange::origin_id, XLogRecordBuffer::origptr, XLogRecordBuffer::record, LogicalDecodingContext::reorder, REORDER_BUFFER_CHANGE_INSERT, REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT, ReorderBufferGetChange(), ReorderBufferGetTupleBuf(), ReorderBufferQueueChange(), SizeOfHeapHeader, LogicalDecodingContext::slot, ReorderBufferChange::tp, XLH_INSERT_CONTAINS_NEW_TUPLE, XLH_INSERT_IS_SPECULATIVE, XLogRecGetBlockData(), XLogRecGetBlockTag(), XLogRecGetData, XLogRecGetOrigin, and XLogRecGetXid.

Referenced by DecodeHeapOp().

{
    XLogReaderState *r = buf->record;
    xl_heap_insert *xlrec;
    ReorderBufferChange *change;
    RelFileNode target_node;

    xlrec = (xl_heap_insert *) XLogRecGetData(r);

    /* only interested in our database */
    XLogRecGetBlockTag(r, 0, &target_node, NULL, NULL);
    if (target_node.dbNode != ctx->slot->data.database)
        return;

    /* output plugin doesn't look for this origin, no need to queue */
    if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
        return;

    change = ReorderBufferGetChange(ctx->reorder);
    if (!(xlrec->flags & XLH_INSERT_IS_SPECULATIVE))
        change->action = REORDER_BUFFER_CHANGE_INSERT;
    else
        change->action = REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT;
    change->origin_id = XLogRecGetOrigin(r);

    memcpy(&change->data.tp.relnode, &target_node, sizeof(RelFileNode));

    if (xlrec->flags & XLH_INSERT_CONTAINS_NEW_TUPLE)
    {
        Size        datalen;
        char       *tupledata = XLogRecGetBlockData(r, 0, &datalen);
        Size        tuplelen = datalen - SizeOfHeapHeader;

        change->data.tp.newtuple =
            ReorderBufferGetTupleBuf(ctx->reorder, tuplelen);

        DecodeXLogTuple(tupledata, datalen, change->data.tp.newtuple);
    }

    change->data.tp.clear_toast_afterwards = true;

    ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r), buf->origptr, change);
}

DecodeLogicalMsgOp()

static void DecodeLogicalMsgOp ( LogicalDecodingContext *  ctx, XLogRecordBuffer *  buf  )

static

Definition at line 509 of file decode.c.

References ReplicationSlot::data, ReplicationSlotPersistentData::database, xl_logical_message::dbId, elog, XLogRecordBuffer::endptr, ERROR, LogicalDecodingContext::fast_forward, FilterByOrigin(), xl_logical_message::message, xl_logical_message::message_size, XLogRecordBuffer::origptr, xl_logical_message::prefix_size, XLogRecordBuffer::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, XLogRecGetXid, and XLR_INFO_MASK.

Referenced by LogicalDecodingProcessRecord().

{
    SnapBuild  *builder = ctx->snapshot_builder;
    XLogReaderState *r = buf->record;
    TransactionId xid = XLogRecGetXid(r);
    uint8       info = XLogRecGetInfo(r) & ~XLR_INFO_MASK;
    RepOriginId origin_id = XLogRecGetOrigin(r);
    Snapshot    snapshot;
    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 or we are just fast-forwarding, there is no
     * point in decoding messages.
     */
    if (SnapBuildCurrentState(builder) < SNAPBUILD_FULL_SNAPSHOT ||
        ctx->fast_forward)
        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;

    snapshot = SnapBuildGetOrBuildSnapshot(builder, xid);
    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);
}

DecodeMultiInsert()

static void DecodeMultiInsert ( LogicalDecodingContext *  ctx, XLogRecordBuffer *  buf  )

static

Definition at line 875 of file decode.c.

References ReorderBufferChange::action, Assert, ReplicationSlot::data, ReorderBufferChange::data, ReplicationSlotPersistentData::database, xl_multi_insert_tuple::datalen, RelFileNode::dbNode, FilterByOrigin(), xl_heap_multi_insert::flags, header(), i, InvalidOid, ItemPointerSetInvalid, xl_heap_multi_insert::ntuples, ReorderBufferChange::origin_id, XLogRecordBuffer::origptr, XLogRecordBuffer::record, LogicalDecodingContext::reorder, REORDER_BUFFER_CHANGE_INSERT, ReorderBufferGetChange(), ReorderBufferGetTupleBuf(), ReorderBufferQueueChange(), SHORTALIGN, SizeofHeapTupleHeader, SizeOfMultiInsertTuple, LogicalDecodingContext::slot, HeapTupleData::t_data, HeapTupleHeaderData::t_hoff, xl_multi_insert_tuple::t_hoff, HeapTupleHeaderData::t_infomask, xl_multi_insert_tuple::t_infomask, HeapTupleHeaderData::t_infomask2, xl_multi_insert_tuple::t_infomask2, HeapTupleData::t_len, HeapTupleData::t_self, HeapTupleData::t_tableOid, ReorderBufferChange::tp, ReorderBufferTupleBuf::tuple, XLH_INSERT_CONTAINS_NEW_TUPLE, XLH_INSERT_LAST_IN_MULTI, XLogRecGetBlockData(), XLogRecGetBlockTag(), XLogRecGetData, XLogRecGetOrigin, and XLogRecGetXid.

Referenced by DecodeHeap2Op().

{
    XLogReaderState *r = buf->record;
    xl_heap_multi_insert *xlrec;
    int         i;
    char       *data;
    char       *tupledata;
    Size        tuplelen;
    RelFileNode rnode;

    xlrec = (xl_heap_multi_insert *) XLogRecGetData(r);

    /* only interested in our database */
    XLogRecGetBlockTag(r, 0, &rnode, NULL, NULL);
    if (rnode.dbNode != ctx->slot->data.database)
        return;

    /* output plugin doesn't look for this origin, no need to queue */
    if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
        return;

    tupledata = XLogRecGetBlockData(r, 0, &tuplelen);

    data = tupledata;
    for (i = 0; i < xlrec->ntuples; i++)
    {
        ReorderBufferChange *change;
        xl_multi_insert_tuple *xlhdr;
        int         datalen;
        ReorderBufferTupleBuf *tuple;

        change = ReorderBufferGetChange(ctx->reorder);
        change->action = REORDER_BUFFER_CHANGE_INSERT;
        change->origin_id = XLogRecGetOrigin(r);

        memcpy(&change->data.tp.relnode, &rnode, sizeof(RelFileNode));

        /*
         * CONTAINS_NEW_TUPLE will always be set currently as multi_insert
         * isn't used for catalogs, but better be future proof.
         *
         * We decode the tuple in pretty much the same way as DecodeXLogTuple,
         * but since the layout is slightly different, we can't use it here.
         */
        if (xlrec->flags & XLH_INSERT_CONTAINS_NEW_TUPLE)
        {
            HeapTupleHeader header;

            xlhdr = (xl_multi_insert_tuple *) SHORTALIGN(data);
            data = ((char *) xlhdr) + SizeOfMultiInsertTuple;
            datalen = xlhdr->datalen;

            change->data.tp.newtuple =
                ReorderBufferGetTupleBuf(ctx->reorder, datalen);

            tuple = change->data.tp.newtuple;
            header = tuple->tuple.t_data;

            /* not a disk based tuple */
            ItemPointerSetInvalid(&tuple->tuple.t_self);

            /*
             * We can only figure this out after reassembling the
             * transactions.
             */
            tuple->tuple.t_tableOid = InvalidOid;

            tuple->tuple.t_len = datalen + SizeofHeapTupleHeader;

            memset(header, 0, SizeofHeapTupleHeader);

            memcpy((char *) tuple->tuple.t_data + SizeofHeapTupleHeader,
                   (char *) data,
                   datalen);
            data += datalen;

            header->t_infomask = xlhdr->t_infomask;
            header->t_infomask2 = xlhdr->t_infomask2;
            header->t_hoff = xlhdr->t_hoff;
        }

        /*
         * Reset toast reassembly state only after the last row in the last
         * xl_multi_insert_tuple record emitted by one heap_multi_insert()
         * call.
         */
        if (xlrec->flags & XLH_INSERT_LAST_IN_MULTI &&
            (i + 1) == xlrec->ntuples)
            change->data.tp.clear_toast_afterwards = true;
        else
            change->data.tp.clear_toast_afterwards = false;

        ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r),
                                 buf->origptr, change);
    }
    Assert(data == tupledata + tuplelen);
}

DecodeSpecConfirm()

static void DecodeSpecConfirm ( LogicalDecodingContext *  ctx, XLogRecordBuffer *  buf  )

static

Definition at line 980 of file decode.c.

References ReorderBufferChange::action, ReplicationSlot::data, ReorderBufferChange::data, ReplicationSlotPersistentData::database, RelFileNode::dbNode, FilterByOrigin(), ReorderBufferChange::origin_id, XLogRecordBuffer::origptr, XLogRecordBuffer::record, LogicalDecodingContext::reorder, REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM, ReorderBufferGetChange(), ReorderBufferQueueChange(), LogicalDecodingContext::slot, ReorderBufferChange::tp, XLogRecGetBlockTag(), XLogRecGetOrigin, and XLogRecGetXid.

Referenced by DecodeHeapOp().

{
    XLogReaderState *r = buf->record;
    ReorderBufferChange *change;
    RelFileNode target_node;

    /* only interested in our database */
    XLogRecGetBlockTag(r, 0, &target_node, NULL, NULL);
    if (target_node.dbNode != ctx->slot->data.database)
        return;

    /* output plugin doesn't look for this origin, no need to queue */
    if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
        return;

    change = ReorderBufferGetChange(ctx->reorder);
    change->action = REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM;
    change->origin_id = XLogRecGetOrigin(r);

    memcpy(&change->data.tp.relnode, &target_node, sizeof(RelFileNode));

    change->data.tp.clear_toast_afterwards = true;

    ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r), buf->origptr, change);
}

DecodeStandbyOp()

static void DecodeStandbyOp ( LogicalDecodingContext *  ctx, XLogRecordBuffer *  buf  )

static

Definition at line 304 of file decode.c.

References elog, ERROR, LogicalDecodingContext::fast_forward, xl_invalidations::msgs, xl_invalidations::nmsgs, xl_running_xacts::oldestRunningXid, XLogRecordBuffer::origptr, XLogRecordBuffer::record, LogicalDecodingContext::reorder, ReorderBufferAbortOld(), ReorderBufferImmediateInvalidation(), ReorderBufferProcessXid(), SnapBuildProcessRunningXacts(), LogicalDecodingContext::snapshot_builder, XLOG_INVALIDATIONS, XLOG_RUNNING_XACTS, XLOG_STANDBY_LOCK, XLogRecGetData, XLogRecGetInfo, XLogRecGetXid, and XLR_INFO_MASK.

Referenced by LogicalDecodingProcessRecord().

{
    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);

                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:
            {
                xl_invalidations *invalidations =
                (xl_invalidations *) XLogRecGetData(r);

                if (!ctx->fast_forward)
                    ReorderBufferImmediateInvalidation(ctx->reorder,
                                                       invalidations->nmsgs,
                                                       invalidations->msgs);
            }
            break;
        default:
            elog(ERROR, "unexpected RM_STANDBY_ID record type: %u", info);
    }
}

DecodeTruncate()

static void DecodeTruncate ( LogicalDecodingContext *  ctx, XLogRecordBuffer *  buf  )

static

Definition at line 838 of file decode.c.

References ReorderBufferChange::action, ReplicationSlot::data, ReorderBufferChange::data, ReplicationSlotPersistentData::database, xl_heap_truncate::dbId, FilterByOrigin(), xl_heap_truncate::flags, xl_heap_truncate::nrelids, ReorderBufferChange::origin_id, XLogRecordBuffer::origptr, palloc(), XLogRecordBuffer::record, xl_heap_truncate::relids, LogicalDecodingContext::reorder, REORDER_BUFFER_CHANGE_TRUNCATE, ReorderBufferGetChange(), ReorderBufferQueueChange(), LogicalDecodingContext::slot, ReorderBufferChange::truncate, XLH_TRUNCATE_CASCADE, XLH_TRUNCATE_RESTART_SEQS, XLogRecGetData, XLogRecGetOrigin, and XLogRecGetXid.

Referenced by DecodeHeapOp().

{
    XLogReaderState *r = buf->record;
    xl_heap_truncate *xlrec;
    ReorderBufferChange *change;

    xlrec = (xl_heap_truncate *) XLogRecGetData(r);

    /* only interested in our database */
    if (xlrec->dbId != ctx->slot->data.database)
        return;

    /* output plugin doesn't look for this origin, no need to queue */
    if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
        return;

    change = ReorderBufferGetChange(ctx->reorder);
    change->action = REORDER_BUFFER_CHANGE_TRUNCATE;
    change->origin_id = XLogRecGetOrigin(r);
    if (xlrec->flags & XLH_TRUNCATE_CASCADE)
        change->data.truncate.cascade = true;
    if (xlrec->flags & XLH_TRUNCATE_RESTART_SEQS)
        change->data.truncate.restart_seqs = true;
    change->data.truncate.nrelids = xlrec->nrelids;
    change->data.truncate.relids = palloc(xlrec->nrelids * sizeof(Oid));
    memcpy(change->data.truncate.relids, xlrec->relids,
           xlrec->nrelids * sizeof(Oid));
    ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r),
                             buf->origptr, change);
}

DecodeUpdate()

static void DecodeUpdate ( LogicalDecodingContext *  ctx, XLogRecordBuffer *  buf  )

static

Definition at line 717 of file decode.c.

References ReorderBufferChange::action, ReplicationSlot::data, ReorderBufferChange::data, ReplicationSlotPersistentData::database, RelFileNode::dbNode, DecodeXLogTuple(), FilterByOrigin(), xl_heap_update::flags, ReorderBufferChange::origin_id, XLogRecordBuffer::origptr, XLogRecordBuffer::record, LogicalDecodingContext::reorder, REORDER_BUFFER_CHANGE_UPDATE, ReorderBufferGetChange(), ReorderBufferGetTupleBuf(), ReorderBufferQueueChange(), SizeOfHeapHeader, SizeOfHeapUpdate, LogicalDecodingContext::slot, ReorderBufferChange::tp, XLH_UPDATE_CONTAINS_NEW_TUPLE, XLH_UPDATE_CONTAINS_OLD, XLogRecGetBlockData(), XLogRecGetBlockTag(), XLogRecGetData, XLogRecGetDataLen, XLogRecGetOrigin, and XLogRecGetXid.

Referenced by DecodeHeapOp().

{
    XLogReaderState *r = buf->record;
    xl_heap_update *xlrec;
    ReorderBufferChange *change;
    char       *data;
    RelFileNode target_node;

    xlrec = (xl_heap_update *) XLogRecGetData(r);

    /* only interested in our database */
    XLogRecGetBlockTag(r, 0, &target_node, NULL, NULL);
    if (target_node.dbNode != ctx->slot->data.database)
        return;

    /* output plugin doesn't look for this origin, no need to queue */
    if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
        return;

    change = ReorderBufferGetChange(ctx->reorder);
    change->action = REORDER_BUFFER_CHANGE_UPDATE;
    change->origin_id = XLogRecGetOrigin(r);
    memcpy(&change->data.tp.relnode, &target_node, sizeof(RelFileNode));

    if (xlrec->flags & XLH_UPDATE_CONTAINS_NEW_TUPLE)
    {
        Size        datalen;
        Size        tuplelen;

        data = XLogRecGetBlockData(r, 0, &datalen);

        tuplelen = datalen - SizeOfHeapHeader;

        change->data.tp.newtuple =
            ReorderBufferGetTupleBuf(ctx->reorder, tuplelen);

        DecodeXLogTuple(data, datalen, change->data.tp.newtuple);
    }

    if (xlrec->flags & XLH_UPDATE_CONTAINS_OLD)
    {
        Size        datalen;
        Size        tuplelen;

        /* caution, remaining data in record is not aligned */
        data = XLogRecGetData(r) + SizeOfHeapUpdate;
        datalen = XLogRecGetDataLen(r) - SizeOfHeapUpdate;
        tuplelen = datalen - SizeOfHeapHeader;

        change->data.tp.oldtuple =
            ReorderBufferGetTupleBuf(ctx->reorder, tuplelen);

        DecodeXLogTuple(data, datalen, change->data.tp.oldtuple);
    }

    change->data.tp.clear_toast_afterwards = true;

    ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r), buf->origptr, change);
}

DecodeXactOp()

static void DecodeXactOp ( LogicalDecodingContext *  ctx, XLogRecordBuffer *  buf  )

static

Definition at line 212 of file decode.c.

References DecodeAbort(), DecodeCommit(), elog, ERROR, i, xl_xact_assignment::nsubxacts, XLogRecordBuffer::origptr, ParseAbortRecord(), ParseCommitRecord(), XLogRecordBuffer::record, LogicalDecodingContext::reorder, ReorderBufferAssignChild(), ReorderBufferProcessXid(), SNAPBUILD_FULL_SNAPSHOT, SnapBuildCurrentState(), LogicalDecodingContext::snapshot_builder, TransactionIdIsValid, xl_xact_parsed_commit::twophase_xid, xl_xact_parsed_abort::twophase_xid, XLOG_XACT_ABORT, XLOG_XACT_ABORT_PREPARED, XLOG_XACT_ASSIGNMENT, XLOG_XACT_COMMIT, XLOG_XACT_COMMIT_PREPARED, XLOG_XACT_OPMASK, XLOG_XACT_PREPARE, XLogRecGetData, XLogRecGetInfo, XLogRecGetXid, xl_xact_assignment::xsub, and xl_xact_assignment::xtop.

Referenced by LogicalDecodingProcessRecord().

{
    SnapBuild  *builder = ctx->snapshot_builder;
    ReorderBuffer *reorder = ctx->reorder;
    XLogReaderState *r = buf->record;
    uint8       info = XLogRecGetInfo(r) & XLOG_XACT_OPMASK;

    /*
     * No point in doing anything yet, data could not be decoded anyway. It's
     * ok not to call ReorderBufferProcessXid() in that case, except in the
     * assignment case there'll not be any later records with the same xid;
     * and in the assignment case we'll not decode those xacts.
     */
    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;

                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;

                DecodeCommit(ctx, buf, &parsed, xid);
                break;
            }
        case XLOG_XACT_ABORT:
        case XLOG_XACT_ABORT_PREPARED:
            {
                xl_xact_abort *xlrec;
                xl_xact_parsed_abort parsed;
                TransactionId xid;

                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;

                DecodeAbort(ctx, buf, &parsed, xid);
                break;
            }
        case XLOG_XACT_ASSIGNMENT:
            {
                xl_xact_assignment *xlrec;
                int         i;
                TransactionId *sub_xid;

                xlrec = (xl_xact_assignment *) XLogRecGetData(r);

                sub_xid = &xlrec->xsub[0];

                for (i = 0; i < xlrec->nsubxacts; i++)
                {
                    ReorderBufferAssignChild(reorder, xlrec->xtop,
                                             *(sub_xid++), buf->origptr);
                }
                break;
            }
        case XLOG_XACT_PREPARE:

            /*
             * Currently decoding ignores PREPARE TRANSACTION and will just
             * decode the transaction when the COMMIT PREPARED is sent or
             * throw away the transaction's contents when a ROLLBACK PREPARED
             * is received. In the future we could add code to expose prepared
             * transactions in the changestream allowing for a kind of
             * distributed 2PC.
             */
            ReorderBufferProcessXid(reorder, XLogRecGetXid(r), buf->origptr);
            break;
        default:
            elog(ERROR, "unexpected RM_XACT_ID record type: %u", info);
    }
}

DecodeXLogOp()

static void DecodeXLogOp ( LogicalDecodingContext *  ctx, XLogRecordBuffer *  buf  )

static

Definition at line 170 of file decode.c.

References elog, ERROR, XLogRecordBuffer::origptr, XLogRecordBuffer::record, LogicalDecodingContext::reorder, ReorderBufferProcessXid(), SnapBuildSerializationPoint(), LogicalDecodingContext::snapshot_builder, XLOG_BACKUP_END, XLOG_CHECKPOINT_ONLINE, XLOG_CHECKPOINT_SHUTDOWN, XLOG_END_OF_RECOVERY, XLOG_FPI, XLOG_FPI_FOR_HINT, XLOG_FPW_CHANGE, XLOG_NEXTOID, XLOG_NOOP, XLOG_PARAMETER_CHANGE, XLOG_RESTORE_POINT, XLOG_SWITCH, XLogRecGetInfo, XLogRecGetXid, and XLR_INFO_MASK.

Referenced by LogicalDecodingProcessRecord().

{
    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_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:
            break;
        default:
            elog(ERROR, "unexpected RM_XLOG_ID record type: %u", info);
    }
}

DecodeXLogTuple()

static void DecodeXLogTuple ( char *  data, Size  len, ReorderBufferTupleBuf *  tup  )

static

Definition at line 1015 of file decode.c.

References Assert, header(), InvalidOid, ItemPointerSetInvalid, SizeOfHeapHeader, SizeofHeapTupleHeader, HeapTupleData::t_data, HeapTupleHeaderData::t_hoff, HeapTupleHeaderData::t_infomask, HeapTupleHeaderData::t_infomask2, HeapTupleData::t_len, HeapTupleData::t_self, HeapTupleData::t_tableOid, and ReorderBufferTupleBuf::tuple.

Referenced by DecodeDelete(), DecodeInsert(), and DecodeUpdate().

{
    xl_heap_header xlhdr;
    int         datalen = len - SizeOfHeapHeader;
    HeapTupleHeader header;

    Assert(datalen >= 0);

    tuple->tuple.t_len = datalen + SizeofHeapTupleHeader;
    header = tuple->tuple.t_data;

    /* not a disk based tuple */
    ItemPointerSetInvalid(&tuple->tuple.t_self);

    /* we can only figure this out after reassembling the transactions */
    tuple->tuple.t_tableOid = InvalidOid;

    /* data is not stored aligned, copy to aligned storage */
    memcpy((char *) &xlhdr,
           data,
           SizeOfHeapHeader);

    memset(header, 0, SizeofHeapTupleHeader);

    memcpy(((char *) tuple->tuple.t_data) + SizeofHeapTupleHeader,
           data + SizeOfHeapHeader,
           datalen);

    header->t_infomask = xlhdr.t_infomask;
    header->t_infomask2 = xlhdr.t_infomask2;
    header->t_hoff = xlhdr.t_hoff;
}

FilterByOrigin()

static bool FilterByOrigin ( LogicalDecodingContext *  ctx, RepOriginId  origin_id  )

inlinestatic

Definition at line 497 of file decode.c.

References LogicalDecodingContext::callbacks, OutputPluginCallbacks::filter_by_origin_cb, and filter_by_origin_cb_wrapper().

Referenced by DecodeCommit(), DecodeDelete(), DecodeInsert(), DecodeLogicalMsgOp(), DecodeMultiInsert(), DecodeSpecConfirm(), DecodeTruncate(), and DecodeUpdate().

{
    if (ctx->callbacks.filter_by_origin_cb == NULL)
        return false;

    return filter_by_origin_cb_wrapper(ctx, origin_id);
}

LogicalDecodingProcessRecord()

void LogicalDecodingProcessRecord	(	LogicalDecodingContext *	ctx,
		XLogReaderState *	record
	)

Definition at line 97 of file decode.c.

References buf, DecodeHeap2Op(), DecodeHeapOp(), DecodeLogicalMsgOp(), DecodeStandbyOp(), DecodeXactOp(), DecodeXLogOp(), elog, XLogRecordBuffer::endptr, XLogReaderState::EndRecPtr, ERROR, XLogRecordBuffer::origptr, LogicalDecodingContext::reader, XLogReaderState::ReadRecPtr, XLogRecordBuffer::record, LogicalDecodingContext::reorder, ReorderBufferProcessXid(), RM_NEXT_ID, XLogRecGetRmid, and XLogRecGetXid.

Referenced by DecodingContextFindStartpoint(), pg_logical_replication_slot_advance(), pg_logical_slot_get_changes_guts(), and XLogSendLogical().

{
    XLogRecordBuffer buf;

    buf.origptr = ctx->reader->ReadRecPtr;
    buf.endptr = ctx->reader->EndRecPtr;
    buf.record = record;

    /* cast so we get a warning when new rmgrs are added */
    switch ((RmgrIds) XLogRecGetRmid(record))
    {
            /*
             * Rmgrs we care about for logical decoding. Add new rmgrs in
             * rmgrlist.h's order.
             */
        case RM_XLOG_ID:
            DecodeXLogOp(ctx, &buf);
            break;

        case RM_XACT_ID:
            DecodeXactOp(ctx, &buf);
            break;

        case RM_STANDBY_ID:
            DecodeStandbyOp(ctx, &buf);
            break;

        case RM_HEAP2_ID:
            DecodeHeap2Op(ctx, &buf);
            break;

        case RM_HEAP_ID:
            DecodeHeapOp(ctx, &buf);
            break;

        case RM_LOGICALMSG_ID:
            DecodeLogicalMsgOp(ctx, &buf);
            break;

            /*
             * Rmgrs irrelevant for logical decoding; they describe stuff not
             * represented in logical decoding. Add new rmgrs in rmgrlist.h's
             * order.
             */
        case RM_SMGR_ID:
        case RM_CLOG_ID:
        case RM_DBASE_ID:
        case RM_TBLSPC_ID:
        case RM_MULTIXACT_ID:
        case RM_RELMAP_ID:
        case RM_BTREE_ID:
        case RM_HASH_ID:
        case RM_GIN_ID:
        case RM_GIST_ID:
        case RM_SEQ_ID:
        case RM_SPGIST_ID:
        case RM_BRIN_ID:
        case RM_COMMIT_TS_ID:
        case RM_REPLORIGIN_ID:
        case RM_GENERIC_ID:
            /* just deal with xid, and done */
            ReorderBufferProcessXid(ctx->reorder, XLogRecGetXid(record),
                                    buf.origptr);
            break;
        case RM_NEXT_ID:
            elog(ERROR, "unexpected RM_NEXT_ID rmgr_id: %u", (RmgrIds) XLogRecGetRmid(buf.record));
    }
}