PostgreSQL Source Code  git master
decode.c File Reference
Include dependency graph for decode.c:

Go to the source code of this file.

Data Structures

struct  XLogRecordBuffer
 

Typedefs

typedef struct XLogRecordBuffer XLogRecordBuffer
 

Functions

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

Typedef Documentation

◆ XLogRecordBuffer

Function Documentation

◆ DecodeAbort()

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

Definition at line 660 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().

662 {
663  int i;
664 
665  for (i = 0; i < parsed->nsubxacts; i++)
666  {
667  ReorderBufferAbort(ctx->reorder, parsed->subxacts[i],
668  buf->record->EndRecPtr);
669  }
670 
671  ReorderBufferAbort(ctx->reorder, xid, buf->record->EndRecPtr);
672 }
struct ReorderBuffer * reorder
Definition: logical.h:42
XLogRecPtr EndRecPtr
Definition: xlogreader.h:176
XLogReaderState * record
Definition: decode.c:50
TransactionId * subxacts
Definition: xact.h:359
void ReorderBufferAbort(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
int i

◆ DecodeCommit()

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

Definition at line 587 of file decode.c.

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

Referenced by DecodeXactOp().

589 {
590  XLogRecPtr origin_lsn = InvalidXLogRecPtr;
591  TimestampTz commit_time = parsed->xact_time;
592  RepOriginId origin_id = XLogRecGetOrigin(buf->record);
593  int i;
594 
595  if (parsed->xinfo & XACT_XINFO_HAS_ORIGIN)
596  {
597  origin_lsn = parsed->origin_lsn;
598  commit_time = parsed->origin_timestamp;
599  }
600 
602  parsed->nsubxacts, parsed->subxacts);
603 
604  /* ----
605  * Check whether we are interested in this specific transaction, and tell
606  * the reorderbuffer to forget the content of the (sub-)transactions
607  * if not.
608  *
609  * There can be several reasons we might not be interested in this
610  * transaction:
611  * 1) We might not be interested in decoding transactions up to this
612  * LSN. This can happen because we previously decoded it and now just
613  * are restarting or if we haven't assembled a consistent snapshot yet.
614  * 2) The transaction happened in another database.
615  * 3) The output plugin is not interested in the origin.
616  * 4) We are doing fast-forwarding
617  *
618  * We can't just use ReorderBufferAbort() here, because we need to execute
619  * the transaction's invalidations. This currently won't be needed if
620  * we're just skipping over the transaction because currently we only do
621  * so during startup, to get to the first transaction the client needs. As
622  * we have reset the catalog caches before starting to read WAL, and we
623  * haven't yet touched any catalogs, there can't be anything to invalidate.
624  * But if we're "forgetting" this commit because it's it happened in
625  * another database, the invalidations might be important, because they
626  * could be for shared catalogs and we might have loaded data into the
627  * relevant syscaches.
628  * ---
629  */
631  (parsed->dbId != InvalidOid && parsed->dbId != ctx->slot->data.database) ||
632  ctx->fast_forward || FilterByOrigin(ctx, origin_id))
633  {
634  for (i = 0; i < parsed->nsubxacts; i++)
635  {
636  ReorderBufferForget(ctx->reorder, parsed->subxacts[i], buf->origptr);
637  }
638  ReorderBufferForget(ctx->reorder, xid, buf->origptr);
639 
640  return;
641  }
642 
643  /* tell the reorderbuffer about the surviving subtransactions */
644  for (i = 0; i < parsed->nsubxacts; i++)
645  {
646  ReorderBufferCommitChild(ctx->reorder, xid, parsed->subxacts[i],
647  buf->origptr, buf->endptr);
648  }
649 
650  /* replay actions of all transaction + subtransactions in order */
651  ReorderBufferCommit(ctx->reorder, xid, buf->origptr, buf->endptr,
652  commit_time, origin_id, origin_lsn);
653 }
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
#define XACT_XINFO_HAS_ORIGIN
Definition: xact.h:171
struct ReorderBuffer * reorder
Definition: logical.h:42
void ReorderBufferForget(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
bool SnapBuildXactNeedsSkip(SnapBuild *builder, XLogRecPtr ptr)
Definition: snapbuild.c:404
void ReorderBufferCommit(ReorderBuffer *rb, TransactionId xid, XLogRecPtr commit_lsn, XLogRecPtr end_lsn, TimestampTz commit_time, RepOriginId origin_id, XLogRecPtr origin_lsn)
TimestampTz origin_timestamp
Definition: xact.h:345
void SnapBuildCommitTxn(SnapBuild *builder, XLogRecPtr lsn, TransactionId xid, int nsubxacts, TransactionId *subxacts)
Definition: snapbuild.c:924
int64 TimestampTz
Definition: timestamp.h:39
TransactionId * subxacts
Definition: xact.h:331
uint16 RepOriginId
Definition: xlogdefs.h:58
XLogRecPtr origin_lsn
Definition: xact.h:344
ReplicationSlotPersistentData data
Definition: slot.h:143
static bool FilterByOrigin(LogicalDecodingContext *ctx, RepOriginId origin_id)
Definition: decode.c:524
#define XLogRecGetOrigin(decoder)
Definition: xlogreader.h:308
XLogRecPtr endptr
Definition: decode.c:49
XLogReaderState * record
Definition: decode.c:50
void ReorderBufferCommitChild(ReorderBuffer *rb, TransactionId xid, TransactionId subxid, XLogRecPtr commit_lsn, XLogRecPtr end_lsn)
#define InvalidOid
Definition: postgres_ext.h:36
struct SnapBuild * snapshot_builder
Definition: logical.h:43
uint64 XLogRecPtr
Definition: xlogdefs.h:21
XLogRecPtr origptr
Definition: decode.c:48
ReplicationSlot * slot
Definition: logical.h:38
int i
TimestampTz xact_time
Definition: xact.h:324

◆ DecodeDelete()

static void DecodeDelete ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)
static

Definition at line 806 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().

807 {
808  XLogReaderState *r = buf->record;
809  xl_heap_delete *xlrec;
810  ReorderBufferChange *change;
811  RelFileNode target_node;
812 
813  xlrec = (xl_heap_delete *) XLogRecGetData(r);
814 
815  /* only interested in our database */
816  XLogRecGetBlockTag(r, 0, &target_node, NULL, NULL);
817  if (target_node.dbNode != ctx->slot->data.database)
818  return;
819 
820  /*
821  * Super deletions are irrelevant for logical decoding, it's driven by the
822  * confirmation records.
823  */
824  if (xlrec->flags & XLH_DELETE_IS_SUPER)
825  return;
826 
827  /* output plugin doesn't look for this origin, no need to queue */
828  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
829  return;
830 
831  change = ReorderBufferGetChange(ctx->reorder);
833  change->origin_id = XLogRecGetOrigin(r);
834 
835  memcpy(&change->data.tp.relnode, &target_node, sizeof(RelFileNode));
836 
837  /* old primary key stored */
838  if (xlrec->flags & XLH_DELETE_CONTAINS_OLD)
839  {
840  Size datalen = XLogRecGetDataLen(r) - SizeOfHeapDelete;
841  Size tuplelen = datalen - SizeOfHeapHeader;
842 
843  Assert(XLogRecGetDataLen(r) > (SizeOfHeapDelete + SizeOfHeapHeader));
844 
845  change->data.tp.oldtuple =
846  ReorderBufferGetTupleBuf(ctx->reorder, tuplelen);
847 
848  DecodeXLogTuple((char *) xlrec + SizeOfHeapDelete,
849  datalen, change->data.tp.oldtuple);
850  }
851 
852  change->data.tp.clear_toast_afterwards = true;
853 
855  change, false);
856 }
#define XLH_DELETE_IS_SUPER
Definition: heapam_xlog.h:96
struct ReorderBuffer * reorder
Definition: logical.h:42
static void DecodeXLogTuple(char *data, Size len, ReorderBufferTupleBuf *tup)
Definition: decode.c:1042
#define XLH_DELETE_CONTAINS_OLD
Definition: heapam_xlog.h:100
ReorderBufferChange * ReorderBufferGetChange(ReorderBuffer *rb)
ReplicationSlotPersistentData data
Definition: slot.h:143
static bool FilterByOrigin(LogicalDecodingContext *ctx, RepOriginId origin_id)
Definition: decode.c:524
enum ReorderBufferChangeType action
Definition: reorderbuffer.h:83
#define XLogRecGetOrigin(decoder)
Definition: xlogreader.h:308
#define XLogRecGetData(decoder)
Definition: xlogreader.h:310
#define XLogRecGetDataLen(decoder)
Definition: xlogreader.h:311
#define SizeOfHeapDelete
Definition: heapam_xlog.h:112
XLogReaderState * record
Definition: decode.c:50
RepOriginId origin_id
Definition: reorderbuffer.h:88
bool XLogRecGetBlockTag(XLogReaderState *record, uint8 block_id, RelFileNode *rnode, ForkNumber *forknum, BlockNumber *blknum)
Definition: xlogreader.c:1498
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:307
void ReorderBufferQueueChange(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, ReorderBufferChange *change, bool toast_insert)
union ReorderBufferChange::@99 data
struct ReorderBufferChange::@99::@100 tp
#define Assert(condition)
Definition: c.h:745
size_t Size
Definition: c.h:473
XLogRecPtr origptr
Definition: decode.c:48
ReplicationSlot * slot
Definition: logical.h:38
#define SizeOfHeapHeader
Definition: heapam_xlog.h:148
ReorderBufferTupleBuf * ReorderBufferGetTupleBuf(ReorderBuffer *rb, Size tuple_len)

◆ DecodeHeap2Op()

static void DecodeHeap2Op ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)
static

Definition at line 381 of file decode.c.

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

Referenced by LogicalDecodingProcessRecord().

382 {
384  TransactionId xid = XLogRecGetXid(buf->record);
385  SnapBuild *builder = ctx->snapshot_builder;
386 
387  ReorderBufferProcessXid(ctx->reorder, xid, buf->origptr);
388 
389  /*
390  * If we don't have snapshot or we are just fast-forwarding, there is no
391  * point in decoding changes.
392  */
394  ctx->fast_forward)
395  return;
396 
397  switch (info)
398  {
400  if (!ctx->fast_forward &&
401  SnapBuildProcessChange(builder, xid, buf->origptr))
402  DecodeMultiInsert(ctx, buf);
403  break;
404  case XLOG_HEAP2_NEW_CID:
405  {
406  xl_heap_new_cid *xlrec;
407 
408  xlrec = (xl_heap_new_cid *) XLogRecGetData(buf->record);
409  SnapBuildProcessNewCid(builder, xid, buf->origptr, xlrec);
410 
411  break;
412  }
413  case XLOG_HEAP2_REWRITE:
414 
415  /*
416  * Although these records only exist to serve the needs of logical
417  * decoding, all the work happens as part of crash or archive
418  * recovery, so we don't need to do anything here.
419  */
420  break;
421 
422  /*
423  * Everything else here is just low level physical stuff we're not
424  * interested in.
425  */
427  case XLOG_HEAP2_CLEAN:
429  case XLOG_HEAP2_VISIBLE:
431  break;
432  default:
433  elog(ERROR, "unexpected RM_HEAP2_ID record type: %u", info);
434  }
435 }
static void DecodeMultiInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:900
struct ReorderBuffer * reorder
Definition: logical.h:42
#define XLOG_HEAP2_LOCK_UPDATED
Definition: heapam_xlog.h:59
uint32 TransactionId
Definition: c.h:520
SnapBuildState SnapBuildCurrentState(SnapBuild *builder)
Definition: snapbuild.c:395
#define XLOG_HEAP2_REWRITE
Definition: heapam_xlog.h:53
unsigned char uint8
Definition: c.h:372
#define XLOG_HEAP_OPMASK
Definition: heapam_xlog.h:41
#define XLOG_HEAP2_MULTI_INSERT
Definition: heapam_xlog.h:58
bool SnapBuildProcessChange(SnapBuild *builder, TransactionId xid, XLogRecPtr lsn)
Definition: snapbuild.c:716
#define XLOG_HEAP2_VISIBLE
Definition: heapam_xlog.h:57
#define XLogRecGetData(decoder)
Definition: xlogreader.h:310
#define ERROR
Definition: elog.h:43
#define XLOG_HEAP2_CLEAN
Definition: heapam_xlog.h:54
#define XLOG_HEAP2_CLEANUP_INFO
Definition: heapam_xlog.h:56
void SnapBuildProcessNewCid(SnapBuild *builder, TransactionId xid, XLogRecPtr lsn, xl_heap_new_cid *xlrec)
Definition: snapbuild.c:766
#define XLOG_HEAP2_NEW_CID
Definition: heapam_xlog.h:60
XLogReaderState * record
Definition: decode.c:50
#define XLogRecGetInfo(decoder)
Definition: xlogreader.h:305
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:307
struct SnapBuild * snapshot_builder
Definition: logical.h:43
#define XLOG_HEAP2_FREEZE_PAGE
Definition: heapam_xlog.h:55
XLogRecPtr origptr
Definition: decode.c:48
#define elog(elevel,...)
Definition: elog.h:214
void ReorderBufferProcessXid(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)

◆ DecodeHeapOp()

static void DecodeHeapOp ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)
static

Definition at line 441 of file decode.c.

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

Referenced by LogicalDecodingProcessRecord().

442 {
444  TransactionId xid = XLogRecGetXid(buf->record);
445  SnapBuild *builder = ctx->snapshot_builder;
446 
447  ReorderBufferProcessXid(ctx->reorder, xid, buf->origptr);
448 
449  /*
450  * If we don't have snapshot or we are just fast-forwarding, there is no
451  * point in decoding data changes.
452  */
454  ctx->fast_forward)
455  return;
456 
457  switch (info)
458  {
459  case XLOG_HEAP_INSERT:
460  if (SnapBuildProcessChange(builder, xid, buf->origptr))
461  DecodeInsert(ctx, buf);
462  break;
463 
464  /*
465  * Treat HOT update as normal updates. There is no useful
466  * information in the fact that we could make it a HOT update
467  * locally and the WAL layout is compatible.
468  */
470  case XLOG_HEAP_UPDATE:
471  if (SnapBuildProcessChange(builder, xid, buf->origptr))
472  DecodeUpdate(ctx, buf);
473  break;
474 
475  case XLOG_HEAP_DELETE:
476  if (SnapBuildProcessChange(builder, xid, buf->origptr))
477  DecodeDelete(ctx, buf);
478  break;
479 
480  case XLOG_HEAP_TRUNCATE:
481  if (SnapBuildProcessChange(builder, xid, buf->origptr))
482  DecodeTruncate(ctx, buf);
483  break;
484 
485  case XLOG_HEAP_INPLACE:
486 
487  /*
488  * Inplace updates are only ever performed on catalog tuples and
489  * can, per definition, not change tuple visibility. Since we
490  * don't decode catalog tuples, we're not interested in the
491  * record's contents.
492  *
493  * In-place updates can be used either by XID-bearing transactions
494  * (e.g. in CREATE INDEX CONCURRENTLY) or by XID-less
495  * transactions (e.g. VACUUM). In the former case, the commit
496  * record will include cache invalidations, so we mark the
497  * transaction as catalog modifying here. Currently that's
498  * redundant because the commit will do that as well, but once we
499  * support decoding in-progress relations, this will be important.
500  */
501  if (!TransactionIdIsValid(xid))
502  break;
503 
504  SnapBuildProcessChange(builder, xid, buf->origptr);
506  break;
507 
508  case XLOG_HEAP_CONFIRM:
509  if (SnapBuildProcessChange(builder, xid, buf->origptr))
510  DecodeSpecConfirm(ctx, buf);
511  break;
512 
513  case XLOG_HEAP_LOCK:
514  /* we don't care about row level locks for now */
515  break;
516 
517  default:
518  elog(ERROR, "unexpected RM_HEAP_ID record type: %u", info);
519  break;
520  }
521 }
#define XLOG_HEAP_INSERT
Definition: heapam_xlog.h:32
struct ReorderBuffer * reorder
Definition: logical.h:42
static void DecodeInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:680
uint32 TransactionId
Definition: c.h:520
SnapBuildState SnapBuildCurrentState(SnapBuild *builder)
Definition: snapbuild.c:395
unsigned char uint8
Definition: c.h:372
#define XLOG_HEAP_OPMASK
Definition: heapam_xlog.h:41
#define XLOG_HEAP_HOT_UPDATE
Definition: heapam_xlog.h:36
static void DecodeUpdate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:739
bool SnapBuildProcessChange(SnapBuild *builder, TransactionId xid, XLogRecPtr lsn)
Definition: snapbuild.c:716
static void DecodeSpecConfirm(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:1006
#define ERROR
Definition: elog.h:43
static void DecodeDelete(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:806
XLogReaderState * record
Definition: decode.c:50
#define XLogRecGetInfo(decoder)
Definition: xlogreader.h:305
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:307
#define XLOG_HEAP_TRUNCATE
Definition: heapam_xlog.h:35
struct SnapBuild * snapshot_builder
Definition: logical.h:43
#define XLOG_HEAP_UPDATE
Definition: heapam_xlog.h:34
static void DecodeTruncate(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:862
#define XLOG_HEAP_INPLACE
Definition: heapam_xlog.h:39
XLogRecPtr origptr
Definition: decode.c:48
#define XLOG_HEAP_LOCK
Definition: heapam_xlog.h:38
#define elog(elevel,...)
Definition: elog.h:214
void ReorderBufferXidSetCatalogChanges(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
#define TransactionIdIsValid(xid)
Definition: transam.h:41
void ReorderBufferProcessXid(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
#define XLOG_HEAP_DELETE
Definition: heapam_xlog.h:33
#define XLOG_HEAP_CONFIRM
Definition: heapam_xlog.h:37

◆ DecodeInsert()

static void DecodeInsert ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)
static

Definition at line 680 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, XLH_INSERT_ON_TOAST_RELATION, XLogRecGetBlockData(), XLogRecGetBlockTag(), XLogRecGetData, XLogRecGetOrigin, and XLogRecGetXid.

Referenced by DecodeHeapOp().

681 {
682  Size datalen;
683  char *tupledata;
684  Size tuplelen;
685  XLogReaderState *r = buf->record;
686  xl_heap_insert *xlrec;
687  ReorderBufferChange *change;
688  RelFileNode target_node;
689 
690  xlrec = (xl_heap_insert *) XLogRecGetData(r);
691 
692  /*
693  * Ignore insert records without new tuples (this does happen when
694  * raw_heap_insert marks the TOAST record as HEAP_INSERT_NO_LOGICAL).
695  */
696  if (!(xlrec->flags & XLH_INSERT_CONTAINS_NEW_TUPLE))
697  return;
698 
699  /* only interested in our database */
700  XLogRecGetBlockTag(r, 0, &target_node, NULL, NULL);
701  if (target_node.dbNode != ctx->slot->data.database)
702  return;
703 
704  /* output plugin doesn't look for this origin, no need to queue */
705  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
706  return;
707 
708  change = ReorderBufferGetChange(ctx->reorder);
709  if (!(xlrec->flags & XLH_INSERT_IS_SPECULATIVE))
711  else
713  change->origin_id = XLogRecGetOrigin(r);
714 
715  memcpy(&change->data.tp.relnode, &target_node, sizeof(RelFileNode));
716 
717  tupledata = XLogRecGetBlockData(r, 0, &datalen);
718  tuplelen = datalen - SizeOfHeapHeader;
719 
720  change->data.tp.newtuple =
721  ReorderBufferGetTupleBuf(ctx->reorder, tuplelen);
722 
723  DecodeXLogTuple(tupledata, datalen, change->data.tp.newtuple);
724 
725  change->data.tp.clear_toast_afterwards = true;
726 
728  change,
730 }
struct ReorderBuffer * reorder
Definition: logical.h:42
static void DecodeXLogTuple(char *data, Size len, ReorderBufferTupleBuf *tup)
Definition: decode.c:1042
#define XLH_INSERT_IS_SPECULATIVE
Definition: heapam_xlog.h:68
#define XLH_INSERT_ON_TOAST_RELATION
Definition: heapam_xlog.h:70
ReorderBufferChange * ReorderBufferGetChange(ReorderBuffer *rb)
ReplicationSlotPersistentData data
Definition: slot.h:143
static bool FilterByOrigin(LogicalDecodingContext *ctx, RepOriginId origin_id)
Definition: decode.c:524
enum ReorderBufferChangeType action
Definition: reorderbuffer.h:83
#define XLogRecGetOrigin(decoder)
Definition: xlogreader.h:308
#define XLogRecGetData(decoder)
Definition: xlogreader.h:310
#define XLH_INSERT_CONTAINS_NEW_TUPLE
Definition: heapam_xlog.h:69
XLogReaderState * record
Definition: decode.c:50
RepOriginId origin_id
Definition: reorderbuffer.h:88
bool XLogRecGetBlockTag(XLogReaderState *record, uint8 block_id, RelFileNode *rnode, ForkNumber *forknum, BlockNumber *blknum)
Definition: xlogreader.c:1498
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:307
char * XLogRecGetBlockData(XLogReaderState *record, uint8 block_id, Size *len)
Definition: xlogreader.c:1522
void ReorderBufferQueueChange(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, ReorderBufferChange *change, bool toast_insert)
union ReorderBufferChange::@99 data
struct ReorderBufferChange::@99::@100 tp
size_t Size
Definition: c.h:473
XLogRecPtr origptr
Definition: decode.c:48
ReplicationSlot * slot
Definition: logical.h:38
#define SizeOfHeapHeader
Definition: heapam_xlog.h:148
ReorderBufferTupleBuf * ReorderBufferGetTupleBuf(ReorderBuffer *rb, Size tuple_len)

◆ DecodeLogicalMsgOp()

static void DecodeLogicalMsgOp ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)
static

Definition at line 536 of file decode.c.

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

Referenced by LogicalDecodingProcessRecord().

537 {
538  SnapBuild *builder = ctx->snapshot_builder;
539  XLogReaderState *r = buf->record;
540  TransactionId xid = XLogRecGetXid(r);
541  uint8 info = XLogRecGetInfo(r) & ~XLR_INFO_MASK;
542  RepOriginId origin_id = XLogRecGetOrigin(r);
543  Snapshot snapshot;
544  xl_logical_message *message;
545 
546  if (info != XLOG_LOGICAL_MESSAGE)
547  elog(ERROR, "unexpected RM_LOGICALMSG_ID record type: %u", info);
548 
550 
551  /*
552  * If we don't have snapshot or we are just fast-forwarding, there is no
553  * point in decoding messages.
554  */
556  ctx->fast_forward)
557  return;
558 
559  message = (xl_logical_message *) XLogRecGetData(r);
560 
561  if (message->dbId != ctx->slot->data.database ||
562  FilterByOrigin(ctx, origin_id))
563  return;
564 
565  if (message->transactional &&
566  !SnapBuildProcessChange(builder, xid, buf->origptr))
567  return;
568  else if (!message->transactional &&
570  SnapBuildXactNeedsSkip(builder, buf->origptr)))
571  return;
572 
573  snapshot = SnapBuildGetOrBuildSnapshot(builder, xid);
574  ReorderBufferQueueMessage(ctx->reorder, xid, snapshot, buf->endptr,
575  message->transactional,
576  message->message, /* first part of message is
577  * prefix */
578  message->message_size,
579  message->message + message->prefix_size);
580 }
void ReorderBufferQueueMessage(ReorderBuffer *rb, TransactionId xid, Snapshot snapshot, XLogRecPtr lsn, bool transactional, const char *prefix, Size message_size, const char *message)
struct ReorderBuffer * reorder
Definition: logical.h:42
uint32 TransactionId
Definition: c.h:520
bool SnapBuildXactNeedsSkip(SnapBuild *builder, XLogRecPtr ptr)
Definition: snapbuild.c:404
SnapBuildState SnapBuildCurrentState(SnapBuild *builder)
Definition: snapbuild.c:395
unsigned char uint8
Definition: c.h:372
uint16 RepOriginId
Definition: xlogdefs.h:58
ReplicationSlotPersistentData data
Definition: slot.h:143
static bool FilterByOrigin(LogicalDecodingContext *ctx, RepOriginId origin_id)
Definition: decode.c:524
#define XLogRecGetOrigin(decoder)
Definition: xlogreader.h:308
char message[FLEXIBLE_ARRAY_MEMBER]
Definition: message.h:27
bool SnapBuildProcessChange(SnapBuild *builder, TransactionId xid, XLogRecPtr lsn)
Definition: snapbuild.c:716
#define XLogRecGetData(decoder)
Definition: xlogreader.h:310
#define ERROR
Definition: elog.h:43
XLogRecPtr endptr
Definition: decode.c:49
XLogReaderState * record
Definition: decode.c:50
#define XLogRecGetInfo(decoder)
Definition: xlogreader.h:305
bool transactional
Definition: message.h:23
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:307
struct SnapBuild * snapshot_builder
Definition: logical.h:43
#define XLR_INFO_MASK
Definition: xlogrecord.h:62
XLogRecPtr origptr
Definition: decode.c:48
ReplicationSlot * slot
Definition: logical.h:38
#define elog(elevel,...)
Definition: elog.h:214
#define XLOG_LOGICAL_MESSAGE
Definition: message.h:36
void ReorderBufferProcessXid(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
Snapshot SnapBuildGetOrBuildSnapshot(SnapBuild *builder, TransactionId xid)
Definition: snapbuild.c:672

◆ DecodeMultiInsert()

static void DecodeMultiInsert ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)
static

Definition at line 900 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().

901 {
902  XLogReaderState *r = buf->record;
903  xl_heap_multi_insert *xlrec;
904  int i;
905  char *data;
906  char *tupledata;
907  Size tuplelen;
908  RelFileNode rnode;
909 
910  xlrec = (xl_heap_multi_insert *) XLogRecGetData(r);
911 
912  /*
913  * Ignore insert records without new tuples. This happens when a
914  * multi_insert is done on a catalog or on a non-persistent relation.
915  */
916  if (!(xlrec->flags & XLH_INSERT_CONTAINS_NEW_TUPLE))
917  return;
918 
919  /* only interested in our database */
920  XLogRecGetBlockTag(r, 0, &rnode, NULL, NULL);
921  if (rnode.dbNode != ctx->slot->data.database)
922  return;
923 
924  /* output plugin doesn't look for this origin, no need to queue */
925  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
926  return;
927 
928  /*
929  * We know that this multi_insert isn't for a catalog, so the block should
930  * always have data even if a full-page write of it is taken.
931  */
932  tupledata = XLogRecGetBlockData(r, 0, &tuplelen);
933  Assert(tupledata != NULL);
934 
935  data = tupledata;
936  for (i = 0; i < xlrec->ntuples; i++)
937  {
938  ReorderBufferChange *change;
939  xl_multi_insert_tuple *xlhdr;
940  int datalen;
941  ReorderBufferTupleBuf *tuple;
943 
944  change = ReorderBufferGetChange(ctx->reorder);
946  change->origin_id = XLogRecGetOrigin(r);
947 
948  memcpy(&change->data.tp.relnode, &rnode, sizeof(RelFileNode));
949 
950  xlhdr = (xl_multi_insert_tuple *) SHORTALIGN(data);
951  data = ((char *) xlhdr) + SizeOfMultiInsertTuple;
952  datalen = xlhdr->datalen;
953 
954  change->data.tp.newtuple =
955  ReorderBufferGetTupleBuf(ctx->reorder, datalen);
956 
957  tuple = change->data.tp.newtuple;
958  header = tuple->tuple.t_data;
959 
960  /* not a disk based tuple */
962 
963  /*
964  * We can only figure this out after reassembling the transactions.
965  */
966  tuple->tuple.t_tableOid = InvalidOid;
967 
968  tuple->tuple.t_len = datalen + SizeofHeapTupleHeader;
969 
970  memset(header, 0, SizeofHeapTupleHeader);
971 
972  memcpy((char *) tuple->tuple.t_data + SizeofHeapTupleHeader,
973  (char *) data,
974  datalen);
975  header->t_infomask = xlhdr->t_infomask;
976  header->t_infomask2 = xlhdr->t_infomask2;
977  header->t_hoff = xlhdr->t_hoff;
978 
979  /*
980  * Reset toast reassembly state only after the last row in the last
981  * xl_multi_insert_tuple record emitted by one heap_multi_insert()
982  * call.
983  */
984  if (xlrec->flags & XLH_INSERT_LAST_IN_MULTI &&
985  (i + 1) == xlrec->ntuples)
986  change->data.tp.clear_toast_afterwards = true;
987  else
988  change->data.tp.clear_toast_afterwards = false;
989 
991  buf->origptr, change, false);
992 
993  /* move to the next xl_multi_insert_tuple entry */
994  data += datalen;
995  }
996  Assert(data == tupledata + tuplelen);
997 }
#define SizeofHeapTupleHeader
Definition: htup_details.h:184
struct ReorderBuffer * reorder
Definition: logical.h:42
ReorderBufferChange * ReorderBufferGetChange(ReorderBuffer *rb)
ReplicationSlotPersistentData data
Definition: slot.h:143
static bool FilterByOrigin(LogicalDecodingContext *ctx, RepOriginId origin_id)
Definition: decode.c:524
enum ReorderBufferChangeType action
Definition: reorderbuffer.h:83
#define XLogRecGetOrigin(decoder)
Definition: xlogreader.h:308
HeapTupleHeader t_data
Definition: htup.h:68
#define XLogRecGetData(decoder)
Definition: xlogreader.h:310
#define XLH_INSERT_CONTAINS_NEW_TUPLE
Definition: heapam_xlog.h:69
ItemPointerData t_self
Definition: htup.h:65
uint32 t_len
Definition: htup.h:64
HeapTupleData tuple
Definition: reorderbuffer.h:29
Oid t_tableOid
Definition: htup.h:66
XLogReaderState * record
Definition: decode.c:50
RepOriginId origin_id
Definition: reorderbuffer.h:88
#define XLH_INSERT_LAST_IN_MULTI
Definition: heapam_xlog.h:67
bool XLogRecGetBlockTag(XLogReaderState *record, uint8 block_id, RelFileNode *rnode, ForkNumber *forknum, BlockNumber *blknum)
Definition: xlogreader.c:1498
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:307
char * XLogRecGetBlockData(XLogReaderState *record, uint8 block_id, Size *len)
Definition: xlogreader.c:1522
void ReorderBufferQueueChange(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, ReorderBufferChange *change, bool toast_insert)
union ReorderBufferChange::@99 data
#define InvalidOid
Definition: postgres_ext.h:36
struct ReorderBufferChange::@99::@100 tp
#define Assert(condition)
Definition: c.h:745
size_t Size
Definition: c.h:473
static void header(const char *fmt,...) pg_attribute_printf(1
Definition: pg_regress.c:210
XLogRecPtr origptr
Definition: decode.c:48
#define ItemPointerSetInvalid(pointer)
Definition: itemptr.h:172
ReplicationSlot * slot
Definition: logical.h:38
int i
#define SHORTALIGN(LEN)
Definition: c.h:694
#define SizeOfMultiInsertTuple
Definition: heapam_xlog.h:190
ReorderBufferTupleBuf * ReorderBufferGetTupleBuf(ReorderBuffer *rb, Size tuple_len)

◆ DecodeSpecConfirm()

static void DecodeSpecConfirm ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)
static

Definition at line 1006 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().

1007 {
1008  XLogReaderState *r = buf->record;
1009  ReorderBufferChange *change;
1010  RelFileNode target_node;
1011 
1012  /* only interested in our database */
1013  XLogRecGetBlockTag(r, 0, &target_node, NULL, NULL);
1014  if (target_node.dbNode != ctx->slot->data.database)
1015  return;
1016 
1017  /* output plugin doesn't look for this origin, no need to queue */
1018  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
1019  return;
1020 
1021  change = ReorderBufferGetChange(ctx->reorder);
1023  change->origin_id = XLogRecGetOrigin(r);
1024 
1025  memcpy(&change->data.tp.relnode, &target_node, sizeof(RelFileNode));
1026 
1027  change->data.tp.clear_toast_afterwards = true;
1028 
1030  change, false);
1031 }
struct ReorderBuffer * reorder
Definition: logical.h:42
ReorderBufferChange * ReorderBufferGetChange(ReorderBuffer *rb)
ReplicationSlotPersistentData data
Definition: slot.h:143
static bool FilterByOrigin(LogicalDecodingContext *ctx, RepOriginId origin_id)
Definition: decode.c:524
enum ReorderBufferChangeType action
Definition: reorderbuffer.h:83
#define XLogRecGetOrigin(decoder)
Definition: xlogreader.h:308
XLogReaderState * record
Definition: decode.c:50
RepOriginId origin_id
Definition: reorderbuffer.h:88
bool XLogRecGetBlockTag(XLogReaderState *record, uint8 block_id, RelFileNode *rnode, ForkNumber *forknum, BlockNumber *blknum)
Definition: xlogreader.c:1498
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:307
void ReorderBufferQueueChange(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, ReorderBufferChange *change, bool toast_insert)
union ReorderBufferChange::@99 data
struct ReorderBufferChange::@99::@100 tp
XLogRecPtr origptr
Definition: decode.c:48
ReplicationSlot * slot
Definition: logical.h:38

◆ DecodeStandbyOp()

static void DecodeStandbyOp ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)
static

Definition at line 335 of file decode.c.

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

Referenced by LogicalDecodingProcessRecord().

336 {
337  SnapBuild *builder = ctx->snapshot_builder;
338  XLogReaderState *r = buf->record;
339  uint8 info = XLogRecGetInfo(r) & ~XLR_INFO_MASK;
340 
342 
343  switch (info)
344  {
345  case XLOG_RUNNING_XACTS:
346  {
348 
349  SnapBuildProcessRunningXacts(builder, buf->origptr, running);
350 
351  /*
352  * Abort all transactions that we keep track of, that are
353  * older than the record's oldestRunningXid. This is the most
354  * convenient spot for doing so since, in contrast to shutdown
355  * or end-of-recovery checkpoints, we have information about
356  * all running transactions which includes prepared ones,
357  * while shutdown checkpoints just know that no non-prepared
358  * transactions are in progress.
359  */
361  }
362  break;
363  case XLOG_STANDBY_LOCK:
364  break;
365  case XLOG_INVALIDATIONS:
366 
367  /*
368  * We are processing the invalidations at the command level via
369  * XLOG_XACT_INVALIDATIONS. So we don't need to do anything here.
370  */
371  break;
372  default:
373  elog(ERROR, "unexpected RM_STANDBY_ID record type: %u", info);
374  }
375 }
struct ReorderBuffer * reorder
Definition: logical.h:42
void ReorderBufferAbortOld(ReorderBuffer *rb, TransactionId oldestRunningXid)
#define XLOG_INVALIDATIONS
Definition: standbydefs.h:36
unsigned char uint8
Definition: c.h:372
#define XLOG_STANDBY_LOCK
Definition: standbydefs.h:34
#define XLogRecGetData(decoder)
Definition: xlogreader.h:310
#define ERROR
Definition: elog.h:43
void SnapBuildProcessRunningXacts(SnapBuild *builder, XLogRecPtr lsn, xl_running_xacts *running)
Definition: snapbuild.c:1091
XLogReaderState * record
Definition: decode.c:50
#define XLogRecGetInfo(decoder)
Definition: xlogreader.h:305
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:307
struct SnapBuild * snapshot_builder
Definition: logical.h:43
#define XLR_INFO_MASK
Definition: xlogrecord.h:62
XLogRecPtr origptr
Definition: decode.c:48
#define elog(elevel,...)
Definition: elog.h:214
TransactionId oldestRunningXid
Definition: standbydefs.h:53
#define XLOG_RUNNING_XACTS
Definition: standbydefs.h:35
void ReorderBufferProcessXid(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)

◆ DecodeTruncate()

static void DecodeTruncate ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)
static

Definition at line 862 of file decode.c.

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

Referenced by DecodeHeapOp().

863 {
864  XLogReaderState *r = buf->record;
865  xl_heap_truncate *xlrec;
866  ReorderBufferChange *change;
867 
868  xlrec = (xl_heap_truncate *) XLogRecGetData(r);
869 
870  /* only interested in our database */
871  if (xlrec->dbId != ctx->slot->data.database)
872  return;
873 
874  /* output plugin doesn't look for this origin, no need to queue */
875  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
876  return;
877 
878  change = ReorderBufferGetChange(ctx->reorder);
880  change->origin_id = XLogRecGetOrigin(r);
881  if (xlrec->flags & XLH_TRUNCATE_CASCADE)
882  change->data.truncate.cascade = true;
883  if (xlrec->flags & XLH_TRUNCATE_RESTART_SEQS)
884  change->data.truncate.restart_seqs = true;
885  change->data.truncate.nrelids = xlrec->nrelids;
886  change->data.truncate.relids = ReorderBufferGetRelids(ctx->reorder,
887  xlrec->nrelids);
888  memcpy(change->data.truncate.relids, xlrec->relids,
889  xlrec->nrelids * sizeof(Oid));
891  buf->origptr, change, false);
892 }
#define XLH_TRUNCATE_CASCADE
Definition: heapam_xlog.h:117
struct ReorderBuffer * reorder
Definition: logical.h:42
struct ReorderBufferChange::@99::@101 truncate
ReorderBufferChange * ReorderBufferGetChange(ReorderBuffer *rb)
ReplicationSlotPersistentData data
Definition: slot.h:143
unsigned int Oid
Definition: postgres_ext.h:31
static bool FilterByOrigin(LogicalDecodingContext *ctx, RepOriginId origin_id)
Definition: decode.c:524
enum ReorderBufferChangeType action
Definition: reorderbuffer.h:83
#define XLogRecGetOrigin(decoder)
Definition: xlogreader.h:308
Oid relids[FLEXIBLE_ARRAY_MEMBER]
Definition: heapam_xlog.h:130
#define XLogRecGetData(decoder)
Definition: xlogreader.h:310
XLogReaderState * record
Definition: decode.c:50
RepOriginId origin_id
Definition: reorderbuffer.h:88
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:307
void ReorderBufferQueueChange(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, ReorderBufferChange *change, bool toast_insert)
union ReorderBufferChange::@99 data
#define XLH_TRUNCATE_RESTART_SEQS
Definition: heapam_xlog.h:118
XLogRecPtr origptr
Definition: decode.c:48
ReplicationSlot * slot
Definition: logical.h:38
Oid * ReorderBufferGetRelids(ReorderBuffer *rb, int nrelids)

◆ DecodeUpdate()

static void DecodeUpdate ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)
static

Definition at line 739 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().

740 {
741  XLogReaderState *r = buf->record;
742  xl_heap_update *xlrec;
743  ReorderBufferChange *change;
744  char *data;
745  RelFileNode target_node;
746 
747  xlrec = (xl_heap_update *) XLogRecGetData(r);
748 
749  /* only interested in our database */
750  XLogRecGetBlockTag(r, 0, &target_node, NULL, NULL);
751  if (target_node.dbNode != ctx->slot->data.database)
752  return;
753 
754  /* output plugin doesn't look for this origin, no need to queue */
755  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
756  return;
757 
758  change = ReorderBufferGetChange(ctx->reorder);
760  change->origin_id = XLogRecGetOrigin(r);
761  memcpy(&change->data.tp.relnode, &target_node, sizeof(RelFileNode));
762 
764  {
765  Size datalen;
766  Size tuplelen;
767 
768  data = XLogRecGetBlockData(r, 0, &datalen);
769 
770  tuplelen = datalen - SizeOfHeapHeader;
771 
772  change->data.tp.newtuple =
773  ReorderBufferGetTupleBuf(ctx->reorder, tuplelen);
774 
775  DecodeXLogTuple(data, datalen, change->data.tp.newtuple);
776  }
777 
778  if (xlrec->flags & XLH_UPDATE_CONTAINS_OLD)
779  {
780  Size datalen;
781  Size tuplelen;
782 
783  /* caution, remaining data in record is not aligned */
784  data = XLogRecGetData(r) + SizeOfHeapUpdate;
785  datalen = XLogRecGetDataLen(r) - SizeOfHeapUpdate;
786  tuplelen = datalen - SizeOfHeapHeader;
787 
788  change->data.tp.oldtuple =
789  ReorderBufferGetTupleBuf(ctx->reorder, tuplelen);
790 
791  DecodeXLogTuple(data, datalen, change->data.tp.oldtuple);
792  }
793 
794  change->data.tp.clear_toast_afterwards = true;
795 
797  change, false);
798 }
struct ReorderBuffer * reorder
Definition: logical.h:42
static void DecodeXLogTuple(char *data, Size len, ReorderBufferTupleBuf *tup)
Definition: decode.c:1042
ReorderBufferChange * ReorderBufferGetChange(ReorderBuffer *rb)
ReplicationSlotPersistentData data
Definition: slot.h:143
static bool FilterByOrigin(LogicalDecodingContext *ctx, RepOriginId origin_id)
Definition: decode.c:524
enum ReorderBufferChangeType action
Definition: reorderbuffer.h:83
#define XLogRecGetOrigin(decoder)
Definition: xlogreader.h:308
#define XLH_UPDATE_CONTAINS_NEW_TUPLE
Definition: heapam_xlog.h:81
#define SizeOfHeapUpdate
Definition: heapam_xlog.h:224
#define XLogRecGetData(decoder)
Definition: xlogreader.h:310
#define XLogRecGetDataLen(decoder)
Definition: xlogreader.h:311
XLogReaderState * record
Definition: decode.c:50
RepOriginId origin_id
Definition: reorderbuffer.h:88
bool XLogRecGetBlockTag(XLogReaderState *record, uint8 block_id, RelFileNode *rnode, ForkNumber *forknum, BlockNumber *blknum)
Definition: xlogreader.c:1498
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:307
char * XLogRecGetBlockData(XLogReaderState *record, uint8 block_id, Size *len)
Definition: xlogreader.c:1522
void ReorderBufferQueueChange(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, ReorderBufferChange *change, bool toast_insert)
union ReorderBufferChange::@99 data
struct ReorderBufferChange::@99::@100 tp
size_t Size
Definition: c.h:473
XLogRecPtr origptr
Definition: decode.c:48
ReplicationSlot * slot
Definition: logical.h:38
#define XLH_UPDATE_CONTAINS_OLD
Definition: heapam_xlog.h:86
#define SizeOfHeapHeader
Definition: heapam_xlog.h:148
ReorderBufferTupleBuf * ReorderBufferGetTupleBuf(ReorderBuffer *rb, Size tuple_len)

◆ DecodeXactOp()

static void DecodeXactOp ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)
static

Definition at line 225 of file decode.c.

References DecodeAbort(), DecodeCommit(), elog, ERROR, LogicalDecodingContext::fast_forward, xl_xact_invals::msgs, xl_xact_invals::nmsgs, XLogRecordBuffer::origptr, ParseAbortRecord(), ParseCommitRecord(), XLogRecordBuffer::record, LogicalDecodingContext::reorder, ReorderBufferAddInvalidations(), ReorderBufferImmediateInvalidation(), ReorderBufferProcessXid(), ReorderBufferXidSetCatalogChanges(), 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_INVALIDATIONS, XLOG_XACT_OPMASK, XLOG_XACT_PREPARE, XLogRecGetData, XLogRecGetInfo, and XLogRecGetXid.

Referenced by LogicalDecodingProcessRecord().

226 {
227  SnapBuild *builder = ctx->snapshot_builder;
228  ReorderBuffer *reorder = ctx->reorder;
229  XLogReaderState *r = buf->record;
231 
232  /*
233  * If the snapshot isn't yet fully built, we cannot decode anything, so
234  * bail out.
235  */
237  return;
238 
239  switch (info)
240  {
241  case XLOG_XACT_COMMIT:
243  {
244  xl_xact_commit *xlrec;
245  xl_xact_parsed_commit parsed;
246  TransactionId xid;
247 
248  xlrec = (xl_xact_commit *) XLogRecGetData(r);
249  ParseCommitRecord(XLogRecGetInfo(buf->record), xlrec, &parsed);
250 
251  if (!TransactionIdIsValid(parsed.twophase_xid))
252  xid = XLogRecGetXid(r);
253  else
254  xid = parsed.twophase_xid;
255 
256  DecodeCommit(ctx, buf, &parsed, xid);
257  break;
258  }
259  case XLOG_XACT_ABORT:
261  {
262  xl_xact_abort *xlrec;
263  xl_xact_parsed_abort parsed;
264  TransactionId xid;
265 
266  xlrec = (xl_xact_abort *) XLogRecGetData(r);
267  ParseAbortRecord(XLogRecGetInfo(buf->record), xlrec, &parsed);
268 
269  if (!TransactionIdIsValid(parsed.twophase_xid))
270  xid = XLogRecGetXid(r);
271  else
272  xid = parsed.twophase_xid;
273 
274  DecodeAbort(ctx, buf, &parsed, xid);
275  break;
276  }
278 
279  /*
280  * We assign subxact to the toplevel xact while processing each
281  * record if required. So, we don't need to do anything here. See
282  * LogicalDecodingProcessRecord.
283  */
284  break;
286  {
287  TransactionId xid;
288  xl_xact_invals *invals;
289 
290  xid = XLogRecGetXid(r);
291  invals = (xl_xact_invals *) XLogRecGetData(r);
292 
293  /*
294  * Execute the invalidations for xid-less transactions,
295  * otherwise, accumulate them so that they can be processed at
296  * the commit time.
297  */
298  if (TransactionIdIsValid(xid))
299  {
300  if (!ctx->fast_forward)
301  ReorderBufferAddInvalidations(reorder, xid,
302  buf->origptr,
303  invals->nmsgs,
304  invals->msgs);
306  buf->origptr);
307  }
308  else if ((!ctx->fast_forward))
310  invals->nmsgs,
311  invals->msgs);
312  }
313  break;
314  case XLOG_XACT_PREPARE:
315 
316  /*
317  * Currently decoding ignores PREPARE TRANSACTION and will just
318  * decode the transaction when the COMMIT PREPARED is sent or
319  * throw away the transaction's contents when a ROLLBACK PREPARED
320  * is received. In the future we could add code to expose prepared
321  * transactions in the changestream allowing for a kind of
322  * distributed 2PC.
323  */
324  ReorderBufferProcessXid(reorder, XLogRecGetXid(r), buf->origptr);
325  break;
326  default:
327  elog(ERROR, "unexpected RM_XACT_ID record type: %u", info);
328  }
329 }
#define XLOG_XACT_COMMIT
Definition: xact.h:147
static void DecodeCommit(LogicalDecodingContext *ctx, XLogRecordBuffer *buf, xl_xact_parsed_commit *parsed, TransactionId xid)
Definition: decode.c:587
struct ReorderBuffer * reorder
Definition: logical.h:42
#define XLOG_XACT_PREPARE
Definition: xact.h:148
uint32 TransactionId
Definition: c.h:520
SnapBuildState SnapBuildCurrentState(SnapBuild *builder)
Definition: snapbuild.c:395
#define XLOG_XACT_INVALIDATIONS
Definition: xact.h:153
TransactionId twophase_xid
Definition: xact.h:339
unsigned char uint8
Definition: c.h:372
static void DecodeAbort(LogicalDecodingContext *ctx, XLogRecordBuffer *buf, xl_xact_parsed_abort *parsed, TransactionId xid)
Definition: decode.c:660
#define XLogRecGetData(decoder)
Definition: xlogreader.h:310
#define ERROR
Definition: elog.h:43
SharedInvalidationMessage msgs[FLEXIBLE_ARRAY_MEMBER]
Definition: xact.h:254
void ReorderBufferImmediateInvalidation(ReorderBuffer *rb, uint32 ninvalidations, SharedInvalidationMessage *invalidations)
void ParseCommitRecord(uint8 info, xl_xact_commit *xlrec, xl_xact_parsed_commit *parsed)
Definition: xactdesc.c:34
#define XLOG_XACT_ABORT_PREPARED
Definition: xact.h:151
XLogReaderState * record
Definition: decode.c:50
#define XLogRecGetInfo(decoder)
Definition: xlogreader.h:305
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:307
#define XLOG_XACT_ASSIGNMENT
Definition: xact.h:152
TransactionId twophase_xid
Definition: xact.h:364
struct SnapBuild * snapshot_builder
Definition: logical.h:43
#define XLOG_XACT_ABORT
Definition: xact.h:149
void ReorderBufferAddInvalidations(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, Size nmsgs, SharedInvalidationMessage *msgs)
#define XLOG_XACT_OPMASK
Definition: xact.h:157
void ParseAbortRecord(uint8 info, xl_xact_abort *xlrec, xl_xact_parsed_abort *parsed)
Definition: xactdesc.c:129
XLogRecPtr origptr
Definition: decode.c:48
#define elog(elevel,...)
Definition: elog.h:214
void ReorderBufferXidSetCatalogChanges(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
#define TransactionIdIsValid(xid)
Definition: transam.h:41
#define XLOG_XACT_COMMIT_PREPARED
Definition: xact.h:150
void ReorderBufferProcessXid(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
int nmsgs
Definition: xact.h:253

◆ DecodeXLogOp()

static void DecodeXLogOp ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)
static

Definition at line 183 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().

184 {
185  SnapBuild *builder = ctx->snapshot_builder;
186  uint8 info = XLogRecGetInfo(buf->record) & ~XLR_INFO_MASK;
187 
189  buf->origptr);
190 
191  switch (info)
192  {
193  /* this is also used in END_OF_RECOVERY checkpoints */
196  SnapBuildSerializationPoint(builder, buf->origptr);
197 
198  break;
200 
201  /*
202  * a RUNNING_XACTS record will have been logged near to this, we
203  * can restart from there.
204  */
205  break;
206  case XLOG_NOOP:
207  case XLOG_NEXTOID:
208  case XLOG_SWITCH:
209  case XLOG_BACKUP_END:
211  case XLOG_RESTORE_POINT:
212  case XLOG_FPW_CHANGE:
213  case XLOG_FPI_FOR_HINT:
214  case XLOG_FPI:
215  break;
216  default:
217  elog(ERROR, "unexpected RM_XLOG_ID record type: %u", info);
218  }
219 }
#define XLOG_FPW_CHANGE
Definition: pg_control.h:75
struct ReorderBuffer * reorder
Definition: logical.h:42
#define XLOG_CHECKPOINT_ONLINE
Definition: pg_control.h:68
unsigned char uint8
Definition: c.h:372
#define XLOG_RESTORE_POINT
Definition: pg_control.h:74
#define XLOG_END_OF_RECOVERY
Definition: pg_control.h:76
#define XLOG_NEXTOID
Definition: pg_control.h:70
#define ERROR
Definition: elog.h:43
#define XLOG_PARAMETER_CHANGE
Definition: pg_control.h:73
#define XLOG_CHECKPOINT_SHUTDOWN
Definition: pg_control.h:67
XLogReaderState * record
Definition: decode.c:50
#define XLOG_BACKUP_END
Definition: pg_control.h:72
#define XLogRecGetInfo(decoder)
Definition: xlogreader.h:305
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:307
#define XLOG_NOOP
Definition: pg_control.h:69
struct SnapBuild * snapshot_builder
Definition: logical.h:43
#define XLR_INFO_MASK
Definition: xlogrecord.h:62
#define XLOG_FPI_FOR_HINT
Definition: pg_control.h:77
#define XLOG_SWITCH
Definition: pg_control.h:71
XLogRecPtr origptr
Definition: decode.c:48
#define XLOG_FPI
Definition: pg_control.h:78
#define elog(elevel,...)
Definition: elog.h:214
void SnapBuildSerializationPoint(SnapBuild *builder, XLogRecPtr lsn)
Definition: snapbuild.c:1468
void ReorderBufferProcessXid(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)

◆ DecodeXLogTuple()

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

Definition at line 1042 of file decode.c.

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

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

1043 {
1044  xl_heap_header xlhdr;
1045  int datalen = len - SizeOfHeapHeader;
1047 
1048  Assert(datalen >= 0);
1049 
1050  tuple->tuple.t_len = datalen + SizeofHeapTupleHeader;
1051  header = tuple->tuple.t_data;
1052 
1053  /* not a disk based tuple */
1054  ItemPointerSetInvalid(&tuple->tuple.t_self);
1055 
1056  /* we can only figure this out after reassembling the transactions */
1057  tuple->tuple.t_tableOid = InvalidOid;
1058 
1059  /* data is not stored aligned, copy to aligned storage */
1060  memcpy((char *) &xlhdr,
1061  data,
1062  SizeOfHeapHeader);
1063 
1064  memset(header, 0, SizeofHeapTupleHeader);
1065 
1066  memcpy(((char *) tuple->tuple.t_data) + SizeofHeapTupleHeader,
1067  data + SizeOfHeapHeader,
1068  datalen);
1069 
1070  header->t_infomask = xlhdr.t_infomask;
1071  header->t_infomask2 = xlhdr.t_infomask2;
1072  header->t_hoff = xlhdr.t_hoff;
1073 }
#define SizeofHeapTupleHeader
Definition: htup_details.h:184
uint16 t_infomask2
Definition: heapam_xlog.h:143
#define InvalidOid
Definition: postgres_ext.h:36
#define Assert(condition)
Definition: c.h:745
uint16 t_infomask
Definition: heapam_xlog.h:144
static void header(const char *fmt,...) pg_attribute_printf(1
Definition: pg_regress.c:210
#define ItemPointerSetInvalid(pointer)
Definition: itemptr.h:172
#define SizeOfHeapHeader
Definition: heapam_xlog.h:148

◆ FilterByOrigin()

static bool FilterByOrigin ( LogicalDecodingContext ctx,
RepOriginId  origin_id 
)
inlinestatic

Definition at line 524 of file decode.c.

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

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

525 {
526  if (ctx->callbacks.filter_by_origin_cb == NULL)
527  return false;
528 
529  return filter_by_origin_cb_wrapper(ctx, origin_id);
530 }
OutputPluginCallbacks callbacks
Definition: logical.h:52
bool filter_by_origin_cb_wrapper(LogicalDecodingContext *ctx, RepOriginId origin_id)
Definition: logical.c:862
LogicalDecodeFilterByOriginCB filter_by_origin_cb

◆ LogicalDecodingProcessRecord()

void LogicalDecodingProcessRecord ( LogicalDecodingContext ctx,
XLogReaderState record 
)

Definition at line 94 of file decode.c.

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

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

95 {
97  TransactionId txid;
98 
99  buf.origptr = ctx->reader->ReadRecPtr;
100  buf.endptr = ctx->reader->EndRecPtr;
101  buf.record = record;
102 
103  txid = XLogRecGetTopXid(record);
104 
105  /*
106  * If the top-level xid is valid, we need to assign the subxact to the
107  * top-level xact. We need to do this for all records, hence we do it
108  * before the switch.
109  */
110  if (TransactionIdIsValid(txid))
111  {
113  txid,
114  record->decoded_record->xl_xid,
115  buf.origptr);
116  }
117 
118  /* cast so we get a warning when new rmgrs are added */
119  switch ((RmgrId) XLogRecGetRmid(record))
120  {
121  /*
122  * Rmgrs we care about for logical decoding. Add new rmgrs in
123  * rmgrlist.h's order.
124  */
125  case RM_XLOG_ID:
126  DecodeXLogOp(ctx, &buf);
127  break;
128 
129  case RM_XACT_ID:
130  DecodeXactOp(ctx, &buf);
131  break;
132 
133  case RM_STANDBY_ID:
134  DecodeStandbyOp(ctx, &buf);
135  break;
136 
137  case RM_HEAP2_ID:
138  DecodeHeap2Op(ctx, &buf);
139  break;
140 
141  case RM_HEAP_ID:
142  DecodeHeapOp(ctx, &buf);
143  break;
144 
145  case RM_LOGICALMSG_ID:
146  DecodeLogicalMsgOp(ctx, &buf);
147  break;
148 
149  /*
150  * Rmgrs irrelevant for logical decoding; they describe stuff not
151  * represented in logical decoding. Add new rmgrs in rmgrlist.h's
152  * order.
153  */
154  case RM_SMGR_ID:
155  case RM_CLOG_ID:
156  case RM_DBASE_ID:
157  case RM_TBLSPC_ID:
158  case RM_MULTIXACT_ID:
159  case RM_RELMAP_ID:
160  case RM_BTREE_ID:
161  case RM_HASH_ID:
162  case RM_GIN_ID:
163  case RM_GIST_ID:
164  case RM_SEQ_ID:
165  case RM_SPGIST_ID:
166  case RM_BRIN_ID:
167  case RM_COMMIT_TS_ID:
168  case RM_REPLORIGIN_ID:
169  case RM_GENERIC_ID:
170  /* just deal with xid, and done */
172  buf.origptr);
173  break;
174  case RM_NEXT_ID:
175  elog(ERROR, "unexpected RM_NEXT_ID rmgr_id: %u", (RmgrIds) XLogRecGetRmid(buf.record));
176  }
177 }
struct ReorderBuffer * reorder
Definition: logical.h:42
uint32 TransactionId
Definition: c.h:520
XLogRecPtr EndRecPtr
Definition: xlogreader.h:176
static void DecodeXactOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:225
static void DecodeLogicalMsgOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:536
#define ERROR
Definition: elog.h:43
XLogRecPtr endptr
Definition: decode.c:49
static char * buf
Definition: pg_test_fsync.c:67
XLogRecPtr ReadRecPtr
Definition: xlogreader.h:175
XLogRecord * decoded_record
Definition: xlogreader.h:186
XLogReaderState * record
Definition: decode.c:50
static void DecodeXLogOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:183
void ReorderBufferAssignChild(ReorderBuffer *rb, TransactionId xid, TransactionId subxid, XLogRecPtr lsn)
static void DecodeHeap2Op(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:381
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:307
uint8 RmgrId
Definition: rmgr.h:11
TransactionId xl_xid
Definition: xlogrecord.h:44
XLogRecPtr origptr
Definition: decode.c:48
XLogReaderState * reader
Definition: logical.h:41
#define elog(elevel,...)
Definition: elog.h:214
static void DecodeHeapOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:441
#define XLogRecGetTopXid(decoder)
Definition: xlogreader.h:309
#define TransactionIdIsValid(xid)
Definition: transam.h:41
void ReorderBufferProcessXid(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
RmgrIds
Definition: rmgr.h:25
static void DecodeStandbyOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:335
#define XLogRecGetRmid(decoder)
Definition: xlogreader.h:306