#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:

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	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

typedef struct XLogRecordBuffer XLogRecordBuffer

Function Documentation

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

static

Definition at line 620 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);
 }

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

static

Definition at line 536 of file decode.c.

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)
     {
         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.
      *
      * 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) ||
         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);
 }

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

static

Definition at line 757 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);
 }

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

static

Definition at line 348 of file decode.c.

References DecodeMultiInsert(), elog, ERROR, 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);
 
     /* no point in doing anything yet */
     if (SnapBuildCurrentState(builder) < SNAPBUILD_FULL_SNAPSHOT)
         return;
 
     switch (info)
     {
         case XLOG_HEAP2_MULTI_INSERT:
             if (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);
     }
 }

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

static

Definition at line 403 of file decode.c.

References DecodeDelete(), DecodeInsert(), DecodeSpecConfirm(), DecodeUpdate(), elog, ERROR, 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_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);
 
     /* no point in doing anything yet */
     if (SnapBuildCurrentState(builder) < SNAPBUILD_FULL_SNAPSHOT)
         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_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;
     }
 }

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

static

Definition at line 640 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);
 }

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

static

Definition at line 489 of file decode.c.

References ReplicationSlot::data, ReplicationSlotPersistentData::database, xl_logical_message::dbId, elog, XLogRecordBuffer::endptr, ERROR, 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);
 
     /* No point in doing anything yet. */
     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;
 
     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);
 }

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

static

Definition at line 814 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);
 }

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

static

Definition at line 919 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);
 }

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

static

Definition at line 300 of file decode.c.

References elog, ERROR, 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);
 
                 ReorderBufferImmediateInvalidation(
                                                    ctx->reorder, invalidations->nmsgs, invalidations->msgs);
             }
             break;
         default:
             elog(ERROR, "unexpected RM_STANDBY_ID record type: %u", info);
     }
 }

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

static

Definition at line 691 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);
 }

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

static

Definition at line 208 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);
     }
 }

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

static

Definition at line 166 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);
     }
 }

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

static

Definition at line 954 of file decode.c.

References Assert, header(), InvalidOid, ItemPointerSetInvalid, 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, 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;
 }

static bool FilterByOrigin	(	LogicalDecodingContext *	ctx,
		RepOriginId	origin_id
	)

inlinestatic

Definition at line 477 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(), and DecodeUpdate().

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

void LogicalDecodingProcessRecord	(	LogicalDecodingContext *	ctx,
		XLogReaderState *	record
	)

Definition at line 93 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_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));
     }
 }

Data Structures

Typedefs

Functions

Typedef Documentation

Function Documentation