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 646 of file decode.c.

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

Referenced by DecodeXactOp().

648 {
649  int i;
650 
651  for (i = 0; i < parsed->nsubxacts; i++)
652  {
653  ReorderBufferAbort(ctx->reorder, parsed->subxacts[i],
654  buf->record->EndRecPtr);
655  }
656 
657  ReorderBufferAbort(ctx->reorder, xid, buf->record->EndRecPtr);
658 }
struct ReorderBuffer * reorder
Definition: logical.h:42
XLogRecPtr EndRecPtr
Definition: xlogreader.h:132
XLogReaderState * record
Definition: decode.c:50
TransactionId * subxacts
Definition: xact.h:355
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 560 of file decode.c.

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

Referenced by DecodeXactOp().

562 {
563  XLogRecPtr origin_lsn = InvalidXLogRecPtr;
564  TimestampTz commit_time = parsed->xact_time;
565  RepOriginId origin_id = XLogRecGetOrigin(buf->record);
566  int i;
567 
568  if (parsed->xinfo & XACT_XINFO_HAS_ORIGIN)
569  {
570  origin_lsn = parsed->origin_lsn;
571  commit_time = parsed->origin_timestamp;
572  }
573 
574  /*
575  * Process invalidation messages, even if we're not interested in the
576  * transaction's contents, since the various caches need to always be
577  * consistent.
578  */
579  if (parsed->nmsgs > 0)
580  {
581  if (!ctx->fast_forward)
583  parsed->nmsgs, parsed->msgs);
585  }
586 
588  parsed->nsubxacts, parsed->subxacts);
589 
590  /* ----
591  * Check whether we are interested in this specific transaction, and tell
592  * the reorderbuffer to forget the content of the (sub-)transactions
593  * if not.
594  *
595  * There can be several reasons we might not be interested in this
596  * transaction:
597  * 1) We might not be interested in decoding transactions up to this
598  * LSN. This can happen because we previously decoded it and now just
599  * are restarting or if we haven't assembled a consistent snapshot yet.
600  * 2) The transaction happened in another database.
601  * 3) The output plugin is not interested in the origin.
602  * 4) We are doing fast-forwarding
603  *
604  * We can't just use ReorderBufferAbort() here, because we need to execute
605  * the transaction's invalidations. This currently won't be needed if
606  * we're just skipping over the transaction because currently we only do
607  * so during startup, to get to the first transaction the client needs. As
608  * we have reset the catalog caches before starting to read WAL, and we
609  * haven't yet touched any catalogs, there can't be anything to invalidate.
610  * But if we're "forgetting" this commit because it's it happened in
611  * another database, the invalidations might be important, because they
612  * could be for shared catalogs and we might have loaded data into the
613  * relevant syscaches.
614  * ---
615  */
617  (parsed->dbId != InvalidOid && parsed->dbId != ctx->slot->data.database) ||
618  ctx->fast_forward || FilterByOrigin(ctx, origin_id))
619  {
620  for (i = 0; i < parsed->nsubxacts; i++)
621  {
622  ReorderBufferForget(ctx->reorder, parsed->subxacts[i], buf->origptr);
623  }
624  ReorderBufferForget(ctx->reorder, xid, buf->origptr);
625 
626  return;
627  }
628 
629  /* tell the reorderbuffer about the surviving subtransactions */
630  for (i = 0; i < parsed->nsubxacts; i++)
631  {
632  ReorderBufferCommitChild(ctx->reorder, xid, parsed->subxacts[i],
633  buf->origptr, buf->endptr);
634  }
635 
636  /* replay actions of all transaction + subtransactions in order */
637  ReorderBufferCommit(ctx->reorder, xid, buf->origptr, buf->endptr,
638  commit_time, origin_id, origin_lsn);
639 }
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
#define XACT_XINFO_HAS_ORIGIN
Definition: xact.h:167
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:341
void SnapBuildCommitTxn(SnapBuild *builder, XLogRecPtr lsn, TransactionId xid, int nsubxacts, TransactionId *subxacts)
Definition: snapbuild.c:923
int64 TimestampTz
Definition: timestamp.h:39
TransactionId * subxacts
Definition: xact.h:327
uint16 RepOriginId
Definition: xlogdefs.h:58
SharedInvalidationMessage * msgs
Definition: xact.h:333
XLogRecPtr origin_lsn
Definition: xact.h:340
ReplicationSlotPersistentData data
Definition: slot.h:132
static bool FilterByOrigin(LogicalDecodingContext *ctx, RepOriginId origin_id)
Definition: decode.c:497
#define XLogRecGetOrigin(decoder)
Definition: xlogreader.h:282
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
void ReorderBufferAddInvalidations(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, Size nmsgs, SharedInvalidationMessage *msgs)
XLogRecPtr origptr
Definition: decode.c:48
ReplicationSlot * slot
Definition: logical.h:38
int i
void ReorderBufferXidSetCatalogChanges(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
TimestampTz xact_time
Definition: xact.h:320

◆ DecodeDelete()

static void DecodeDelete ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)
static

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

790 {
791  XLogReaderState *r = buf->record;
792  xl_heap_delete *xlrec;
793  ReorderBufferChange *change;
794  RelFileNode target_node;
795 
796  xlrec = (xl_heap_delete *) XLogRecGetData(r);
797 
798  /* only interested in our database */
799  XLogRecGetBlockTag(r, 0, &target_node, NULL, NULL);
800  if (target_node.dbNode != ctx->slot->data.database)
801  return;
802 
803  /*
804  * Super deletions are irrelevant for logical decoding, it's driven by the
805  * confirmation records.
806  */
807  if (xlrec->flags & XLH_DELETE_IS_SUPER)
808  return;
809 
810  /* output plugin doesn't look for this origin, no need to queue */
811  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
812  return;
813 
814  change = ReorderBufferGetChange(ctx->reorder);
816  change->origin_id = XLogRecGetOrigin(r);
817 
818  memcpy(&change->data.tp.relnode, &target_node, sizeof(RelFileNode));
819 
820  /* old primary key stored */
821  if (xlrec->flags & XLH_DELETE_CONTAINS_OLD)
822  {
823  Size datalen = XLogRecGetDataLen(r) - SizeOfHeapDelete;
824  Size tuplelen = datalen - SizeOfHeapHeader;
825 
826  Assert(XLogRecGetDataLen(r) > (SizeOfHeapDelete + SizeOfHeapHeader));
827 
828  change->data.tp.oldtuple =
829  ReorderBufferGetTupleBuf(ctx->reorder, tuplelen);
830 
831  DecodeXLogTuple((char *) xlrec + SizeOfHeapDelete,
832  datalen, change->data.tp.oldtuple);
833  }
834 
835  change->data.tp.clear_toast_afterwards = true;
836 
837  ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r), buf->origptr, change);
838 }
#define XLH_DELETE_IS_SUPER
Definition: heapam_xlog.h:95
struct ReorderBuffer * reorder
Definition: logical.h:42
struct ReorderBufferChange::@101::@102 tp
static void DecodeXLogTuple(char *data, Size len, ReorderBufferTupleBuf *tup)
Definition: decode.c:1028
#define XLH_DELETE_CONTAINS_OLD
Definition: heapam_xlog.h:99
ReorderBufferChange * ReorderBufferGetChange(ReorderBuffer *rb)
ReplicationSlotPersistentData data
Definition: slot.h:132
static bool FilterByOrigin(LogicalDecodingContext *ctx, RepOriginId origin_id)
Definition: decode.c:497
enum ReorderBufferChangeType action
Definition: reorderbuffer.h:83
#define XLogRecGetOrigin(decoder)
Definition: xlogreader.h:282
#define XLogRecGetData(decoder)
Definition: xlogreader.h:283
#define XLogRecGetDataLen(decoder)
Definition: xlogreader.h:284
#define SizeOfHeapDelete
Definition: heapam_xlog.h:111
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:1460
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:281
#define Assert(condition)
Definition: c.h:739
union ReorderBufferChange::@101 data
size_t Size
Definition: c.h:467
void ReorderBufferQueueChange(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, ReorderBufferChange *change)
XLogRecPtr origptr
Definition: decode.c:48
ReplicationSlot * slot
Definition: logical.h:38
#define SizeOfHeapHeader
Definition: heapam_xlog.h:149
ReorderBufferTupleBuf * ReorderBufferGetTupleBuf(ReorderBuffer *rb, Size tuple_len)

◆ DecodeHeap2Op()

static void DecodeHeap2Op ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)
static

Definition at line 354 of file decode.c.

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

Referenced by LogicalDecodingProcessRecord().

355 {
357  TransactionId xid = XLogRecGetXid(buf->record);
358  SnapBuild *builder = ctx->snapshot_builder;
359 
360  ReorderBufferProcessXid(ctx->reorder, xid, buf->origptr);
361 
362  /*
363  * If we don't have snapshot or we are just fast-forwarding, there is no
364  * point in decoding changes.
365  */
367  ctx->fast_forward)
368  return;
369 
370  switch (info)
371  {
373  if (!ctx->fast_forward &&
374  SnapBuildProcessChange(builder, xid, buf->origptr))
375  DecodeMultiInsert(ctx, buf);
376  break;
377  case XLOG_HEAP2_NEW_CID:
378  {
379  xl_heap_new_cid *xlrec;
380 
381  xlrec = (xl_heap_new_cid *) XLogRecGetData(buf->record);
382  SnapBuildProcessNewCid(builder, xid, buf->origptr, xlrec);
383 
384  break;
385  }
386  case XLOG_HEAP2_REWRITE:
387 
388  /*
389  * Although these records only exist to serve the needs of logical
390  * decoding, all the work happens as part of crash or archive
391  * recovery, so we don't need to do anything here.
392  */
393  break;
394 
395  /*
396  * Everything else here is just low level physical stuff we're not
397  * interested in.
398  */
400  case XLOG_HEAP2_CLEAN:
402  case XLOG_HEAP2_VISIBLE:
404  break;
405  default:
406  elog(ERROR, "unexpected RM_HEAP2_ID record type: %u", info);
407  }
408 }
static void DecodeMultiInsert(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:882
struct ReorderBuffer * reorder
Definition: logical.h:42
#define XLOG_HEAP2_LOCK_UPDATED
Definition: heapam_xlog.h:59
uint32 TransactionId
Definition: c.h:514
SnapBuildState SnapBuildCurrentState(SnapBuild *builder)
Definition: snapbuild.c:395
#define XLOG_HEAP2_REWRITE
Definition: heapam_xlog.h:53
unsigned char uint8
Definition: c.h:357
#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:715
#define XLOG_HEAP2_VISIBLE
Definition: heapam_xlog.h:57
#define XLogRecGetData(decoder)
Definition: xlogreader.h:283
#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:765
#define XLOG_HEAP2_NEW_CID
Definition: heapam_xlog.h:60
XLogReaderState * record
Definition: decode.c:50
#define XLogRecGetInfo(decoder)
Definition: xlogreader.h:279
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:281
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:228
void ReorderBufferProcessXid(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)

◆ DecodeHeapOp()

static void DecodeHeapOp ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)
static

Definition at line 414 of file decode.c.

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

Referenced by LogicalDecodingProcessRecord().

415 {
417  TransactionId xid = XLogRecGetXid(buf->record);
418  SnapBuild *builder = ctx->snapshot_builder;
419 
420  ReorderBufferProcessXid(ctx->reorder, xid, buf->origptr);
421 
422  /*
423  * If we don't have snapshot or we are just fast-forwarding, there is no
424  * point in decoding data changes.
425  */
427  ctx->fast_forward)
428  return;
429 
430  switch (info)
431  {
432  case XLOG_HEAP_INSERT:
433  if (SnapBuildProcessChange(builder, xid, buf->origptr))
434  DecodeInsert(ctx, buf);
435  break;
436 
437  /*
438  * Treat HOT update as normal updates. There is no useful
439  * information in the fact that we could make it a HOT update
440  * locally and the WAL layout is compatible.
441  */
443  case XLOG_HEAP_UPDATE:
444  if (SnapBuildProcessChange(builder, xid, buf->origptr))
445  DecodeUpdate(ctx, buf);
446  break;
447 
448  case XLOG_HEAP_DELETE:
449  if (SnapBuildProcessChange(builder, xid, buf->origptr))
450  DecodeDelete(ctx, buf);
451  break;
452 
453  case XLOG_HEAP_TRUNCATE:
454  if (SnapBuildProcessChange(builder, xid, buf->origptr))
455  DecodeTruncate(ctx, buf);
456  break;
457 
458  case XLOG_HEAP_INPLACE:
459 
460  /*
461  * Inplace updates are only ever performed on catalog tuples and
462  * can, per definition, not change tuple visibility. Since we
463  * don't decode catalog tuples, we're not interested in the
464  * record's contents.
465  *
466  * In-place updates can be used either by XID-bearing transactions
467  * (e.g. in CREATE INDEX CONCURRENTLY) or by XID-less
468  * transactions (e.g. VACUUM). In the former case, the commit
469  * record will include cache invalidations, so we mark the
470  * transaction as catalog modifying here. Currently that's
471  * redundant because the commit will do that as well, but once we
472  * support decoding in-progress relations, this will be important.
473  */
474  if (!TransactionIdIsValid(xid))
475  break;
476 
477  SnapBuildProcessChange(builder, xid, buf->origptr);
479  break;
480 
481  case XLOG_HEAP_CONFIRM:
482  if (SnapBuildProcessChange(builder, xid, buf->origptr))
483  DecodeSpecConfirm(ctx, buf);
484  break;
485 
486  case XLOG_HEAP_LOCK:
487  /* we don't care about row level locks for now */
488  break;
489 
490  default:
491  elog(ERROR, "unexpected RM_HEAP_ID record type: %u", info);
492  break;
493  }
494 }
#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:666
uint32 TransactionId
Definition: c.h:514
SnapBuildState SnapBuildCurrentState(SnapBuild *builder)
Definition: snapbuild.c:395
unsigned char uint8
Definition: c.h:357
#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:723
bool SnapBuildProcessChange(SnapBuild *builder, TransactionId xid, XLogRecPtr lsn)
Definition: snapbuild.c:715
static void DecodeSpecConfirm(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:993
#define ERROR
Definition: elog.h:43
static void DecodeDelete(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:789
XLogReaderState * record
Definition: decode.c:50
#define XLogRecGetInfo(decoder)
Definition: xlogreader.h:279
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:281
#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:844
#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:228
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 666 of file decode.c.

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

Referenced by DecodeHeapOp().

667 {
668  Size datalen;
669  char *tupledata;
670  Size tuplelen;
671  XLogReaderState *r = buf->record;
672  xl_heap_insert *xlrec;
673  ReorderBufferChange *change;
674  RelFileNode target_node;
675 
676  xlrec = (xl_heap_insert *) XLogRecGetData(r);
677 
678  /*
679  * Ignore insert records without new tuples (this does happen when
680  * raw_heap_insert marks the TOAST record as HEAP_INSERT_NO_LOGICAL).
681  */
682  if (!(xlrec->flags & XLH_INSERT_CONTAINS_NEW_TUPLE))
683  return;
684 
685  /* only interested in our database */
686  XLogRecGetBlockTag(r, 0, &target_node, NULL, NULL);
687  if (target_node.dbNode != ctx->slot->data.database)
688  return;
689 
690  /* output plugin doesn't look for this origin, no need to queue */
691  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
692  return;
693 
694  change = ReorderBufferGetChange(ctx->reorder);
695  if (!(xlrec->flags & XLH_INSERT_IS_SPECULATIVE))
697  else
699  change->origin_id = XLogRecGetOrigin(r);
700 
701  memcpy(&change->data.tp.relnode, &target_node, sizeof(RelFileNode));
702 
703  tupledata = XLogRecGetBlockData(r, 0, &datalen);
704  tuplelen = datalen - SizeOfHeapHeader;
705 
706  change->data.tp.newtuple =
707  ReorderBufferGetTupleBuf(ctx->reorder, tuplelen);
708 
709  DecodeXLogTuple(tupledata, datalen, change->data.tp.newtuple);
710 
711  change->data.tp.clear_toast_afterwards = true;
712 
713  ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r), buf->origptr, change);
714 }
struct ReorderBuffer * reorder
Definition: logical.h:42
struct ReorderBufferChange::@101::@102 tp
static void DecodeXLogTuple(char *data, Size len, ReorderBufferTupleBuf *tup)
Definition: decode.c:1028
#define XLH_INSERT_IS_SPECULATIVE
Definition: heapam_xlog.h:68
ReorderBufferChange * ReorderBufferGetChange(ReorderBuffer *rb)
ReplicationSlotPersistentData data
Definition: slot.h:132
static bool FilterByOrigin(LogicalDecodingContext *ctx, RepOriginId origin_id)
Definition: decode.c:497
enum ReorderBufferChangeType action
Definition: reorderbuffer.h:83
#define XLogRecGetOrigin(decoder)
Definition: xlogreader.h:282
#define XLogRecGetData(decoder)
Definition: xlogreader.h:283
#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:1460
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:281
char * XLogRecGetBlockData(XLogReaderState *record, uint8 block_id, Size *len)
Definition: xlogreader.c:1484
union ReorderBufferChange::@101 data
size_t Size
Definition: c.h:467
void ReorderBufferQueueChange(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, ReorderBufferChange *change)
XLogRecPtr origptr
Definition: decode.c:48
ReplicationSlot * slot
Definition: logical.h:38
#define SizeOfHeapHeader
Definition: heapam_xlog.h:149
ReorderBufferTupleBuf * ReorderBufferGetTupleBuf(ReorderBuffer *rb, Size tuple_len)

◆ DecodeLogicalMsgOp()

static void DecodeLogicalMsgOp ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)
static

Definition at line 509 of file decode.c.

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

Referenced by LogicalDecodingProcessRecord().

510 {
511  SnapBuild *builder = ctx->snapshot_builder;
512  XLogReaderState *r = buf->record;
513  TransactionId xid = XLogRecGetXid(r);
514  uint8 info = XLogRecGetInfo(r) & ~XLR_INFO_MASK;
515  RepOriginId origin_id = XLogRecGetOrigin(r);
516  Snapshot snapshot;
517  xl_logical_message *message;
518 
519  if (info != XLOG_LOGICAL_MESSAGE)
520  elog(ERROR, "unexpected RM_LOGICALMSG_ID record type: %u", info);
521 
523 
524  /*
525  * If we don't have snapshot or we are just fast-forwarding, there is no
526  * point in decoding messages.
527  */
529  ctx->fast_forward)
530  return;
531 
532  message = (xl_logical_message *) XLogRecGetData(r);
533 
534  if (message->dbId != ctx->slot->data.database ||
535  FilterByOrigin(ctx, origin_id))
536  return;
537 
538  if (message->transactional &&
539  !SnapBuildProcessChange(builder, xid, buf->origptr))
540  return;
541  else if (!message->transactional &&
543  SnapBuildXactNeedsSkip(builder, buf->origptr)))
544  return;
545 
546  snapshot = SnapBuildGetOrBuildSnapshot(builder, xid);
547  ReorderBufferQueueMessage(ctx->reorder, xid, snapshot, buf->endptr,
548  message->transactional,
549  message->message, /* first part of message is
550  * prefix */
551  message->message_size,
552  message->message + message->prefix_size);
553 }
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:514
bool SnapBuildXactNeedsSkip(SnapBuild *builder, XLogRecPtr ptr)
Definition: snapbuild.c:404
SnapBuildState SnapBuildCurrentState(SnapBuild *builder)
Definition: snapbuild.c:395
unsigned char uint8
Definition: c.h:357
uint16 RepOriginId
Definition: xlogdefs.h:58
ReplicationSlotPersistentData data
Definition: slot.h:132
static bool FilterByOrigin(LogicalDecodingContext *ctx, RepOriginId origin_id)
Definition: decode.c:497
#define XLogRecGetOrigin(decoder)
Definition: xlogreader.h:282
char message[FLEXIBLE_ARRAY_MEMBER]
Definition: message.h:26
bool SnapBuildProcessChange(SnapBuild *builder, TransactionId xid, XLogRecPtr lsn)
Definition: snapbuild.c:715
#define XLogRecGetData(decoder)
Definition: xlogreader.h:283
#define ERROR
Definition: elog.h:43
XLogRecPtr endptr
Definition: decode.c:49
XLogReaderState * record
Definition: decode.c:50
#define XLogRecGetInfo(decoder)
Definition: xlogreader.h:279
bool transactional
Definition: message.h:23
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:281
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:228
#define XLOG_LOGICAL_MESSAGE
Definition: message.h:37
void ReorderBufferProcessXid(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
Snapshot SnapBuildGetOrBuildSnapshot(SnapBuild *builder, TransactionId xid)
Definition: snapbuild.c:671

◆ DecodeMultiInsert()

static void DecodeMultiInsert ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)
static

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

883 {
884  XLogReaderState *r = buf->record;
885  xl_heap_multi_insert *xlrec;
886  int i;
887  char *data;
888  char *tupledata;
889  Size tuplelen;
890  RelFileNode rnode;
891 
892  xlrec = (xl_heap_multi_insert *) XLogRecGetData(r);
893 
894  /* only interested in our database */
895  XLogRecGetBlockTag(r, 0, &rnode, NULL, NULL);
896  if (rnode.dbNode != ctx->slot->data.database)
897  return;
898 
899  /* output plugin doesn't look for this origin, no need to queue */
900  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
901  return;
902 
903  /*
904  * As multi_insert is not used for catalogs yet, the block should always
905  * have data even if a full-page write of it is taken.
906  */
907  tupledata = XLogRecGetBlockData(r, 0, &tuplelen);
908  Assert(tupledata != NULL);
909 
910  data = tupledata;
911  for (i = 0; i < xlrec->ntuples; i++)
912  {
913  ReorderBufferChange *change;
914  xl_multi_insert_tuple *xlhdr;
915  int datalen;
916  ReorderBufferTupleBuf *tuple;
917 
918  change = ReorderBufferGetChange(ctx->reorder);
920  change->origin_id = XLogRecGetOrigin(r);
921 
922  memcpy(&change->data.tp.relnode, &rnode, sizeof(RelFileNode));
923 
924  xlhdr = (xl_multi_insert_tuple *) SHORTALIGN(data);
925  data = ((char *) xlhdr) + SizeOfMultiInsertTuple;
926  datalen = xlhdr->datalen;
927 
928  /*
929  * CONTAINS_NEW_TUPLE will always be set currently as multi_insert
930  * isn't used for catalogs, but better be future proof.
931  *
932  * We decode the tuple in pretty much the same way as DecodeXLogTuple,
933  * but since the layout is slightly different, we can't use it here.
934  */
936  {
938 
939  change->data.tp.newtuple =
940  ReorderBufferGetTupleBuf(ctx->reorder, datalen);
941 
942  tuple = change->data.tp.newtuple;
943  header = tuple->tuple.t_data;
944 
945  /* not a disk based tuple */
947 
948  /*
949  * We can only figure this out after reassembling the
950  * transactions.
951  */
952  tuple->tuple.t_tableOid = InvalidOid;
953 
954  tuple->tuple.t_len = datalen + SizeofHeapTupleHeader;
955 
956  memset(header, 0, SizeofHeapTupleHeader);
957 
958  memcpy((char *) tuple->tuple.t_data + SizeofHeapTupleHeader,
959  (char *) data,
960  datalen);
961  header->t_infomask = xlhdr->t_infomask;
962  header->t_infomask2 = xlhdr->t_infomask2;
963  header->t_hoff = xlhdr->t_hoff;
964  }
965 
966  /*
967  * Reset toast reassembly state only after the last row in the last
968  * xl_multi_insert_tuple record emitted by one heap_multi_insert()
969  * call.
970  */
971  if (xlrec->flags & XLH_INSERT_LAST_IN_MULTI &&
972  (i + 1) == xlrec->ntuples)
973  change->data.tp.clear_toast_afterwards = true;
974  else
975  change->data.tp.clear_toast_afterwards = false;
976 
978  buf->origptr, change);
979 
980  /* move to the next xl_multi_insert_tuple entry */
981  data += datalen;
982  }
983  Assert(data == tupledata + tuplelen);
984 }
#define SizeofHeapTupleHeader
Definition: htup_details.h:184
struct ReorderBuffer * reorder
Definition: logical.h:42
struct ReorderBufferChange::@101::@102 tp
ReorderBufferChange * ReorderBufferGetChange(ReorderBuffer *rb)
ReplicationSlotPersistentData data
Definition: slot.h:132
static bool FilterByOrigin(LogicalDecodingContext *ctx, RepOriginId origin_id)
Definition: decode.c:497
enum ReorderBufferChangeType action
Definition: reorderbuffer.h:83
#define XLogRecGetOrigin(decoder)
Definition: xlogreader.h:282
HeapTupleHeader t_data
Definition: htup.h:68
#define XLogRecGetData(decoder)
Definition: xlogreader.h:283
#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:1460
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:281
char * XLogRecGetBlockData(XLogReaderState *record, uint8 block_id, Size *len)
Definition: xlogreader.c:1484
#define InvalidOid
Definition: postgres_ext.h:36
#define Assert(condition)
Definition: c.h:739
union ReorderBufferChange::@101 data
size_t Size
Definition: c.h:467
static void header(const char *fmt,...) pg_attribute_printf(1
Definition: pg_regress.c:209
void ReorderBufferQueueChange(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, ReorderBufferChange *change)
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:688
#define SizeOfMultiInsertTuple
Definition: heapam_xlog.h:191
ReorderBufferTupleBuf * ReorderBufferGetTupleBuf(ReorderBuffer *rb, Size tuple_len)

◆ DecodeSpecConfirm()

static void DecodeSpecConfirm ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)
static

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

994 {
995  XLogReaderState *r = buf->record;
996  ReorderBufferChange *change;
997  RelFileNode target_node;
998 
999  /* only interested in our database */
1000  XLogRecGetBlockTag(r, 0, &target_node, NULL, NULL);
1001  if (target_node.dbNode != ctx->slot->data.database)
1002  return;
1003 
1004  /* output plugin doesn't look for this origin, no need to queue */
1005  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
1006  return;
1007 
1008  change = ReorderBufferGetChange(ctx->reorder);
1010  change->origin_id = XLogRecGetOrigin(r);
1011 
1012  memcpy(&change->data.tp.relnode, &target_node, sizeof(RelFileNode));
1013 
1014  change->data.tp.clear_toast_afterwards = true;
1015 
1016  ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r), buf->origptr, change);
1017 }
struct ReorderBuffer * reorder
Definition: logical.h:42
struct ReorderBufferChange::@101::@102 tp
ReorderBufferChange * ReorderBufferGetChange(ReorderBuffer *rb)
ReplicationSlotPersistentData data
Definition: slot.h:132
static bool FilterByOrigin(LogicalDecodingContext *ctx, RepOriginId origin_id)
Definition: decode.c:497
enum ReorderBufferChangeType action
Definition: reorderbuffer.h:83
#define XLogRecGetOrigin(decoder)
Definition: xlogreader.h:282
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:1460
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:281
union ReorderBufferChange::@101 data
void ReorderBufferQueueChange(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, ReorderBufferChange *change)
XLogRecPtr origptr
Definition: decode.c:48
ReplicationSlot * slot
Definition: logical.h:38

◆ DecodeStandbyOp()

static void DecodeStandbyOp ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)
static

Definition at line 304 of file decode.c.

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

Referenced by LogicalDecodingProcessRecord().

305 {
306  SnapBuild *builder = ctx->snapshot_builder;
307  XLogReaderState *r = buf->record;
308  uint8 info = XLogRecGetInfo(r) & ~XLR_INFO_MASK;
309 
311 
312  switch (info)
313  {
314  case XLOG_RUNNING_XACTS:
315  {
317 
318  SnapBuildProcessRunningXacts(builder, buf->origptr, running);
319 
320  /*
321  * Abort all transactions that we keep track of, that are
322  * older than the record's oldestRunningXid. This is the most
323  * convenient spot for doing so since, in contrast to shutdown
324  * or end-of-recovery checkpoints, we have information about
325  * all running transactions which includes prepared ones,
326  * while shutdown checkpoints just know that no non-prepared
327  * transactions are in progress.
328  */
330  }
331  break;
332  case XLOG_STANDBY_LOCK:
333  break;
334  case XLOG_INVALIDATIONS:
335  {
336  xl_invalidations *invalidations =
338 
339  if (!ctx->fast_forward)
341  invalidations->nmsgs,
342  invalidations->msgs);
343  }
344  break;
345  default:
346  elog(ERROR, "unexpected RM_STANDBY_ID record type: %u", info);
347  }
348 }
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:357
#define XLOG_STANDBY_LOCK
Definition: standbydefs.h:34
#define XLogRecGetData(decoder)
Definition: xlogreader.h:283
#define ERROR
Definition: elog.h:43
void ReorderBufferImmediateInvalidation(ReorderBuffer *rb, uint32 ninvalidations, SharedInvalidationMessage *invalidations)
void SnapBuildProcessRunningXacts(SnapBuild *builder, XLogRecPtr lsn, xl_running_xacts *running)
Definition: snapbuild.c:1090
XLogReaderState * record
Definition: decode.c:50
#define XLogRecGetInfo(decoder)
Definition: xlogreader.h:279
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:281
struct SnapBuild * snapshot_builder
Definition: logical.h:43
#define XLR_INFO_MASK
Definition: xlogrecord.h:62
SharedInvalidationMessage msgs[FLEXIBLE_ARRAY_MEMBER]
Definition: standbydefs.h:69
XLogRecPtr origptr
Definition: decode.c:48
#define elog(elevel,...)
Definition: elog.h:228
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 844 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().

845 {
846  XLogReaderState *r = buf->record;
847  xl_heap_truncate *xlrec;
848  ReorderBufferChange *change;
849 
850  xlrec = (xl_heap_truncate *) XLogRecGetData(r);
851 
852  /* only interested in our database */
853  if (xlrec->dbId != ctx->slot->data.database)
854  return;
855 
856  /* output plugin doesn't look for this origin, no need to queue */
857  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
858  return;
859 
860  change = ReorderBufferGetChange(ctx->reorder);
862  change->origin_id = XLogRecGetOrigin(r);
863  if (xlrec->flags & XLH_TRUNCATE_CASCADE)
864  change->data.truncate.cascade = true;
865  if (xlrec->flags & XLH_TRUNCATE_RESTART_SEQS)
866  change->data.truncate.restart_seqs = true;
867  change->data.truncate.nrelids = xlrec->nrelids;
868  change->data.truncate.relids = ReorderBufferGetRelids(ctx->reorder,
869  xlrec->nrelids);
870  memcpy(change->data.truncate.relids, xlrec->relids,
871  xlrec->nrelids * sizeof(Oid));
873  buf->origptr, change);
874 }
#define XLH_TRUNCATE_CASCADE
Definition: heapam_xlog.h:116
struct ReorderBuffer * reorder
Definition: logical.h:42
ReorderBufferChange * ReorderBufferGetChange(ReorderBuffer *rb)
ReplicationSlotPersistentData data
Definition: slot.h:132
unsigned int Oid
Definition: postgres_ext.h:31
static bool FilterByOrigin(LogicalDecodingContext *ctx, RepOriginId origin_id)
Definition: decode.c:497
enum ReorderBufferChangeType action
Definition: reorderbuffer.h:83
#define XLogRecGetOrigin(decoder)
Definition: xlogreader.h:282
Oid relids[FLEXIBLE_ARRAY_MEMBER]
Definition: heapam_xlog.h:129
#define XLogRecGetData(decoder)
Definition: xlogreader.h:283
XLogReaderState * record
Definition: decode.c:50
RepOriginId origin_id
Definition: reorderbuffer.h:88
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:281
#define XLH_TRUNCATE_RESTART_SEQS
Definition: heapam_xlog.h:117
union ReorderBufferChange::@101 data
void ReorderBufferQueueChange(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, ReorderBufferChange *change)
XLogRecPtr origptr
Definition: decode.c:48
ReplicationSlot * slot
Definition: logical.h:38
Oid * ReorderBufferGetRelids(ReorderBuffer *rb, int nrelids)
struct ReorderBufferChange::@101::@103 truncate

◆ DecodeUpdate()

static void DecodeUpdate ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)
static

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

724 {
725  XLogReaderState *r = buf->record;
726  xl_heap_update *xlrec;
727  ReorderBufferChange *change;
728  char *data;
729  RelFileNode target_node;
730 
731  xlrec = (xl_heap_update *) XLogRecGetData(r);
732 
733  /* only interested in our database */
734  XLogRecGetBlockTag(r, 0, &target_node, NULL, NULL);
735  if (target_node.dbNode != ctx->slot->data.database)
736  return;
737 
738  /* output plugin doesn't look for this origin, no need to queue */
739  if (FilterByOrigin(ctx, XLogRecGetOrigin(r)))
740  return;
741 
742  change = ReorderBufferGetChange(ctx->reorder);
744  change->origin_id = XLogRecGetOrigin(r);
745  memcpy(&change->data.tp.relnode, &target_node, sizeof(RelFileNode));
746 
748  {
749  Size datalen;
750  Size tuplelen;
751 
752  data = XLogRecGetBlockData(r, 0, &datalen);
753 
754  tuplelen = datalen - SizeOfHeapHeader;
755 
756  change->data.tp.newtuple =
757  ReorderBufferGetTupleBuf(ctx->reorder, tuplelen);
758 
759  DecodeXLogTuple(data, datalen, change->data.tp.newtuple);
760  }
761 
762  if (xlrec->flags & XLH_UPDATE_CONTAINS_OLD)
763  {
764  Size datalen;
765  Size tuplelen;
766 
767  /* caution, remaining data in record is not aligned */
768  data = XLogRecGetData(r) + SizeOfHeapUpdate;
769  datalen = XLogRecGetDataLen(r) - SizeOfHeapUpdate;
770  tuplelen = datalen - SizeOfHeapHeader;
771 
772  change->data.tp.oldtuple =
773  ReorderBufferGetTupleBuf(ctx->reorder, tuplelen);
774 
775  DecodeXLogTuple(data, datalen, change->data.tp.oldtuple);
776  }
777 
778  change->data.tp.clear_toast_afterwards = true;
779 
780  ReorderBufferQueueChange(ctx->reorder, XLogRecGetXid(r), buf->origptr, change);
781 }
struct ReorderBuffer * reorder
Definition: logical.h:42
struct ReorderBufferChange::@101::@102 tp
static void DecodeXLogTuple(char *data, Size len, ReorderBufferTupleBuf *tup)
Definition: decode.c:1028
ReorderBufferChange * ReorderBufferGetChange(ReorderBuffer *rb)
ReplicationSlotPersistentData data
Definition: slot.h:132
static bool FilterByOrigin(LogicalDecodingContext *ctx, RepOriginId origin_id)
Definition: decode.c:497
enum ReorderBufferChangeType action
Definition: reorderbuffer.h:83
#define XLogRecGetOrigin(decoder)
Definition: xlogreader.h:282
#define XLH_UPDATE_CONTAINS_NEW_TUPLE
Definition: heapam_xlog.h:80
#define SizeOfHeapUpdate
Definition: heapam_xlog.h:225
#define XLogRecGetData(decoder)
Definition: xlogreader.h:283
#define XLogRecGetDataLen(decoder)
Definition: xlogreader.h:284
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:1460
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:281
char * XLogRecGetBlockData(XLogReaderState *record, uint8 block_id, Size *len)
Definition: xlogreader.c:1484
union ReorderBufferChange::@101 data
size_t Size
Definition: c.h:467
void ReorderBufferQueueChange(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn, ReorderBufferChange *change)
XLogRecPtr origptr
Definition: decode.c:48
ReplicationSlot * slot
Definition: logical.h:38
#define XLH_UPDATE_CONTAINS_OLD
Definition: heapam_xlog.h:85
#define SizeOfHeapHeader
Definition: heapam_xlog.h:149
ReorderBufferTupleBuf * ReorderBufferGetTupleBuf(ReorderBuffer *rb, Size tuple_len)

◆ DecodeXactOp()

static void DecodeXactOp ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)
static

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

210 {
211  SnapBuild *builder = ctx->snapshot_builder;
212  ReorderBuffer *reorder = ctx->reorder;
213  XLogReaderState *r = buf->record;
215 
216  /*
217  * If the snapshot isn't yet fully built, we cannot decode anything, so
218  * bail out.
219  *
220  * However, it's critical to process XLOG_XACT_ASSIGNMENT records even
221  * when the snapshot is being built: it is possible to get later records
222  * that require subxids to be properly assigned.
223  */
225  info != XLOG_XACT_ASSIGNMENT)
226  return;
227 
228  switch (info)
229  {
230  case XLOG_XACT_COMMIT:
232  {
233  xl_xact_commit *xlrec;
234  xl_xact_parsed_commit parsed;
235  TransactionId xid;
236 
237  xlrec = (xl_xact_commit *) XLogRecGetData(r);
238  ParseCommitRecord(XLogRecGetInfo(buf->record), xlrec, &parsed);
239 
240  if (!TransactionIdIsValid(parsed.twophase_xid))
241  xid = XLogRecGetXid(r);
242  else
243  xid = parsed.twophase_xid;
244 
245  DecodeCommit(ctx, buf, &parsed, xid);
246  break;
247  }
248  case XLOG_XACT_ABORT:
250  {
251  xl_xact_abort *xlrec;
252  xl_xact_parsed_abort parsed;
253  TransactionId xid;
254 
255  xlrec = (xl_xact_abort *) XLogRecGetData(r);
256  ParseAbortRecord(XLogRecGetInfo(buf->record), xlrec, &parsed);
257 
258  if (!TransactionIdIsValid(parsed.twophase_xid))
259  xid = XLogRecGetXid(r);
260  else
261  xid = parsed.twophase_xid;
262 
263  DecodeAbort(ctx, buf, &parsed, xid);
264  break;
265  }
267  {
268  xl_xact_assignment *xlrec;
269  int i;
270  TransactionId *sub_xid;
271 
272  xlrec = (xl_xact_assignment *) XLogRecGetData(r);
273 
274  sub_xid = &xlrec->xsub[0];
275 
276  for (i = 0; i < xlrec->nsubxacts; i++)
277  {
278  ReorderBufferAssignChild(reorder, xlrec->xtop,
279  *(sub_xid++), buf->origptr);
280  }
281  break;
282  }
283  case XLOG_XACT_PREPARE:
284 
285  /*
286  * Currently decoding ignores PREPARE TRANSACTION and will just
287  * decode the transaction when the COMMIT PREPARED is sent or
288  * throw away the transaction's contents when a ROLLBACK PREPARED
289  * is received. In the future we could add code to expose prepared
290  * transactions in the changestream allowing for a kind of
291  * distributed 2PC.
292  */
293  ReorderBufferProcessXid(reorder, XLogRecGetXid(r), buf->origptr);
294  break;
295  default:
296  elog(ERROR, "unexpected RM_XACT_ID record type: %u", info);
297  }
298 }
#define XLOG_XACT_COMMIT
Definition: xact.h:143
static void DecodeCommit(LogicalDecodingContext *ctx, XLogRecordBuffer *buf, xl_xact_parsed_commit *parsed, TransactionId xid)
Definition: decode.c:560
struct ReorderBuffer * reorder
Definition: logical.h:42
#define XLOG_XACT_PREPARE
Definition: xact.h:144
uint32 TransactionId
Definition: c.h:514
SnapBuildState SnapBuildCurrentState(SnapBuild *builder)
Definition: snapbuild.c:395
TransactionId twophase_xid
Definition: xact.h:335
unsigned char uint8
Definition: c.h:357
TransactionId xtop
Definition: xact.h:193
TransactionId xsub[FLEXIBLE_ARRAY_MEMBER]
Definition: xact.h:195
static void DecodeAbort(LogicalDecodingContext *ctx, XLogRecordBuffer *buf, xl_xact_parsed_abort *parsed, TransactionId xid)
Definition: decode.c:646
#define XLogRecGetData(decoder)
Definition: xlogreader.h:283
#define ERROR
Definition: elog.h:43
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:147
XLogReaderState * record
Definition: decode.c:50
#define XLogRecGetInfo(decoder)
Definition: xlogreader.h:279
void ReorderBufferAssignChild(ReorderBuffer *rb, TransactionId xid, TransactionId subxid, XLogRecPtr lsn)
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:281
#define XLOG_XACT_ASSIGNMENT
Definition: xact.h:148
TransactionId twophase_xid
Definition: xact.h:360
struct SnapBuild * snapshot_builder
Definition: logical.h:43
#define XLOG_XACT_ABORT
Definition: xact.h:145
#define XLOG_XACT_OPMASK
Definition: xact.h:153
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:228
int i
#define TransactionIdIsValid(xid)
Definition: transam.h:41
#define XLOG_XACT_COMMIT_PREPARED
Definition: xact.h:146
void ReorderBufferProcessXid(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)

◆ DecodeXLogOp()

static void DecodeXLogOp ( LogicalDecodingContext ctx,
XLogRecordBuffer buf 
)
static

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

168 {
169  SnapBuild *builder = ctx->snapshot_builder;
170  uint8 info = XLogRecGetInfo(buf->record) & ~XLR_INFO_MASK;
171 
173  buf->origptr);
174 
175  switch (info)
176  {
177  /* this is also used in END_OF_RECOVERY checkpoints */
180  SnapBuildSerializationPoint(builder, buf->origptr);
181 
182  break;
184 
185  /*
186  * a RUNNING_XACTS record will have been logged near to this, we
187  * can restart from there.
188  */
189  break;
190  case XLOG_NOOP:
191  case XLOG_NEXTOID:
192  case XLOG_SWITCH:
193  case XLOG_BACKUP_END:
195  case XLOG_RESTORE_POINT:
196  case XLOG_FPW_CHANGE:
197  case XLOG_FPI_FOR_HINT:
198  case XLOG_FPI:
199  break;
200  default:
201  elog(ERROR, "unexpected RM_XLOG_ID record type: %u", info);
202  }
203 }
#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:357
#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:279
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:281
#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:228
void SnapBuildSerializationPoint(SnapBuild *builder, XLogRecPtr lsn)
Definition: snapbuild.c:1467
void ReorderBufferProcessXid(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)

◆ DecodeXLogTuple()

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

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

1029 {
1030  xl_heap_header xlhdr;
1031  int datalen = len - SizeOfHeapHeader;
1033 
1034  Assert(datalen >= 0);
1035 
1036  tuple->tuple.t_len = datalen + SizeofHeapTupleHeader;
1037  header = tuple->tuple.t_data;
1038 
1039  /* not a disk based tuple */
1040  ItemPointerSetInvalid(&tuple->tuple.t_self);
1041 
1042  /* we can only figure this out after reassembling the transactions */
1043  tuple->tuple.t_tableOid = InvalidOid;
1044 
1045  /* data is not stored aligned, copy to aligned storage */
1046  memcpy((char *) &xlhdr,
1047  data,
1048  SizeOfHeapHeader);
1049 
1050  memset(header, 0, SizeofHeapTupleHeader);
1051 
1052  memcpy(((char *) tuple->tuple.t_data) + SizeofHeapTupleHeader,
1053  data + SizeOfHeapHeader,
1054  datalen);
1055 
1056  header->t_infomask = xlhdr.t_infomask;
1057  header->t_infomask2 = xlhdr.t_infomask2;
1058  header->t_hoff = xlhdr.t_hoff;
1059 }
#define SizeofHeapTupleHeader
Definition: htup_details.h:184
uint16 t_infomask2
Definition: heapam_xlog.h:144
#define InvalidOid
Definition: postgres_ext.h:36
#define Assert(condition)
Definition: c.h:739
uint16 t_infomask
Definition: heapam_xlog.h:145
static void header(const char *fmt,...) pg_attribute_printf(1
Definition: pg_regress.c:209
#define ItemPointerSetInvalid(pointer)
Definition: itemptr.h:172
#define SizeOfHeapHeader
Definition: heapam_xlog.h:149

◆ FilterByOrigin()

static bool FilterByOrigin ( LogicalDecodingContext ctx,
RepOriginId  origin_id 
)
inlinestatic

Definition at line 497 of file decode.c.

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

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

498 {
499  if (ctx->callbacks.filter_by_origin_cb == NULL)
500  return false;
501 
502  return filter_by_origin_cb_wrapper(ctx, origin_id);
503 }
OutputPluginCallbacks callbacks
Definition: logical.h:52
bool filter_by_origin_cb_wrapper(LogicalDecodingContext *ctx, RepOriginId origin_id)
Definition: logical.c:801
LogicalDecodeFilterByOriginCB filter_by_origin_cb

◆ LogicalDecodingProcessRecord()

void LogicalDecodingProcessRecord ( LogicalDecodingContext ctx,
XLogReaderState record 
)

Definition at line 94 of file decode.c.

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

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

95 {
97 
98  buf.origptr = ctx->reader->ReadRecPtr;
99  buf.endptr = ctx->reader->EndRecPtr;
100  buf.record = record;
101 
102  /* cast so we get a warning when new rmgrs are added */
103  switch ((RmgrIds) XLogRecGetRmid(record))
104  {
105  /*
106  * Rmgrs we care about for logical decoding. Add new rmgrs in
107  * rmgrlist.h's order.
108  */
109  case RM_XLOG_ID:
110  DecodeXLogOp(ctx, &buf);
111  break;
112 
113  case RM_XACT_ID:
114  DecodeXactOp(ctx, &buf);
115  break;
116 
117  case RM_STANDBY_ID:
118  DecodeStandbyOp(ctx, &buf);
119  break;
120 
121  case RM_HEAP2_ID:
122  DecodeHeap2Op(ctx, &buf);
123  break;
124 
125  case RM_HEAP_ID:
126  DecodeHeapOp(ctx, &buf);
127  break;
128 
129  case RM_LOGICALMSG_ID:
130  DecodeLogicalMsgOp(ctx, &buf);
131  break;
132 
133  /*
134  * Rmgrs irrelevant for logical decoding; they describe stuff not
135  * represented in logical decoding. Add new rmgrs in rmgrlist.h's
136  * order.
137  */
138  case RM_SMGR_ID:
139  case RM_CLOG_ID:
140  case RM_DBASE_ID:
141  case RM_TBLSPC_ID:
142  case RM_MULTIXACT_ID:
143  case RM_RELMAP_ID:
144  case RM_BTREE_ID:
145  case RM_HASH_ID:
146  case RM_GIN_ID:
147  case RM_GIST_ID:
148  case RM_SEQ_ID:
149  case RM_SPGIST_ID:
150  case RM_BRIN_ID:
151  case RM_COMMIT_TS_ID:
152  case RM_REPLORIGIN_ID:
153  case RM_GENERIC_ID:
154  /* just deal with xid, and done */
156  buf.origptr);
157  break;
158  case RM_NEXT_ID:
159  elog(ERROR, "unexpected RM_NEXT_ID rmgr_id: %u", (RmgrIds) XLogRecGetRmid(buf.record));
160  }
161 }
struct ReorderBuffer * reorder
Definition: logical.h:42
XLogRecPtr EndRecPtr
Definition: xlogreader.h:132
static void DecodeXactOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:209
static void DecodeLogicalMsgOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:509
#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:131
XLogReaderState * record
Definition: decode.c:50
static void DecodeXLogOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:167
static void DecodeHeap2Op(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:354
#define XLogRecGetXid(decoder)
Definition: xlogreader.h:281
XLogRecPtr origptr
Definition: decode.c:48
XLogReaderState * reader
Definition: logical.h:41
#define elog(elevel,...)
Definition: elog.h:228
static void DecodeHeapOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:414
void ReorderBufferProcessXid(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
RmgrIds
Definition: rmgr.h:25
static void DecodeStandbyOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
Definition: decode.c:304
#define XLogRecGetRmid(decoder)
Definition: xlogreader.h:280