#include "postgres.h"
#include "access/heapam_xlog.h"
#include "access/transam.h"
#include "access/xact.h"
#include "access/xlog_internal.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/snapbuild.h"
#include "storage/standbydefs.h"

Include dependency graph for decode.c:

Functions
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, bool two_phase)

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

static void	DecodePrepare (LogicalDecodingContext ctx, XLogRecordBuffer buf, xl_xact_parsed_prepare *parsed)

static void	DecodeXLogTuple (char *data, Size len, HeapTuple tuple)

static bool	FilterPrepare (LogicalDecodingContext ctx, TransactionId xid, const char gid)

static bool	DecodeTXNNeedSkip (LogicalDecodingContext ctx, XLogRecordBuffer buf, Oid txn_dbid, RepOriginId origin_id)

void	LogicalDecodingProcessRecord (LogicalDecodingContext ctx, XLogReaderState record)

void	xlog_decode (LogicalDecodingContext ctx, XLogRecordBuffer buf)

void	xact_decode (LogicalDecodingContext ctx, XLogRecordBuffer buf)

void	standby_decode (LogicalDecodingContext ctx, XLogRecordBuffer buf)

void	heap2_decode (LogicalDecodingContext ctx, XLogRecordBuffer buf)

void	heap_decode (LogicalDecodingContext ctx, XLogRecordBuffer buf)

static bool	FilterByOrigin (LogicalDecodingContext *ctx, RepOriginId origin_id)

void	logicalmsg_decode (LogicalDecodingContext ctx, XLogRecordBuffer buf)

Function Documentation

◆ DecodeAbort()

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

static

Definition at line 847 of file decode.c.

{
    int         i;
    XLogRecPtr  origin_lsn = InvalidXLogRecPtr;
    TimestampTz abort_time = parsed->xact_time;
    RepOriginId origin_id = XLogRecGetOrigin(buf->record);
    bool        skip_xact;
 
    if (parsed->xinfo & XACT_XINFO_HAS_ORIGIN)
    {
        origin_lsn = parsed->origin_lsn;
        abort_time = parsed->origin_timestamp;
    }
 
    /*
     * Check whether we need to process this transaction. See
     * DecodeTXNNeedSkip for the reasons why we sometimes want to skip the
     * transaction.
     */
    skip_xact = DecodeTXNNeedSkip(ctx, buf, parsed->dbId, origin_id);
 
    /*
     * Send the final rollback record for a prepared transaction unless we
     * need to skip it. For non-two-phase xacts, simply forget the xact.
     */
    if (two_phase && !skip_xact)
    {
        ReorderBufferFinishPrepared(ctx->reorder, xid, buf->origptr, buf->endptr,
                                    InvalidXLogRecPtr,
                                    abort_time, origin_id, origin_lsn,
                                    parsed->twophase_gid, false);
    }
    else
    {
        for (i = 0; i < parsed->nsubxacts; i++)
        {
            ReorderBufferAbort(ctx->reorder, parsed->subxacts[i],
                               buf->record->EndRecPtr, abort_time);
        }
 
        ReorderBufferAbort(ctx->reorder, xid, buf->record->EndRecPtr,
                           abort_time);
    }
 
    /* update the decoding stats */
    UpdateDecodingStats(ctx);
}

References buf, xl_xact_parsed_abort::dbId, DecodeTXNNeedSkip(), i, InvalidXLogRecPtr, xl_xact_parsed_abort::nsubxacts, xl_xact_parsed_abort::origin_lsn, xl_xact_parsed_abort::origin_timestamp, LogicalDecodingContext::reorder, ReorderBufferAbort(), ReorderBufferFinishPrepared(), xl_xact_parsed_abort::subxacts, two_phase, xl_xact_parsed_abort::twophase_gid, UpdateDecodingStats(), xl_xact_parsed_abort::xact_time, XACT_XINFO_HAS_ORIGIN, xl_xact_parsed_abort::xinfo, and XLogRecGetOrigin.

Referenced by xact_decode().

◆ DecodeCommit()

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

static

Definition at line 676 of file decode.c.

{
    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;
    }
 
    SnapBuildCommitTxn(ctx->snapshot_builder, buf->origptr, xid,
                       parsed->nsubxacts, parsed->subxacts,
                       parsed->xinfo);
 
    /* ----
     * Check whether we are interested in this specific transaction, and tell
     * the reorderbuffer to forget the content of the (sub-)transactions
     * if not.
     *
     * 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 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 (DecodeTXNNeedSkip(ctx, buf, parsed->dbId, 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);
    }
 
    /*
     * Send the final commit record if the transaction data is already
     * decoded, otherwise, process the entire transaction.
     */
    if (two_phase)
    {
        ReorderBufferFinishPrepared(ctx->reorder, xid, buf->origptr, buf->endptr,
                                    SnapBuildGetTwoPhaseAt(ctx->snapshot_builder),
                                    commit_time, origin_id, origin_lsn,
                                    parsed->twophase_gid, true);
    }
    else
    {
        ReorderBufferCommit(ctx->reorder, xid, buf->origptr, buf->endptr,
                            commit_time, origin_id, origin_lsn);
    }
 
    /*
     * Update the decoding stats at transaction prepare/commit/abort.
     * Additionally we send the stats when we spill or stream the changes to
     * avoid losing them in case the decoding is interrupted. It is not clear
     * that sending more or less frequently than this would be better.
     */
    UpdateDecodingStats(ctx);
}

Referenced by xact_decode().

◆ DecodeDelete()

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

static

Definition at line 1029 of file decode.c.

{
    XLogReaderState *r = buf->record;
    xl_heap_delete *xlrec;
    ReorderBufferChange *change;
    RelFileLocator target_locator;
 
    xlrec = (xl_heap_delete *) XLogRecGetData(r);
 
    /* only interested in our database */
    XLogRecGetBlockTag(r, 0, &target_locator, NULL, NULL);
    if (target_locator.dbOid != ctx->slot->data.database)
        return;
 
    /* output plugin doesn't look for this origin, no need to queue */
    if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
        return;
 
    change = ReorderBufferAllocChange(ctx->reorder);
 
    if (xlrec->flags & XLH_DELETE_IS_SUPER)
        change->action = REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT;
    else
        change->action = REORDER_BUFFER_CHANGE_DELETE;
 
    change->origin_id = XLogRecGetOrigin(r);
 
    memcpy(&change->data.tp.rlocator, &target_locator, sizeof(RelFileLocator));
 
    /* 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 =
            ReorderBufferAllocTupleBuf(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, false);
}

References ReorderBufferChange::action, Assert(), buf, ReorderBufferChange::clear_toast_afterwards, ReorderBufferChange::data, ReplicationSlot::data, ReplicationSlotPersistentData::database, RelFileLocator::dbOid, DecodeXLogTuple(), FilterByOrigin(), xl_heap_delete::flags, ReorderBufferChange::oldtuple, ReorderBufferChange::origin_id, LogicalDecodingContext::reorder, REORDER_BUFFER_CHANGE_DELETE, REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT, ReorderBufferAllocChange(), ReorderBufferAllocTupleBuf(), ReorderBufferQueueChange(), ReorderBufferChange::rlocator, SizeOfHeapDelete, SizeOfHeapHeader, LogicalDecodingContext::slot, ReorderBufferChange::tp, XLH_DELETE_CONTAINS_OLD, XLH_DELETE_IS_SUPER, XLogRecGetBlockTag(), XLogRecGetData, XLogRecGetDataLen, XLogRecGetOrigin, and XLogRecGetXid.

Referenced by heap_decode().

◆ DecodeInsert()

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

static

Definition at line 903 of file decode.c.

{
    Size        datalen;
    char       *tupledata;
    Size        tuplelen;
    XLogReaderState *r = buf->record;
    xl_heap_insert *xlrec;
    ReorderBufferChange *change;
    RelFileLocator target_locator;
 
    xlrec = (xl_heap_insert *) XLogRecGetData(r);
 
    /*
     * Ignore insert records without new tuples (this does happen when
     * raw_heap_insert marks the TOAST record as HEAP_INSERT_NO_LOGICAL).
     */
    if (!(xlrec->flags & XLH_INSERT_CONTAINS_NEW_TUPLE))
        return;
 
    /* only interested in our database */
    XLogRecGetBlockTag(r, 0, &target_locator, NULL, NULL);
    if (target_locator.dbOid != ctx->slot->data.database)
        return;
 
    /* output plugin doesn't look for this origin, no need to queue */
    if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
        return;
 
    change = ReorderBufferAllocChange(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.rlocator, &target_locator, sizeof(RelFileLocator));
 
    tupledata = XLogRecGetBlockData(r, 0, &datalen);
    tuplelen = datalen - SizeOfHeapHeader;
 
    change->data.tp.newtuple =
        ReorderBufferAllocTupleBuf(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,
                             xlrec->flags & XLH_INSERT_ON_TOAST_RELATION);
}

References ReorderBufferChange::action, buf, ReorderBufferChange::clear_toast_afterwards, ReorderBufferChange::data, ReplicationSlot::data, ReplicationSlotPersistentData::database, RelFileLocator::dbOid, DecodeXLogTuple(), FilterByOrigin(), xl_heap_insert::flags, ReorderBufferChange::newtuple, ReorderBufferChange::origin_id, LogicalDecodingContext::reorder, REORDER_BUFFER_CHANGE_INSERT, REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT, ReorderBufferAllocChange(), ReorderBufferAllocTupleBuf(), ReorderBufferQueueChange(), ReorderBufferChange::rlocator, SizeOfHeapHeader, LogicalDecodingContext::slot, ReorderBufferChange::tp, XLH_INSERT_CONTAINS_NEW_TUPLE, XLH_INSERT_IS_SPECULATIVE, XLH_INSERT_ON_TOAST_RELATION, XLogRecGetBlockData(), XLogRecGetBlockTag(), XLogRecGetData, XLogRecGetOrigin, and XLogRecGetXid.

Referenced by heap_decode().

◆ DecodeMultiInsert()

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

static

Definition at line 1121 of file decode.c.

{
    XLogReaderState *r = buf->record;
    xl_heap_multi_insert *xlrec;
    int         i;
    char       *data;
    char       *tupledata;
    Size        tuplelen;
    RelFileLocator rlocator;
 
    xlrec = (xl_heap_multi_insert *) XLogRecGetData(r);
 
    /*
     * Ignore insert records without new tuples.  This happens when a
     * multi_insert is done on a catalog or on a non-persistent relation.
     */
    if (!(xlrec->flags & XLH_INSERT_CONTAINS_NEW_TUPLE))
        return;
 
    /* only interested in our database */
    XLogRecGetBlockTag(r, 0, &rlocator, NULL, NULL);
    if (rlocator.dbOid != ctx->slot->data.database)
        return;
 
    /* output plugin doesn't look for this origin, no need to queue */
    if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
        return;
 
    /*
     * We know that this multi_insert isn't for a catalog, so the block should
     * always have data even if a full-page write of it is taken.
     */
    tupledata = XLogRecGetBlockData(r, 0, &tuplelen);
    Assert(tupledata != NULL);
 
    data = tupledata;
    for (i = 0; i < xlrec->ntuples; i++)
    {
        ReorderBufferChange *change;
        xl_multi_insert_tuple *xlhdr;
        int         datalen;
        HeapTuple   tuple;
        HeapTupleHeader header;
 
        change = ReorderBufferAllocChange(ctx->reorder);
        change->action = REORDER_BUFFER_CHANGE_INSERT;
        change->origin_id = XLogRecGetOrigin(r);
 
        memcpy(&change->data.tp.rlocator, &rlocator, sizeof(RelFileLocator));
 
        xlhdr = (xl_multi_insert_tuple *) SHORTALIGN(data);
        data = ((char *) xlhdr) + SizeOfMultiInsertTuple;
        datalen = xlhdr->datalen;
 
        change->data.tp.newtuple =
            ReorderBufferAllocTupleBuf(ctx->reorder, datalen);
 
        tuple = change->data.tp.newtuple;
        header = tuple->t_data;
 
        /* not a disk based tuple */
        ItemPointerSetInvalid(&tuple->t_self);
 
        /*
         * We can only figure this out after reassembling the transactions.
         */
        tuple->t_tableOid = InvalidOid;
 
        tuple->t_len = datalen + SizeofHeapTupleHeader;
 
        memset(header, 0, SizeofHeapTupleHeader);
 
        memcpy((char *) tuple->t_data + SizeofHeapTupleHeader, 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, false);
 
        /* move to the next xl_multi_insert_tuple entry */
        data += datalen;
    }
    Assert(data == tupledata + tuplelen);
}

References ReorderBufferChange::action, Assert(), buf, ReorderBufferChange::clear_toast_afterwards, ReorderBufferChange::data, ReplicationSlot::data, data, ReplicationSlotPersistentData::database, xl_multi_insert_tuple::datalen, RelFileLocator::dbOid, FilterByOrigin(), xl_heap_multi_insert::flags, i, InvalidOid, ItemPointerSetInvalid(), ReorderBufferChange::newtuple, xl_heap_multi_insert::ntuples, ReorderBufferChange::origin_id, LogicalDecodingContext::reorder, REORDER_BUFFER_CHANGE_INSERT, ReorderBufferAllocChange(), ReorderBufferAllocTupleBuf(), ReorderBufferQueueChange(), ReorderBufferChange::rlocator, SHORTALIGN, SizeofHeapTupleHeader, SizeOfMultiInsertTuple, LogicalDecodingContext::slot, HeapTupleData::t_data, xl_multi_insert_tuple::t_hoff, HeapTupleHeaderData::t_hoff, xl_multi_insert_tuple::t_infomask, HeapTupleHeaderData::t_infomask, xl_multi_insert_tuple::t_infomask2, HeapTupleHeaderData::t_infomask2, HeapTupleData::t_len, HeapTupleData::t_self, HeapTupleData::t_tableOid, ReorderBufferChange::tp, XLH_INSERT_CONTAINS_NEW_TUPLE, XLH_INSERT_LAST_IN_MULTI, XLogRecGetBlockData(), XLogRecGetBlockTag(), XLogRecGetData, XLogRecGetOrigin, and XLogRecGetXid.

Referenced by heap2_decode().

◆ DecodePrepare()

static void DecodePrepare	(	LogicalDecodingContext *	ctx,
		XLogRecordBuffer *	buf,
		xl_xact_parsed_prepare *	parsed
	)

static

Definition at line 772 of file decode.c.

{
    SnapBuild  *builder = ctx->snapshot_builder;
    XLogRecPtr  origin_lsn = parsed->origin_lsn;
    TimestampTz prepare_time = parsed->xact_time;
    RepOriginId origin_id = XLogRecGetOrigin(buf->record);
    int         i;
    TransactionId xid = parsed->twophase_xid;
 
    if (parsed->origin_timestamp != 0)
        prepare_time = parsed->origin_timestamp;
 
    /*
     * Remember the prepare info for a txn so that it can be used later in
     * commit prepared if required. See ReorderBufferFinishPrepared.
     */
    if (!ReorderBufferRememberPrepareInfo(ctx->reorder, xid, buf->origptr,
                                          buf->endptr, prepare_time, origin_id,
                                          origin_lsn))
        return;
 
    /* We can't start streaming unless a consistent state is reached. */
    if (SnapBuildCurrentState(builder) < SNAPBUILD_CONSISTENT)
    {
        ReorderBufferSkipPrepare(ctx->reorder, xid);
        return;
    }
 
    /*
     * Check whether we need to process this transaction. See
     * DecodeTXNNeedSkip for the reasons why we sometimes want to skip the
     * transaction.
     *
     * We can't call ReorderBufferForget as we did in DecodeCommit as the txn
     * hasn't yet been committed, removing this txn before a commit might
     * result in the computation of an incorrect restart_lsn. See
     * SnapBuildProcessRunningXacts. But we need to process cache
     * invalidations if there are any for the reasons mentioned in
     * DecodeCommit.
     */
    if (DecodeTXNNeedSkip(ctx, buf, parsed->dbId, origin_id))
    {
        ReorderBufferSkipPrepare(ctx->reorder, xid);
        ReorderBufferInvalidate(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 */
    ReorderBufferPrepare(ctx->reorder, xid, parsed->twophase_gid);
 
    /*
     * Update the decoding stats at transaction prepare/commit/abort.
     * Additionally we send the stats when we spill or stream the changes to
     * avoid losing them in case the decoding is interrupted. It is not clear
     * that sending more or less frequently than this would be better.
     */
    UpdateDecodingStats(ctx);
}

Referenced by xact_decode().

◆ DecodeSpecConfirm()

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

static

Definition at line 1225 of file decode.c.

{
    XLogReaderState *r = buf->record;
    ReorderBufferChange *change;
    RelFileLocator target_locator;
 
    /* only interested in our database */
    XLogRecGetBlockTag(r, 0, &target_locator, NULL, NULL);
    if (target_locator.dbOid != ctx->slot->data.database)
        return;
 
    /* output plugin doesn't look for this origin, no need to queue */
    if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
        return;
 
    change = ReorderBufferAllocChange(ctx->reorder);
    change->action = REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM;
    change->origin_id = XLogRecGetOrigin(r);
 
    memcpy(&change->data.tp.rlocator, &target_locator, sizeof(RelFileLocator));
 
    change->data.tp.clear_toast_afterwards = true;
 
    ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r), buf->origptr,
                             change, false);
}

References ReorderBufferChange::action, buf, ReorderBufferChange::clear_toast_afterwards, ReorderBufferChange::data, ReplicationSlot::data, ReplicationSlotPersistentData::database, RelFileLocator::dbOid, FilterByOrigin(), ReorderBufferChange::origin_id, LogicalDecodingContext::reorder, REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM, ReorderBufferAllocChange(), ReorderBufferQueueChange(), ReorderBufferChange::rlocator, LogicalDecodingContext::slot, ReorderBufferChange::tp, XLogRecGetBlockTag(), XLogRecGetOrigin, and XLogRecGetXid.

Referenced by heap_decode().

◆ DecodeTruncate()

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

static

Definition at line 1083 of file decode.c.

{
    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 = ReorderBufferAllocChange(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 = ReorderBufferAllocRelids(ctx->reorder,
                                                            xlrec->nrelids);
    memcpy(change->data.truncate.relids, xlrec->relids,
           xlrec->nrelids * sizeof(Oid));
    ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r),
                             buf->origptr, change, false);
}

Referenced by heap_decode().

◆ DecodeTXNNeedSkip()

static bool DecodeTXNNeedSkip	(	LogicalDecodingContext *	ctx,
		XLogRecordBuffer *	buf,
		Oid	txn_dbid,
		RepOriginId	origin_id
	)

static

Definition at line 1305 of file decode.c.

{
    if (SnapBuildXactNeedsSkip(ctx->snapshot_builder, buf->origptr) ||
        (txn_dbid != InvalidOid && txn_dbid != ctx->slot->data.database) ||
        FilterByOrigin(ctx, origin_id))
        return true;
 
    /*
     * We also skip decoding in fast_forward mode. In passing set the
     * processing_required flag to indicate that if it were not for
     * fast_forward mode, processing would have been required.
     */
    if (ctx->fast_forward)
    {
        ctx->processing_required = true;
        return true;
    }
 
    return false;
}

References buf, ReplicationSlot::data, ReplicationSlotPersistentData::database, LogicalDecodingContext::fast_forward, FilterByOrigin(), InvalidOid, LogicalDecodingContext::processing_required, LogicalDecodingContext::slot, SnapBuildXactNeedsSkip(), and LogicalDecodingContext::snapshot_builder.

Referenced by DecodeAbort(), DecodeCommit(), and DecodePrepare().

◆ DecodeUpdate()

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

static

Definition at line 962 of file decode.c.

{
    XLogReaderState *r = buf->record;
    xl_heap_update *xlrec;
    ReorderBufferChange *change;
    char       *data;
    RelFileLocator target_locator;
 
    xlrec = (xl_heap_update *) XLogRecGetData(r);
 
    /* only interested in our database */
    XLogRecGetBlockTag(r, 0, &target_locator, NULL, NULL);
    if (target_locator.dbOid != ctx->slot->data.database)
        return;
 
    /* output plugin doesn't look for this origin, no need to queue */
    if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
        return;
 
    change = ReorderBufferAllocChange(ctx->reorder);
    change->action = REORDER_BUFFER_CHANGE_UPDATE;
    change->origin_id = XLogRecGetOrigin(r);
    memcpy(&change->data.tp.rlocator, &target_locator, sizeof(RelFileLocator));
 
    if (xlrec->flags & XLH_UPDATE_CONTAINS_NEW_TUPLE)
    {
        Size        datalen;
        Size        tuplelen;
 
        data = XLogRecGetBlockData(r, 0, &datalen);
 
        tuplelen = datalen - SizeOfHeapHeader;
 
        change->data.tp.newtuple =
            ReorderBufferAllocTupleBuf(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 =
            ReorderBufferAllocTupleBuf(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, false);
}

Referenced by heap_decode().

◆ DecodeXLogTuple()

static void DecodeXLogTuple	(	char *	data,
		Size	len,
		HeapTuple	tuple
	)

static

Definition at line 1261 of file decode.c.

{
    xl_heap_header xlhdr;
    int         datalen = len - SizeOfHeapHeader;
    HeapTupleHeader header;
 
    Assert(datalen >= 0);
 
    tuple->t_len = datalen + SizeofHeapTupleHeader;
    header = tuple->t_data;
 
    /* not a disk based tuple */
    ItemPointerSetInvalid(&tuple->t_self);
 
    /* we can only figure this out after reassembling the transactions */
    tuple->t_tableOid = InvalidOid;
 
    /* data is not stored aligned, copy to aligned storage */
    memcpy(&xlhdr, data, SizeOfHeapHeader);
 
    memset(header, 0, SizeofHeapTupleHeader);
 
    memcpy(((char *) 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;
}

References Assert(), data, InvalidOid, ItemPointerSetInvalid(), len, SizeOfHeapHeader, SizeofHeapTupleHeader, HeapTupleData::t_data, xl_heap_header::t_hoff, HeapTupleHeaderData::t_hoff, xl_heap_header::t_infomask, HeapTupleHeaderData::t_infomask, xl_heap_header::t_infomask2, HeapTupleHeaderData::t_infomask2, HeapTupleData::t_len, HeapTupleData::t_self, and HeapTupleData::t_tableOid.

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

◆ FilterByOrigin()

static bool FilterByOrigin	(	LogicalDecodingContext *	ctx,
		RepOriginId	origin_id
	)

inlinestatic

Definition at line 583 of file decode.c.

{
    if (ctx->callbacks.filter_by_origin_cb == NULL)
        return false;
 
    return filter_by_origin_cb_wrapper(ctx, origin_id);
}

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

Referenced by DecodeDelete(), DecodeInsert(), DecodeMultiInsert(), DecodeSpecConfirm(), DecodeTruncate(), DecodeTXNNeedSkip(), DecodeUpdate(), and logicalmsg_decode().

◆ FilterPrepare()

static bool FilterPrepare	(	LogicalDecodingContext *	ctx,
		TransactionId	xid,
		const char *	gid
	)

inlinestatic

Definition at line 560 of file decode.c.

{
    /*
     * Skip if decoding of two-phase transactions at PREPARE time is not
     * enabled. In that case, all two-phase transactions are considered
     * filtered out and will be applied as regular transactions at COMMIT
     * PREPARED.
     */
    if (!ctx->twophase)
        return true;
 
    /*
     * The filter_prepare callback is optional. When not supplied, all
     * prepared transactions should go through.
     */
    if (ctx->callbacks.filter_prepare_cb == NULL)
        return false;
 
    return filter_prepare_cb_wrapper(ctx, xid, gid);
}

References LogicalDecodingContext::callbacks, OutputPluginCallbacks::filter_prepare_cb, filter_prepare_cb_wrapper(), and LogicalDecodingContext::twophase.

Referenced by xact_decode().

◆ heap2_decode()

void heap2_decode	(	LogicalDecodingContext *	ctx,
		XLogRecordBuffer *	buf
	)

Definition at line 405 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)
                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_VISIBLE:
        case XLOG_HEAP2_LOCK_UPDATED:
            break;
        default:
            elog(ERROR, "unexpected RM_HEAP2_ID record type: %u", info);
    }
}

References buf, DecodeMultiInsert(), elog, ERROR, LogicalDecodingContext::fast_forward, 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_HEAP2_VISIBLE, XLOG_HEAP_OPMASK, XLogRecGetData, XLogRecGetInfo, and XLogRecGetXid.

◆ heap_decode()

void heap_decode	(	LogicalDecodingContext *	ctx,
		XLogRecordBuffer *	buf
	)

Definition at line 469 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)
                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)
                DecodeUpdate(ctx, buf);
            break;
 
        case XLOG_HEAP_DELETE:
            if (SnapBuildProcessChange(builder, xid, buf->origptr) &&
                !ctx->fast_forward)
                DecodeDelete(ctx, buf);
            break;
 
        case XLOG_HEAP_TRUNCATE:
            if (SnapBuildProcessChange(builder, xid, buf->origptr) &&
                !ctx->fast_forward)
                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.  Inplace
             * updates don't affect storage or interpretation of table rows,
             * so they don't affect logicalrep_write_tuple() outcomes.  Hence,
             * we don't process invalidations from the original operation.  If
             * inplace updates did affect those things, invalidations wouldn't
             * make it work, since there are no snapshot-specific versions of
             * inplace-updated values.  Since we also don't decode catalog
             * tuples, we're not interested in the record's contents.
             *
             * WAL contains likely-unnecessary commit-time invals from the
             * CacheInvalidateHeapTuple() call in
             * heap_inplace_update_and_unlock(). Excess invalidation is safe.
             */
            break;
 
        case XLOG_HEAP_CONFIRM:
            if (SnapBuildProcessChange(builder, xid, buf->origptr) &&
                !ctx->fast_forward)
                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, 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
	)

Definition at line 88 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, GetRmgr(), LogicalDecodingContext::reader, XLogReaderState::ReadRecPtr, LogicalDecodingContext::reorder, ReorderBufferAssignChild(), ReorderBufferProcessXid(), RmgrData::rm_decode, TransactionIdIsValid, XLogRecGetRmid, XLogRecGetTopXid, and XLogRecGetXid.

Referenced by DecodingContextFindStartpoint(), LogicalReplicationSlotHasPendingWal(), LogicalSlotAdvanceAndCheckSnapState(), pg_logical_slot_get_changes_guts(), and XLogSendLogical().

◆ logicalmsg_decode()

void logicalmsg_decode	(	LogicalDecodingContext *	ctx,
		XLogRecordBuffer *	buf
	)

Definition at line 595 of file decode.c.

{
    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 = 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, FilterByOrigin(), xl_logical_message::message, xl_logical_message::message_size, xl_logical_message::prefix_size, LogicalDecodingContext::processing_required, 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
	)

Definition at line 359 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);
 
                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, xl_running_xacts::oldestRunningXid, 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
	)

Definition at line 201 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, FilterPrepare(), xl_xact_invals::msgs, xl_xact_invals::nmsgs, ParseAbortRecord(), ParseCommitRecord(), ParsePrepareRecord(), LogicalDecodingContext::reorder, ReorderBufferAddInvalidations(), ReorderBufferImmediateInvalidation(), ReorderBufferProcessXid(), ReorderBufferXidSetCatalogChanges(), SNAPBUILD_FULL_SNAPSHOT, SnapBuildCurrentState(), LogicalDecodingContext::snapshot_builder, TransactionIdIsValid, two_phase, xl_xact_parsed_commit::twophase_gid, xl_xact_parsed_abort::twophase_gid, 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_INVALIDATIONS, XLOG_XACT_OPMASK, XLOG_XACT_PREPARE, XLogRecGetData, XLogRecGetInfo, and XLogRecGetXid.

◆ xlog_decode()

void xlog_decode	(	LogicalDecodingContext *	ctx,
		XLogRecordBuffer *	buf
	)

Definition at line 129 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_PARAMETER_CHANGE:
            {
                xl_parameter_change *xlrec =
                    (xl_parameter_change *) XLogRecGetData(buf->record);
 
                /*
                 * If wal_level on the primary is reduced to less than
                 * logical, we want to prevent existing logical slots from
                 * being used.  Existing logical slots on the standby get
                 * invalidated when this WAL record is replayed; and further,
                 * slot creation fails when 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)
                {
                    /*
                     * This can occur only on a standby, as a primary would
                     * not allow to restart after changing wal_level < logical
                     * if there is pre-existing logical slot.
                     */
                    Assert(RecoveryInProgress());
                    ereport(ERROR,
                            (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                             errmsg("logical decoding on standby requires \"wal_level\" >= \"logical\" on the primary")));
                }
                break;
            }
        case XLOG_NOOP:
        case XLOG_NEXTOID:
        case XLOG_SWITCH:
        case XLOG_BACKUP_END:
        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 Assert(), buf, elog, ereport, errcode(), errmsg(), ERROR, RecoveryInProgress(), LogicalDecodingContext::reorder, ReorderBufferProcessXid(), SnapBuildSerializationPoint(), LogicalDecodingContext::snapshot_builder, xl_parameter_change::wal_level, WAL_LEVEL_LOGICAL, 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_NEXTOID, XLOG_NOOP, XLOG_OVERWRITE_CONTRECORD, XLOG_PARAMETER_CHANGE, XLOG_RESTORE_POINT, XLOG_SWITCH, XLogRecGetData, XLogRecGetInfo, and XLogRecGetXid.

Functions

Function Documentation

◆ DecodeAbort()

◆ DecodeCommit()

◆ DecodeDelete()

◆ DecodeInsert()

◆ DecodeMultiInsert()

◆ DecodePrepare()

◆ DecodeSpecConfirm()

◆ DecodeTruncate()

◆ DecodeTXNNeedSkip()

◆ DecodeUpdate()

◆ DecodeXLogTuple()

◆ FilterByOrigin()

◆ FilterPrepare()

◆ heap2_decode()

◆ heap_decode()

◆ LogicalDecodingProcessRecord()

◆ logicalmsg_decode()

◆ standby_decode()

◆ xact_decode()

◆ xlog_decode()